diff --git a/alembic/versions/2026_05_27_1605-0d28973280f5_add_caller_type_to_workflow_tag_events.py b/alembic/versions/2026_05_27_1605-0d28973280f5_add_caller_type_to_workflow_tag_events.py
new file mode 100644
index 000000000..520f765ab
--- /dev/null
+++ b/alembic/versions/2026_05_27_1605-0d28973280f5_add_caller_type_to_workflow_tag_events.py
@@ -0,0 +1,35 @@
+"""add caller_type to workflow_tag_events
+
+Revision ID: 0d28973280f5
+Revises: 4b254743ea85
+Create Date: 2026-05-27T16:05:27.572803+00:00
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "0d28973280f5"
+down_revision: Union[str, None] = "4b254743ea85"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("workflow_tag_events", sa.Column("caller_type", sa.String(), nullable=True))
+    # Values must stay in sync with CallerType (skyvern/forge/sdk/workflow/models/tags.py).
+    # Adding a new CallerType variant requires a companion migration to widen this constraint.
+    op.create_check_constraint(
+        "ck_workflow_tag_events_caller_type",
+        "workflow_tag_events",
+        "caller_type IS NULL OR caller_type IN ('user', 'api_key', 'system')",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("ck_workflow_tag_events_caller_type", "workflow_tag_events", type_="check")
+    op.drop_column("workflow_tag_events", "caller_type")
diff --git a/skyvern-frontend/src/routes/workflows/editor/nodes/LoginNode/LoginBlockCredentialSelector.tsx b/skyvern-frontend/src/routes/workflows/editor/nodes/LoginNode/LoginBlockCredentialSelector.tsx
index c6d6aab61..f1082045a 100644
--- a/skyvern-frontend/src/routes/workflows/editor/nodes/LoginNode/LoginBlockCredentialSelector.tsx
+++ b/skyvern-frontend/src/routes/workflows/editor/nodes/LoginNode/LoginBlockCredentialSelector.tsx
@@ -205,17 +205,22 @@ function LoginBlockCredentialSelector({
     (option) => option.type === "parameter" && option.value === value,
   );
 
+  // Always pass a defined string so Radix Select stays controlled; uncontrolled
+  // mode caches the picked value and silently skips re-firing onValueChange.
+  let selectValue: string;
+  if (isCredentialMissing) {
+    selectValue = "";
+  } else if (valueIsParameterOption) {
+    selectValue = value ?? "";
+  } else {
+    selectValue = selectedCredentialId ?? value ?? "";
+  }
+
   return (
     <>
       <Select
         key={value ?? "no-credential"}
-        value={
-          isCredentialMissing
-            ? undefined
-            : valueIsParameterOption
-              ? value
-              : (selectedCredentialId ?? value)
-        }
+        value={selectValue}
         onValueChange={(newValue) => {
           if (newValue === "new") {
             setIsOpen(true);
diff --git a/skyvern/forge/sdk/copilot/agent.py b/skyvern/forge/sdk/copilot/agent.py
index 938ead164..9997f1b23 100644
--- a/skyvern/forge/sdk/copilot/agent.py
+++ b/skyvern/forge/sdk/copilot/agent.py
@@ -9,6 +9,7 @@ import asyncio
 import contextlib
 import json
 import re
+import uuid
 from collections.abc import Iterator
 from dataclasses import dataclass
 from datetime import datetime, timezone
@@ -79,6 +80,11 @@ from skyvern.forge.sdk.copilot.request_policy import (
     build_request_policy,
     redact_raw_secrets_for_prompt,
 )
+from skyvern.forge.sdk.copilot.streaming_adapter import (
+    emit_turn_start,
+    emit_workflow_draft,
+    maybe_emit_design_end,
+)
 from skyvern.forge.sdk.copilot.tracing_setup import _copilot_model_name, ensure_tracing_initialized, is_tracing_enabled
 from skyvern.forge.sdk.copilot.turn_context import TurnContextAssembler, TurnContextInputs, TurnContextPacket
 from skyvern.forge.sdk.copilot.turn_intent import (
@@ -126,11 +132,12 @@ def _derive_turn_index(
     chat_history: list[WorkflowCopilotChatHistoryMessage],
     explicit: int | None,
 ) -> int:
-    # `chat_history` may be a truncated tail of the full message log, so this
+    # Zero-based to match the wire contract (``WorkflowCopilotTurnStartUpdate``).
+    # ``chat_history`` may be a truncated tail of the full message log, so this
     # fallback can undercount long sessions; prefer the explicit count.
     if explicit is not None:
         return explicit
-    return sum(1 for m in chat_history if m.sender == WorkflowCopilotChatSender.USER) + 1
+    return sum(1 for m in chat_history if m.sender == WorkflowCopilotChatSender.USER)
 
 
 @contextlib.contextmanager
@@ -139,11 +146,14 @@ def _copilot_turn_span(
     chat_request: WorkflowCopilotChatRequest,
     chat_history: list[WorkflowCopilotChatHistoryMessage],
     turn_index: int | None,
+    turn_id: str | None = None,
 ) -> Iterator[Any]:
     tracer = otel_trace.get_tracer("skyvern")
     with tracer.start_as_current_span(_COPILOT_TURN_SPAN_NAME) as span:
         span.set_attribute("skyvern.span.role", "wrapper")
         span.set_attribute("copilot.turn_index", _derive_turn_index(chat_history, turn_index))
+        if turn_id is not None:
+            span.set_attribute("copilot.turn_id", turn_id)
         preview = _build_user_message_preview(chat_request.message)
         if preview:
             span.set_attribute("copilot.user_message_preview", preview)
@@ -734,6 +744,8 @@ def _build_exit_result(
             total_tokens=ctx.total_tokens_used,
             cancelled=cancelled,
             turn_outcome=outcome,
+            turn_id=ctx.turn_id,
+            narrative_summary=ctx.narrative_summary,
         ),
         exit_site="exit_result",
     )
@@ -950,6 +962,8 @@ def _finalize_result_with_blocker_override(
         proposal_disposition="review_untested" if preserved_proposal else "no_proposal",
         output_policy_diagnostics=rendered_diagnostics,
         turn_outcome=turn_outcome,
+        turn_id=ctx.turn_id,
+        narrative_summary=ctx.narrative_summary,
     )
 
 
@@ -1170,6 +1184,8 @@ def _build_wip_exit_result(
                 proposal_disposition="review_tested",
                 cancelled=cancelled,
                 turn_outcome=outcome,
+                turn_id=ctx.turn_id,
+                narrative_summary=ctx.narrative_summary,
             ),
             exit_site="wip_last_good_workflow",
         )
@@ -1198,6 +1214,8 @@ def _build_wip_exit_result(
                 proposal_disposition="review_untested" if unvalidated else "auto_applicable",
                 cancelled=cancelled,
                 turn_outcome=outcome,
+                turn_id=ctx.turn_id,
+                narrative_summary=ctx.narrative_summary,
             ),
             exit_site="wip_last_workflow",
         )
@@ -1624,6 +1642,8 @@ def _translate_to_agent_result(
             ),
             output_policy_diagnostics=output_policy_diagnostics,
             turn_outcome=turn_outcome,
+            turn_id=ctx.turn_id,
+            narrative_summary=ctx.narrative_summary,
         ),
         exit_site="translate_to_agent_result",
     )
@@ -1669,6 +1689,8 @@ def _build_feasibility_clarification_result(
             workflow_was_persisted=False,
             clear_proposed_workflow=True,
             turn_outcome=outcome,
+            turn_id=ctx.turn_id,
+            narrative_summary=ctx.narrative_summary,
         ),
         exit_site="feasibility_clarification",
     )
@@ -1750,6 +1772,8 @@ def _build_request_policy_clarification_result(
             workflow_was_persisted=False,
             clear_proposed_workflow=True,
             turn_outcome=outcome,
+            turn_id=ctx.turn_id,
+            narrative_summary=ctx.narrative_summary,
         ),
         exit_site="request_policy_clarification",
     )
@@ -2070,6 +2094,8 @@ def _build_output_policy_blocked_result(
             soft_rewrite_reason_codes=[],
         ),
         turn_outcome=output_policy_outcome,
+        turn_id=ctx.turn_id,
+        narrative_summary=ctx.narrative_summary,
     )
 
 
@@ -2085,8 +2111,15 @@ async def run_copilot_agent(
     security_rules: str = "",
     config: CopilotConfig | None = None,
     turn_index: int | None = None,
+    turn_id: str | None = None,
     prior_copilot_workflow_yaml: str | None = None,
 ) -> AgentResult:
+    # One id per turn — passed to every downstream AgentResult and
+    # CopilotContext so the envelope and terminal frames correlate. The
+    # default_factory on CopilotContext is only the per-construction fallback.
+    if turn_id is None:
+        turn_id = uuid.uuid4().hex
+    normalized_turn_index = turn_index if turn_index is not None else 0
     try:
         # Initialize tracing before opening the turn span so Logfire's OTel provider
         # is installed; otherwise the very first turn lands the parent span on
@@ -2096,6 +2129,7 @@ async def run_copilot_agent(
             chat_request=chat_request,
             chat_history=chat_history,
             turn_index=turn_index,
+            turn_id=turn_id,
         ) as turn_span:
             try:
                 return await _run_copilot_turn_impl(
@@ -2109,6 +2143,8 @@ async def run_copilot_agent(
                     api_key=api_key,
                     security_rules=security_rules,
                     config=config,
+                    turn_id=turn_id,
+                    turn_index=normalized_turn_index,
                     prior_copilot_workflow_yaml=prior_copilot_workflow_yaml,
                 )
             except Exception as exc:
@@ -2130,6 +2166,8 @@ async def run_copilot_agent(
                     api_key=api_key,
                     user_message=chat_request.message,
                     workflow_copilot_chat_id=chat_request.workflow_copilot_chat_id,
+                    turn_id=turn_id,
+                    turn_index=normalized_turn_index,
                 )
                 return _build_unexpected_error_exit_result(ctx, global_llm_context, error=exc, span=turn_span)
     except Exception as exc:
@@ -2150,6 +2188,8 @@ async def run_copilot_agent(
             api_key=api_key,
             user_message=chat_request.message,
             workflow_copilot_chat_id=chat_request.workflow_copilot_chat_id,
+            turn_id=turn_id,
+            turn_index=normalized_turn_index,
         )
         return _build_unexpected_error_exit_result(ctx, global_llm_context, error=exc)
 
@@ -2166,6 +2206,8 @@ async def _run_copilot_turn_impl(
     api_key: str | None,
     security_rules: str,
     config: CopilotConfig | None,
+    turn_id: str,
+    turn_index: int,
     prior_copilot_workflow_yaml: str | None = None,
 ) -> AgentResult:
     copilot_config = config or CopilotConfig(security_rules=security_rules)
@@ -2211,6 +2253,7 @@ async def _run_copilot_turn_impl(
                 global_llm_context=global_llm_context,
                 workflow_yaml=chat_request.workflow_yaml or None,
                 turn_outcome=missing_sdk_outcome,
+                turn_id=turn_id,
             )
         raise
 
@@ -2224,7 +2267,17 @@ async def _run_copilot_turn_impl(
         api_key=api_key,
         user_message=chat_request.message,
         workflow_copilot_chat_id=chat_request.workflow_copilot_chat_id,
+        turn_id=turn_id,
+        turn_index=turn_index,
     )
+    # Fail loud if a future caller skips the kwarg and gets a fresh UUID from
+    # the default_factory — the envelope and terminal frames would then carry
+    # different ids and correlation would silently break. Uses a real
+    # conditional so the check survives ``python -O``.
+    if ctx.turn_id != turn_id:
+        raise RuntimeError(
+            f"CopilotContext.turn_id ({ctx.turn_id!r}) diverged from route-supplied turn_id ({turn_id!r})"
+        )
     policy_inputs = RequestPolicyGuardrailInputs(
         user_message=chat_request.message,
         workflow_yaml=safe_workflow_yaml,
@@ -2253,6 +2306,14 @@ async def _run_copilot_turn_impl(
         chat_request.message,
         RunContextWrapper(context=ctx),
     )
+    # Emit TURN_START after the guardrail runs so the envelope carries an
+    # accurate ``mode`` when ``ctx.turn_intent`` is populated, and falls back
+    # to ``UNKNOWN`` defensively otherwise. Best-effort — an emission failure
+    # must not abort an otherwise-runnable turn.
+    try:
+        await emit_turn_start(stream, ctx)
+    except Exception as emit_err:
+        LOG.warning("copilot_narrative_turn_start_emit_failed", error=str(emit_err))
     request_policy = ctx.request_policy if isinstance(ctx.request_policy, RequestPolicy) else None
     if request_policy is not None:
         _store_turn_context_packet_on_context(
@@ -2493,13 +2554,30 @@ async def _run_copilot_turn_impl(
                     )
                     ctx.supports_vision = fallback_supports_vision
                     result = await _run_attempt(fallback_model_name, fallback_run_config, fallback_resolved_key)
-                return _translate_to_agent_result(
+                agent_result = _translate_to_agent_result(
                     result,
                     ctx,
                     global_llm_context,
                     chat_request,
                     organization_id,
                 )
+                # Inline ``REPLACE_WORKFLOW`` bypasses the ``update_workflow``
+                # tool, so the envelope fires here instead — keeps the FE
+                # bubble identical regardless of which path produced the
+                # draft. Best-effort.
+                if (
+                    agent_result.response_type == "REPLACE_WORKFLOW"
+                    and agent_result.updated_workflow is not None
+                    and ctx.stream is not None
+                ):
+                    try:
+                        await maybe_emit_design_end(ctx.stream, ctx)
+                        await emit_workflow_draft(ctx.stream, ctx, agent_result.updated_workflow)
+                    except Exception as emit_err:
+                        LOG.warning("copilot_narrative_inline_replace_emit_failed", error=str(emit_err))
+                    ctx.design_start_emitted = False
+                    ctx.design_end_emitted = False
+                return agent_result
             except asyncio.CancelledError:
                 # Re-raising would leave the route with ``agent_result is None``
                 # and skip its ``workflow_was_persisted`` rollback decision.
@@ -2562,6 +2640,8 @@ async def _run_copilot_turn_impl(
                         workflow_was_persisted=ctx.workflow_persisted,
                         total_tokens=ctx.total_tokens_used,
                         turn_outcome=nav_outcome,
+                        turn_id=ctx.turn_id,
+                        narrative_summary=ctx.narrative_summary,
                     ),
                     exit_site="non_retriable_nav",
                 )
diff --git a/skyvern/forge/sdk/copilot/context.py b/skyvern/forge/sdk/copilot/context.py
index 2bf689b4f..5edabccec 100644
--- a/skyvern/forge/sdk/copilot/context.py
+++ b/skyvern/forge/sdk/copilot/context.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import re
+import uuid
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Literal, get_args
 
@@ -169,6 +170,8 @@ class AgentResult:
     proposal_disposition: ProposalDisposition = "auto_applicable"
     output_policy_diagnostics: dict[str, Any] | None = None
     turn_outcome: TurnOutcome | None = None
+    turn_id: str | None = None
+    narrative_summary: str | None = None
 
 
 @dataclass
@@ -329,3 +332,14 @@ class CopilotContext(AgentContext):
     # Set in `_run_attempt` after SkyvernOverlayMCPServer is constructed.
     # The discovery tool reaches the connected FastMCP client through this.
     discovery_mcp_server: Any | None = None
+
+    # default_factory is the safety net — Python dataclass inheritance
+    # disallows non-default fields after default ones, and the parent
+    # ``AgentContext`` has many defaulted fields. The route generates the
+    # canonical turn_id and passes it as an explicit kwarg at every
+    # construction site, overriding this default.
+    turn_id: str = field(default_factory=lambda: uuid.uuid4().hex)
+    turn_index: int = 0
+    design_start_emitted: bool = False
+    design_end_emitted: bool = False
+    narrative_summary: str | None = None
diff --git a/skyvern/forge/sdk/copilot/streaming_adapter.py b/skyvern/forge/sdk/copilot/streaming_adapter.py
index 1e52f48fb..28fd4976a 100644
--- a/skyvern/forge/sdk/copilot/streaming_adapter.py
+++ b/skyvern/forge/sdk/copilot/streaming_adapter.py
@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import json
+from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
 
 import structlog
@@ -23,15 +24,22 @@ from skyvern.forge.sdk.copilot.narration import (
 )
 from skyvern.forge.sdk.copilot.output_utils import format_tool_result_for_user, summarize_tool_result_detail
 from skyvern.forge.sdk.schemas.workflow_copilot import (
+    WorkflowCopilotDesignEndUpdate,
+    WorkflowCopilotDesignStartUpdate,
     WorkflowCopilotStreamMessageType,
     WorkflowCopilotToolCallUpdate,
     WorkflowCopilotToolResultUpdate,
+    WorkflowCopilotTurnMode,
+    WorkflowCopilotTurnStartUpdate,
+    WorkflowCopilotWorkflowDraftUpdate,
 )
 
 if TYPE_CHECKING:
     from agents.result import RunResultStreaming
 
+    from skyvern.forge.sdk.copilot.context import CopilotContext
     from skyvern.forge.sdk.routes.event_source_stream import EventSourceStream
+    from skyvern.forge.sdk.workflow.models.workflow import Workflow
 
 LOG = structlog.get_logger()
 
@@ -110,6 +118,16 @@ async def stream_to_sse(
             # finish. stream.send below would drop the payload anyway.
             client_gone = await stream.is_disconnected()
 
+            # Edge-trigger DESIGN_START on the first user-visible agent event —
+            # message_output_created is the canonical signal, tool_called the
+            # fallback when the agent goes straight to a tool. Best-effort: a
+            # serialization failure here cannot abort the agent run.
+            if event.name in ("message_output_created", "tool_called") and not client_gone:
+                try:
+                    await maybe_emit_design_start(stream, ctx)
+                except Exception as emit_err:
+                    LOG.warning("copilot_narrative_design_start_emit_failed", error=str(emit_err))
+
             if event.name == "tool_called":
                 raw = event.item.raw_item
                 call_id = _get_raw_field(raw, "call_id") or _get_raw_field(raw, "id") or ""
@@ -316,3 +334,57 @@ def _sanitize_input(raw_args: dict[str, Any]) -> dict[str, Any]:
     if isinstance(redacted, dict):
         return redacted
     return trimmed
+
+
+async def emit_turn_start(stream: EventSourceStream, ctx: CopilotContext) -> None:
+    mode_value = ctx.turn_intent.mode.value if ctx.turn_intent is not None else WorkflowCopilotTurnMode.UNKNOWN.value
+    await stream.send(
+        WorkflowCopilotTurnStartUpdate(
+            turn_id=ctx.turn_id,
+            turn_index=ctx.turn_index,
+            mode=WorkflowCopilotTurnMode(mode_value),
+            timestamp=datetime.now(timezone.utc),
+        )
+    )
+
+
+async def maybe_emit_design_start(stream: EventSourceStream, ctx: CopilotContext) -> None:
+    if ctx.design_start_emitted:
+        return
+    ctx.design_start_emitted = True
+    await stream.send(WorkflowCopilotDesignStartUpdate(timestamp=datetime.now(timezone.utc)))
+
+
+async def maybe_emit_design_end(stream: EventSourceStream, ctx: CopilotContext) -> None:
+    # Guard: never emit DESIGN_END without a matching DESIGN_START. Both flags
+    # are turn-scoped, so a turn that exits before the streaming adapter sees
+    # its first user-visible event simply skips the design phase entirely.
+    if not ctx.design_start_emitted or ctx.design_end_emitted:
+        return
+    ctx.design_end_emitted = True
+    await stream.send(WorkflowCopilotDesignEndUpdate(timestamp=datetime.now(timezone.utc)))
+
+
+async def emit_workflow_draft(stream: EventSourceStream, ctx: CopilotContext, workflow: Workflow) -> None:
+    # Summary payload only: the full workflow definition is transported via
+    # terminal updated_workflow on RESPONSE or via the chat's proposed_workflow.
+    block_count = 0
+    block_labels: list[str] = []
+    try:
+        for block in workflow.workflow_definition.blocks:
+            block_count += 1
+            label = getattr(block, "label", None)
+            if isinstance(label, str) and label:
+                block_labels.append(label)
+    except AttributeError:
+        # Defensive: workflow without a blocks-bearing definition. Emit with
+        # whatever we could collect rather than skipping the event entirely.
+        pass
+    await stream.send(
+        WorkflowCopilotWorkflowDraftUpdate(
+            block_count=block_count,
+            block_labels=block_labels,
+            summary=None,
+            timestamp=datetime.now(timezone.utc),
+        )
+    )
diff --git a/skyvern/forge/sdk/copilot/tools.py b/skyvern/forge/sdk/copilot/tools.py
index 624ef3c13..c0f3f9556 100644
--- a/skyvern/forge/sdk/copilot/tools.py
+++ b/skyvern/forge/sdk/copilot/tools.py
@@ -86,6 +86,7 @@ from skyvern.forge.sdk.copilot.output_utils import (
 from skyvern.forge.sdk.copilot.request_policy import CREDENTIAL_DEFERRED_DRAFT_REASONS, RequestPolicy
 from skyvern.forge.sdk.copilot.runtime import AgentContext, ensure_browser_session
 from skyvern.forge.sdk.copilot.screenshot_utils import enqueue_screenshot_from_result
+from skyvern.forge.sdk.copilot.streaming_adapter import emit_workflow_draft, maybe_emit_design_end
 from skyvern.forge.sdk.copilot.tracing_setup import copilot_span
 from skyvern.forge.sdk.copilot.turn_intent import (
     NO_MUTATION_TURN_INTENT_MODES,
@@ -1582,6 +1583,22 @@ async def _update_workflow(
             edited_by="copilot",
         )
         ctx.workflow_yaml = workflow_yaml
+        # Best-effort — narrative emit failures must never abort an
+        # otherwise-successful update_workflow tool call. ``isinstance``
+        # narrows the parameter's declared ``AgentContext`` to the
+        # envelope-aware ``CopilotContext`` for mypy.
+        if isinstance(ctx, CopilotContext) and ctx.stream is not None:
+            try:
+                await maybe_emit_design_end(ctx.stream, ctx)
+                await emit_workflow_draft(ctx.stream, ctx, workflow)
+            except Exception as emit_err:
+                LOG.warning("copilot_narrative_workflow_draft_emit_failed", error=str(emit_err))
+            # Reset phase flags so a subsequent ``tool_called`` re-edge-triggers
+            # ``design_start`` and a subsequent ``update_workflow`` re-emits
+            # ``design_end`` — multi-iteration designs (draft → test → redraft)
+            # would otherwise collapse into a single phase.
+            ctx.design_start_emitted = False
+            ctx.design_end_emitted = False
         return {
             "ok": True,
             "data": {
diff --git a/skyvern/forge/sdk/db/agent_db.py b/skyvern/forge/sdk/db/agent_db.py
index bc644954b..382b7d21b 100644
--- a/skyvern/forge/sdk/db/agent_db.py
+++ b/skyvern/forge/sdk/db/agent_db.py
@@ -26,6 +26,7 @@ from skyvern.forge.sdk.db.repositories.organizations import OrganizationsReposit
 from skyvern.forge.sdk.db.repositories.otp import OTPRepository
 from skyvern.forge.sdk.db.repositories.schedules import SchedulesRepository
 from skyvern.forge.sdk.db.repositories.scripts import ScriptsRepository
+from skyvern.forge.sdk.db.repositories.tags import TagsRepository
 from skyvern.forge.sdk.db.repositories.tasks import TasksRepository
 from skyvern.forge.sdk.db.repositories.workflow_parameters import WorkflowParametersRepository
 from skyvern.forge.sdk.db.repositories.workflow_runs import WorkflowRunsRepository
@@ -130,6 +131,7 @@ class AgentDB(BaseAlchemyDB):
         self.debug = DebugRepository(self.Session, debug_enabled, self.is_retryable_error)
         self.organizations = OrganizationsRepository(self.Session, debug_enabled, self.is_retryable_error)
         self.scripts = ScriptsRepository(self.Session, debug_enabled, self.is_retryable_error)
+        self.tags = TagsRepository(self.Session, debug_enabled, self.is_retryable_error)
         self.browser_sessions = BrowserSessionsRepository(self.Session, debug_enabled, self.is_retryable_error)
         self.google_oauth = GoogleOAuthRepository(self.Session, debug_enabled, self.is_retryable_error)
         self.schedules = SchedulesRepository(
diff --git a/skyvern/forge/sdk/db/models.py b/skyvern/forge/sdk/db/models.py
index 262ea63bf..dc8b39de9 100644
--- a/skyvern/forge/sdk/db/models.py
+++ b/skyvern/forge/sdk/db/models.py
@@ -356,6 +356,10 @@ class WorkflowTagEventModel(Base):
             "source IN ('manual', 'bulk_apply', 'backfill', 'inherited', 'import')",
             name="ck_workflow_tag_events_source",
         ),
+        CheckConstraint(
+            "caller_type IS NULL OR caller_type IN ('user', 'api_key', 'system')",
+            name="ck_workflow_tag_events_caller_type",
+        ),
         CheckConstraint("event_type != 'delete' OR value IS NULL", name="ck_workflow_tag_events_delete_null_value"),
     )
 
@@ -368,6 +372,7 @@ class WorkflowTagEventModel(Base):
     set_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
     set_by = Column(String, nullable=False)
     source = Column(String, nullable=False)
+    caller_type = Column(String, nullable=True)
     superseded_at = Column(DateTime, nullable=True)
 
     created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
@@ -381,11 +386,9 @@ class WorkflowTagEventModel(Base):
 
 
 class TagKeyModel(Base):
-    """Org-scoped registry of tag keys and their descriptions.
-
-    Auto-upserted on first use of a new key via INSERT ... ON CONFLICT
-    (organization_id, key) WHERE deleted_at IS NULL DO UPDATE.
-    """
+    """Org-scoped registry of tag keys and their descriptions. Auto-registered
+    on first use; partial UNIQUE on (org, key) WHERE deleted_at IS NULL races
+    concurrent first-use writers — the losing writer surfaces IntegrityError to the caller."""
 
     __tablename__ = "tag_keys"
     __table_args__ = (
diff --git a/skyvern/forge/sdk/db/repositories/tags.py b/skyvern/forge/sdk/db/repositories/tags.py
new file mode 100644
index 000000000..94849ef96
--- /dev/null
+++ b/skyvern/forge/sdk/db/repositories/tags.py
@@ -0,0 +1,224 @@
+"""Workflow-tag write/read path. Named ``tags`` (not ``workflow_tags``) so run-tag events can land here later."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+import structlog
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from skyvern.forge.sdk.db._error_handling import db_operation, register_passthrough_exception
+from skyvern.forge.sdk.db.base_repository import BaseRepository
+from skyvern.forge.sdk.db.models import TagKeyModel, WorkflowTagEventModel
+from skyvern.forge.sdk.workflow.models.tags import TagEventType, TagWriteContext
+from skyvern.forge.sdk.workflow.models.validators import RUN_METADATA_MAX_KEYS
+
+LOG = structlog.get_logger()
+
+# Aliased to RUN_METADATA_MAX_KEYS so tag and run_metadata caps stay coupled.
+MAX_TAGS_PER_WORKFLOW = RUN_METADATA_MAX_KEYS
+
+
+class TagCountLimitExceeded(ValueError):
+    """Raised when an apply would push a workflow over MAX_TAGS_PER_WORKFLOW."""
+
+
+# Cap breaches are user input, not infra failures — log as BusinessLogicError (WARN).
+register_passthrough_exception(TagCountLimitExceeded)
+
+
+@dataclass(frozen=True)
+class TagChange:
+    """One concrete state change derived from a caller's set/delete request."""
+
+    key: str
+    new_value: str | None
+    event_type: TagEventType
+    superseded_event_id: str | None
+
+
+class TagsRepository(BaseRepository):
+    """Database operations for workflow tag events and the tag-key registry."""
+
+    @db_operation("apply_tag_changes")
+    async def apply_tag_changes(
+        self,
+        workflow_permanent_id: str,
+        organization_id: str,
+        sets: dict[str, str],
+        deletes: set[str],
+        context: TagWriteContext,
+    ) -> list[TagChange]:
+        """Atomically apply SET/DELETE events for a workflow's tags. Sets win
+        over deletes on same-key collision; same-value SETs are no-ops. Same-key
+        concurrent writers race the partial UNIQUE — the loser's IntegrityError
+        is not caught here, so it surfaces to the caller as a 5xx."""
+        effective_deletes = {k for k in deletes if k not in sets}
+        if not sets and not effective_deletes:
+            return []
+
+        now = datetime.now(timezone.utc)
+
+        async with self.Session() as session:
+            current = await self._get_current_active_set_events(
+                session,
+                workflow_permanent_id=workflow_permanent_id,
+                organization_id=organization_id,
+            )
+
+            # Soft cap: concurrent distinct-key writers can each pass; partial
+            # UNIQUE catches only same-key races. Cap is a UX rule, acceptable.
+            projected_keys = (set(current.keys()) | set(sets.keys())) - effective_deletes
+            if len(projected_keys) > MAX_TAGS_PER_WORKFLOW:
+                raise TagCountLimitExceeded(
+                    f"workflow {workflow_permanent_id} would have {len(projected_keys)} tags; "
+                    f"max is {MAX_TAGS_PER_WORKFLOW}"
+                )
+
+            changes: list[TagChange] = []
+
+            for key, new_value in sets.items():
+                existing = current.get(key)
+                if existing is not None and existing.value == new_value:
+                    continue
+                if existing is not None:
+                    existing.superseded_at = now
+                event = WorkflowTagEventModel(
+                    workflow_permanent_id=workflow_permanent_id,
+                    organization_id=organization_id,
+                    key=key,
+                    value=new_value,
+                    event_type=TagEventType.SET.value,
+                    set_at=now,
+                    set_by=context.caller_id,
+                    source=context.source.value,
+                    caller_type=context.caller_type.value if context.caller_type else None,
+                )
+                session.add(event)
+                changes.append(
+                    TagChange(
+                        key=key,
+                        new_value=new_value,
+                        event_type=TagEventType.SET,
+                        superseded_event_id=existing.tag_event_id if existing else None,
+                    )
+                )
+
+            for key in effective_deletes:
+                existing = current.get(key)
+                if existing is None:
+                    continue
+                existing.superseded_at = now
+                event = WorkflowTagEventModel(
+                    workflow_permanent_id=workflow_permanent_id,
+                    organization_id=organization_id,
+                    key=key,
+                    value=None,
+                    event_type=TagEventType.DELETE.value,
+                    set_at=now,
+                    set_by=context.caller_id,
+                    source=context.source.value,
+                    caller_type=context.caller_type.value if context.caller_type else None,
+                )
+                session.add(event)
+                changes.append(
+                    TagChange(
+                        key=key,
+                        new_value=None,
+                        event_type=TagEventType.DELETE,
+                        superseded_event_id=existing.tag_event_id,
+                    )
+                )
+
+            # Flush UPDATEs-before-INSERTs so the partial UNIQUE sees a
+            # consistent state when the new SET row hits it.
+            await session.flush()
+
+            # Auto-register TagKeyModel rows only for keys whose SET actually
+            # wrote a new event — idempotent no-op SETs don't need to re-touch
+            # the registry. Partial UNIQUE on (org, key) WHERE deleted_at IS
+            # NULL races concurrent first-use writers; the loser surfaces
+            # IntegrityError to the caller.
+            changed_set_keys = {c.key for c in changes if c.event_type == TagEventType.SET}
+            if changed_set_keys:
+                existing_keys_stmt = select(TagKeyModel.key).where(
+                    TagKeyModel.organization_id == organization_id,
+                    TagKeyModel.key.in_(changed_set_keys),
+                    TagKeyModel.deleted_at.is_(None),
+                )
+                existing_keys = set((await session.execute(existing_keys_stmt)).scalars().all())
+                for key in changed_set_keys - existing_keys:
+                    session.add(TagKeyModel(organization_id=organization_id, key=key))
+
+            await session.commit()
+            return changes
+
+    async def _get_current_active_set_events(
+        self,
+        session: AsyncSession,
+        *,
+        workflow_permanent_id: str,
+        organization_id: str,
+    ) -> dict[str, WorkflowTagEventModel]:
+        stmt = select(WorkflowTagEventModel).where(
+            and_(
+                WorkflowTagEventModel.organization_id == organization_id,
+                WorkflowTagEventModel.workflow_permanent_id == workflow_permanent_id,
+                WorkflowTagEventModel.superseded_at.is_(None),
+                WorkflowTagEventModel.event_type == TagEventType.SET.value,
+                WorkflowTagEventModel.deleted_at.is_(None),
+            )
+        )
+        result = await session.execute(stmt)
+        return {row.key: row for row in result.scalars().all()}
+
+    @db_operation("get_active_tags_for_workflow")
+    async def get_active_tags_for_workflow(
+        self,
+        workflow_permanent_id: str,
+        organization_id: str,
+    ) -> dict[str, str]:
+        """Return the current {key: value} map for a workflow."""
+        async with self.Session() as session:
+            rows = await self._get_current_active_set_events(
+                session,
+                workflow_permanent_id=workflow_permanent_id,
+                organization_id=organization_id,
+            )
+            # SET events always carry a non-null value, but defend against
+            # data drift: skip-and-log instead of crashing the request.
+            result: dict[str, str] = {}
+            for key, row in rows.items():
+                if row.value is None:
+                    LOG.warning(
+                        "active SET tag row has null value; skipping",
+                        tag_event_id=row.tag_event_id,
+                        organization_id=organization_id,
+                        workflow_permanent_id=workflow_permanent_id,
+                        key=key,
+                    )
+                    continue
+                result[key] = row.value
+            return result
+
+    @db_operation("get_tag_event_history")
+    async def get_tag_event_history(
+        self,
+        workflow_permanent_id: str,
+        organization_id: str,
+        limit: int = 100,
+    ) -> list[WorkflowTagEventModel]:
+        """Return tag events newest-first for a workflow. Includes DELETE and superseded SET rows."""
+        async with self.Session() as session:
+            stmt = (
+                select(WorkflowTagEventModel)
+                .where(WorkflowTagEventModel.organization_id == organization_id)
+                .where(WorkflowTagEventModel.workflow_permanent_id == workflow_permanent_id)
+                .where(WorkflowTagEventModel.deleted_at.is_(None))
+                .order_by(WorkflowTagEventModel.set_at.desc(), WorkflowTagEventModel.tag_event_id.desc())
+                .limit(limit)
+            )
+            result = await session.execute(stmt)
+            return list(result.scalars().all())
diff --git a/skyvern/forge/sdk/routes/workflow_copilot.py b/skyvern/forge/sdk/routes/workflow_copilot.py
index ed489b3f9..5611023eb 100644
--- a/skyvern/forge/sdk/routes/workflow_copilot.py
+++ b/skyvern/forge/sdk/routes/workflow_copilot.py
@@ -1,6 +1,7 @@
 import asyncio
 import contextlib
 import time
