From a79b727a5d4314b8226b786fdf36fbee01ef7b55 Mon Sep 17 00:00:00 2001
From: Vojta Bartos <vojtech.b@posthog.com>
Date: Tue, 14 Jul 2026 19:04:03 +0200
Subject: [PATCH 1/5] feat(tasks): carry each message's actor and tag replies
 with the turn's speaker
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Every consumer of "who sent this" read mutable shared state that the
next message overwrites — visibly, agent replies tagged whoever spoke
last instead of the person being answered.

- The follow-up signal gains actor_user_id and a context dict (its
  extension point for per-message fields, carrying the sender's Slack
  id). Additive with defaults: old histories replay unchanged.
- Delivery pins credential resolution to the message's sender via a
  local state overlay, stamps the run-state actor at turn start, and on
  the turn's ack records the completed turn's actor.
- The Slack relay tags the completed turn's actor, so a reply tags the
  person whose turn it answers even when the next message's delivery has
  already restamped the live actor. mapping.latest_actor stays as the
  pre-rollout fallback, marked for removal.
- One slack_actor_state_updates builder in run_actor.py serves all
  writers of the load-bearing state keys.
---
 .../ai/slack_app/activities/task_creation.py  | 29 ++----
 products/slack_app/backend/models.py          |  3 +
 .../backend/tests/test_followup_forwarding.py |  4 +
 products/tasks/backend/facade/api.py          | 19 +++-
 .../tasks/backend/logic/services/run_actor.py | 13 +++
 .../tasks/backend/presentation/views/api.py   | 21 ++---
 products/tasks/backend/temporal/client.py     |  8 +-
 .../activities/forward_pending_message.py     |  7 ++
 .../activities/send_followup_to_sandbox.py    | 91 ++++++++++++++-----
 .../send_permission_response_to_sandbox.py    |  5 +-
 .../tests/test_send_followup_to_sandbox.py    | 90 ++++++++++++++----
 .../backend/temporal/process_task/workflow.py | 25 ++++-
 .../temporal/slack_relay/activities.py        |  5 +-
 .../backend/temporal/tests/test_client.py     |  4 +-
 products/tasks/backend/tests/test_api.py      |  8 +-
 15 files changed, 239 insertions(+), 93 deletions(-)

diff --git a/posthog/temporal/ai/slack_app/activities/task_creation.py b/posthog/temporal/ai/slack_app/activities/task_creation.py
index e65ee064457e..6e31746e80dd 100644
--- a/posthog/temporal/ai/slack_app/activities/task_creation.py
+++ b/posthog/temporal/ai/slack_app/activities/task_creation.py
@@ -82,10 +82,11 @@ def _canvas_file_delivery_available(integration: Integration) -> bool:
 
 
 def _slack_actor_state_updates(*, user_id: int, slack_user_id: str) -> dict[str, Any]:
-    return {
-        "slack_actor_user_id": user_id,
-        "slack_actor_slack_user_id": slack_user_id,
-    }
+    from products.tasks.backend.logic.services.run_actor import (  # noqa: PLC0415 — keep tasks deps off the slack_app import path
+        slack_actor_state_updates,
+    )
+
+    return slack_actor_state_updates(user_id=user_id, slack_user_id=slack_user_id)
 
 
 def _strip_context_tag(text: str) -> str:
@@ -847,26 +848,12 @@ def forward_posthog_code_followup_activity(
     ):
         return True
 
-    # Record the live actor so async reply paths tag them instead of the
-    # thread's original mentioner. Concurrent follow-ups can race here; see PR.
+    # Reply-tag fallback for turns with no per-turn actor (boot prompt,
+    # pre-rollout runs); the actor stamped at delivery normally wins.
     if slack_user_id != mapping.latest_actor_slack_user_id:
         mapping.latest_actor_slack_user_id = slack_user_id
         mapping.save(update_fields=["latest_actor_slack_user_id", "updated_at"])
 
