feat(tasks): rebind sandbox GitHub identity on actor transitions

products/tasks/backend/temporal/execute_sandbox/workflow.py

@@ -1203,6 +1203,7 @@ async def _send_followup_to_sandbox(self, message: str | None, artifact_ids: lis
                 posthog_mcp_scopes=self._posthog_mcp_scopes,
                 artifact_ids=artifact_ids,
                 message_id=str(workflow.uuid4()),
+                context=self.context,
             ),
             start_to_close_timeout=timedelta(minutes=35),
             # See process_task: heartbeat detects worker restarts, message_id

products/tasks/backend/temporal/process_task/activities/get_task_processing_context.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import timedelta
 
 from django.conf import settings
@@ -94,6 +94,19 @@ class TaskProcessingContext:
     # (the user's toggle) applies. Captured at workflow start so it's stable across retries.
     rtk_enabled: bool = True
 
+    def with_state(self, state: dict | None) -> "TaskProcessingContext":
+        """Copy of this context rebased onto the given (typically live) run state.
+
+        The context is captured at workflow start, so its ``state`` is a
+        boot-time snapshot while the run's actual state moves on (multiplayer
+        follow-ups update the actor mid-run). Anything resolving the *current*
+        actor must go through a rebased copy. The other fields intentionally
+        stay boot-time: they are either immutable for the run (ids, repository)
+        or last-resort fallbacks (``github_user_integration_id``,
+        ``distinct_id``) that actor-aware resolution overrides via state.
+        """
+        return replace(self, state=state)
+
     @property
     def mode(self) -> str:
         """Get the execution mode from state. Defaults to 'background'."""

products/tasks/backend/temporal/process_task/activities/refresh_sandbox_credentials.py

@@ -11,7 +11,6 @@
     SandboxNotRunningError,
     TaskNotFoundError,
 )
-from products.tasks.backend.logic.services.agent_command import send_refresh_session
 from products.tasks.backend.logic.services.connection_token import create_sandbox_connection_token
 from products.tasks.backend.logic.services.sandbox import Sandbox
 from products.tasks.backend.models import Task, TaskRun