+import uuid
 from contextlib import contextmanager
 from copy import deepcopy
 from dataclasses import dataclass
@@ -240,7 +241,11 @@ async def _watch_for_cancel(
             return
 
 
-async def _ensure_terminal_frame(stream: EventSourceStream, already_emitted: bool) -> None:
+async def _ensure_terminal_frame(
+    stream: EventSourceStream,
+    already_emitted: bool,
+    turn_id: str | None = None,
+) -> None:
     """Emit a fallback ERROR frame if the turn hasn't sent a terminal one.
 
     Shielded so cancellation on the outer scope doesn't abort the send;
@@ -254,6 +259,8 @@ async def _ensure_terminal_frame(stream: EventSourceStream, already_emitted: boo
                 WorkflowCopilotStreamErrorUpdate(
                     type=WorkflowCopilotStreamMessageType.ERROR,
                     error="The assistant didn't finish this turn. Please try again.",
+                    turn_id=turn_id,
+                    narrative_summary=None,
                 )
             )
         )
@@ -309,6 +316,7 @@ def _build_recoverable_route_agent_result(
     workflow_modified: bool,
     clear_proposed_workflow: bool,
     global_llm_context: str | None,
+    turn_id: str | None = None,
 ) -> tuple[AgentResult, RecoverableFailure]:
     failure = build_recoverable_failure(error, workflow_modified=workflow_modified)
     agent_result = AgentResult(
@@ -318,6 +326,7 @@ def _build_recoverable_route_agent_result(
         workflow_was_persisted=False,
         proposal_disposition="no_proposal",
         clear_proposed_workflow=clear_proposed_workflow,
+        turn_id=turn_id,
     )
     _record_recoverable_failure_span_attrs(failure, proposal_disposition="no_proposal")
     return agent_result, failure
@@ -382,6 +391,7 @@ async def _persist_cancel_turn(
     original_workflow: Workflow | None,
     user_message: str,
     agent_result: AgentResult | None,
+    turn_id: str | None = None,
 ) -> None:
     """Persist a cancelled turn and emit a terminal SSE response frame.
 
@@ -449,6 +459,8 @@ async def _persist_cancel_turn(
                     proposal_disposition=proposal_disposition,
                     cancelled=True,
                     output_policy_diagnostics=output_policy_diagnostics,
+                    turn_id=turn_id or getattr(agent_result, "turn_id", None),
+                    narrative_summary=getattr(agent_result, "narrative_summary", None),
                 )
             )
         )
