fix: make coder approvals replay-safe

Author: tylerslaton

agent/coding/repository_tools.py

@@ -30,6 +30,7 @@ class PreparedRepository:
     head_branch: str
     identity: GitHubIdentity
     pr_number: int | None
+    request_signature: tuple[Any, ...]
     approved_signature: tuple[Any, ...] | None = None
     pushed_commit: str | None = None
 
@@ -71,6 +72,20 @@ def _run_git(backend: PerJobDaytonaBackend, path: str, args: str):
     return backend.execute(f"git -C {shlex.quote(path)} {args}")
 
 
+def _prepared_result(prepared: PreparedRepository, *, replayed: bool = False) -> str:
+    status = "already_prepared" if replayed else "prepared"
+    return (
+        f"status: {status}\n"
+        f"repository: {prepared.repo}\n"
+        f"push_repository: {prepared.push_repo}\n"
+        f"working_directory: {prepared.path}\n"
+        f"base_branch: {prepared.base_branch}\n"
+        f"head_branch: {prepared.head_branch}\n"
+        f"actor: {prepared.identity.login}\n"
+        f"pr_number: {prepared.pr_number or ''}"
+    )
+
+
 def build_repository_tools(
     backend: PerJobDaytonaBackend,
     provider: GitHubCredentialProvider,
@@ -94,16 +109,29 @@ def prepare_repository(
             _validate_branch(base_branch)
         if head_branch:
             _validate_branch(head_branch)
+        if pr_number is not None and pr_number < 1:
+            raise RuntimeError("pr_number must be positive")
+
+        request_signature = (
+            repo.casefold(),
+            base_branch,
+            head_branch,
+            pr_number,
+            sync_base,
+        )
 
         state = backend.job_state()
-        if state.get("repository") is not None:
+        prior = state.get("repository")
+        if isinstance(prior, PreparedRepository):
+            if prior.request_signature == request_signature:
+                return _prepared_result(prior, replayed=True)
             raise RuntimeError("prepare_repository may be called only once per coder job")
+        if prior is not None:
+            raise RuntimeError("coder job contains invalid repository state")
 
         push_repo = repo
         existing_pr: dict[str, Any] | None = None
         if pr_number is not None:
-            if pr_number < 1:
-                raise RuntimeError("pr_number must be positive")
             existing_pr = provider.request_json(
                 "GET", f"/repos/{repo}/pulls/{pr_number}"
             )
@@ -187,18 +215,10 @@ def prepare_repository(
             head_branch=head_branch,
             identity=identity,
             pr_number=pr_number,
+            request_signature=request_signature,
         )
         state["repository"] = prepared
-        return (
-            "status: prepared\n"
-            f"repository: {repo}\n"
-            f"push_repository: {push_repo}\n"
-            f"working_directory: {path}\n"
-            f"base_branch: {base_branch}\n"
-            f"head_branch: {head_branch}\n"
-            f"actor: {identity.login}\n"
-            f"pr_number: {pr_number or ''}"
-        )
+        return _prepared_result(prepared)
 
     @tool
     def publish_changes(

agent/coding/sandbox.py

@@ -60,20 +60,24 @@
 
 
 def current_run_id() -> str:
-    """LangGraph run id, or thread id, or `unknown`."""
+    """Stable coder-job id across nested tool calls and approval resumes."""
     try:
         from langgraph.config import get_config
 
         config = get_config()
         configurable = config.get("configurable") or {}
     except Exception:
         return "unknown"
-    return str(
-        config.get("run_id")
-        or configurable.get("run_id")
-        or configurable.get("thread_id")
-        or "unknown"
-    )
+    run_id = config.get("run_id") or configurable.get("run_id")
+    if run_id:
+        return str(run_id)
+
+    thread_id = configurable.get("thread_id")
+    checkpoint_ns = str(configurable.get("checkpoint_ns") or "")
+    job_ns = checkpoint_ns.partition("|")[0]
+    if job_ns:
+        return f"{thread_id}:{job_ns}" if thread_id else job_ns
+    return str(thread_id or "unknown")
 
 
 def _raw_sandbox(box):

agent/tests/test_coder_approval_resume.py

@@ -0,0 +1,223 @@
+import asyncio
+from types import SimpleNamespace
+from typing import Any
+
+from ag_ui.core import RunAgentInput
+from copilotkit import CopilotKitMiddleware
+from deepagents import create_deep_agent
+from langchain_core.language_models import BaseChatModel
+from langchain_core.messages import AIMessage, BaseMessage, ToolMessage
+from langchain_core.outputs import ChatGeneration, ChatResult
+from langgraph.checkpoint.memory import MemorySaver
+from pydantic import Field
+
+from agui import build_agui_agent
+from coding.github_credentials import GitHubIdentity
+from coding.subagent import build_coder_subagent
+
+
+class ApprovalResumeModel(BaseChatModel):
+    tool_names: frozenset[str] = Field(default_factory=frozenset)
+
+    @property
+    def _llm_type(self):
+        return "coder-approval-resume"
+
+    def bind_tools(self, tools, **_kwargs):
+        return self.model_copy(
+            update={"tool_names": frozenset(tool.name for tool in tools)}
+        )
+
+    def _generate(
+        self,
+        messages: list[BaseMessage],
+        stop=None,
+        run_manager=None,
+        **_kwargs: Any,
+    ):
+        del stop, run_manager
+        results = {
+            message.tool_call_id
+            for message in messages
+            if isinstance(message, ToolMessage)
+        }
+        if "prepare_repository" not in self.tool_names:
+            message = (
+                AIMessage(content="done")
+                if "task-1" in results
+                else AIMessage(
+                    content="",
+                    tool_calls=[
+                        {
+                            "id": "task-1",
+                            "name": "task",
+                            "args": {
+                                "description": "Make the requested change",
+                                "subagent_type": "coder",
+                            },
+                        }
+                    ],
+                )
+            )
+        elif "prepare-1" not in results:
+            message = AIMessage(
+                content="",
+                tool_calls=[
+                    {
+                        "id": "prepare-1",
+                        "name": "prepare_repository",
+                        "args": {
+                            "repo": "org/repo",
+                            "base_branch": "main",
+                            "head_branch": "opentag/test",
+                        },
+                    }
+                ],
+            )
+        elif "publish-1" not in results:
+            message = AIMessage(
+                content="",
+                tool_calls=[
+                    {
+                        "id": "publish-1",
+                        "name": "publish_changes",
+                        "args": {
+                            "repo": "org/repo",
+                            "base_branch": "main",
+                            "head_branch": "opentag/test",
+                            "title": "Test approval resume",
+                            "body": "Test body",
+                            "test_command": "true",
+                            "test_exit_code": 0,
+                        },
+                    }
+                ],
+            )
+        else:
+            message = AIMessage(content="coder done")
+        return ChatResult(generations=[ChatGeneration(message=message)])
+
+
+class ApprovalResumeProvider:
+    git_username = "x-access-token"
+
+    def __init__(self):
+        self.requests = []
+
+    def token(self):
+        return "operation-secret"
+
+    def identity(self):
+        return GitHubIdentity(
+            "open-tag[bot]",
+            42,
+            "42+open-tag[bot]@users.noreply.github.com",
+        )
+
+    def request_json(self, method, path, *, json=None):
+        self.requests.append((method, path, json))
+        if method == "POST":
+            return {"html_url": "https://github.com/org/repo/pull/9"}
+        raise AssertionError((method, path, json))
+
+
+class ApprovalResumeBackend:
+    def __init__(self):
+        self.state = {}
+        self.branch = ""
+        self.pushes = 0
+
+    @property
+    def id(self):
+        return "approval-resume"
+
+    def job_state(self):
+        return self.state
+
+    def clone_repository(self, **kwargs):
+        self.branch = kwargs["branch"]
+
+    def set_git_identity(self, **_kwargs):
+        pass
+
+    def push_repository(self, **_kwargs):
+        self.pushes += 1
+
+    def execute(self, command, **_kwargs):
+        if "switch -c" in command:
+            self.branch = "opentag/test"
+            return SimpleNamespace(output="", exit_code=0)
+        if "branch --show-current" in command:
+            return SimpleNamespace(output=self.branch, exit_code=0)
+        if "status --porcelain" in command:
+            return SimpleNamespace(output="", exit_code=0)
+        if "rev-parse HEAD" in command:
+            return SimpleNamespace(output="abc123", exit_code=0)
+        raise AssertionError(command)
+
+    def stop_current(self):
+        pass
+
+
+def test_coder_confirmation_survives_subagent_tool_replay():
+    model = ApprovalResumeModel()
+    checkpointer = MemorySaver()
+    backend = ApprovalResumeBackend()
+    provider = ApprovalResumeProvider()
+    coder = build_coder_subagent(
+        model=model,
+        checkpointer=checkpointer,
+        provider=provider,
+        backend=backend,
+    )
+    graph = create_deep_agent(
+        model=model,
+        middleware=[CopilotKitMiddleware()],
+        subagents=[coder],
+        checkpointer=checkpointer,
+    )
+    agent = build_agui_agent(graph, recursion_limit=80)
+    request = {
+        "threadId": "approval-resume-thread",
+        "state": {},
+        "messages": [{"id": "user-1", "role": "user", "content": "go"}],
+        "tools": [],
+        "context": [],
+    }
+
+    first = asyncio.run(
+        _collect(
+            agent.run(
+                RunAgentInput(runId="run-1", forwardedProps={}, **request)
+            )
+        )
+    )
+    assert any(getattr(event, "name", None) == "on_interrupt" for event in first)
+
+    # A nested interrupt can replay the parent task if its subgraph checkpoint
+    # is unavailable. The prepared sandbox state still survives that replay.
+    namespaces = checkpointer.storage["approval-resume-thread"]
+    for namespace in list(namespaces):
+        if namespace:
+            del namespaces[namespace]
+
+    asyncio.run(
+        _collect(
+            agent.run(
+                RunAgentInput(
+                    runId="run-2",
+                    forwardedProps={
+                        "command": {"resume": {"confirmed": True}}
+                    },
+                    **request,
+                )
+            )
+        )
+    )
+
+    assert backend.pushes == 1
+    assert provider.requests[-1][:2] == ("POST", "/repos/org/repo/pulls")
+
+
+async def _collect(stream):
+    return [event async for event in stream]

agent/tests/test_repository_tools.py

@@ -134,6 +134,26 @@ def test_prepare_clones_with_operation_token_and_configures_local_identity():
     assert backend.branch == "opentag/fix"
 
 
+def test_identical_prepare_replay_reuses_the_existing_repository():
+    prepare, _publish_tool, backend, _provider = _tools()
+
+    _prepare(prepare)
+    result = _prepare(prepare)
+
+    assert "status: already_prepared" in result
+    assert len(backend.clone_calls) == 1
+
+
+def test_prepare_replay_cannot_change_the_target():
+    prepare, _publish_tool, backend, _provider = _tools()
+
+    _prepare(prepare)
+
+    with pytest.raises(RuntimeError, match="may be called only once"):
+        _prepare(prepare, head_branch="opentag/other")
+    assert len(backend.clone_calls) == 1
+
+
 def test_prepare_existing_fork_pr_and_syncs_base_through_daytona():
     pr = {
         "base": {"ref": "main", "repo": {"full_name": "org/repo"}},

agent/tests/test_sandbox.py

@@ -66,6 +66,24 @@ def test_current_run_id_falls_back_to_thread_id(monkeypatch):
     assert current_run_id() == "thread-a"
 
 
+def test_current_run_id_uses_the_parent_task_namespace_for_a_coder_job(monkeypatch):
+    namespace = {"value": "tools:job-a|tools:prepare"}
+    monkeypatch.setattr(
+        "langgraph.config.get_config",
+        lambda: {
+            "configurable": {
+                "thread_id": "thread-a",
+                "checkpoint_ns": namespace["value"],
+            }
+        },
+    )
+
+    prepared_job = current_run_id()
+    namespace["value"] = "tools:job-a|tools:publish"
+
+    assert current_run_id() == prepared_job == "thread-a:tools:job-a"
+
+
 def test_redact_secrets_strips_github_token_prefixes():
     text = "auth ghp_ABCDEFG123 github_pat_ZZ gho_YY"
     redacted = redact_secrets(text)