@@ -20,6 +19,7 @@
 from products.tasks.backend.temporal.process_task.sandbox_credentials import (
     DEFAULT_REFRESH_INTERVAL_SECONDS,
     build_sandbox_credentials,
+    notify_sandbox_credentials_refreshed,
 )
 from products.tasks.backend.temporal.process_task.utils import (
     get_actor_distinct_id,
@@ -32,12 +32,13 @@
 logger = get_logger(__name__)
 
 
-def _notify_agent_server_of_refresh(ctx: TaskProcessingContext, task: Task, refreshed_kinds: list[str]) -> None:
+def _notify_agent_server_of_refresh(
+    ctx: TaskProcessingContext, task: Task, task_run: TaskRun, refreshed_kinds: list[str]
+) -> None:
     """Tell the running agent-server which credentials were re-injected so it logs them.
     This is best-effort since the sandbox may be unreachable, so a failure here never fails the refresh itself.
     """
     try:
-        task_run = TaskRun.objects.get(id=ctx.run_id)
         auth_token = None
         actor_user = get_task_run_credential_user(task, ctx.state)
         if is_slack_interaction_state(ctx.state) and actor_user is None:
@@ -47,10 +48,7 @@ def _notify_agent_server_of_refresh(ctx: TaskProcessingContext, task: Task, refr
             auth_token = create_sandbox_connection_token(
                 task_run, user_id=actor_user.id, distinct_id=get_actor_distinct_id(actor_user)
             )
-        authorship = (ctx.state or {}).get("pr_authorship_mode")
-        send_refresh_session(
-            task_run, [], auth_token=auth_token, refreshed_credentials=refreshed_kinds, authorship=authorship
-        )
+        notify_sandbox_credentials_refreshed(task_run, refreshed_kinds, auth_token=auth_token)
     except Exception:
         logger.warning("sandbox_credentials_refresh_notify_failed", run_id=ctx.run_id, exc_info=True)
 
@@ -98,6 +96,16 @@ def refresh_sandbox_credentials(input: RefreshSandboxCredentialsInput) -> Refres
         except Task.DoesNotExist as e:
             raise TaskNotFoundError(f"Task {ctx.task_id} not found", {"task_id": ctx.task_id}, cause=e)
 
+        # Credential resolution must follow the *current* actor (multiplayer
+        # follow-ups update it mid-run), so rebase the boot-time context onto
+        # the live run state.
+        task_run = None
+        try:
+            task_run = TaskRun.objects.get(id=ctx.run_id)
+            ctx = ctx.with_state(task_run.state)
+        except TaskRun.DoesNotExist:
+            logger.warning("sandbox_credentials_refresh_run_missing", run_id=ctx.run_id)
+
         refreshed_kinds: list[str] = []
         orphaned_kinds: list[str] = []
         next_refresh = DEFAULT_REFRESH_INTERVAL_SECONDS
@@ -185,8 +193,8 @@ def refresh_sandbox_credentials(input: RefreshSandboxCredentialsInput) -> Refres
         if intervals:
             next_refresh = min(intervals)
 
-        if refreshed_kinds:
-            _notify_agent_server_of_refresh(ctx, task, refreshed_kinds)
+        if refreshed_kinds and task_run is not None:
+            _notify_agent_server_of_refresh(ctx, task, task_run, refreshed_kinds)
 
         track_event(
             "sandbox_credentials_refreshed",

products/tasks/backend/temporal/process_task/activities/send_followup_to_sandbox.py

@@ -1,5 +1,4 @@
 import json
-import time
 import threading
 import contextvars
 from dataclasses import dataclass
@@ -12,35 +11,24 @@
 from posthog.temporal.common.utils import close_db_connections
 from posthog.temporal.oauth import PosthogMcpScopes
 
-from products.tasks.backend.logic.services.agent_command import (
-    FOLLOWUP_TIMEOUT_SECONDS,
-    REFRESH_TIMEOUT_SECONDS,
-    CommandResult,
-    send_refresh_session,
-    send_user_message,
-)
+from products.tasks.backend.logic.services.agent_command import FOLLOWUP_TIMEOUT_SECONDS, send_user_message
 from products.tasks.backend.logic.services.connection_token import create_sandbox_connection_token
 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
 from products.tasks.backend.redis import get_tasks_stream_redis_sync, run_uses_dedicated_stream
-from products.tasks.backend.temporal.oauth import create_oauth_access_token_for_run
+from products.tasks.backend.temporal.process_task.activities.get_task_processing_context import TaskProcessingContext
+from products.tasks.backend.temporal.process_task.sandbox_identity import ensure_sandbox_identity
 from products.tasks.backend.temporal.process_task.utils import (
     get_actor_distinct_id,
-    get_sandbox_ph_mcp_configs,
     get_task_run_credential_user,
-    get_user_mcp_server_configs,
     is_slack_interaction_state,
-    mark_mcp_token_issued,
-    should_refresh_mcp_token,
 )
 
 from ee.hogai.sandbox import STOP_REASON_END_TURN, TURN_COMPLETE_METHOD
 
 logger = structlog.get_logger(__name__)
 
-REFRESH_RETRY_DELAY_SECONDS = 0.5
-
 # Retries exist for attempt-level deaths (worker restart kills the in-flight
 # attempt, detected via heartbeat timeout) and for delivery-unknown failures.
 # Application failures that write an error sentinel raise non-retryable.
@@ -57,6 +45,11 @@ class SendFollowupToSandboxInput:
     # Workflow-generated idempotency key. Stable across activity retries, so
     # the agent-server can drop a redelivery of a message it already accepted.
     message_id: str | None = None
+    # The run's processing context, needed to rebind GitHub credentials on an
+    # actor transition. Optional so activities scheduled by pre-rollout
+    # workflow histories (without it) still deserialize; they skip the GitHub
+    # surface until the run's next workflow.
+    context: TaskProcessingContext | None = None
 
 
 @activity.defn
@@ -128,10 +121,16 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
             task_run, user_id=actor_user.id, distinct_id=get_actor_distinct_id(actor_user)
         )
 
-    # 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
+    # Rebind the sandbox session (MCP OAuth token, GitHub credentials) to this
+    # message's actor before the turn runs. Non-fatal: if the rebind fails we
     # still deliver the follow-up with the existing (possibly stale) creds.
-    _refresh_sandbox_mcp(task_run, input.posthog_mcp_scopes, auth_token)
+    ensure_sandbox_identity(
+        task_run,
+        actor_user,
+        posthog_mcp_scopes=input.posthog_mcp_scopes,
+        auth_token=auth_token,
+        processing_context=input.context,
+    )
     artifacts = None
     artifact_ids = input.artifact_ids or []
     if artifact_ids:
@@ -216,92 +215,6 @@ def _deliver_followup(input: SendFollowupToSandboxInput) -> None:
         raise ApplicationError(f"send_followup failed: {error_msg}", non_retryable=True)
 
 
-def _refresh_sandbox_mcp(
-    task_run: TaskRun,
-    scopes: PosthogMcpScopes,
-    auth_token: str | 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.
-
-    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.
-    """
-    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
-
-    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)
-    except Exception as e:
-        logger.warning("refresh_mcp_token_mint_failed", run_id=run_id, error=str(e))
-        return
-
-    mcp_configs = get_sandbox_ph_mcp_configs(
-        token=access_token,
-        project_id=task_run.team_id,
-        scopes=scopes,
-        interaction_origin=(task_run.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
-
-    if not mcp_configs:
-        logger.info("refresh_mcp_skipped_no_configs", run_id=run_id)
-        return
-
-    mcp_servers = [config.to_dict() for config in mcp_configs]
-
-    result = send_refresh_session(
-        task_run,
-        mcp_servers,
-        auth_token=auth_token,
-        timeout=REFRESH_TIMEOUT_SECONDS,
-    )
-    if result.success:
-        mark_mcp_token_issued(run_id)
-        logger.info("refresh_mcp_delivered", run_id=run_id, attempts=1)
-        return
-
-    logger.info(
-        "refresh_mcp_retrying",
-        run_id=run_id,
-        error=result.error,
-        status_code=result.status_code,
-    )
-    time.sleep(REFRESH_RETRY_DELAY_SECONDS)
-    retry: CommandResult = send_refresh_session(
-        task_run,
-        mcp_servers,
-        auth_token=auth_token,
-        timeout=REFRESH_TIMEOUT_SECONDS,
-    )
-    if retry.success:
-        mark_mcp_token_issued(run_id)
-        logger.info("refresh_mcp_delivered", run_id=run_id, attempts=2)
-        return
-
-    logger.warning(
-        "refresh_mcp_failed",
-        run_id=run_id,
-        error=retry.error,
-        status_code=retry.status_code,
-    )
-
-
 def _get_stop_reason(result_data: dict[str, Any] | None) -> str:
     if not isinstance(result_data, dict):
         return STOP_REASON_END_TURN

products/tasks/backend/temporal/process_task/activities/start_agent_server.py

@@ -28,7 +28,8 @@
     get_sandbox_ph_mcp_configs,
     get_task_run_credential_user,
     get_user_mcp_server_configs,
-    mark_mcp_token_issued,
+    mark_mcp_session,
+    mark_sandbox_identity,
 )
 
 from .get_task_processing_context import TaskProcessingContext
@@ -164,6 +165,10 @@ class StartAgentServerOutput:
 @dataclass
 class _LaunchParams:
     mcp_configs: list[McpServerConfig]
+    # The user the boot-time credentials (MCP OAuth token, GitHub token) were
+    # minted for, recorded as the sandbox's initial session identities once
+    # the agent server starts.
+    actor_user_id: int | None
     agentsh_domains: list[str] | None
     protected_base_branch: str | None
     event_ingest_token: str | None
@@ -275,6 +280,7 @@ def _prepare_launch(ctx: TaskProcessingContext, scopes: PosthogMcpScopes) -> _La
 
     return _LaunchParams(
         mcp_configs=mcp_configs,
+        actor_user_id=actor_user.id if actor_user else None,
         agentsh_domains=agentsh_domains,
         protected_base_branch=protected_base_branch,
         event_ingest_token=event_ingest_token,
@@ -316,10 +322,15 @@ def _invoke_start_agent_server(
             rtk_enabled=ctx.rtk_enabled,
         )
 
-        # Mark startup-time token issuance so follow-ups within the next
-        # 30m window skip the redundant refresh.
-        if params.mcp_configs:
-            mark_mcp_token_issued(ctx.run_id)
+        # Record the sandbox's boot-time session identities and start the MCP
+        # freshness window, so follow-ups from the same actor within
+        # MCP_TOKEN_REFRESH_INTERVAL_SECONDS skip the redundant refresh.
+        # Keyed on the sandbox id: a replacement sandbox starts unmarked.
+        if params.actor_user_id is not None:
+            if params.mcp_configs:
+                mark_mcp_session(sandbox.id, params.actor_user_id)
+            if ctx.has_github_credentials:
+                mark_sandbox_identity(sandbox.id, "github", params.actor_user_id)
 
         # Persist the effective rtk posture the agent launched with, so terminal
         # analytics can cohort runs by it (the state override alone misses the