feat(code): make /offload server-owned

libs/code/deepagents_code/_cli_context.py

@@ -70,8 +70,6 @@ class CLIContextSchema:
 
     turn_id: str | None = None
 
-    offload_tool_call_id: str | None = None
-
     hooks_snapshot_id: str | None = None
 
     hooks_server_events: list[str] = field(default_factory=list)
@@ -139,13 +137,6 @@ class CLIContext(TypedDict, total=False):
     turn_id: str | None
     """Current user-turn ID for binding trusted interactive responses."""
 
-    offload_tool_call_id: str | None
-    """The sole tool-call ID authorized during a server-driven `/offload` run.
-
-    This is set by the client, not graph state, so model-generated calls cannot
-    grant themselves permission to execute during the hidden compaction turn.
-    """
-
     hooks_snapshot_id: str | None
     """Canonical Hooks v2 configuration hash for this session.
 

libs/code/deepagents_code/_testing_models.py

@@ -20,6 +20,17 @@
     from langchain_core.callbacks import CallbackManagerForLLMRun
 
 
+DCA_TEST_OFFLOAD_GATE_ENV = "DCA_TEST_OFFLOAD_GATE_DIR"
+"""Env var pointing at a directory used to gate summary generation.
+
+When set, a summary request writes `<dir>/entered` and then polls for
+`<dir>/release` before replying. File-based so the test process can hold the
+server's compaction model call open without sharing Python state across the
+server subprocess boundary. Only summary prompts are gated; ordinary turns pass
+through, which is what lets a test launch a concurrent run *while* `/offload`
+is blocked here.
+"""
+
 # Prompt markers that drive `ToolCallingIntegrationChatModel`. Each marker is the
 # full token (including the trailing `=`); the file path follows on the same line,
 # e.g. `DCA_TEST_WRITE_FILE=/tmp/out.txt`. These are the single source of truth
@@ -94,6 +105,7 @@ def _generate(
             if (text := self._stringify_message(message)).strip()
         )
         if self._looks_like_summary_request(prompt):
+            self._wait_at_summary_gate()
             content = "integration summary"
         else:
             excerpt = " ".join(prompt.split()[-18:])
@@ -103,14 +115,58 @@ def _generate(
                 content = "integration reply"
 
         return ChatResult(
-            generations=[ChatGeneration(message=AIMessage(content=content))]
+            generations=[
+                ChatGeneration(
+                    message=AIMessage(
+                        content=content,
+                        usage_metadata={
+                            "input_tokens": 100,
+                            "output_tokens": 20,
+                            "total_tokens": 120,
+                        },
+                    )
+                )
+            ]
         )
 
     @property
     def _llm_type(self) -> str:
         """LangChain model type identifier."""
         return "deterministic-integration"
 
+    @staticmethod
+    def _wait_at_summary_gate() -> None:
+        """Hold the summary call open until the test releases it.
+
+        No-op unless `DCA_TEST_OFFLOAD_GATE_DIR` names a directory. When set,
+        write `<dir>/entered` (the test's signal that the offload operation is
+        mid-summary) and then poll for `<dir>/release`. Every summary request
+        rewrites the marker, so the test reads it as "a summary is in flight"
+        rather than "the first summary started". Bounded so a crashed test
+        cannot wedge the server subprocess indefinitely.
+
+        Raises:
+            TimeoutError: If the gate is not released within 120 seconds.
+        """
+        import os
+        import time
+        from pathlib import Path
+
+        gate_dir = os.environ.get(DCA_TEST_OFFLOAD_GATE_ENV)
+        if not gate_dir:
+            return
+        gate = Path(gate_dir)
+        (gate / "entered").write_text("1")
+        deadline = time.monotonic() + 120
+        while not (gate / "release").exists():
+            if time.monotonic() > deadline:
+                msg = (
+                    "Offload test gate was never released; refusing to block "
+                    "the server summary call forever."
+                )
+                raise TimeoutError(msg)
+            time.sleep(0.05)
+
     @staticmethod
     def _stringify_message(message: BaseMessage) -> str:
         """Flatten message content into plain text for deterministic responses.

libs/code/deepagents_code/agent.py

@@ -100,7 +100,11 @@
     _artifacts_root,
     _offload_fallback_root,
 )
-from deepagents_code.offload_middleware import _create_cli_compaction_middleware
+from deepagents_code.offload_middleware import (
+    OffloadOperation,
+    _create_cli_compaction_middleware,
+    attach_offload_operation,
+)
 from deepagents_code.plugins.adapters.skills_middleware import PluginSkillsMiddleware
 from deepagents_code.project_utils import ProjectContext, get_server_project_context
 from deepagents_code.reliable_rubric import ReliableRubricMiddleware
@@ -116,6 +120,7 @@
 
 logger = logging.getLogger(__name__)
 
+
 _MEMORY_READONLY_SYSTEM_PROMPT = (
     "<agent_memory>\n"
     "{agent_memory}\n\n"
@@ -2943,7 +2948,7 @@ def _subagent_cli_middleware(
         trusted_root, narrow_allow_list = auto_mode_config
         # An explicit argument wins; otherwise the env var / `config.toml`
         # preference is read here, where agent construction already runs off the
-        # blockbuster-guarded server loop (see `server_graph._make_graph`).
+        # blockbuster-guarded server loop (see `server_graph._make_graphs`).
         classifier_model = (
             auto_classifier_model
             if auto_classifier_model is not None
@@ -2973,7 +2978,16 @@ def _subagent_cli_middleware(
     from deepagents_code.hooks.server_middleware import ServerHooksMiddleware
 
     hooks_cwd = Path(effective_cwd) if effective_cwd is not None else Path.cwd()
-    agent_middleware.append(ServerHooksMiddleware(cwd=hooks_cwd, mcp_tools=mcp_tools))
+    server_hooks_middleware = ServerHooksMiddleware(cwd=hooks_cwd, mcp_tools=mcp_tools)
+    agent_middleware.append(server_hooks_middleware)
+
+    # Publish the server operation on the backend shared with `server_graph`.
+    # The custom HTTP route owns checkpoint access and persistence, while this
+    # object retains the exact compaction and hook instances used by the agent.
+    attach_offload_operation(
+        composite_backend,
+        OffloadOperation(compaction_middleware, server_hooks_middleware),
+    )
 
     if fs_tools is not None:
         # `fs_tools` is an explicit allowlist here (`--allow-fs-tools all` and an