@@ -524,6 +536,8 @@ async def _finalise_normal_turn(
             response_type=getattr(agent_result, "response_type", "REPLY"),
             proposal_disposition=proposal_disposition,
             output_policy_diagnostics=getattr(agent_result, "output_policy_diagnostics", None),
+            turn_id=getattr(agent_result, "turn_id", None),
+            narrative_summary=getattr(agent_result, "narrative_summary", None),
         )
     )
 
@@ -1344,6 +1358,11 @@ async def _new_copilot_chat_post(
             organization_id=organization.organization_id,
         )
 
+        # Canonical turn_id for the whole HTTP request. Generated before any
+        # try-block so route-level error paths and the agent's TURN_START
+        # envelope all carry the same identifier.
+        turn_id = uuid.uuid4().hex
+
         original_workflow: Workflow | None = None
         chat = None
         agent_result: AgentResult | None = None
@@ -1474,6 +1493,8 @@ async def _new_copilot_chat_post(
                     WorkflowCopilotStreamErrorUpdate(
                         type=WorkflowCopilotStreamMessageType.ERROR,
                         error="Copilot is not configured for this organization. Contact support.",
+                        turn_id=turn_id,
+                        narrative_summary=None,
                     )
                 )
                 return
@@ -1498,8 +1519,11 @@ async def _new_copilot_chat_post(
                         )
                     )
 