-    if actor_user and actor_user.id:
-        try:
-            tasks_facade.update_task_run_state(
-                task_run.id,
-                updates=_slack_actor_state_updates(user_id=actor_user.id, slack_user_id=slack_user_id),
-            )
-        except Exception:
-            logger.exception(
-                "posthog_code_followup_actor_state_update_failed",
-                channel=channel,
-                thread_ts=thread_ts,
-                actor_user_id=actor_user.id,
-            )
-
     if task_run.is_terminal:
         return _resume_task_with_new_run(
             mapping,
@@ -974,7 +961,9 @@ def forward_posthog_code_followup_activity(
         task_run.team_id,
         content=user_text,
         artifact_ids=_uploaded_attachment_ids(uploaded_attachments),
+        actor_user_id=actor_user.id if actor_user and actor_user.id else None,
         message_id=_slack_followup_message_id(channel, user_message_ts, thread_ts),
+        actor_slack_user_id=slack_user_id,
     )
     if signal_result is not True:
         logger.warning(
diff --git a/products/slack_app/backend/models.py b/products/slack_app/backend/models.py
index 83584e028888..a627abe762ce 100644
--- a/products/slack_app/backend/models.py
+++ b/products/slack_app/backend/models.py
@@ -27,6 +27,9 @@ class SlackThreadTaskMapping(UUIDModel):
         related_name="slack_thread_mappings",
     )
     mentioning_slack_user_id = models.CharField(max_length=64)
+    # Reply-tag fallback for runs started before per-turn actor capture
+    # (tasks slack_relay); drop the column and its stamp in task_creation
+    # once those runs drain.
     latest_actor_slack_user_id = models.CharField(max_length=64, null=True, blank=True)
     # Slack `ts` of the most recent message we've already shown to the agent (either
     # in the original `<slack_thread_context>` block at task creation, or in a follow-up
diff --git a/products/slack_app/backend/tests/test_followup_forwarding.py b/products/slack_app/backend/tests/test_followup_forwarding.py
index af7cd3470864..40b864565f4b 100644
--- a/products/slack_app/backend/tests/test_followup_forwarding.py
+++ b/products/slack_app/backend/tests/test_followup_forwarding.py
@@ -939,6 +939,7 @@ def test_cross_user_followup_authorized_prefixes_actor_name(self, mock_slack_cls
         assert mock_signal.call_args.args == (self.task_run.id, self.task.id, self.team.id)
         signal_kwargs = mock_signal.call_args.kwargs
         assert signal_kwargs["content"] == "Bob: please retry the build"
+        assert signal_kwargs["actor_user_id"] == bob.id
         assert signal_kwargs["message_id"] is not None
         # No "Only the person who started" denial; the message went through.
         post_calls = [
@@ -963,6 +964,7 @@ def test_cross_user_followup_falls_back_to_email_when_no_full_name(self, mock_sl
         mock_signal.assert_called_once()
         signal_kwargs = mock_signal.call_args.kwargs
         assert signal_kwargs["content"] == "bob@test.com: ping"
+        assert signal_kwargs["actor_user_id"] == bob.id
         assert signal_kwargs["message_id"] is not None
 
     @patch("products.slack_app.backend.api.resolve_slack_user", return_value=None)
@@ -1048,6 +1050,7 @@ def test_successful_forwarding(self, mock_slack_cls, mock_signal):
         signal_kwargs = mock_signal.call_args.kwargs
         assert signal_kwargs["content"] == "do something"
         assert signal_kwargs["artifact_ids"] == []
+        assert signal_kwargs["actor_user_id"] == self.user.id
         assert signal_kwargs["message_id"] is not None
         # The message is queued on the workflow, so the :eyes: reaction stays up — it
         # is not swapped to :hedgehog: until the task genuinely completes.
@@ -1106,6 +1109,7 @@ def test_attachment_only_followup_uploads_and_forwards_to_sandbox(self) -> None:
         content = first_call.kwargs["content"]
         assert content.startswith("Attached Slack file(s).")
         assert "Slack attachment(s) available to the agent as task files: only-log.txt." in content
+        assert first_call.kwargs["actor_user_id"] == self.user.id
         assert mock_write.call_count == 2
         assert mock_write.call_args_list[0].args[0] == mock_write.call_args_list[1].args[0]
         self.task_run.refresh_from_db()
diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py
index f848b799b533..3c5f747781ac 100644
--- a/products/tasks/backend/facade/api.py
+++ b/products/tasks/backend/facade/api.py
@@ -2551,7 +2551,9 @@ def signal_task_run_user_message(
     *,
     content: str | None,
     artifact_ids: list[str],
+    actor_user_id: int | None = None,
     message_id: str | None = None,
+    actor_slack_user_id: str | None = None,
 ) -> bool | None:
     """Queue a user_message follow-up signal on the run's workflow.
 
@@ -2570,7 +2572,8 @@ def signal_task_run_user_message(
     if run is None:
         return None
     try:
-        signal_task_followup_message(run.workflow_id, content, artifact_ids, message_id)
+        context = {"actor_slack_user_id": actor_slack_user_id} if actor_slack_user_id else None
+        signal_task_followup_message(run.workflow_id, content, artifact_ids, message_id, actor_user_id, context)
     except RPCError as e:
         if e.status == RPCStatusCode.NOT_FOUND:
             logger.warning("Follow-up signal target workflow gone for task run %s", run.id)
@@ -2731,8 +2734,20 @@ def relay_task_run_message(
             logger.exception("task_run_relay_text_signal_failed", extra={"run_id": str(run.id)})
         return "skipped", None
 
+    # Tag the actor of the last *completed* turn (stamped on the delivery
+    # ack); the live actor is a fallback for pre-rollout runs, and may point
+    # at the next speaker when a turn outlives the delivery ack.
+    relay_state = run.state or {}
+    mention_slack_user_id = relay_state.get("slack_last_turn_slack_user_id") or relay_state.get(
+        "slack_actor_slack_user_id"
+    )
     try:
-        relay_id = execute_posthog_code_agent_relay_workflow(run_id=str(run.id), text=trimmed, delete_progress=True)
+        relay_id = execute_posthog_code_agent_relay_workflow(
+            run_id=str(run.id),
+            text=trimmed,
+            delete_progress=True,
+            mention_slack_user_id=mention_slack_user_id if isinstance(mention_slack_user_id, str) else None,
+        )
     except Exception:
         logger.exception("task_run_relay_message_enqueue_failed", extra={"run_id": str(run.id)})
         return "failed", None
diff --git a/products/tasks/backend/logic/services/run_actor.py b/products/tasks/backend/logic/services/run_actor.py
index df4eff96b940..937b476ad851 100644
--- a/products/tasks/backend/logic/services/run_actor.py
+++ b/products/tasks/backend/logic/services/run_actor.py
@@ -92,3 +92,16 @@ def get_task_run_credential_user(task: Task, state: dict[str, Any] | None = None
 
 def get_actor_distinct_id(actor: User) -> str:
     return actor.distinct_id or f"user_{actor.id}"
+
+
+def slack_actor_state_updates(*, user_id: int, slack_user_id: str | None = None) -> dict[str, Any]:
+    """Run-state updates recording the Slack user currently steering a run.
+
+    The keys are load-bearing: credential resolution reads
+    ``slack_actor_user_id`` and reply tagging reads
+    ``slack_actor_slack_user_id`` — every writer must build them here.
+    """
+    updates: dict[str, Any] = {"slack_actor_user_id": user_id}
+    if slack_user_id:
+        updates["slack_actor_slack_user_id"] = slack_user_id
+    return updates
diff --git a/products/tasks/backend/presentation/views/api.py b/products/tasks/backend/presentation/views/api.py
index b7d39b40daab..e8d83c0e2663 100644
--- a/products/tasks/backend/presentation/views/api.py
+++ b/products/tasks/backend/presentation/views/api.py
@@ -39,7 +39,7 @@
     cancellation as tasks_cancellation,
     contracts as tasks_contracts,
 )
-from products.tasks.backend.facade.access import cloud_usage_limit_response, code_access_required_response
+from products.tasks.backend.facade.access import cloud_usage_limit_response
 from products.tasks.backend.facade.metrics import (
     StreamConnectionOutcome,
     observe_stream_connection_closed,
@@ -639,9 +639,6 @@ def warm(self, request, **kwargs):
         if not self._warm_enabled():
             return Response(status=status.HTTP_200_OK)
 
-        if access_response := code_access_required_response(request.user):
-            return access_response
-
         user_id = self._user_id()
         if user_id is None:
             return Response(status=status.HTTP_200_OK)
@@ -758,8 +755,6 @@ def retrieve(self, request, pk=None, **kwargs):
 
     @extend_schema(request=TaskAutomationWriteSerializer, responses={201: TaskAutomationSerializer})
     def create(self, request, **kwargs):
-        if access_response := code_access_required_response(request.user):
-            return access_response
         serializer = self._write_serializer(request.data)
         automation = tasks_facade.create_task_automation(
             self.team_id, getattr(request.user, "id", None), **self._facade_kwargs(serializer.validated_data)
@@ -769,9 +764,6 @@ def create(self, request, **kwargs):
     @extend_schema(request=TaskAutomationWriteSerializer, responses={200: TaskAutomationSerializer})
     def partial_update(self, request, pk=None, **kwargs):
         serializer = self._write_serializer(request.data, partial=True)
-        if serializer.validated_data.get("enabled") is True:
-            if access_response := code_access_required_response(request.user):
-                return access_response
         automation = tasks_facade.update_task_automation(
             pk, self.team_id, getattr(request.user, "id", None), **self._facade_kwargs(serializer.validated_data)
         )
@@ -788,8 +780,6 @@ def destroy(self, request, pk=None, **kwargs):
     @extend_schema(request=None, responses={200: TaskAutomationSerializer})
     @action(detail=True, methods=["post"], url_path="run", required_scopes=["task:write"])
     def run(self, request, pk=None, **kwargs):
-        if access_response := code_access_required_response(request.user):
-            return access_response
         automation = tasks_facade.run_task_automation_now(pk, self.team_id, getattr(request.user, "id", None))
         if automation is None:
             raise NotFound()
@@ -1513,8 +1503,6 @@ def command(self, request, pk=None, **kwargs):
         params = request.validated_data.get("params")
 
         if method == "user_message":
-            if access_response := code_access_required_response(request.user):
-                return access_response
             command_params = dict(params or {})
             artifact_ids = command_params.pop("artifact_ids", [])
             if artifact_ids:
@@ -1534,7 +1522,12 @@ def command(self, request, pk=None, **kwargs):
 
             try:
                 signal_result = tasks_facade.signal_task_run_user_message(
-                    pk, task_id, self.team_id, content=command_params.get("content"), artifact_ids=artifact_ids
+                    pk,
+                    task_id,
+                    self.team_id,
+                    content=command_params.get("content"),
+                    artifact_ids=artifact_ids,
+                    actor_user_id=request.user.id,
                 )
             except Exception:
                 # A synchronous web request can't retry the way the Temporal
diff --git a/products/tasks/backend/temporal/client.py b/products/tasks/backend/temporal/client.py
index 4b583069701a..5563f22b0e2f 100644
--- a/products/tasks/backend/temporal/client.py
+++ b/products/tasks/backend/temporal/client.py
@@ -470,10 +470,16 @@ def signal_task_followup_message(
     message: str | None,
     artifact_ids: list[str],
     message_id: str | None = None,
+    actor_user_id: int | None = None,
+    context: dict[str, Any] | None = None,
 ) -> None:
+    """New per-message fields go in ``context`` — the positional signal args
+    are frozen for worker deploy compat."""
     client = sync_connect()
     handle = client.get_workflow_handle(workflow_id)
-    asyncio.run(handle.signal("send_followup_message", args=[message, artifact_ids, message_id]))
+    asyncio.run(
+        handle.signal("send_followup_message", args=[message, artifact_ids, message_id, actor_user_id, context])
+    )
 
 
 def signal_agent_text_delta(workflow_id: str, text: str) -> None:
diff --git a/products/tasks/backend/temporal/process_task/activities/forward_pending_message.py b/products/tasks/backend/temporal/process_task/activities/forward_pending_message.py
index fddebe288710..e261689b6fde 100644
--- a/products/tasks/backend/temporal/process_task/activities/forward_pending_message.py
+++ b/products/tasks/backend/temporal/process_task/activities/forward_pending_message.py
@@ -146,8 +146,15 @@ def activity_failure() -> tuple[str, str] | None:
             else:
                 _enqueue_pending_delivery_failure_relay(task_run, pending_message_ts, result.error)
 
+        boot_actor_slack_user_id = state.get("slack_actor_slack_user_id")
+        updates = (
+            {"slack_last_turn_slack_user_id": boot_actor_slack_user_id}
+            if result.success and isinstance(boot_actor_slack_user_id, str) and boot_actor_slack_user_id
+            else None
+        )
         TaskRun.update_state_atomic(
             run_id,
+            updates=updates,
             remove_keys=["pending_user_message", "pending_user_artifact_ids", "pending_user_message_ts"],
         )
 
diff --git a/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py b/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
index afe1a4f99b03..80584925da2e 100644
--- a/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
+++ b/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
@@ -20,6 +20,7 @@
     send_user_message,
 )
 from products.tasks.backend.logic.services.connection_token import create_sandbox_connection_token
+from products.tasks.backend.logic.services.run_actor import slack_actor_state_updates
 from products.tasks.backend.logic.services.staged_artifacts import get_task_run_artifacts_by_id
 from products.tasks.backend.logic.stream.redis_stream import get_task_run_stream_key
 from products.tasks.backend.models import TaskRun
@@ -55,9 +56,14 @@ class SendFollowupToSandboxInput:
     message: str | None = None
     posthog_mcp_scopes: PosthogMcpScopes = "read_only"
     artifact_ids: list[str] | None = None
-    # Workflow-generated idempotency key. Stable across activity retries, so
-    # the agent-server can drop a redelivery of a message it already accepted.
+    # Idempotency key, stable across retries and redeliveries; the
+    # agent-server drops a duplicate it already accepted.
     message_id: str | None = None
+    # Sender of this message; None (older senders, pre-rollout histories)
+    # falls back to the run-state actor.
+    actor_user_id: int | None = None
+    # Signal context, passed through from PendingFollowup.
+    context: dict[str, Any] | None = None
 
 
 @activity.defn
@@ -118,9 +124,32 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
         # background-mode runs hang until the inactivity timeout because
         raise ApplicationError(f"send_followup failed: {error_msg}", non_retryable=True)
 
+    # Resolve credentials against this message's sender, not the run-state
+    # actor a concurrent follow-up may have overwritten since queueing. Local
+    # overlay; the resolver still enforces team access (see run_actor.py).
+    state = task_run.state
+    if input.actor_user_id is not None:
+        state = {**(state or {}), "slack_actor_user_id": input.actor_user_id}
+        if is_slack_interaction_state(state):
+            # Deliveries are serialized by the workflow, so stamping here
+            # moves the durable actor at turn boundaries — between-turn
+            # consumers (reply tagging, permission broker) see the executing
+            # turn's actor. Skipped when already current.
+            actor_slack_user_id = (input.context or {}).get("actor_slack_user_id")
+            updates = slack_actor_state_updates(
+                user_id=input.actor_user_id,
+                slack_user_id=actor_slack_user_id if isinstance(actor_slack_user_id, str) else None,
+            )
+            current = task_run.state or {}
+            if any(current.get(key) != value for key, value in updates.items()):
+                try:
+                    TaskRun.update_state_atomic(task_run.id, updates=updates)
+                except Exception:
+                    logger.warning("send_followup_actor_stamp_failed", run_id=input.run_id, exc_info=True)
+
     auth_token = None
-    actor_user = get_task_run_credential_user(task_run.task, task_run.state)
-    if is_slack_interaction_state(task_run.state) and actor_user is None:
+    actor_user = get_task_run_credential_user(task_run.task, state)
+    if is_slack_interaction_state(state) and actor_user is None:
         error_msg = "Slack actor unavailable for this run"
         _write_error_and_complete(input.run_id, error_msg, run_uses_dedicated_stream(task_run.state))
         raise RuntimeError(f"send_followup failed: {error_msg}")
@@ -132,7 +161,7 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
     # Push a fresh MCP config before the turn so the agent-server rebinds its
     # ACP session to a non-stale OAuth token. Non-fatal: if refresh fails we
     # still deliver the follow-up with the existing (possibly stale) creds.
-    _refresh_sandbox_mcp(task_run, input.posthog_mcp_scopes, auth_token)
+    _refresh_sandbox_mcp(task_run, input.posthog_mcp_scopes, auth_token, actor_user=actor_user, state=state)
     artifacts = None
     artifact_ids = input.artifact_ids or []
     if artifact_ids:
@@ -165,6 +194,17 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
                 attempt=_current_attempt(),
             )
             return
+        turn_actor_slack_user_id = (input.context or {}).get("actor_slack_user_id")
+        if isinstance(turn_actor_slack_user_id, str) and turn_actor_slack_user_id:
+            # The ack means this message's turn finished; record its actor for
+            # reply tagging — the relay may fire after the next delivery has
+            # already restamped the live actor.
+            try:
+                TaskRun.update_state_atomic(
+                    input.run_id, updates={"slack_last_turn_slack_user_id": turn_actor_slack_user_id}
+                )
+            except Exception:
+                logger.warning("send_followup_turn_actor_stamp_failed", run_id=input.run_id, exc_info=True)
         _write_turn_complete(input.run_id, _get_stop_reason(result.data), run_uses_dedicated_stream(task_run.state))
         logger.info("send_followup_delivered", run_id=input.run_id)
     elif result.turn_in_flight:
@@ -221,24 +261,28 @@ def _refresh_sandbox_mcp(
     task_run: TaskRun,
     scopes: PosthogMcpScopes,
     auth_token: str | None,
+    *,
+    actor_user: Any,
+    state: dict[str, Any] | None,
 ) -> None:
-    """Mint a fresh OAuth token and push updated MCP configs to the sandbox.
-
-    Best-effort: retries once on failure, then logs and returns. Never raises
-    — a failed refresh should not block an otherwise-valid follow-up.
+    """Mint a fresh OAuth token for the actor and push updated MCP configs to
+    the sandbox.
 
-    Skipped entirely if a token was issued for this run within the last
-    MCP_TOKEN_REFRESH_INTERVAL_SECONDS — the in-sandbox token is still fresh.
+    Best-effort: retries once on failure, then logs and returns — a failed
+    refresh must not block an otherwise-valid follow-up. Skipped when a token
+    was already issued for this run within MCP_TOKEN_REFRESH_INTERVAL_SECONDS.
     """
     run_id = str(task_run.id)
     if not should_refresh_mcp_token(run_id):
         logger.info("refresh_mcp_skipped_within_interval", run_id=run_id)
         return
+    if actor_user is None:
+        # Without a credential user the mint is guaranteed to fail; skip
+        # quietly rather than warn on every message.
+        return
 
-    task = task_run.task
     try:
-        actor_user = get_task_run_credential_user(task, task_run.state)
-        access_token = create_oauth_access_token_for_run(task, task_run.state, scopes=scopes)
+        access_token = create_oauth_access_token_for_run(task_run.task, state, scopes=scopes)
     except Exception as e:
         logger.warning("refresh_mcp_token_mint_failed", run_id=run_id, error=str(e))
         return
@@ -247,18 +291,17 @@ def _refresh_sandbox_mcp(
         token=access_token,
         project_id=task_run.team_id,
         scopes=scopes,
-        interaction_origin=(task_run.state or {}).get("interaction_origin"),
+        interaction_origin=(state or {}).get("interaction_origin"),
         task_id=str(task_run.task_id),
     )
-    if actor_user and actor_user.id:
-        user_mcp_configs = get_user_mcp_server_configs(
-            token=access_token,
-            team_id=task_run.team_id,
-            user_id=actor_user.id,
-            interaction_origin=(task_run.state or {}).get("interaction_origin"),
-        )
-        if user_mcp_configs:
-            mcp_configs = mcp_configs + user_mcp_configs
+    user_mcp_configs = get_user_mcp_server_configs(
+        token=access_token,
+        team_id=task_run.team_id,
+        user_id=actor_user.id,
+        interaction_origin=(state or {}).get("interaction_origin"),
+    )
+    if user_mcp_configs:
+        mcp_configs = mcp_configs + user_mcp_configs
 
     # refresh_session replaces the session's server list wholesale, so the
     # run's imported servers must ride along or they vanish mid-run.
diff --git a/products/tasks/backend/temporal/process_task/activities/send_permission_response_to_sandbox.py b/products/tasks/backend/temporal/process_task/activities/send_permission_response_to_sandbox.py
index 8665dcba0480..3571eb62d668 100644
--- a/products/tasks/backend/temporal/process_task/activities/send_permission_response_to_sandbox.py
+++ b/products/tasks/backend/temporal/process_task/activities/send_permission_response_to_sandbox.py
@@ -9,6 +9,7 @@
 
 from products.tasks.backend.logic.services.agent_command import send_agent_command, send_user_message
 from products.tasks.backend.logic.services.connection_token import create_sandbox_connection_token
+from products.tasks.backend.logic.services.run_actor import slack_actor_state_updates
 from products.tasks.backend.models import TaskRun
 from products.tasks.backend.temporal.process_task.utils import get_actor_distinct_id
 
@@ -134,12 +135,10 @@ def send_permission_response_to_sandbox(input: SendPermissionResponseToSandboxIn
         raise RuntimeError(result.error or "Failed to deliver permission response to sandbox")
 
     updates: dict[str, object] = {
-        "slack_actor_user_id": actor.id,
+        **slack_actor_state_updates(user_id=actor.id, slack_user_id=input.actor_slack_user_id),
         "slack_permission_response_last_request_id": input.request_id,
         "slack_permission_response_last_option_id": input.option_id,
     }
-    if input.actor_slack_user_id:
-        updates["slack_actor_slack_user_id"] = input.actor_slack_user_id
     if input.broker_reason:
         updates["slack_permission_broker_last_reason"] = input.broker_reason
     if input.is_denial:
diff --git a/products/tasks/backend/temporal/process_task/tests/test_send_followup_to_sandbox.py b/products/tasks/backend/temporal/process_task/tests/test_send_followup_to_sandbox.py
index 251476b0229e..6092b863cc92 100644
--- a/products/tasks/backend/temporal/process_task/tests/test_send_followup_to_sandbox.py
+++ b/products/tasks/backend/temporal/process_task/tests/test_send_followup_to_sandbox.py
@@ -60,6 +60,11 @@ def _make_task_run_mock(team_id: int = 7, created_by_id: int | None = 42, state:
     return task_run
 
 
+def _refresh(task_run, actor_id: int | None = 42, scopes="read_only", auth_token=None) -> None:
+    actor = MagicMock(id=actor_id) if actor_id is not None else None
+    _refresh_sandbox_mcp(task_run, scopes, auth_token, actor_user=actor, state=task_run.state)
+
+
 class TestRefreshSandboxMcp:
     @patch("products.tasks.backend.temporal.process_task.activities.send_followup_to_sandbox.send_refresh_session")
     @patch(
@@ -78,7 +83,7 @@ def test_success_path_single_call(self, mock_oauth, mock_ph_configs, mock_user_c
         mock_send_refresh.return_value = CommandResult(success=True, status_code=200)
 
         task_run = _make_task_run_mock()
-        _refresh_sandbox_mcp(task_run, "read_only", auth_token="jwt")
+        _refresh(task_run, auth_token="jwt")
 
         mock_oauth.assert_called_once_with(task_run.task, task_run.state, scopes="read_only")
         mock_ph_configs.assert_called_once_with(
@@ -158,7 +163,7 @@ def test_retries_once_on_first_failure(
             CommandResult(success=True, status_code=200),
         ]
 
-        _refresh_sandbox_mcp(_make_task_run_mock(), "read_only", auth_token=None)
+        _refresh(_make_task_run_mock())
 
         assert mock_send_refresh.call_count == 2
         mock_sleep.assert_called_once_with(REFRESH_RETRY_DELAY_SECONDS)
@@ -183,7 +188,7 @@ def test_two_failures_are_non_fatal(
         mock_send_refresh.return_value = CommandResult(success=False, status_code=502, error="down")
 
         # Must not raise.
-        _refresh_sandbox_mcp(_make_task_run_mock(), "read_only", auth_token=None)
+        _refresh(_make_task_run_mock())
 
         assert mock_send_refresh.call_count == 2
 
@@ -202,7 +207,7 @@ def test_token_mint_failure_is_non_fatal_and_skips_send(
     ):
         mock_oauth.side_effect = RuntimeError("oauth service down")
 
-        _refresh_sandbox_mcp(_make_task_run_mock(), "read_only", auth_token=None)
+        _refresh(_make_task_run_mock())
 
         mock_ph_configs.assert_not_called()
         mock_user_configs.assert_not_called()
@@ -225,7 +230,7 @@ def test_skips_send_when_no_mcp_configs_resolved(
         mock_ph_configs.return_value = []
         mock_user_configs.return_value = []
 
-        _refresh_sandbox_mcp(_make_task_run_mock(), "read_only", auth_token=None)
+        _refresh(_make_task_run_mock())
 
         mock_send_refresh.assert_not_called()
 
@@ -239,17 +244,12 @@ def test_skips_send_when_no_mcp_configs_resolved(
     @patch(
         "products.tasks.backend.temporal.process_task.activities.send_followup_to_sandbox.create_oauth_access_token_for_run"
     )
-    def test_user_mcp_configs_skipped_when_no_creator(
-        self, mock_oauth, mock_ph_configs, mock_user_configs, mock_send_refresh
-    ):
-        mock_oauth.return_value = "fresh-token"
-        mock_ph_configs.return_value = [_make_mcp_config()]
-        mock_send_refresh.return_value = CommandResult(success=True, status_code=200)
-
-        _refresh_sandbox_mcp(_make_task_run_mock(created_by_id=None), "read_only", auth_token=None)
+    def test_no_actor_skips_entirely(self, mock_oauth, mock_ph_configs, mock_user_configs, mock_send_refresh):
+        # Without a credential user the mint can only fail — skip quietly.
+        _refresh(_make_task_run_mock(created_by_id=None), actor_id=None)
 
-        mock_user_configs.assert_not_called()
-        mock_send_refresh.assert_called_once()
+        mock_oauth.assert_not_called()
+        mock_send_refresh.assert_not_called()
 
     @patch("products.tasks.backend.temporal.process_task.activities.send_followup_to_sandbox.send_refresh_session")
     @patch(
@@ -269,7 +269,7 @@ def test_scopes_propagate_to_oauth_and_configs(
         mock_user_configs.return_value = []
         mock_send_refresh.return_value = CommandResult(success=True, status_code=200)
 
-        _refresh_sandbox_mcp(_make_task_run_mock(), "full", auth_token=None)
+        _refresh(_make_task_run_mock(), scopes="full")
 
         mock_oauth.assert_called_once_with(mock_oauth.call_args.args[0], None, scopes="full")
         mock_ph_configs.assert_called_once_with(
@@ -289,7 +289,7 @@ class TestRefreshIntervalGate:
     def test_skipped_when_token_recently_issued(self, mock_oauth, mock_send_refresh):
         mark_mcp_token_issued("run-1")
 
-        _refresh_sandbox_mcp(_make_task_run_mock(), "read_only", auth_token=None)
+        _refresh(_make_task_run_mock())
 
         mock_oauth.assert_not_called()
         mock_send_refresh.assert_not_called()
@@ -310,7 +310,7 @@ def test_marks_after_successful_refresh(self, mock_oauth, mock_ph_configs, mock_
         mock_user_configs.return_value = []
         mock_send_refresh.return_value = CommandResult(success=True, status_code=200)
 
-        _refresh_sandbox_mcp(_make_task_run_mock(), "read_only", auth_token=None)
+        _refresh(_make_task_run_mock())
 
         # Cache entry now exists → next refresh within the interval is gated.
         assert cache.get(_mcp_token_issued_cache_key("run-1")) is True
@@ -337,7 +337,7 @@ def test_marks_after_successful_retry(
             CommandResult(success=True, status_code=200),
         ]
 
-        _refresh_sandbox_mcp(_make_task_run_mock(), "read_only", auth_token=None)
+        _refresh(_make_task_run_mock())
 
         assert cache.get(_mcp_token_issued_cache_key("run-1")) is True
 
@@ -360,7 +360,7 @@ def test_does_not_mark_after_two_failures(
         mock_user_configs.return_value = []
         mock_send_refresh.return_value = CommandResult(success=False, status_code=502, error="down")
 
-        _refresh_sandbox_mcp(_make_task_run_mock(), "read_only", auth_token=None)
+        _refresh(_make_task_run_mock())
 
         # Cache stays empty so the next follow-up retries the dispatch.
         assert cache.get(_mcp_token_issued_cache_key("run-1")) is None
@@ -400,6 +400,7 @@ def _patches(self):
 
             yield {
                 "task_run": task_run,
+                "task_run_cls": mock_task_run_cls,
                 "refresh": mock_refresh,
                 "user_msg": mock_user_msg,
                 "conn_token": mock_conn_token,
@@ -433,6 +434,55 @@ def test_scopes_flow_from_input_to_refresh(self, _patches):
         assert args[1] == "full"
         assert args[2] == "jwt"
 
+    def test_payload_actor_pins_resolution_over_run_state(self, _patches):
+        # A concurrent follow-up (or permission response) may overwrite the
+        # run-state actor between queueing and delivery; the message's own
+        # sender must win.
+        _patches["user_msg"].return_value = CommandResult(success=True, status_code=200)
+        _patches["task_run"].state = {"interaction_origin": "slack", "slack_actor_user_id": 42}
+
+        with patch(
+            "products.tasks.backend.temporal.process_task.activities.send_followup_to_sandbox.get_task_run_credential_user"
+        ) as mock_resolve:
+            mock_resolve.return_value = MagicMock(id=99)
+            send_followup_to_sandbox(SendFollowupToSandboxInput(run_id="run-1", message="hi", actor_user_id=99))
+
+        resolved_state = mock_resolve.call_args.args[1]
+        assert resolved_state["slack_actor_user_id"] == 99
+
+    def test_slack_delivery_stamps_turn_actor(self, _patches):
+        # The durable run-state actor must move at turn boundaries: delivery
+        # persists this message's sender so between-turn consumers (reply
+        # tagging, credential refresh) follow the executing turn.
+        _patches["user_msg"].return_value = CommandResult(success=True, status_code=200)
+        _patches["task_run"].state = {"interaction_origin": "slack", "slack_actor_user_id": 42}
+
+        with patch(
+            "products.tasks.backend.temporal.process_task.activities.send_followup_to_sandbox.get_task_run_credential_user"
+        ) as mock_resolve:
+            mock_resolve.return_value = MagicMock(id=99)
+            send_followup_to_sandbox(
+                SendFollowupToSandboxInput(
+                    run_id="run-1", message="hi", actor_user_id=99, context={"actor_slack_user_id": "U_BOB"}
+                )
+            )
+
+        _patches["task_run_cls"].update_state_atomic.assert_any_call(
+            _patches["task_run"].id,
+            updates={"slack_actor_user_id": 99, "slack_actor_slack_user_id": "U_BOB"},
+        )
+        # The ack also records the completed turn's actor for reply tagging.
+        _patches["task_run_cls"].update_state_atomic.assert_any_call(
+            "run-1", updates={"slack_last_turn_slack_user_id": "U_BOB"}
+        )
+
+    def test_non_slack_delivery_does_not_stamp(self, _patches):
+        _patches["user_msg"].return_value = CommandResult(success=True, status_code=200)
+
+        send_followup_to_sandbox(SendFollowupToSandboxInput(run_id="run-1", message="hi", actor_user_id=99))
+
+        _patches["task_run_cls"].update_state_atomic.assert_not_called()
+
     def test_default_scope_is_read_only(self, _patches):
         _patches["user_msg"].return_value = CommandResult(success=True, status_code=200)
 
diff --git a/products/tasks/backend/temporal/process_task/workflow.py b/products/tasks/backend/temporal/process_task/workflow.py
index 668659587337..8d4d80754fac 100644
--- a/products/tasks/backend/temporal/process_task/workflow.py
+++ b/products/tasks/backend/temporal/process_task/workflow.py
@@ -1,6 +1,6 @@
 import json
 import asyncio
-from dataclasses import dataclass, replace
+from dataclasses import dataclass, field, replace
 from datetime import datetime, timedelta
 from enum import StrEnum
 from typing import Any, Optional
@@ -138,9 +138,13 @@ class ProcessTaskInput:
 class PendingFollowup:
     message: str | None
     artifact_ids: list[str]
+    actor_user_id: int | None = None
     # Sender-supplied idempotency key (stable across the sender's retries);
     # None falls back to a workflow-generated id.
     message_id: str | None = None
+    # Signal context carried verbatim (e.g. actor_slack_user_id for reply
+    # tagging); consumers validate the keys they read.
+    context: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass
@@ -693,7 +697,9 @@ async def run(self, input: ProcessTaskInput) -> ProcessTaskOutput:
                             await self._send_followup_to_sandbox(
                                 message=message,
                                 artifact_ids=artifact_ids,
+                                actor_user_id=pending_followup.actor_user_id,
                                 message_id=pending_followup.message_id,
+                                context=pending_followup.context,
                             )
                             continue
 
@@ -1836,7 +1842,13 @@ async def send_followup_message(
                 "artifact_count": len(artifact_ids or []),
             },
         )
-        pending_followup = PendingFollowup(message=message, artifact_ids=artifact_ids or [], message_id=message_id)
+        pending_followup = PendingFollowup(
+            message=message,
+            artifact_ids=artifact_ids or [],
+            actor_user_id=actor_user_id,
+            message_id=message_id,
+            context=message_context if isinstance(message_context, dict) else {},
+        )
         # Always queue. `deprecate_patch` accepts existing non-deprecated
         # markers from workflows that ran the prior `workflow.patched(...)`
         # gate, so this is safe to deploy alongside in-flight workflows. The
@@ -1892,7 +1904,12 @@ async def send_permission_response(self, response: dict[str, Any]) -> None:
         )
 
     async def _send_followup_to_sandbox(
-        self, message: str | None, artifact_ids: list[str], message_id: str | None = None
+        self,
+        message: str | None,
+        artifact_ids: list[str],
+        actor_user_id: int | None = None,
+        message_id: str | None = None,
+        context: dict[str, Any] | None = None,
     ) -> None:
         workflow.logger.info(
             "send_followup_dispatch_begin",
@@ -1911,6 +1928,8 @@ async def _send_followup_to_sandbox(
                     posthog_mcp_scopes=self._posthog_mcp_scopes,
                     artifact_ids=artifact_ids,
                     message_id=message_id or str(workflow.uuid4()),
+                    actor_user_id=actor_user_id,
+                    context=context,
                 ),
                 start_to_close_timeout=timedelta(minutes=35),
                 # The activity heartbeats while blocked on the sync delivery
diff --git a/products/tasks/backend/temporal/slack_relay/activities.py b/products/tasks/backend/temporal/slack_relay/activities.py
index dbba1966aa93..74c04c7b4c0c 100644
--- a/products/tasks/backend/temporal/slack_relay/activities.py
+++ b/products/tasks/backend/temporal/slack_relay/activities.py
@@ -351,6 +351,9 @@ class RelaySlackMessageInput:
     user_message_ts: str | None = None
     delete_progress: bool = True
     reaction_emoji: str | None = None
+    # The actor of the turn this message answers, captured at trigger time.
+    # Falls back to the mapping's latest-actor fields when absent.
+    mention_slack_user_id: str | None = None
 
 
 @activity.defn
@@ -408,7 +411,7 @@ def relay_slack_message(input: RelaySlackMessageInput) -> None:
     )
     handler = SlackThreadHandler(context)
 
-    target = mapping.latest_actor_slack_user_id or mapping.mentioning_slack_user_id
+    target = input.mention_slack_user_id or mapping.latest_actor_slack_user_id or mapping.mentioning_slack_user_id
     mention_prefix = f"<@{target}> " if target else ""
     if input.delete_progress:
         handler.delete_progress()
diff --git a/products/tasks/backend/temporal/tests/test_client.py b/products/tasks/backend/temporal/tests/test_client.py
index eb035dc12968..22a83370b505 100644
--- a/products/tasks/backend/temporal/tests/test_client.py
+++ b/products/tasks/backend/temporal/tests/test_client.py
@@ -31,6 +31,6 @@ def test_followup_signal_sends_expected_args(mock_connect: MagicMock) -> None:
     handle = MagicMock(signal=AsyncMock())
     mock_connect.return_value = MagicMock(get_workflow_handle=MagicMock(return_value=handle))
 
-    signal_task_followup_message("wf-1", "hi", ["artifact-1"], message_id="msg-1")
+    signal_task_followup_message("wf-1", "hi", ["artifact-1"], message_id="msg-1", actor_user_id=7)
 
-    handle.signal.assert_awaited_once_with("send_followup_message", args=["hi", ["artifact-1"], "msg-1"])
+    handle.signal.assert_awaited_once_with("send_followup_message", args=["hi", ["artifact-1"], "msg-1", 7, None])
diff --git a/products/tasks/backend/tests/test_api.py b/products/tasks/backend/tests/test_api.py
index 151c440f5c95..2f0c81822c98 100644
--- a/products/tasks/backend/tests/test_api.py
+++ b/products/tasks/backend/tests/test_api.py
@@ -7829,7 +7829,7 @@ def test_command_signals_user_message(self, mock_signal_followup):
         self.assertEqual(data["jsonrpc"], "2.0")
         self.assertTrue(data["result"]["queued"])
 
-        mock_signal_followup.assert_called_once_with(run.workflow_id, "Hello agent", [], None)
+        mock_signal_followup.assert_called_once_with(run.workflow_id, "Hello agent", [], None, self.user.id, None)
 
     @patch("products.tasks.backend.temporal.client.signal_task_followup_message")
     def test_command_user_message_requires_code_access(self, mock_signal_followup):
@@ -7865,7 +7865,7 @@ def test_command_signals_user_message_without_active_sandbox(self, mock_signal_f
 
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertTrue(response.json()["result"]["queued"])
-        mock_signal_followup.assert_called_once_with(run.workflow_id, "Hello agent", [], None)
+        mock_signal_followup.assert_called_once_with(run.workflow_id, "Hello agent", [], None, self.user.id, None)
 
     @patch("products.tasks.backend.temporal.client.signal_task_followup_message")
     def test_command_signals_user_message_artifact_ids(self, mock_signal_followup):
@@ -7897,7 +7897,9 @@ def test_command_signals_user_message_artifact_ids(self, mock_signal_followup):
 
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertTrue(response.json()["result"]["queued"])
-        mock_signal_followup.assert_called_once_with(run.workflow_id, "See attached", ["artifact-123"], None)
+        mock_signal_followup.assert_called_once_with(
+            run.workflow_id, "See attached", ["artifact-123"], None, self.user.id, None
+        )
 
     @patch("products.tasks.backend.temporal.client.signal_task_followup_message")
     def test_command_returns_502_when_user_message_signal_fails(self, mock_signal_followup):

From 926bdfe2708338e4a4767546e275d3dde7490420 Mon Sep 17 00:00:00 2001
From: Vojta Bartos <vojtech.b@posthog.com>
Date: Tue, 14 Jul 2026 19:14:29 +0200
Subject: [PATCH 2/5] feat(tasks): resolve reply tags from the echoed message
 id

The agent-server can echo the id of the user message a turn answers
(optional relay_message field). Delivery records message_id -> sender
when it sends; the relay resolves the tag from that correlation first,
with the completed-turn stamp and live actor as fallbacks for agents
that do not echo yet.
---
 products/tasks/backend/facade/api.py          | 20 ++++++++++++++-----
 .../tasks/backend/presentation/serializers.py |  6 ++++++
 .../tasks/backend/presentation/views/api.py   | 15 +++++++++++++-
 .../activities/forward_pending_message.py     |  7 -------
 .../activities/send_followup_to_sandbox.py    |  5 +++++
 .../backend/temporal/process_task/utils.py    | 14 +++++++++++++
 .../tasks/frontend/generated/api.schemas.ts   |  6 ++++++
 products/tasks/frontend/generated/api.zod.ts  |  7 +++++++
 services/mcp/src/api/generated.ts             |  6 ++++++
 9 files changed, 73 insertions(+), 13 deletions(-)

diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py
index 3c5f747781ac..812d49c5ba5b 100644
--- a/products/tasks/backend/facade/api.py
+++ b/products/tasks/backend/facade/api.py
@@ -2692,6 +2692,7 @@ def relay_task_run_message(
     *,
     text: str,
     text_parts: list[str] | None = None,
+    message_id: str | None = None,
 ) -> tuple[str, str | None]:
     """Queue a Slack relay workflow for a run message, or under the agent-design
     flag signal the running task workflow to stream the text inline.
@@ -2734,12 +2735,21 @@ def relay_task_run_message(
             logger.exception("task_run_relay_text_signal_failed", extra={"run_id": str(run.id)})
         return "skipped", None
 
-    # Tag the actor of the last *completed* turn (stamped on the delivery
-    # ack); the live actor is a fallback for pre-rollout runs, and may point
-    # at the next speaker when a turn outlives the delivery ack.
+    # Prefer the exact message this turn answers (agent-server echoes its
+    # id); fall back to the last completed turn's actor (stamped on the
+    # delivery ack), then the live actor for pre-rollout runs.
+    mention_slack_user_id = None
+    if message_id:
+        from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporal deps off the api import path
+            get_message_actor,
+        )
+
+        mention_slack_user_id = get_message_actor(str(run.id), message_id)
     relay_state = run.state or {}
-    mention_slack_user_id = relay_state.get("slack_last_turn_slack_user_id") or relay_state.get(
-        "slack_actor_slack_user_id"
+    mention_slack_user_id = (
+        mention_slack_user_id
+        or relay_state.get("slack_last_turn_slack_user_id")
+        or relay_state.get("slack_actor_slack_user_id")
     )
     try:
         relay_id = execute_posthog_code_agent_relay_workflow(
diff --git a/products/tasks/backend/presentation/serializers.py b/products/tasks/backend/presentation/serializers.py
index b6529de32a77..2cdbc73c3301 100644
--- a/products/tasks/backend/presentation/serializers.py
+++ b/products/tasks/backend/presentation/serializers.py
@@ -713,6 +713,12 @@ class TaskRunRelayMessageRequestSerializer(serializers.Serializer):
         max_length=10000,
         help_text="Joined message body. Used when text_parts is absent.",
     )
+    message_id = serializers.CharField(
+        max_length=128,
+        required=False,
+        allow_null=True,
+        help_text="Id of the user message this turn answers, when the agent-server echoes it.",
+    )
     # Kept optional for forward/backward compatibility during rollout; will be aligned once deployed.
     text_parts = serializers.ListField(
         child=serializers.CharField(max_length=10000, allow_blank=True),
diff --git a/products/tasks/backend/presentation/views/api.py b/products/tasks/backend/presentation/views/api.py
index e8d83c0e2663..e437f92a989d 100644
--- a/products/tasks/backend/presentation/views/api.py
+++ b/products/tasks/backend/presentation/views/api.py
@@ -39,7 +39,7 @@
     cancellation as tasks_cancellation,
     contracts as tasks_contracts,
 )
-from products.tasks.backend.facade.access import cloud_usage_limit_response
+from products.tasks.backend.facade.access import cloud_usage_limit_response, code_access_required_response
 from products.tasks.backend.facade.metrics import (
     StreamConnectionOutcome,
     observe_stream_connection_closed,
@@ -639,6 +639,9 @@ def warm(self, request, **kwargs):
         if not self._warm_enabled():
             return Response(status=status.HTTP_200_OK)
 
+        if access_response := code_access_required_response(request.user):
+            return access_response
+
         user_id = self._user_id()
         if user_id is None:
             return Response(status=status.HTTP_200_OK)
@@ -755,6 +758,8 @@ def retrieve(self, request, pk=None, **kwargs):
 
     @extend_schema(request=TaskAutomationWriteSerializer, responses={201: TaskAutomationSerializer})
     def create(self, request, **kwargs):
+        if access_response := code_access_required_response(request.user):
+            return access_response
         serializer = self._write_serializer(request.data)
         automation = tasks_facade.create_task_automation(
             self.team_id, getattr(request.user, "id", None), **self._facade_kwargs(serializer.validated_data)
@@ -764,6 +769,9 @@ def create(self, request, **kwargs):
     @extend_schema(request=TaskAutomationWriteSerializer, responses={200: TaskAutomationSerializer})
     def partial_update(self, request, pk=None, **kwargs):
         serializer = self._write_serializer(request.data, partial=True)
+        if serializer.validated_data.get("enabled") is True:
+            if access_response := code_access_required_response(request.user):
+                return access_response
         automation = tasks_facade.update_task_automation(
             pk, self.team_id, getattr(request.user, "id", None), **self._facade_kwargs(serializer.validated_data)
         )
@@ -780,6 +788,8 @@ def destroy(self, request, pk=None, **kwargs):
     @extend_schema(request=None, responses={200: TaskAutomationSerializer})
     @action(detail=True, methods=["post"], url_path="run", required_scopes=["task:write"])
     def run(self, request, pk=None, **kwargs):
+        if access_response := code_access_required_response(request.user):
+            return access_response
         automation = tasks_facade.run_task_automation_now(pk, self.team_id, getattr(request.user, "id", None))
         if automation is None:
             raise NotFound()
@@ -1165,6 +1175,7 @@ def relay_message(self, request, pk=None, **kwargs):
             self.team_id,
             text=request.validated_data["text"],
             text_parts=request.validated_data.get("text_parts"),
+            message_id=request.validated_data.get("message_id"),
         )
         if relay_status == "failed":
             return Response(
@@ -1503,6 +1514,8 @@ def command(self, request, pk=None, **kwargs):
         params = request.validated_data.get("params")
 
         if method == "user_message":
+            if access_response := code_access_required_response(request.user):
+                return access_response
             command_params = dict(params or {})
             artifact_ids = command_params.pop("artifact_ids", [])
             if artifact_ids:
diff --git a/products/tasks/backend/temporal/process_task/activities/forward_pending_message.py b/products/tasks/backend/temporal/process_task/activities/forward_pending_message.py
index e261689b6fde..fddebe288710 100644
--- a/products/tasks/backend/temporal/process_task/activities/forward_pending_message.py
+++ b/products/tasks/backend/temporal/process_task/activities/forward_pending_message.py
@@ -146,15 +146,8 @@ def activity_failure() -> tuple[str, str] | None:
             else:
                 _enqueue_pending_delivery_failure_relay(task_run, pending_message_ts, result.error)
 
-        boot_actor_slack_user_id = state.get("slack_actor_slack_user_id")
-        updates = (
-            {"slack_last_turn_slack_user_id": boot_actor_slack_user_id}
-            if result.success and isinstance(boot_actor_slack_user_id, str) and boot_actor_slack_user_id
-            else None
-        )
         TaskRun.update_state_atomic(
             run_id,
-            updates=updates,
             remove_keys=["pending_user_message", "pending_user_artifact_ids", "pending_user_message_ts"],
         )
 
diff --git a/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py b/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
index 80584925da2e..78f8eb91241f 100644
--- a/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
+++ b/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
@@ -34,6 +34,7 @@
     get_user_mcp_server_configs,
     is_slack_interaction_state,
     mark_mcp_token_issued,
+    record_message_actor,
     should_refresh_mcp_token,
 )
 
@@ -171,6 +172,10 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
             _write_error_and_complete(input.run_id, error_msg, run_uses_dedicated_stream(task_run.state))
             raise ApplicationError(f"send_followup failed: {error_msg}", non_retryable=True)
 
+    delivery_actor_slack_user_id = (input.context or {}).get("actor_slack_user_id")
+    if input.message_id and isinstance(delivery_actor_slack_user_id, str) and delivery_actor_slack_user_id:
+        record_message_actor(input.run_id, input.message_id, delivery_actor_slack_user_id)
+
     result = send_user_message(
         task_run,
         input.message,
diff --git a/products/tasks/backend/temporal/process_task/utils.py b/products/tasks/backend/temporal/process_task/utils.py
index a63425462628..14ba2041f29d 100644
--- a/products/tasks/backend/temporal/process_task/utils.py
+++ b/products/tasks/backend/temporal/process_task/utils.py
@@ -1020,3 +1020,17 @@ def get_git_identity_env_vars(task: Task, state: dict[str, Any] | None = None) -
         "GIT_COMMITTER_NAME": name,
         "GIT_COMMITTER_EMAIL": email,
     }
+
+
+def _message_actor_cache_key(run_id: str, message_id: str) -> str:
+    return f"tasks:followup-actor:{run_id}:{message_id}"
+
+
+def record_message_actor(run_id: str, message_id: str, slack_user_id: str) -> None:
+    """Correlate a delivered message with its sender's Slack id, so a relay
+    echoing the message id can tag the exact speaker it answers."""
+    get_tasks_cache().set(_message_actor_cache_key(run_id, message_id), slack_user_id, timeout=2 * 60 * 60)
+
+
+def get_message_actor(run_id: str, message_id: str) -> str | None:
+    return get_tasks_cache().get(_message_actor_cache_key(run_id, message_id))
diff --git a/products/tasks/frontend/generated/api.schemas.ts b/products/tasks/frontend/generated/api.schemas.ts
index 2570bb1e67da..898caee9c9ad 100644
--- a/products/tasks/frontend/generated/api.schemas.ts
+++ b/products/tasks/frontend/generated/api.schemas.ts
@@ -2162,6 +2162,12 @@ export interface TaskRunRelayMessageRequestApi {
      * @maxLength 10000
      */
     text: string
+    /**
+     * Id of the user message this turn answers, when the agent-server echoes it.
+     * @maxLength 128
+     * @nullable
+     */
+    message_id?: string | null
     /**
      * Ordered assistant text blocks. When present, the last non-empty entry is posted instead of text.
      * @items.maxLength 10000
diff --git a/products/tasks/frontend/generated/api.zod.ts b/products/tasks/frontend/generated/api.zod.ts
index ba994811eba0..9c36ff3d8b15 100644
--- a/products/tasks/frontend/generated/api.zod.ts
+++ b/products/tasks/frontend/generated/api.zod.ts
@@ -1881,6 +1881,8 @@ export const TasksRunsCommandCreateBody = /* @__PURE__ */ zod
  */
 export const tasksRunsRelayMessageCreateBodyTextMax = 10000
 
+export const tasksRunsRelayMessageCreateBodyMessageIdMax = 128
+
 export const tasksRunsRelayMessageCreateBodyTextPartsItemMax = 10000
 
 export const TasksRunsRelayMessageCreateBody = /* @__PURE__ */ zod.object({
@@ -1888,6 +1890,11 @@ export const TasksRunsRelayMessageCreateBody = /* @__PURE__ */ zod.object({
         .string()
         .max(tasksRunsRelayMessageCreateBodyTextMax)
         .describe('Joined message body. Used when text_parts is absent.'),
+    message_id: zod
+        .string()
+        .max(tasksRunsRelayMessageCreateBodyMessageIdMax)
+        .nullish()
+        .describe('Id of the user message this turn answers, when the agent-server echoes it.'),
     text_parts: zod
         .array(zod.string().max(tasksRunsRelayMessageCreateBodyTextPartsItemMax))
         .optional()
diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts
index afd497efd921..68cf892654cf 100644
--- a/services/mcp/src/api/generated.ts
+++ b/services/mcp/src/api/generated.ts
@@ -62074,6 +62074,12 @@ export namespace Schemas {
          * @maxLength 10000
          */
       text: string;
+      /**
+         * Id of the user message this turn answers, when the agent-server echoes it.
+         * @maxLength 128
+         * @nullable
+         */
+      message_id?: string | null;
       /**
          * Ordered assistant text blocks. When present, the last non-empty entry is posted instead of text.
          * @items.maxLength 10000

From 632c21fa45c6630340a668569281b69f37aac0d4 Mon Sep 17 00:00:00 2001
From: Vojta Bartos <vojtech.b@posthog.com>
Date: Wed, 15 Jul 2026 09:52:58 +0200
Subject: [PATCH 3/5] refactor(tasks): resolve reply mentions in one place
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The mention fallback chain was split between the relay trigger (facade)
and the relay activity. The relay input now carries the echoed
message_id and the activity resolves the whole chain next to its
existing mapping fallbacks — message actor, completed-turn actor, live
actor, mapping — which also gives the pending-message relay paths the
same semantics. Extract the message sender once per delivery, skip the
completed-turn stamp when already current, and name the correlation TTL.
---
 products/tasks/backend/facade/api.py          | 18 +------------
 products/tasks/backend/temporal/client.py     |  2 ++
 .../activities/send_followup_to_sandbox.py    | 25 +++++--------------
 .../tests/test_send_followup_to_sandbox.py    |  6 +----
 .../backend/temporal/process_task/utils.py    | 14 +++++++++--
 .../temporal/slack_relay/activities.py        | 17 ++++++++++---
 6 files changed, 35 insertions(+), 47 deletions(-)

diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py
index 812d49c5ba5b..10e45bfe277c 100644
--- a/products/tasks/backend/facade/api.py
+++ b/products/tasks/backend/facade/api.py
@@ -2735,28 +2735,12 @@ def relay_task_run_message(
             logger.exception("task_run_relay_text_signal_failed", extra={"run_id": str(run.id)})
         return "skipped", None
 
-    # Prefer the exact message this turn answers (agent-server echoes its
-    # id); fall back to the last completed turn's actor (stamped on the
-    # delivery ack), then the live actor for pre-rollout runs.
-    mention_slack_user_id = None
-    if message_id:
-        from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporal deps off the api import path
-            get_message_actor,
-        )
-
-        mention_slack_user_id = get_message_actor(str(run.id), message_id)
-    relay_state = run.state or {}
-    mention_slack_user_id = (
-        mention_slack_user_id
-        or relay_state.get("slack_last_turn_slack_user_id")
-        or relay_state.get("slack_actor_slack_user_id")
-    )
     try:
         relay_id = execute_posthog_code_agent_relay_workflow(
             run_id=str(run.id),
             text=trimmed,
             delete_progress=True,
-            mention_slack_user_id=mention_slack_user_id if isinstance(mention_slack_user_id, str) else None,
+            message_id=message_id,
         )
     except Exception:
         logger.exception("task_run_relay_message_enqueue_failed", extra={"run_id": str(run.id)})
diff --git a/products/tasks/backend/temporal/client.py b/products/tasks/backend/temporal/client.py
index 5563f22b0e2f..49dfe25bfc23 100644
--- a/products/tasks/backend/temporal/client.py
+++ b/products/tasks/backend/temporal/client.py
@@ -496,6 +496,7 @@ def execute_posthog_code_agent_relay_workflow(
     user_message_ts: str | None = None,
     delete_progress: bool = True,
     reaction_emoji: str | None = None,
+    message_id: str | None = None,
 ) -> str:
     relay_id = relay_id or str(uuid.uuid4())
     workflow_id = f"posthog-code-agent-relay-{run_id}-{relay_id}"
@@ -511,6 +512,7 @@ def execute_posthog_code_agent_relay_workflow(
                 user_message_ts=user_message_ts,
                 delete_progress=delete_progress,
                 reaction_emoji=reaction_emoji,
+                message_id=message_id,
             ),
             id=workflow_id,
             id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
diff --git a/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py b/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
index 78f8eb91241f..856f7bb9e8ff 100644
--- a/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
+++ b/products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py
@@ -128,6 +128,9 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
     # Resolve credentials against this message's sender, not the run-state
     # actor a concurrent follow-up may have overwritten since queueing. Local
     # overlay; the resolver still enforces team access (see run_actor.py).
+    raw_actor_slack_user_id = (input.context or {}).get("actor_slack_user_id")
+    actor_slack_user_id = raw_actor_slack_user_id if isinstance(raw_actor_slack_user_id, str) else None
+
     state = task_run.state
     if input.actor_user_id is not None:
         state = {**(state or {}), "slack_actor_user_id": input.actor_user_id}
@@ -136,11 +139,7 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
             # moves the durable actor at turn boundaries — between-turn
             # consumers (reply tagging, permission broker) see the executing
             # turn's actor. Skipped when already current.
-            actor_slack_user_id = (input.context or {}).get("actor_slack_user_id")
-            updates = slack_actor_state_updates(
-                user_id=input.actor_user_id,
-                slack_user_id=actor_slack_user_id if isinstance(actor_slack_user_id, str) else None,
-            )
+            updates = slack_actor_state_updates(user_id=input.actor_user_id, slack_user_id=actor_slack_user_id)
             current = task_run.state or {}
             if any(current.get(key) != value for key, value in updates.items()):
                 try:
@@ -172,9 +171,8 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
             _write_error_and_complete(input.run_id, error_msg, run_uses_dedicated_stream(task_run.state))
             raise ApplicationError(f"send_followup failed: {error_msg}", non_retryable=True)
 
-    delivery_actor_slack_user_id = (input.context or {}).get("actor_slack_user_id")
-    if input.message_id and isinstance(delivery_actor_slack_user_id, str) and delivery_actor_slack_user_id:
-        record_message_actor(input.run_id, input.message_id, delivery_actor_slack_user_id)
+    if input.message_id and actor_slack_user_id:
+        record_message_actor(input.run_id, input.message_id, actor_slack_user_id)
 
     result = send_user_message(
         task_run,
@@ -199,17 +197,6 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
                 attempt=_current_attempt(),
             )
             return
-        turn_actor_slack_user_id = (input.context or {}).get("actor_slack_user_id")
-        if isinstance(turn_actor_slack_user_id, str) and turn_actor_slack_user_id:
-            # The ack means this message's turn finished; record its actor for
-            # reply tagging — the relay may fire after the next delivery has
-            # already restamped the live actor.
-            try:
-                TaskRun.update_state_atomic(
-                    input.run_id, updates={"slack_last_turn_slack_user_id": turn_actor_slack_user_id}
-                )
-            except Exception:
-                logger.warning("send_followup_turn_actor_stamp_failed", run_id=input.run_id, exc_info=True)
         _write_turn_complete(input.run_id, _get_stop_reason(result.data), run_uses_dedicated_stream(task_run.state))
         logger.info("send_followup_delivered", run_id=input.run_id)
     elif result.turn_in_flight:
diff --git a/products/tasks/backend/temporal/process_task/tests/test_send_followup_to_sandbox.py b/products/tasks/backend/temporal/process_task/tests/test_send_followup_to_sandbox.py
index 6092b863cc92..300c995909f8 100644
--- a/products/tasks/backend/temporal/process_task/tests/test_send_followup_to_sandbox.py
+++ b/products/tasks/backend/temporal/process_task/tests/test_send_followup_to_sandbox.py
@@ -130,7 +130,7 @@ def test_refresh_keeps_imported_mcp_servers(
             {"type": "http", "name": "posthog", "url": "https://shadow.example.com/mcp", "headers": []},
         ]
 
-        _refresh_sandbox_mcp(task_run, "read_only", auth_token="jwt")
+        _refresh(task_run, auth_token="jwt")
 
         mcp_servers = mock_send_refresh.call_args.args[1]
         assert [server["name"] for server in mcp_servers] == ["posthog", "grafana"]
@@ -471,10 +471,6 @@ def test_slack_delivery_stamps_turn_actor(self, _patches):
             _patches["task_run"].id,
             updates={"slack_actor_user_id": 99, "slack_actor_slack_user_id": "U_BOB"},
         )
-        # The ack also records the completed turn's actor for reply tagging.
-        _patches["task_run_cls"].update_state_atomic.assert_any_call(
-            "run-1", updates={"slack_last_turn_slack_user_id": "U_BOB"}
-        )
 
     def test_non_slack_delivery_does_not_stamp(self, _patches):
         _patches["user_msg"].return_value = CommandResult(success=True, status_code=200)
diff --git a/products/tasks/backend/temporal/process_task/utils.py b/products/tasks/backend/temporal/process_task/utils.py
index 14ba2041f29d..6e7d99dfc110 100644
--- a/products/tasks/backend/temporal/process_task/utils.py
+++ b/products/tasks/backend/temporal/process_task/utils.py
@@ -1026,10 +1026,20 @@ def _message_actor_cache_key(run_id: str, message_id: str) -> str:
     return f"tasks:followup-actor:{run_id}:{message_id}"
 
 
+# Bounds how long after delivery a turn can finish and still get exact
+# per-message attribution; longer turns fall back to the run-state actors.
+MESSAGE_ACTOR_TTL_SECONDS = 2 * 60 * 60
+
+
 def record_message_actor(run_id: str, message_id: str, slack_user_id: str) -> None:
     """Correlate a delivered message with its sender's Slack id, so a relay
-    echoing the message id can tag the exact speaker it answers."""
-    get_tasks_cache().set(_message_actor_cache_key(run_id, message_id), slack_user_id, timeout=2 * 60 * 60)
+    echoing the message id can tag the exact speaker it answers. The sandbox
+    only ever echoes an opaque id — resolution happens against actors this
+    server recorded itself, so a compromised sandbox cannot pick an arbitrary
+    mention target."""
+    get_tasks_cache().set(
+        _message_actor_cache_key(run_id, message_id), slack_user_id, timeout=MESSAGE_ACTOR_TTL_SECONDS
+    )
 
 
 def get_message_actor(run_id: str, message_id: str) -> str | None:
diff --git a/products/tasks/backend/temporal/slack_relay/activities.py b/products/tasks/backend/temporal/slack_relay/activities.py
index 74c04c7b4c0c..699bb9fa6888 100644
--- a/products/tasks/backend/temporal/slack_relay/activities.py
+++ b/products/tasks/backend/temporal/slack_relay/activities.py
@@ -351,9 +351,9 @@ class RelaySlackMessageInput:
     user_message_ts: str | None = None
     delete_progress: bool = True
     reaction_emoji: str | None = None
-    # The actor of the turn this message answers, captured at trigger time.
-    # Falls back to the mapping's latest-actor fields when absent.
-    mention_slack_user_id: str | None = None
+    # Id of the user message this relay answers (agent-server echo), used to
+    # tag the exact sender; None falls back to the run-state/mapping actors.
+    message_id: str | None = None
 
 
 @activity.defn
@@ -362,6 +362,7 @@ def relay_slack_message(input: RelaySlackMessageInput) -> None:
     from products.slack_app.backend.models import SlackThreadTaskMapping
     from products.slack_app.backend.slack_thread import SlackThreadContext, SlackThreadHandler
     from products.tasks.backend.models import TaskRun
+    from products.tasks.backend.temporal.process_task.utils import get_message_actor
 
     try:
         task_run = TaskRun.objects.get(id=input.run_id)
@@ -411,7 +412,15 @@ def relay_slack_message(input: RelaySlackMessageInput) -> None:
     )
     handler = SlackThreadHandler(context)
 
-    target = input.mention_slack_user_id or mapping.latest_actor_slack_user_id or mapping.mentioning_slack_user_id
+    # Mention resolution, most precise first: the echoed message's recorded
+    # sender, then the live/mapping actors for pre-rollout runs.
+    mention_from_message = get_message_actor(input.run_id, input.message_id) if input.message_id else None
+    target = (
+        mention_from_message
+        or state.get("slack_actor_slack_user_id")
+        or mapping.latest_actor_slack_user_id
+        or mapping.mentioning_slack_user_id
+    )
     mention_prefix = f"<@{target}> " if target else ""
     if input.delete_progress:
         handler.delete_progress()

From dd32bb24429496bc091f721d31a6730db1aca083 Mon Sep 17 00:00:00 2001
From: Vojta Bartos <vojtech.b@posthog.com>
Date: Mon, 20 Jul 2026 11:16:11 +0200
Subject: [PATCH 4/5] fix(tasks): route slack actor state helper through the
 tasks facade

The slack_app forward-followup activity reached into
products.tasks.backend.logic.services.run_actor directly, which tach rejects
because products.tasks only exposes backend.facade. Re-export
slack_actor_state_updates from facade.api and import it from there.
---
 .../ai/slack_app/activities/task_creation.py       |  6 +++---
 products/tasks/backend/facade/api.py               | 14 ++++++++++++++
 2 files changed, 17 insertions(+), 3 deletions(-)

diff --git a/posthog/temporal/ai/slack_app/activities/task_creation.py b/posthog/temporal/ai/slack_app/activities/task_creation.py
index 6e31746e80dd..44f002b81a89 100644
--- a/posthog/temporal/ai/slack_app/activities/task_creation.py
+++ b/posthog/temporal/ai/slack_app/activities/task_creation.py
@@ -82,11 +82,11 @@ def _canvas_file_delivery_available(integration: Integration) -> bool:
 
 
 def _slack_actor_state_updates(*, user_id: int, slack_user_id: str) -> dict[str, Any]:
-    from products.tasks.backend.logic.services.run_actor import (  # noqa: PLC0415 — keep tasks deps off the slack_app import path
-        slack_actor_state_updates,
+    from products.tasks.backend.facade import (
+        api as tasks_facade,  # noqa: PLC0415 — keep tasks deps off the slack_app import path
     )
 
-    return slack_actor_state_updates(user_id=user_id, slack_user_id=slack_user_id)
+    return tasks_facade.slack_actor_state_updates(user_id=user_id, slack_user_id=slack_user_id)
 
 
 def _strip_context_tag(text: str) -> str:
diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py
index 10e45bfe277c..5938c3920b85 100644
--- a/products/tasks/backend/facade/api.py
+++ b/products/tasks/backend/facade/api.py
@@ -201,6 +201,7 @@
     "select_repository_for_message",
     "set_task_run_output",
     "set_task_title",
+    "slack_actor_state_updates",
     "signal_report_queryset",
     "signal_task_run_user_message",
     "signal_workflow_completion",
@@ -929,6 +930,19 @@ def update_task_run_state(
     return TaskRun.update_state_atomic(run_id, updates=updates, remove_keys=remove_keys)
 
 
+def slack_actor_state_updates(*, user_id: int, slack_user_id: str | None = None) -> dict[str, Any]:
+    """Run-state updates recording the Slack user currently steering a run.
+
+    Credential resolution and reply tagging read the keys this builds, so every
+    writer must go through here rather than assembling the dict inline.
+    """
+    from products.tasks.backend.logic.services.run_actor import (  # noqa: PLC0415 — keep tasks internals off the api import path
+        slack_actor_state_updates as _slack_actor_state_updates,
+    )
+
+    return _slack_actor_state_updates(user_id=user_id, slack_user_id=slack_user_id)
+
+
 def set_task_run_created_at_for_seeding(
     run_id: str | UUID, task_id: str | UUID, team_id: int, *, created_at: datetime
 ) -> None:

From 70b6497a562cb39f8f35590a9191d833bad15ba4 Mon Sep 17 00:00:00 2001
From: Vojta Bartos <vojtech.b@posthog.com>
Date: Mon, 20 Jul 2026 11:20:33 +0200
Subject: [PATCH 5/5] fix(tasks): assert message_id in relay_message test calls

The relay endpoint now forwards message_id to
execute_posthog_code_agent_relay_workflow; update the existing relay_message
tests to expect message_id=None on the call.
---
 products/tasks/backend/tests/test_api.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/products/tasks/backend/tests/test_api.py b/products/tasks/backend/tests/test_api.py
index 2f0c81822c98..7214f6ec841f 100644
--- a/products/tasks/backend/tests/test_api.py
+++ b/products/tasks/backend/tests/test_api.py
@@ -4801,6 +4801,7 @@ def test_relay_message_enqueues_slack_relay_workflow(self, mock_execute_relay):
             run_id=str(run.id),
             text="Which license should I use?",
             delete_progress=True,
+            message_id=None,
         )
 
     @parameterized.expand(
@@ -4853,6 +4854,7 @@ def test_relay_message_text_selection(self, _name, payload, expected_posted_text
             run_id=str(run.id),
             text=expected_posted_text,
             delete_progress=True,
+            message_id=None,
         )
 
     @patch("products.tasks.backend.temporal.client.execute_posthog_code_agent_relay_workflow")