diff --git a/posthog/temporal/ai/slack_app/activities/task_creation.py b/posthog/temporal/ai/slack_app/activities/task_creation.py
index e65ee064457e..44f002b81a89 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.facade import (
+        api as tasks_facade,  # noqa: PLC0415 — keep tasks deps off the slack_app import path
+    )
+
+    return tasks_facade.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..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:
@@ -2551,7 +2565,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 +2586,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)
@@ -2689,6 +2706,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.
@@ -2732,7 +2750,12 @@ def relay_task_run_message(
         return "skipped", None
 
     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,
+            message_id=message_id,
+        )
     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/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 b7d39b40daab..e437f92a989d 100644
--- a/products/tasks/backend/presentation/views/api.py
+++ b/products/tasks/backend/presentation/views/api.py
@@ -1175,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(
@@ -1534,7 +1535,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..49dfe25bfc23 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:
@@ -490,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}"
@@ -505,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 afe1a4f99b03..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
@@ -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
@@ -33,6 +34,7 @@
     get_user_mcp_server_configs,
     is_slack_interaction_state,
     mark_mcp_token_issued,
+    record_message_actor,
     should_refresh_mcp_token,
 )
 
@@ -55,9 +57,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 +125,31 @@ 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).
+    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}
+        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.
+            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:
+                    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:
@@ -142,6 +171,9 @@ 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)
 
+    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,
         input.message,
@@ -221,24 +253,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.
+    """Mint a fresh OAuth token for the actor 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.
-
-    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 +283,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..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
@@ -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(
@@ -125,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"]
@@ -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,51 @@ 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"},
+        )
+
+    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/utils.py b/products/tasks/backend/temporal/process_task/utils.py
index a63425462628..6e7d99dfc110 100644
--- a/products/tasks/backend/temporal/process_task/utils.py
+++ b/products/tasks/backend/temporal/process_task/utils.py
@@ -1020,3 +1020,27 @@ 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}"
+
+
+# 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. 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:
+    return get_tasks_cache().get(_message_actor_cache_key(run_id, message_id))
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..699bb9fa6888 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
+    # 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
@@ -359,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)
@@ -408,7 +412,15 @@ def relay_slack_message(input: RelaySlackMessageInput) -> None:
     )
     handler = SlackThreadHandler(context)
 
-    target = 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()
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..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")
@@ -7829,7 +7831,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 +7867,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 +7899,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):
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