-            # Count from the full message log; chat_history below is truncated.
-            turn_index = sum(1 for m in chat_messages if m.sender == WorkflowCopilotChatSender.USER) + 1
+            # Zero-based turn ordinal. The current user message has not been
+            # appended to chat_messages at this point, so ``sum(...)`` already
+            # counts only prior user turns and equals the index of the
+            # about-to-start turn.
+            turn_index = sum(1 for m in chat_messages if m.sender == WorkflowCopilotChatSender.USER)
 
             with bind_copilot_session_id(chat.workflow_copilot_chat_id):
                 agent_result = await run_copilot_agent(
@@ -1513,6 +1537,7 @@ async def _new_copilot_chat_post(
                     api_key=api_key,
                     config=copilot_config,
                     turn_index=turn_index,
+                    turn_id=turn_id,
                     prior_copilot_workflow_yaml=prior_copilot_workflow_yaml,
                 )
 
@@ -1526,6 +1551,7 @@ async def _new_copilot_chat_post(
                     original_workflow=original_workflow,
                     user_message=chat_request.message,
                     agent_result=agent_result,
+                    turn_id=turn_id,
                 )
                 terminal_frame_emitted = True
                 LOG.info(
@@ -1556,6 +1582,8 @@ async def _new_copilot_chat_post(
                 WorkflowCopilotStreamErrorUpdate(
                     type=WorkflowCopilotStreamMessageType.ERROR,
                     error=exc.detail,
+                    turn_id=turn_id,
+                    narrative_summary=None,
                 )
             )
         except LLMProviderError as exc:
@@ -1569,6 +1597,7 @@ async def _new_copilot_chat_post(
                     workflow_modified=workflow_modified,
                     clear_proposed_workflow=restored or workflow_modified,
                     global_llm_context=global_llm_context,
+                    turn_id=turn_id,
                 )
                 LOG.error(
                     "LLM provider error translated to recoverable workflow copilot v2 reply",
@@ -1602,6 +1631,8 @@ async def _new_copilot_chat_post(
                     WorkflowCopilotStreamErrorUpdate(
                         type=WorkflowCopilotStreamMessageType.ERROR,
                         error="Failed to process your request. Please try again.",
+                        turn_id=turn_id,
+                        narrative_summary=None,
                     )
                 )
         except asyncio.CancelledError:
@@ -1619,6 +1650,7 @@ async def _new_copilot_chat_post(
                         original_workflow=None,
                         user_message=chat_request.message,
                         agent_result=None,
+                        turn_id=turn_id,
                     )
                 )
                 terminal_frame_emitted = True
@@ -1650,6 +1682,7 @@ async def _new_copilot_chat_post(
                     workflow_modified=workflow_modified,
                     clear_proposed_workflow=restored or workflow_modified,
                     global_llm_context=global_llm_context,
+                    turn_id=turn_id,
                 )
                 LOG.error(
                     "Unexpected workflow copilot v2 error translated to recoverable reply",
@@ -1683,6 +1716,8 @@ async def _new_copilot_chat_post(
                     WorkflowCopilotStreamErrorUpdate(
                         type=WorkflowCopilotStreamMessageType.ERROR,
                         error="An error occurred. Please try again.",
+                        turn_id=turn_id,
+                        narrative_summary=None,
                     )
                 )
         finally:
@@ -1690,7 +1725,7 @@ async def _new_copilot_chat_post(
                 cancel_watcher.cancel()
                 with contextlib.suppress(asyncio.CancelledError, Exception):
                     await cancel_watcher
-            await _ensure_terminal_frame(stream, terminal_frame_emitted)
+            await _ensure_terminal_frame(stream, terminal_frame_emitted, turn_id=turn_id)
 
     return FastAPIEventSourceStream.create(request, stream_handler)
 
diff --git a/skyvern/forge/sdk/schemas/workflow_copilot.py b/skyvern/forge/sdk/schemas/workflow_copilot.py
index 2bf3c3732..02a11cb1e 100644
--- a/skyvern/forge/sdk/schemas/workflow_copilot.py
+++ b/skyvern/forge/sdk/schemas/workflow_copilot.py
@@ -25,6 +25,21 @@ class WorkflowCopilotChatSender(StrEnum):
     AI = "ai"
 
 
+# Wire-format mirror of ``copilot.turn_intent.TurnIntentMode`` — lives here
+# rather than importing the source enum because ``turn_intent`` already
+# imports this module (a back-edge would be circular). Values MUST stay in
+# lockstep; a cross-enum equality test in the unit tests catches drift.
+class WorkflowCopilotTurnMode(StrEnum):
+    BUILD = "build"
+    EDIT = "edit"
+    DIAGNOSE = "diagnose"
+    DOCS_ANSWER = "docs_answer"
+    DRAFT_ONLY = "draft_only"
+    CLARIFY = "clarify"
+    REFUSE = "refuse"
+    UNKNOWN = "unknown"
+
+
 class WorkflowCopilotChatMessage(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
@@ -98,6 +113,10 @@ class WorkflowCopilotStreamMessageType(StrEnum):
     CONDENSING = "condensing"
     NARRATION = "narration"
     BLOCK_PROGRESS = "block_progress"
+    TURN_START = "turn_start"
+    DESIGN_START = "design_start"
+    DESIGN_END = "design_end"
+    WORKFLOW_DRAFT = "workflow_draft"
 
 
 class WorkflowCopilotProcessingUpdate(BaseModel):
@@ -133,11 +152,27 @@ class WorkflowCopilotStreamResponseUpdate(BaseModel):
         None,
         description="Diagnostic output-policy labels for raw-vs-final quality reporting.",
     )
+    turn_id: str | None = Field(
+        None,
+        description="UUID generated by the route at turn start; correlates this terminal frame to the matching turn_start envelope.",
+    )
+    narrative_summary: str | None = Field(
+        None,
+        description="One-line accomplishment summary for the turn. Optional; the frontend falls back to the response message when absent.",
+    )
 
 
 class WorkflowCopilotStreamErrorUpdate(BaseModel):
     type: WorkflowCopilotStreamMessageType = Field(WorkflowCopilotStreamMessageType.ERROR, description="Message type")
     error: str = Field(..., description="Error message")
+    turn_id: str | None = Field(
+        None,
+        description="UUID generated by the route at turn start; correlates this terminal frame to the matching turn_start envelope.",
+    )
+    narrative_summary: str | None = Field(
+        None,
+        description="One-line accomplishment summary; None for route-level error frames where no agent context was available.",
+    )
 
 
 class WorkflowCopilotToolCallUpdate(BaseModel):
@@ -204,6 +239,44 @@ class WorkflowCopilotBlockProgressUpdate(BaseModel):
     timestamp: datetime = Field(..., description="Server timestamp")
 
 
+class WorkflowCopilotTurnStartUpdate(BaseModel):
+    type: WorkflowCopilotStreamMessageType = Field(
+        WorkflowCopilotStreamMessageType.TURN_START, description="Message type"
+    )
+    turn_id: str = Field(..., description="UUID for this turn; correlates with the matching terminal frame")
+    turn_index: int = Field(..., description="Zero-based ordinal of this turn within the chat")
+    mode: WorkflowCopilotTurnMode = Field(..., description="TurnIntent mode for this turn")
+    timestamp: datetime = Field(..., description="Server timestamp")
+
+
+class WorkflowCopilotDesignStartUpdate(BaseModel):
+    type: WorkflowCopilotStreamMessageType = Field(
+        WorkflowCopilotStreamMessageType.DESIGN_START, description="Message type"
+    )
+    timestamp: datetime = Field(..., description="Server timestamp")
+
+
+class WorkflowCopilotDesignEndUpdate(BaseModel):
+    type: WorkflowCopilotStreamMessageType = Field(
+        WorkflowCopilotStreamMessageType.DESIGN_END, description="Message type"
+    )
+    timestamp: datetime = Field(..., description="Server timestamp")
+
+
+class WorkflowCopilotWorkflowDraftUpdate(BaseModel):
+    # Summary-only payload. The staged/persisted workflow definition is
+    # transported elsewhere (terminal `updated_workflow` on RESPONSE, or
+    # the chat's `proposed_workflow` field); this event signals only that a
+    # draft was produced and what shape it has.
+    type: WorkflowCopilotStreamMessageType = Field(
+        WorkflowCopilotStreamMessageType.WORKFLOW_DRAFT, description="Message type"
+    )
+    block_count: int = Field(..., description="Number of blocks in the drafted workflow")
+    block_labels: list[str] = Field(default_factory=list, description="Ordered block labels in the drafted workflow")
+    summary: str | None = Field(None, description="Optional one-line description; populated by a follow-up PR")
+    timestamp: datetime = Field(..., description="Server timestamp")
+
+
 class WorkflowYAMLConversionRequest(BaseModel):
     workflow_definition_yaml: str = Field(..., description="Workflow definition YAML to convert to blocks")
     workflow_id: str = Field(..., description="Workflow ID")
diff --git a/skyvern/forge/sdk/services/org_auth_service.py b/skyvern/forge/sdk/services/org_auth_service.py
index d2d1b38e8..14ff721b7 100644
--- a/skyvern/forge/sdk/services/org_auth_service.py
+++ b/skyvern/forge/sdk/services/org_auth_service.py
@@ -1,3 +1,4 @@
+import asyncio
 import time
 from dataclasses import dataclass
 from typing import Annotated, Sequence
@@ -17,12 +18,14 @@ from skyvern.forge.sdk.core import skyvern_context
 from skyvern.forge.sdk.db.agent_db import AgentDB
 from skyvern.forge.sdk.models import TokenPayload
 from skyvern.forge.sdk.schemas.organizations import Organization, OrganizationAuthToken, OrganizationAuthTokenType
+from skyvern.forge.sdk.workflow.models.tags import CallerType
 
 LOG = structlog.get_logger()
 
 AUTHENTICATION_TTL = 60 * 60  # one hour
 CACHE_SIZE = 128
 ALGORITHM = "HS256"
+SKYVERN_UI_USER_AGENT = "skyvern-ui"
 _SAFE_JWT_ERROR_REASONS = {
     "Not enough segments",
     "Invalid payload padding",
@@ -160,31 +163,12 @@ async def get_current_org(
         )
     organization = None
     if x_api_key:
-        organization = await _get_current_org_cached(x_api_key, app.DATABASE)
+        organization = await get_current_org_cached(x_api_key, app.DATABASE)
     elif authorization:
-        organization = await _authenticate_helper(authorization)
+        organization = await authenticate_helper(authorization)
 
     if organization:
-        try:
-            # Set in context
-            curr_ctx = skyvern_context.current()
-            if curr_ctx:
-                curr_ctx.organization_id = organization.organization_id
-                curr_ctx.organization_name = organization.organization_name
-
-            # Set organization info on OTEL span for tracing
-            if settings.OTEL_ENABLED:
-                try:
-                    span = trace.get_current_span()
-                    if span:
-                        span.set_attribute("organization_id", organization.organization_id)
-                        if organization.organization_name:
-                            span.set_attribute("organization_name", organization.organization_name)
-                except Exception:
-                    pass  # Silently ignore OTEL errors
-        except Exception:
-            pass
-
+        apply_request_org_context(organization)
         return organization
     raise HTTPException(
         status_code=status.HTTP_403_FORBIDDEN,
@@ -192,6 +176,32 @@ async def get_current_org(
     )
 
 
+def apply_request_org_context(organization: Organization) -> None:
+    """Populate skyvern_context + OTEL span with the resolved organization.
+
+    Shared by get_current_org and get_current_caller_context so the cached
+    helper paths (which short-circuit the per-helper context-setting on cache
+    hits) still leave a consistent request-scoped context for downstream routes.
+    """
+    try:
+        ctx = skyvern_context.current()
+        if ctx:
+            ctx.organization_id = organization.organization_id
+            ctx.organization_name = organization.organization_name
+    except Exception:
+        pass
+    if not settings.OTEL_ENABLED:
+        return
+    try:
+        span = trace.get_current_span()
+        if span:
+            span.set_attribute("organization_id", organization.organization_id)
+            if organization.organization_name:
+                span.set_attribute("organization_name", organization.organization_name)
+    except Exception:
+        pass  # OTEL must never fail auth.
+
+
 async def get_current_org_with_api_key(
     x_api_key: Annotated[str | None, Header()] = None,
 ) -> Organization:
@@ -200,7 +210,7 @@ async def get_current_org_with_api_key(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Invalid credentials",
         )
-    return await _get_current_org_cached(x_api_key, app.DATABASE)
+    return await get_current_org_cached(x_api_key, app.DATABASE)
 
 
 async def get_current_org_with_authentication(
@@ -211,10 +221,10 @@ async def get_current_org_with_authentication(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Invalid credentials",
         )
-    return await _authenticate_helper(authorization)
+    return await authenticate_helper(authorization)
 
 
-async def _authenticate_helper(authorization: str) -> Organization:
+async def authenticate_helper(authorization: str) -> Organization:
     parts = authorization.split(" ", 1)
     if len(parts) < 2 or not parts[1]:
         raise HTTPException(
@@ -250,11 +260,11 @@ async def get_current_user_id(
 ) -> str:
     # Try authorization header first, but only if the authentication function is configured
     if authorization and app.authenticate_user_function:
-        return await _authenticate_user_helper(authorization)
+        return await authenticate_user_helper(authorization)
 
     # Fall back to API key + skyvern-ui user agent
-    if x_api_key and x_user_agent == "skyvern-ui":
-        organization = await _get_current_org_cached(x_api_key, app.DATABASE)
+    if x_api_key and x_user_agent == SKYVERN_UI_USER_AGENT:
+        organization = await get_current_org_cached(x_api_key, app.DATABASE)
         if organization:
             return f"{organization.organization_id}_user"
 
@@ -272,10 +282,10 @@ async def get_current_user_id_with_authentication(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Invalid credentials",
         )
-    return await _authenticate_user_helper(authorization)
+    return await authenticate_user_helper(authorization)
 
 
-async def _authenticate_user_helper(authorization: str) -> str:
+async def authenticate_user_helper(authorization: str) -> str:
     parts = authorization.split(" ", 1)
     if len(parts) < 2 or not parts[1]:
         raise HTTPException(
@@ -380,7 +390,7 @@ _current_org_cache: TTLCache = TTLCache(maxsize=CACHE_SIZE, ttl=AUTHENTICATION_T
 
 
 @cached(cache=_current_org_cache)
-async def _get_current_org_cached(x_api_key: str, db: AgentDB) -> Organization:
+async def get_current_org_cached(x_api_key: str, db: AgentDB) -> Organization:
     """Authentication is cached for one hour."""
     validation = await resolve_org_from_api_key(x_api_key, db)
 
@@ -403,3 +413,59 @@ def invalidate_cached_org(organization_id: str) -> None:
     ]
     for key in keys_to_remove:
         _current_org_cache.pop(key, None)
+
+
+@dataclass(frozen=True)
+class CallerContext:
+    organization: Organization
+    caller_id: str
+    caller_type: CallerType
+
+
+async def get_current_caller_context(
+    x_api_key: Annotated[str | None, Header(include_in_schema=False)] = None,
+    authorization: Annotated[str | None, Header(include_in_schema=False)] = None,
+    x_user_agent: Annotated[str | None, Header(include_in_schema=False)] = None,
+) -> CallerContext:
+    """Resolve the caller identity for write-attribution. Mirrors get_current_org's
+    x-api-key-first precedence and OTEL side effects."""
+    # x-api-key path FIRST — mirrors get_current_org so clients with both
+    # headers (valid key + stale JWT) keep authenticating via the key.
+    if x_api_key:
+        organization = await get_current_org_cached(x_api_key, app.DATABASE)
+        apply_request_org_context(organization)
+        # x-user-agent is spoofable and is NOT an access-control check —
+        # it only flips set_by attribution from API_KEY to USER. Real auth
+        # is already validated by get_current_org_cached above.
+        if x_user_agent == SKYVERN_UI_USER_AGENT:
+            return CallerContext(
+                organization=organization,
+                caller_id=f"{organization.organization_id}_user",
+                caller_type=CallerType.USER,
+            )
+        return CallerContext(
+            organization=organization,
+            caller_id=organization.organization_id,
+            caller_type=CallerType.API_KEY,
+        )
+
+    # JWT path needs both callbacks; otherwise we'd accept user-auth then 403
+    # on org-auth, confusing the caller with a rejection on valid credentials.
+    if authorization and app.authenticate_user_function and app.authentication_function:
+        # Both helpers re-decode the same token independently — run them
+        # concurrently so JWT validation cost is paid in parallel, not serially.
+        user_id, organization = await asyncio.gather(
+            authenticate_user_helper(authorization),
+            authenticate_helper(authorization),
+        )
+        apply_request_org_context(organization)
+        return CallerContext(
+            organization=organization,
+            caller_id=user_id,
+            caller_type=CallerType.USER,
+        )
+
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail="Invalid credentials",
+    )
diff --git a/skyvern/forge/sdk/workflow/models/block.py b/skyvern/forge/sdk/workflow/models/block.py
index 3a04f29d0..21196fd9e 100644
--- a/skyvern/forge/sdk/workflow/models/block.py
+++ b/skyvern/forge/sdk/workflow/models/block.py
@@ -17,7 +17,7 @@ from datetime import date, datetime, time, timezone
 from email.message import EmailMessage
 from pathlib import Path
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Annotated, Any, Awaitable, Callable, ClassVar, Literal, Union, cast
+from typing import Annotated, Any, Awaitable, Callable, ClassVar, Literal, Union, cast
 from urllib.parse import quote, urlparse
 
 import aiofiles
@@ -141,9 +141,6 @@ from skyvern.webeye.utils.page import SkyvernFrame
 
 LOG = structlog.get_logger()
 
-if TYPE_CHECKING:
-    from skyvern.forge.sdk.workflow.models.workflow import Workflow, WorkflowDefinition
-
 
 # SKY-8818: observability threshold for under-configured file_download blocks.
 # Warning fires when `max_steps_per_run` is set below this value. Not a behavior change —
@@ -438,13 +435,6 @@ class Block(BaseModel, abc.ABC):
             is_synthetic_loop_failure=is_synthetic_loop_failure,
         )
 
-    def allow_content_blocking_extensions_for_browser_launch(
-        self, workflow: Workflow | WorkflowDefinition | None = None
-    ) -> bool:
-        if workflow is not None:
-            return workflow.allow_content_blocking_extensions_for_browser_launch()
-        return self.block_type != BlockType.LOGIN
-
     async def get_or_create_browser_state(
         self,
         workflow_run_id: str,
@@ -472,36 +462,11 @@ class Block(BaseModel, abc.ABC):
                 organization_id=organization_id,
             )
             try:
-                workflow_run_context = self.get_workflow_run_context(workflow_run_id)
-                workflow = workflow_run_context.workflow
-                if workflow is None:
-                    LOG.warning(
-                        "Workflow missing from context while resolving browser launch extension policy; fetching workflow",
-                        workflow_run_id=workflow_run_id,
-                        workflow_id=workflow_run.workflow_id,
-                        organization_id=workflow_run.organization_id,
-                    )
-                    workflow = await app.WORKFLOW_SERVICE.get_workflow(
-                        workflow_id=workflow_run.workflow_id,
-                        organization_id=workflow_run.organization_id,
-                    )
-                    workflow_run_context.set_workflow(workflow)
-                if workflow is None:
-                    LOG.warning(
-                        "Workflow unavailable while resolving browser launch extension policy; using block-level fallback",
-                        workflow_run_id=workflow_run_id,
-                        workflow_id=workflow_run.workflow_id,
-                        organization_id=workflow_run.organization_id,
-                        block_type=self.block_type,
-                    )
                 browser_state = await app.BROWSER_MANAGER.get_or_create_for_workflow_run(
                     workflow_run=workflow_run,
                     url=None,
                     browser_session_id=browser_session_id,
                     browser_profile_id=workflow_run.browser_profile_id,
-                    allow_content_blocking_extensions=self.allow_content_blocking_extensions_for_browser_launch(
-                        workflow=workflow
-                    ),
                 )
                 await browser_state.check_and_fix_state(
                     url=None,
@@ -1204,9 +1169,6 @@ class BaseTaskBlock(Block):
                         url=_bm_url,
                         browser_session_id=browser_session_id,
                         browser_profile_id=workflow_run.browser_profile_id,
-                        allow_content_blocking_extensions=self.allow_content_blocking_extensions_for_browser_launch(
-                            workflow=workflow
-                        ),
                     )
                     working_page = await browser_state.get_working_page()
                     if not working_page:
diff --git a/skyvern/forge/sdk/workflow/models/tags.py b/skyvern/forge/sdk/workflow/models/tags.py
index 8b4bcb462..ca31cf603 100644
--- a/skyvern/forge/sdk/workflow/models/tags.py
+++ b/skyvern/forge/sdk/workflow/models/tags.py
@@ -31,8 +31,8 @@ class CallerType(StrEnum):
 
 @dataclass(frozen=True)
 class TagWriteContext:
-    """Attribution carried from a write call into the repository layer."""
+    """Attribution persisted on each tag event row. caller_type is nullable for backfill scripts."""
 
     caller_id: str
-    caller_type: CallerType
     source: TagSource
+    caller_type: CallerType | None = None
diff --git a/skyvern/forge/sdk/workflow/models/workflow.py b/skyvern/forge/sdk/workflow/models/workflow.py
index 32d2f6717..73508beaa 100644
--- a/skyvern/forge/sdk/workflow/models/workflow.py
+++ b/skyvern/forge/sdk/workflow/models/workflow.py
@@ -21,7 +21,7 @@ from skyvern.forge.sdk.workflow.models.run_limits import (
 )
 from skyvern.forge.sdk.workflow.models.validators import normalize_run_metadata, normalize_run_with
 from skyvern.schemas.runs import ProxyLocationInput, ScriptRunResponse
-from skyvern.schemas.workflows import BlockType, WorkflowStatus
+from skyvern.schemas.workflows import WorkflowStatus
 from skyvern.utils.secret_headers import mask_header_values
 from skyvern.utils.url_validators import validate_url
 
@@ -76,9 +76,6 @@ class WorkflowDefinition(BaseModel):
     error_code_mapping: dict[str, str] | None = None
     workflow_system_prompt: str | None = None
 
-    def allow_content_blocking_extensions_for_browser_launch(self) -> bool:
-        return all(block.block_type != BlockType.LOGIN for block in get_all_blocks(self.blocks))
-
     def validate(self) -> None:
         all_labels: set[str] = set()
         duplicate_labels: set[str] = set()
@@ -167,9 +164,6 @@ class Workflow(BaseModel):
                 return parameter
         return None
 
-    def allow_content_blocking_extensions_for_browser_launch(self) -> bool:
-        return self.workflow_definition.allow_content_blocking_extensions_for_browser_launch()
-
 
 class WorkflowRunStatus(StrEnum):
     created = "created"
diff --git a/skyvern/webeye/browser_manager.py b/skyvern/webeye/browser_manager.py
index faa53c1e5..b6cd40805 100644
--- a/skyvern/webeye/browser_manager.py
+++ b/skyvern/webeye/browser_manager.py
@@ -23,7 +23,6 @@ class BrowserManager(Protocol):
         url: str | None = None,
         browser_session_id: str | None = None,
         browser_profile_id: str | None = None,
-        allow_content_blocking_extensions: bool = True,
     ) -> BrowserState: ...
 
     async def cleanup_for_task(
diff --git a/skyvern/webeye/browser_state.py b/skyvern/webeye/browser_state.py
index 3dd05c311..72dbd24fd 100644
--- a/skyvern/webeye/browser_state.py
+++ b/skyvern/webeye/browser_state.py
@@ -16,7 +16,6 @@ class BrowserState(Protocol):
     browser_context: BrowserContext | None
     browser_artifacts: BrowserArtifacts
     browser_cleanup: BrowserCleanupFunc
-    allow_content_blocking_extensions: bool
     pw: Playwright
 
     async def check_and_fix_state(
diff --git a/skyvern/webeye/real_browser_manager.py b/skyvern/webeye/real_browser_manager.py
index a506b90b9..7e3852140 100644
--- a/skyvern/webeye/real_browser_manager.py
+++ b/skyvern/webeye/real_browser_manager.py
@@ -37,7 +37,6 @@ class RealBrowserManager(BrowserManager):
         cdp_connect_headers: dict[str, str] | None = None,
         browser_address: str | None = None,
         browser_profile_id: str | None = None,
-        allow_content_blocking_extensions: bool = True,
     ) -> BrowserState:
         pw = await async_playwright().start()
         (
@@ -57,7 +56,6 @@ class RealBrowserManager(BrowserManager):
             cdp_connect_headers=cdp_connect_headers,
             browser_address=browser_address,
             browser_profile_id=browser_profile_id,
-            allow_content_blocking_extensions=allow_content_blocking_extensions,
         )
         return RealBrowserState(
             pw=pw,
@@ -65,21 +63,11 @@ class RealBrowserManager(BrowserManager):
             page=None,
             browser_artifacts=browser_artifacts,
             browser_cleanup=browser_cleanup,
-            allow_content_blocking_extensions=allow_content_blocking_extensions,
         )
 
     def evict_page(self, page_id: str) -> None:
         self.pages.pop(page_id, None)
 
-    @staticmethod
-    def _browser_state_satisfies_content_blocking_policy(
-        browser_state: BrowserState,
-        allow_content_blocking_extensions: bool,
-    ) -> bool:
-        if allow_content_blocking_extensions:
-            return True
-        return getattr(browser_state, "allow_content_blocking_extensions", True) is False
-
     def get_for_task(self, task_id: str, workflow_run_id: str | None = None) -> BrowserState | None:
         if task_id in self.pages:
             return self.pages[task_id]
@@ -176,35 +164,19 @@ class RealBrowserManager(BrowserManager):
         url: str | None = None,
         browser_session_id: str | None = None,
         browser_profile_id: str | None = None,
-        allow_content_blocking_extensions: bool = True,
     ) -> BrowserState:
         parent_workflow_run_id = workflow_run.parent_workflow_run_id
         workflow_run_id = workflow_run.workflow_run_id
         if browser_profile_id is None:
             browser_profile_id = workflow_run.browser_profile_id
-        effective_allow_content_blocking_extensions = allow_content_blocking_extensions and not bool(browser_profile_id)
-        sync_parent_browser_state = bool(parent_workflow_run_id and not browser_session_id)
 
         # Check own cache entry first so navigate_to_url is only called on the first step.
         # Don't pass parent_workflow_run_id here — that lookup is deferred to the block
         # below so PBS runs don't accidentally inherit the parent's browser.
         browser_state = self.get_for_workflow_run(workflow_run_id=workflow_run_id)
         if browser_state:
-            if self._browser_state_satisfies_content_blocking_policy(
-                browser_state,
-                allow_content_blocking_extensions=effective_allow_content_blocking_extensions,
-            ):
-                LOG.debug("Returning cached browser state for workflow run", workflow_run_id=workflow_run_id)
-                return browser_state
-            LOG.warning(
-                "Skipping cached browser state because requested extension policy is stricter than launch policy",
-                workflow_run_id=workflow_run_id,
-                requested_allow_content_blocking_extensions=effective_allow_content_blocking_extensions,
-                cached_allow_content_blocking_extensions=getattr(
-                    browser_state, "allow_content_blocking_extensions", None
-                ),
-            )
-            self.pages.pop(workflow_run_id, None)
+            LOG.debug("Returning cached browser state for workflow run", workflow_run_id=workflow_run_id)
+            return browser_state
 
         # When an explicit browser_session_id is provided (e.g. from a workflow
         # trigger block), skip the parent workflow lookup so the child uses the
@@ -216,27 +188,11 @@ class RealBrowserManager(BrowserManager):
                 workflow_run_id=workflow_run_id, parent_workflow_run_id=parent_workflow_run_id
             )
             if browser_state:
-                if not self._browser_state_satisfies_content_blocking_policy(
-                    browser_state,
-                    allow_content_blocking_extensions=effective_allow_content_blocking_extensions,
-                ):
-                    LOG.warning(
-                        "Skipping inherited browser state because requested extension policy is stricter than launch policy",
-                        workflow_run_id=workflow_run_id,
-                        parent_workflow_run_id=parent_workflow_run_id,
-                        requested_allow_content_blocking_extensions=effective_allow_content_blocking_extensions,
-                        inherited_allow_content_blocking_extensions=getattr(
-                            browser_state, "allow_content_blocking_extensions", None
-                        ),
-                    )
-                    browser_state = None
-                    sync_parent_browser_state = False
-                else:
-                    # always keep the browser state for the workflow run and the parent workflow run synced
-                    self.pages[workflow_run_id] = browser_state
-                    if parent_workflow_run_id:
-                        self.pages[parent_workflow_run_id] = browser_state
-                    return browser_state
+                # always keep the browser state for the workflow run and the parent workflow run synced
+                self.pages[workflow_run_id] = browser_state
+                if parent_workflow_run_id:
+                    self.pages[parent_workflow_run_id] = browser_state
+                return browser_state
 
         if browser_session_id:
             LOG.info(
@@ -251,28 +207,13 @@ class RealBrowserManager(BrowserManager):
                     "Browser state not found in persistent sessions manager", browser_session_id=browser_session_id
                 )
             else:
-                if not self._browser_state_satisfies_content_blocking_policy(
-                    browser_state,
-                    allow_content_blocking_extensions=effective_allow_content_blocking_extensions,
-                ):
-                    LOG.warning(
-                        "Ignoring browser session state because requested extension policy is stricter than launch policy",
-                        browser_session_id=browser_session_id,
-                        workflow_run_id=workflow_run.workflow_run_id,
-                        requested_allow_content_blocking_extensions=effective_allow_content_blocking_extensions,
-                        session_allow_content_blocking_extensions=getattr(
-                            browser_state, "allow_content_blocking_extensions", None
-                        ),
-                    )
-                    browser_state = None
+                LOG.info("Used to occupy browser session here", browser_session_id=browser_session_id)
+                page = await browser_state.get_working_page()
+                if page:
+                    if url:
+                        await browser_state.navigate_to_url(page=page, url=url)
                 else:
-                    LOG.info("Used to occupy browser session here", browser_session_id=browser_session_id)
-                    page = await browser_state.get_working_page()
-                    if page:
-                        if url:
-                            await browser_state.navigate_to_url(page=page, url=url)
-                    else:
-                        LOG.warning("Browser state has no page", workflow_run_id=workflow_run.workflow_run_id)
+                    LOG.warning("Browser state has no page", workflow_run_id=workflow_run.workflow_run_id)
 
         if browser_state is None:
             LOG.info(
@@ -286,15 +227,6 @@ class RealBrowserManager(BrowserManager):
                 )
                 if session and session.proxy_location is not None:
                     proxy_location = session.proxy_location
-            if browser_profile_id:
-                LOG.info(
-                    "Content-blocking extensions disabled for browser profile launch",
-                    workflow_run_id=workflow_run.workflow_run_id,
-                    workflow_permanent_id=workflow_run.workflow_permanent_id,
-                    browser_profile_id=browser_profile_id,
-                    requested_allow_content_blocking_extensions=allow_content_blocking_extensions,
-                    allow_content_blocking_extensions=effective_allow_content_blocking_extensions,
-                )
             browser_state = await self._create_browser_state(
                 proxy_location=proxy_location,
                 url=url,
@@ -305,7 +237,6 @@ class RealBrowserManager(BrowserManager):
                 cdp_connect_headers=workflow_run.cdp_connect_headers,
                 browser_address=workflow_run.browser_address,
                 browser_profile_id=browser_profile_id,
-                allow_content_blocking_extensions=effective_allow_content_blocking_extensions,
             )
 
             if browser_session_id:
@@ -319,7 +250,7 @@ class RealBrowserManager(BrowserManager):
         # browser.  When an explicit browser_session_id is provided the child
         # has its own browser, and overwriting the parent's entry would break
         # subsequent parent blocks.
-        if sync_parent_browser_state and parent_workflow_run_id:
+        if parent_workflow_run_id and not browser_session_id:
             self.pages[parent_workflow_run_id] = browser_state
 
         # The URL here is only used when creating a new page, and not when using an existing page.
diff --git a/skyvern/webeye/real_browser_state.py b/skyvern/webeye/real_browser_state.py
index bc2477308..937944610 100644
--- a/skyvern/webeye/real_browser_state.py
+++ b/skyvern/webeye/real_browser_state.py
@@ -43,14 +43,12 @@ class RealBrowserState(BrowserState):
         page: Page | None = None,
         browser_artifacts: BrowserArtifacts = BrowserArtifacts(),
         browser_cleanup: BrowserCleanupFunc = None,
-        allow_content_blocking_extensions: bool = True,
     ):
         self.__page = page
         self.pw = pw
         self.browser_context = browser_context
         self.browser_artifacts = browser_artifacts
         self.browser_cleanup = browser_cleanup
-        self.allow_content_blocking_extensions = allow_content_blocking_extensions
 
     async def __assert_page(self) -> Page:
         page = await self.get_working_page()
@@ -108,7 +106,6 @@ class RealBrowserState(BrowserState):
                 cdp_connect_headers=cdp_connect_headers,
                 browser_address=browser_address,
                 browser_profile_id=browser_profile_id,
-                allow_content_blocking_extensions=self.allow_content_blocking_extensions,
             )
             self.browser_context = browser_context
             self.browser_artifacts = browser_artifacts
diff --git a/tests/unit/test_block_browser_extension_policy.py b/tests/unit/test_block_browser_extension_policy.py
deleted file mode 100644
index 1d5e92cc6..000000000
--- a/tests/unit/test_block_browser_extension_policy.py
+++ /dev/null
@@ -1,49 +0,0 @@
-from __future__ import annotations
-
-from datetime import UTC, datetime
-
-from skyvern.forge.sdk.workflow.models.block import LoginBlock, NavigationBlock
-from skyvern.forge.sdk.workflow.models.parameter import OutputParameter
-from skyvern.forge.sdk.workflow.models.workflow import WorkflowDefinition
-
-_NOW = datetime.now(UTC)
-
-
-def _output_parameter(label: str) -> OutputParameter:
-    return OutputParameter(
-        parameter_type="output",
-        key=f"{label}_output",
-        workflow_id="test_wf",
-        output_parameter_id=f"op_{label}",
-        created_at=_NOW,
-        modified_at=_NOW,
-    )
-
-
-def test_login_block_uses_pristine_browser_launch() -> None:
-    block = LoginBlock(label="login", output_parameter=_output_parameter("login"))
-
-    assert not block.allow_content_blocking_extensions_for_browser_launch()
-
-
-def test_navigation_block_allows_content_blocking_extensions() -> None:
-    block = NavigationBlock(
-        label="navigate",
-        output_parameter=_output_parameter("navigate"),
-        navigation_goal="Open the page",
-    )
-
-    assert block.allow_content_blocking_extensions_for_browser_launch()
-
-
-def test_workflow_definition_with_later_login_block_uses_pristine_browser_launch() -> None:
-    navigation_block = NavigationBlock(
-        label="navigate",
-        output_parameter=_output_parameter("navigate"),
-        navigation_goal="Open the login page",
-    )
-    login_block = LoginBlock(label="login", output_parameter=_output_parameter("login"))
-    workflow_definition = WorkflowDefinition(parameters=[], blocks=[navigation_block, login_block])
-
-    assert not workflow_definition.allow_content_blocking_extensions_for_browser_launch()
-    assert not navigation_block.allow_content_blocking_extensions_for_browser_launch(workflow=workflow_definition)
diff --git a/tests/unit/test_caller_context.py b/tests/unit/test_caller_context.py
new file mode 100644
index 000000000..e60a08001
--- /dev/null
+++ b/tests/unit/test_caller_context.py
@@ -0,0 +1,180 @@
+"""Tests for the get_current_caller_context auth dependency.
+
+Focused on the dispatch contracts:
+  - x-api-key takes precedence over Authorization (matches get_current_org).
+  - JWT path requires BOTH auth callbacks to be configured.
+  - x-user-agent: skyvern-ui flips API key callers to CallerType.USER.
+  - 403 (not 401) on auth failure.
+
+The underlying helpers (authenticate_helper, authenticate_user_helper,
+get_current_org_cached) are stubbed via monkeypatch so these tests only
+exercise the dispatch logic.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+from fastapi import HTTPException
+
+from skyvern.forge.sdk.schemas.organizations import Organization
+from skyvern.forge.sdk.services import org_auth_service as caller_context_module
+from skyvern.forge.sdk.services.org_auth_service import (
+    CallerContext,
+    get_current_caller_context,
+)
+from skyvern.forge.sdk.workflow.models.tags import CallerType
+
+
+def _org(org_id: str = "o_test", name: str = "Test Org") -> Organization:
+    now = datetime.now(timezone.utc)
+    return Organization(
+        organization_id=org_id,
+        organization_name=name,
+        webhook_callback_url=None,
+        max_steps_per_run=None,
+        max_retries_per_step=None,
+        domain=None,
+        created_at=now,
+        modified_at=now,
+    )
+
+
+async def _async_return(value):
+    return value
+
+
+@pytest.fixture
+def patched_app(monkeypatch: pytest.MonkeyPatch):
+    """Stub app.authentication_function / authenticate_user_function /
+    get_current_org_cached / authenticate_helper / authenticate_user_helper.
+    """
+    from skyvern.forge import app
+
+    # AppHolder proxies setattr to its inner instance but doesn't support
+    # delattr cleanly, so use save/restore instead of monkeypatch for the
+    # auth-callback attributes on app itself.
+    prior_auth = getattr(app, "authentication_function", None) if hasattr(app, "authentication_function") else None
+    prior_user_auth = (
+        getattr(app, "authenticate_user_function", None) if hasattr(app, "authenticate_user_function") else None
+    )
+    app.authentication_function = lambda token: _async_return(_org())
+    app.authenticate_user_function = lambda token: _async_return("user_42")
+
+    async def fakeget_current_org_cached(x_api_key: str, db) -> Organization:
+        return _org()
+
+    async def fakeauthenticate_helper(authorization: str) -> Organization:
+        return _org()
+
+    async def fakeauthenticate_user_helper(authorization: str) -> str:
+        return "user_42"
+
+    monkeypatch.setattr(
+        caller_context_module,
+        "get_current_org_cached",
+        fakeget_current_org_cached,
+    )
+    monkeypatch.setattr(
+        caller_context_module,
+        "authenticate_helper",
+        fakeauthenticate_helper,
+    )
+    monkeypatch.setattr(
+        caller_context_module,
+        "authenticate_user_helper",
+        fakeauthenticate_user_helper,
+    )
+
+    yield
+    app.authentication_function = prior_auth
+    app.authenticate_user_function = prior_user_auth
+
+
+@pytest.mark.asyncio
+async def test_no_headers_raises_403(patched_app) -> None:
+    with pytest.raises(HTTPException) as exc:
+        await get_current_caller_context(x_api_key=None, authorization=None, x_user_agent=None)
+    assert exc.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_api_key_only_returns_api_key_caller(patched_app) -> None:
+    ctx = await get_current_caller_context(x_api_key="key", authorization=None, x_user_agent=None)
+    assert isinstance(ctx, CallerContext)
+    assert ctx.caller_type == CallerType.API_KEY
+    assert ctx.caller_id == "o_test"
+    assert ctx.organization.organization_id == "o_test"
+
+
+@pytest.mark.asyncio
+async def test_api_key_plus_skyvern_ui_returns_user_caller(patched_app) -> None:
+    ctx = await get_current_caller_context(
+        x_api_key="key",
+        authorization=None,
+        x_user_agent="skyvern-ui",
+    )
+    assert ctx.caller_type == CallerType.USER
+    assert ctx.caller_id == "o_test_user"
+
+
+@pytest.mark.asyncio
+async def test_jwt_only_returns_user_caller(patched_app) -> None:
+    ctx = await get_current_caller_context(
+        x_api_key=None,
+        authorization="Bearer abc",
+        x_user_agent=None,
+    )
+    assert ctx.caller_type == CallerType.USER
+    assert ctx.caller_id == "user_42"
+
+
+@pytest.mark.asyncio
+async def test_x_api_key_takes_precedence_when_both_headers_present(
+    patched_app, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A client sending both x-api-key (valid) and Authorization (anything)
+    must continue to auth via the API key, matching get_current_org's
+    precedence. The JWT helpers must not be invoked.
+    """
+    user_calls = {"n": 0}
+    org_jwt_calls = {"n": 0}
+
+    async def should_not_run_user(_: str) -> str:
+        user_calls["n"] += 1
+        raise AssertionError("JWT user-auth path must not run when x-api-key is present")
+
+    async def should_not_run_org_jwt(_: str) -> Organization:
+        org_jwt_calls["n"] += 1
+        raise AssertionError("JWT org-auth path must not run when x-api-key is present")
+
+    monkeypatch.setattr(caller_context_module, "authenticate_user_helper", should_not_run_user)
+    monkeypatch.setattr(caller_context_module, "authenticate_helper", should_not_run_org_jwt)
+
+    ctx = await get_current_caller_context(
+        x_api_key="key",
+        authorization="Bearer something",
+        x_user_agent=None,
+    )
+    assert ctx.caller_type == CallerType.API_KEY
+    assert user_calls["n"] == 0
+    assert org_jwt_calls["n"] == 0
+
+
+@pytest.mark.asyncio
+async def test_jwt_path_requires_both_callbacks_configured(patched_app, monkeypatch: pytest.MonkeyPatch) -> None:
+    """If only authenticate_user_function is wired (no authentication_function),
+    the JWT path must not be entered — otherwise we'd resolve a user_id and
+    then 403 on the org step, confusing the caller.
+    """
+    from skyvern.forge import app
+
+    app.authentication_function = None
+    with pytest.raises(HTTPException) as exc:
+        await get_current_caller_context(
+            x_api_key=None,
+            authorization="Bearer abc",
+            x_user_agent=None,
+        )
+    assert exc.value.status_code == 403
diff --git a/tests/unit/test_copilot_turn_span.py b/tests/unit/test_copilot_turn_span.py
index b6ee526d8..38b11f8b9 100644
--- a/tests/unit/test_copilot_turn_span.py
+++ b/tests/unit/test_copilot_turn_span.py
@@ -107,8 +107,8 @@ async def test_copilot_turn_span_parents_inner_spans(
     assert attrs.get("skyvern.span.role") == "wrapper"
     assert attrs.get("copilot.session_id") == "chat_abc"
     assert attrs.get("workflow_permanent_id") == "wpid_xyz"
-    # one prior user msg in history + this turn = 2.
-    assert attrs.get("copilot.turn_index") == 2
+    # Zero-based: one prior user msg in history → this turn is index 1.
+    assert attrs.get("copilot.turn_index") == 1
     preview = attrs.get("copilot.user_message_preview")
     assert isinstance(preview, str) and preview.startswith("Hello")
 
diff --git a/tests/unit/test_real_browser_manager.py b/tests/unit/test_real_browser_manager.py
index 78dd309ec..a7e09c179 100644
--- a/tests/unit/test_real_browser_manager.py
+++ b/tests/unit/test_real_browser_manager.py
@@ -120,7 +120,6 @@ async def test_non_pbs_workflow_run_inherits_parent_browser() -> None:
     """Non-PBS child runs must still inherit the parent's browser when no browser_session_id."""
     manager = RealBrowserManager()
     parent_state = MagicMock()
-    parent_state.allow_content_blocking_extensions = True
     manager.pages["wfr_parent"] = parent_state
 
     workflow_run = make_workflow_run("wfr_child", parent_workflow_run_id="wfr_parent")
@@ -137,56 +136,6 @@ async def test_non_pbs_workflow_run_inherits_parent_browser() -> None:
     assert manager.pages["wfr_parent"] is parent_state
 
 
-@pytest.mark.asyncio
-async def test_child_workflow_does_not_inherit_parent_browser_when_extension_policy_is_stricter() -> None:
-    manager = RealBrowserManager()
-    parent_state = MagicMock()
-    parent_state.allow_content_blocking_extensions = True
-    manager.pages["wfr_parent"] = parent_state
-
-    child_state = MagicMock()
-    child_state.allow_content_blocking_extensions = False
-    child_state.get_or_create_page = AsyncMock()
-    workflow_run = make_workflow_run("wfr_child", parent_workflow_run_id="wfr_parent")
-
-    with patch.object(manager, "_create_browser_state", new=AsyncMock(return_value=child_state)) as mock_create:
-        result = await manager.get_or_create_for_workflow_run(
-            workflow_run=workflow_run,
-            url=None,
-            browser_session_id=None,
-            allow_content_blocking_extensions=False,
-        )
-
-    assert result is child_state
-    assert result is not parent_state
-    assert manager.pages["wfr_child"] is child_state
-    assert manager.pages["wfr_parent"] is parent_state
-    mock_create.assert_awaited_once()
-    _, kwargs = mock_create.call_args
-    assert kwargs["allow_content_blocking_extensions"] is False
-
-
-@pytest.mark.asyncio
-async def test_child_workflow_can_inherit_parent_browser_when_extension_policy_matches() -> None:
-    manager = RealBrowserManager()
-    parent_state = MagicMock()
-    parent_state.allow_content_blocking_extensions = False
-    manager.pages["wfr_parent"] = parent_state
-
-    workflow_run = make_workflow_run("wfr_child", parent_workflow_run_id="wfr_parent")
-
-    result = await manager.get_or_create_for_workflow_run(
-        workflow_run=workflow_run,
-        url=None,
-        browser_session_id=None,
-        allow_content_blocking_extensions=False,
-    )
-
-    assert result is parent_state
-    assert manager.pages["wfr_child"] is parent_state
-    assert manager.pages["wfr_parent"] is parent_state
-
-
 def make_task(
     task_id: str,
     organization_id: str = "org_test",
diff --git a/tests/unit/test_tags_repository.py b/tests/unit/test_tags_repository.py
new file mode 100644
index 000000000..3f5f62406
--- /dev/null
+++ b/tests/unit/test_tags_repository.py
@@ -0,0 +1,408 @@
+"""TagsRepository unit tests against an in-memory SQLite database."""
+
+from __future__ import annotations
+
+from typing import AsyncGenerator
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
+
+from skyvern.forge.sdk.db.base_alchemy_db import BaseAlchemyDB
+from skyvern.forge.sdk.db.models import (
+    Base,
+    TagKeyModel,
+    WorkflowTagEventModel,
+)
+from skyvern.forge.sdk.db.repositories.tags import (
+    MAX_TAGS_PER_WORKFLOW,
+    TagCountLimitExceeded,
+    TagsRepository,
+)
+from skyvern.forge.sdk.workflow.models.tags import (
+    CallerType,
+    TagEventType,
+    TagSource,
+    TagWriteContext,
+)
+
+ORG_ID = "o_test"
+WPID = "wpid_alpha"
+
+
+async def _enable_partial_unique_indexes(engine: AsyncEngine) -> None:
+    """SQLite ignores postgresql_where, so the model's partial unique indexes
+    become full unique indexes. Drop them and re-create with WHERE clauses so
+    test behavior matches Postgres semantics.
+    """
+    async with engine.begin() as conn:
+        await conn.exec_driver_sql("DROP INDEX IF EXISTS workflow_tag_events_active_set_unique")
+        await conn.exec_driver_sql(
+            "CREATE UNIQUE INDEX workflow_tag_events_active_set_unique "
+            "ON workflow_tag_events (organization_id, workflow_permanent_id, key) "
+            "WHERE superseded_at IS NULL AND event_type = 'set'"
+        )
+        await conn.exec_driver_sql("DROP INDEX IF EXISTS ix_tag_keys_org_key_active")
+        await conn.exec_driver_sql(
+            "CREATE UNIQUE INDEX ix_tag_keys_org_key_active ON tag_keys (organization_id, key) WHERE deleted_at IS NULL"
+        )
+
+
+@pytest_asyncio.fixture
+async def engine() -> AsyncGenerator[AsyncEngine]:
+    eng = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with eng.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    await _enable_partial_unique_indexes(eng)
+    yield eng
+    await eng.dispose()
+
+
+@pytest_asyncio.fixture
+async def repo(engine: AsyncEngine) -> TagsRepository:
+    db = BaseAlchemyDB(engine)
+    return TagsRepository(db.Session, debug_enabled=False)
+
+
+async def _all_events(repo: TagsRepository) -> list[WorkflowTagEventModel]:
+    async with repo.Session() as session:
+        result = await session.execute(
+            select(WorkflowTagEventModel).order_by(WorkflowTagEventModel.set_at, WorkflowTagEventModel.tag_event_id)
+        )
+        return list(result.scalars().all())
+
+
+async def _registered_keys(repo: TagsRepository) -> list[str]:
+    async with repo.Session() as session:
+        result = await session.execute(select(TagKeyModel.key).order_by(TagKeyModel.key))
+        return list(result.scalars().all())
+
+
+async def _event_count(repo: TagsRepository) -> int:
+    async with repo.Session() as session:
+        return (await session.execute(select(func.count(WorkflowTagEventModel.tag_event_id)))).scalar_one()
+
+
+def _ctx(caller_id: str = "user_abc") -> TagWriteContext:
+    return TagWriteContext(
+        caller_id=caller_id,
+        source=TagSource.MANUAL,
+        caller_type=CallerType.USER,
+    )
+
+
+@pytest.mark.asyncio
+async def test_no_op_when_sets_and_deletes_empty(repo: TagsRepository) -> None:
+    changes = await repo.apply_tag_changes(WPID, ORG_ID, sets={}, deletes=set(), context=_ctx())
+    assert changes == []
+
+
+@pytest.mark.asyncio
+async def test_initial_set_creates_event_and_registers_key(repo: TagsRepository) -> None:
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "prod"}, deletes=set(), context=_ctx())
+
+    assert await repo.get_active_tags_for_workflow(WPID, ORG_ID) == {"env": "prod"}
+    assert await _registered_keys(repo) == ["env"]
+
+
+@pytest.mark.asyncio
+async def test_set_existing_key_supersedes_prior_row(repo: TagsRepository) -> None:
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "prod"}, deletes=set(), context=_ctx())
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "stg"}, deletes=set(), context=_ctx())
+
+    assert await repo.get_active_tags_for_workflow(WPID, ORG_ID) == {"env": "stg"}
+
+    rows = await _all_events(repo)
+    assert len(rows) == 2
+    assert rows[0].value == "prod"
+    assert rows[0].superseded_at is not None
+    assert rows[1].value == "stg"
+    assert rows[1].superseded_at is None
+
+
+@pytest.mark.asyncio
+async def test_set_with_same_value_is_no_op(repo: TagsRepository) -> None:
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "prod"}, deletes=set(), context=_ctx())
+    changes = await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "prod"}, deletes=set(), context=_ctx())
+
+    assert changes == []
+    assert await _event_count(repo) == 1
+
+
+@pytest.mark.asyncio
+async def test_delete_writes_delete_event_with_null_value(repo: TagsRepository) -> None:
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "prod"}, deletes=set(), context=_ctx())
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={}, deletes={"env"}, context=_ctx())
+
+    assert await repo.get_active_tags_for_workflow(WPID, ORG_ID) == {}
+
+    rows = await _all_events(repo)
+    assert len(rows) == 2
+    assert rows[0].event_type == TagEventType.SET.value
+    assert rows[0].superseded_at is not None
+    assert rows[1].event_type == TagEventType.DELETE.value
+    assert rows[1].value is None
+
+
+@pytest.mark.asyncio
+async def test_delete_of_absent_key_is_no_op(repo: TagsRepository) -> None:
+    changes = await repo.apply_tag_changes(WPID, ORG_ID, sets={}, deletes={"never_set"}, context=_ctx())
+    assert changes == []
+
+
+@pytest.mark.asyncio
+async def test_sets_win_on_same_key_collision(repo: TagsRepository) -> None:
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "prod"}, deletes=set(), context=_ctx())
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "stg"}, deletes={"env"}, context=_ctx())
+
+    assert await repo.get_active_tags_for_workflow(WPID, ORG_ID) == {"env": "stg"}
+    rows = await _all_events(repo)
+    # No DELETE event got emitted for the colliding key.
+    assert [r.event_type for r in rows] == [TagEventType.SET.value, TagEventType.SET.value]
+
+
+@pytest.mark.asyncio
+async def test_per_workflow_cap_enforced(repo: TagsRepository) -> None:
+    tags = {f"k{i}": f"v{i}" for i in range(MAX_TAGS_PER_WORKFLOW)}
+    await repo.apply_tag_changes(WPID, ORG_ID, sets=tags, deletes=set(), context=_ctx())
+
+    with pytest.raises(TagCountLimitExceeded):
+        await repo.apply_tag_changes(WPID, ORG_ID, sets={"one_too_many": "v"}, deletes=set(), context=_ctx())
+
+
+@pytest.mark.asyncio
+async def test_cap_allows_replace_at_limit(repo: TagsRepository) -> None:
+    tags = {f"k{i}": f"v{i}" for i in range(MAX_TAGS_PER_WORKFLOW)}
+    await repo.apply_tag_changes(WPID, ORG_ID, sets=tags, deletes=set(), context=_ctx())
+
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"k0": "new"}, deletes=set(), context=_ctx())
+
+    current = await repo.get_active_tags_for_workflow(WPID, ORG_ID)
+    assert current["k0"] == "new"
+    assert len(current) == MAX_TAGS_PER_WORKFLOW
+
+
+@pytest.mark.asyncio
+async def test_history_returns_superseded_and_delete_events(
+    repo: TagsRepository,
+) -> None:
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "prod"}, deletes=set(), context=_ctx())
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"env": "stg"}, deletes=set(), context=_ctx())
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={}, deletes={"env"}, context=_ctx())
+
+    history = await repo.get_tag_event_history(WPID, ORG_ID)
+    assert len(history) == 3
+    types = [row.event_type for row in history]
+    assert types[0] == TagEventType.DELETE.value
+
+
+@pytest.mark.asyncio
+async def test_attribution_carries_caller_and_source(repo: TagsRepository) -> None:
+    ctx = TagWriteContext(
+        caller_id="user_42",
+        source=TagSource.BULK_APPLY,
+        caller_type=CallerType.API_KEY,
+    )
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"team": "core"}, deletes=set(), context=ctx)
+
+    history = await repo.get_tag_event_history(WPID, ORG_ID)
+    assert history[0].set_by == "user_42"
+    assert history[0].source == TagSource.BULK_APPLY.value
+    assert history[0].caller_type == CallerType.API_KEY.value
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "caller_type",
+    [CallerType.USER, CallerType.API_KEY, CallerType.SYSTEM],
+    ids=["user", "api_key", "system"],
+)
+async def test_caller_type_persists_for_every_value(repo: TagsRepository, caller_type: CallerType) -> None:
+    """The tag audit row records whether the writer was a person, an API key,
+    or a system actor. Lets downstream queries ("show me API-driven tag
+    activity this week") run as a column predicate.
+    """
+    ctx = TagWriteContext(
+        caller_id="actor",
+        source=TagSource.MANUAL,
+        caller_type=caller_type,
+    )
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"team": "core"}, deletes=set(), context=ctx)
+    history = await repo.get_tag_event_history(WPID, ORG_ID)
+    assert history[0].caller_type == caller_type.value
+
+
+@pytest.mark.asyncio
+async def test_caller_type_persists_as_null_when_unset(repo: TagsRepository) -> None:
+    """Backfill scripts that lack a request context can omit caller_type;
+    the column is nullable so the audit log remains writable.
+    """
+    ctx = TagWriteContext(caller_id="backfill_script", source=TagSource.BACKFILL)
+    await repo.apply_tag_changes(WPID, ORG_ID, sets={"team": "core"}, deletes=set(), context=ctx)
+    history = await repo.get_tag_event_history(WPID, ORG_ID)
+    assert history[0].caller_type is None
+
+
+@pytest.mark.asyncio
+async def test_isolation_between_workflows_in_same_org(repo: TagsRepository) -> None:
+    await repo.apply_tag_changes("wpid_a", ORG_ID, sets={"env": "prod"}, deletes=set(), context=_ctx())
+    await repo.apply_tag_changes("wpid_b", ORG_ID, sets={"env": "stg"}, deletes=set(), context=_ctx())
+
+    assert await repo.get_active_tags_for_workflow("wpid_a", ORG_ID) == {"env": "prod"}
+    assert await repo.get_active_tags_for_workflow("wpid_b", ORG_ID) == {"env": "stg"}
+
+
+def test_tag_count_limit_is_registered_as_passthrough_exception() -> None:
+    """Cap breaches must log as BusinessLogicError (WARN), not
+    UnexpectedError (ERROR). Mirrors ScheduleLimitExceededError.
+    """
+    from skyvern.forge.sdk.db import _error_handling
+
+    assert TagCountLimitExceeded in _error_handling._PASSTHROUGH_EXCEPTIONS
+
+
+# ---------------------------------------------------------------------------
+# Realistic tag examples gallery
+#
+# Doubles as documentation for what tags look like in practice. Mirrors the
+# "Tag examples gallery" table in the Notion PRD's Design section. If you
+# change a row here, change the PRD too.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "tags,description",
+    [
+        pytest.param(
+            {"team": "payments"},
+            "single-dimension ownership",
+            id="team",
+        ),
+        pytest.param(
+            {"environment": "prod"},
+            "lifecycle stage",
+            id="environment",
+        ),
+        pytest.param(
+            {"customer": "acme-corp"},
+            "external customer this workflow serves",
+            id="customer",
+        ),
+        pytest.param(
+            {"team": "payments", "environment": "prod"},
+            "two orthogonal dimensions on one workflow",
+            id="team-and-environment",
+        ),
+        pytest.param(
+            {
+                "team": "payments",
+                "environment": "prod",
+                "customer": "acme-corp",
+                "cost_center": "R&D",
+                "priority": "p0",
+            },
+            "five-axis tagging — what a fully-categorized workflow looks like",
+            id="five-axis",
+        ),
+        pytest.param(
+            {"use_case": "form-fill", "team": "platform"},
+            "workflow archetype + ownership",
+            id="archetype-plus-team",
+        ),
+    ],
+)
+async def test_realistic_tag_examples(repo: TagsRepository, tags: dict[str, str], description: str) -> None:
+    """Each row is a realistic tagging pattern a customer might use.
+
+    Reading these tests is the fastest way to learn what tags are for and
+    what shapes the system supports.
+    """
+    await repo.apply_tag_changes(WPID, ORG_ID, sets=tags, deletes=set(), context=_ctx())
+    assert await repo.get_active_tags_for_workflow(WPID, ORG_ID) == tags
+
+
+@pytest.mark.asyncio
+async def test_dotted_key_hierarchy_works_without_schema_change(repo: TagsRepository) -> None:
+    """Dotted keys (`team.payments.billing`) are the canonical lightweight
+    hierarchy convention. The key regex already allows `.`, so deep nesting
+    is expressible today without any schema work.
+
+    Parent and child keys live in the same flat namespace; a workflow can
+    carry both at once.
+    """
+    await repo.apply_tag_changes(
+        WPID,
+        ORG_ID,
+        sets={
+            "team": "payments",  # parent dimension
+            "team.subteam": "billing",  # nested child as a separate key
+            "team.subteam.squad": "checkout",  # arbitrary depth
+        },
+        deletes=set(),
+        context=_ctx(),
+    )
+    active = await repo.get_active_tags_for_workflow(WPID, ORG_ID)
+    assert active == {
+        "team": "payments",
+        "team.subteam": "billing",
+        "team.subteam.squad": "checkout",
+    }
+
+
+@pytest.mark.asyncio
+async def test_same_key_value_across_many_workflows_n_to_m(repo: TagsRepository) -> None:
+    """The data model is N:M: one (key, value) like `team:payments` can
+    apply across many workflows, AND a single workflow can carry many keys.
+    There is no `tags` join table — the (org, key, value) triple in the
+    event log IS the tag identity.
+    """
+    # Three workflows in the same org, all tagged team:payments
+    for wpid in ("wpid_invoice_processor", "wpid_refund_handler", "wpid_billing_sync"):
+        await repo.apply_tag_changes(wpid, ORG_ID, sets={"team": "payments"}, deletes=set(), context=_ctx())
+
+    # And one of them has additional tags — orthogonal dimensions stack freely
+    await repo.apply_tag_changes(
+        "wpid_invoice_processor",
+        ORG_ID,
+        sets={"environment": "prod", "priority": "p0"},
+        deletes=set(),
+        context=_ctx(),
+    )
+
+    assert await repo.get_active_tags_for_workflow("wpid_invoice_processor", ORG_ID) == {
+        "team": "payments",
+        "environment": "prod",
+        "priority": "p0",
+    }
+    assert await repo.get_active_tags_for_workflow("wpid_refund_handler", ORG_ID) == {
+        "team": "payments",
+    }
+    assert await repo.get_active_tags_for_workflow("wpid_billing_sync", ORG_ID) == {
+        "team": "payments",
+    }
+
+
+@pytest.mark.asyncio
+async def test_value_can_contain_colons(repo: TagsRepository) -> None:
+    """Colons inside values are data, not delimiters. URLs, timestamps,
+    and external IDs commonly contain colons — they're stored verbatim.
+
+    (The comma-as-pair-separator rule lives in the URL parser, not the
+    repository; the repository accepts any string value < 256 chars.)
+    """
+    await repo.apply_tag_changes(
+        WPID,
+        ORG_ID,
+        sets={
+            "jira_ticket": "PROJ-1234:bugfix",
+            "captured_at": "2026-05-25T18:30:00Z",
+        },
+        deletes=set(),
+        context=_ctx(),
+    )
+    assert await repo.get_active_tags_for_workflow(WPID, ORG_ID) == {
+        "jira_ticket": "PROJ-1234:bugfix",
+        "captured_at": "2026-05-25T18:30:00Z",
+    }