refactor(code): run /offload as a server operation

libs/code/deepagents_code/agent.py

@@ -2842,6 +2842,9 @@ def _subagent_cli_middleware(
         )
 
     compaction_middleware = _create_cli_compaction_middleware(model, composite_backend)
+    # The dedicated server-side `/offload` graph reuses this exact middleware
+    # through the composite backend, avoiding a second summarizer/backend setup.
+    composite_backend._cli_compaction_middleware = compaction_middleware  # ty: ignore[unresolved-attribute]
     if auto_mode_config is not None and resolved_interrupt_on is not None:
         from deepagents_code.auto_mode import AutoModeHITLMiddleware
 
@@ -2872,9 +2875,14 @@ 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
         )
+        # The dedicated `/offload` graph has no model/tool nodes, so retain the
+        # same lifecycle implementation on the shared backend for its explicit
+        # in-memory `PreCompact` dispatch.
+        composite_backend._cli_server_hooks_middleware = server_hooks_middleware  # ty: ignore[unresolved-attribute]
+        agent_middleware.append(server_hooks_middleware)
 
     if fs_tools is not None:
         # `fs_tools` is an explicit allowlist here (`--allow-fs-tools all` and an

libs/code/deepagents_code/app.py

@@ -14530,13 +14530,16 @@ async def _get_conversation_token_count(self) -> int | None:
     async def _handle_offload(self) -> None:
         """Offload older messages to free context window space.
 
-        Runs offload SERVER-SIDE by driving the agent's own
-        `compact_conversation` tool (with `force=True`) instead of
-        reimplementing summarization + persistence client-side. This keeps the
-        offloaded archive in the agent's composite backend so it is readable
-        via `read_file` in every run mode (server, sandbox, in-process). The
-        client only seeds the tool call, approves the resulting HITL interrupt,
-        drains the run, and renders the persisted `_summarization_event`.
+        Server-backed agents run the dedicated `offload` operation graph (no
+        model node, no synthetic tool call); local in-process `Pregel` agents
+        drive the agent's own `compact_conversation` tool (with `force=True`)
+        via a seeded tool call instead. Either way the offloaded archive lands
+        in the agent's composite backend so it is readable via `read_file` in
+        every run mode (server, sandbox, in-process). For the seeded path, the
+        Auto-mode HITL middleware recognizes the seeded call from the run
+        context and lets it through without an approval round trip; if a HITL
+        interrupt still surfaces (e.g. a graph without that middleware), the
+        driver approves the seed itself as a fail-closed fallback.
         """
         from langchain_core.messages.utils import count_tokens_approximately
 
@@ -14580,14 +14583,23 @@ async def _handle_offload(self) -> None:
                 _effective_conversation(before_messages, prior_event)
             )
 
+            # Local `Pregel` agents have no server operation graph, so they
+            # drive the seeded `compact_conversation` tool call in-process
+            # instead.
+            local_seeded = self._remote_agent() is None
             # Own the seeded tool-call id here so a failed run can clean up the
             # committed-but-unanswered seed (see `_remove_unanswered_offload_seed`).
-            seed_tool_call_id = str(uuid.uuid4())
+            seed_tool_call_id = str(uuid.uuid4()) if local_seeded else None
 
             try:
-                tool_error = await self._drive_server_side_compaction(
-                    config, seed_tool_call_id
-                )
+                if seed_tool_call_id is not None:
+                    tool_error = await self._drive_legacy_seeded_compaction(
+                        config, seed_tool_call_id
+                    )
+                else:
+                    tool_error = await self._drive_offload_operation_graph(
+                        config, state_values
+                    )
             except ClientHookStopError:
                 return
             except Exception as stream_error:
@@ -14605,19 +14617,26 @@ async def _handle_offload(self) -> None:
                         "Failed to reconcile state after offload stream error",
                         exc_info=True,
                     )
-                    if not await self._remove_unanswered_offload_seed(
-                        config, seed_tool_call_id
+                    if (
+                        seed_tool_call_id is not None
+                        and not await self._remove_unanswered_offload_seed(
+                            config, seed_tool_call_id
+                        )
                     ):
                         await self._mount_message(ErrorMessage(_OFFLOAD_WEDGE_WARNING))
                     raise stream_error from state_error
                 reconciled_event = new_state.get("_summarization_event")
                 if _summarization_cutoff(reconciled_event) <= prior_cutoff:
-                    # Compaction did not commit, so the seeded tool call was
-                    # never answered. Remove it before re-raising so a failed
-                    # `/offload` cannot wedge the thread with a dangling
+                    # Compaction did not commit. The seeded driver may have left
+                    # its tool call unanswered; remove it before re-raising so a
+                    # failed `/offload` cannot wedge the thread with a dangling
                     # `tool_use` that the model API rejects on the next turn.
-                    if not await self._remove_unanswered_offload_seed(
-                        config, seed_tool_call_id
+                    # The operation graph commits no seed to clean up.
+                    if (
+                        seed_tool_call_id is not None
+                        and not await self._remove_unanswered_offload_seed(
+                            config, seed_tool_call_id
+                        )
                     ):
                         await self._mount_message(ErrorMessage(_OFFLOAD_WEDGE_WARNING))
                     raise
@@ -14651,12 +14670,15 @@ async def _handle_offload(self) -> None:
                 if failure is not None:
                     await self._mount_message(ErrorMessage(failure))
                     return
-                # A no-op still commits the synthetic assistant seed and its
-                # tool result. Restore the exact pre-run conversation so an
-                # operation reported as doing nothing truly changes nothing.
-                await self._remove_offload_artifacts(
-                    config, current_messages, prior_event
-                )
+                if local_seeded:
+                    # A no-op seeded run still commits the synthetic assistant
+                    # seed and its tool result. Restore the exact pre-run
+                    # conversation so an operation reported as doing nothing
+                    # truly changes nothing. The operation graph commits no
+                    # such artifacts.
+                    await self._remove_offload_artifacts(
+                        config, current_messages, prior_event
+                    )
                 # `force=True` bypasses the eligibility gate, so this branch is
                 # reached when there is nothing older than the retention window
                 # to summarize (effective cutoff 0). It also absorbs the
@@ -14759,17 +14781,84 @@ async def _handle_offload(self) -> None:
             except Exception:  # best-effort spinner cleanup
                 logger.exception("Failed to dismiss spinner after offload")
 
-    async def _drive_server_side_compaction(
+    async def _drive_offload_operation_graph(
+        self, config: RunnableConfig, state_values: dict[str, Any]
+    ) -> str | None:
+        """Run the explicit server-side `/offload` operation graph.
+
+        The operation graph shares the interactive agent's checkpoint and
+        composite backend but has no model node or HITL middleware. The slash
+        command is the user's authorization, so it persists only the resulting
+        summarization event and cannot create a synthetic assistant tool call.
+
+        Args:
+            config: Config with `configurable.thread_id`.
+            state_values: Current thread state, replayed as the run input. A
+                bare `{}` would *overwrite* the checkpointed message list with
+                an empty one before the node reads it (and `astream(None)`
+                never starts a run on `RemoteAgent`). `_summarization_event` is
+                stripped: its embedded summary message is a private-shape dict
+                the server cannot deserialize, and the event survives through
+                the checkpoint channels without being replayed.
+
+        Returns:
+            `None` always: the graph has no model or tool nodes, so failure
+                surfaces only as a raised stream error, which the caller
+                reconciles against the checkpointed `_summarization_event`.
+                The `str | None` shape matches the seeded driver so
+                `_handle_offload` handles both paths uniformly.
+
+        Raises:
+            RuntimeError: If the app is not connected to its server graph.
+        """
+        from deepagents_code.config import settings
+
+        agent = self._agent
+        if agent is None:
+            return None
+        remote = self._remote_agent()
+        if remote is None:
+            msg = "The explicit /offload operation requires the server graph."
+            raise RuntimeError(msg)
+        await remote.aensure_thread(dict(config))
+
+        stream_context = CLIContext(
+            model=self._effective_model_spec(),
+            model_params=self._model_params_override or {},
+            profile_overrides=self._profile_override or {},
+            model_context_limit=settings.model_context_limit,
+            thread_id=self._lc_thread_id,
+        )
+        self._hooks.apply_graph_context(stream_context)
+        stream_input = {
+            key: value
+            for key, value in state_values.items()
+            if key != "_summarization_event"
+        }
+        async for _chunk in remote.for_graph("offload").astream(
+            cast("Any", stream_input),
+            stream_mode=["updates"],
+            config=config,
+            context=stream_context,
+            durability="exit",
+        ):
+            pass
+        return None
+
+    async def _drive_legacy_seeded_compaction(
         self, config: RunnableConfig, seed_tool_call_id: str | None = None
     ) -> str | None:
         """Trigger the server-side `compact_conversation` tool with `force=True`.
 
         Seeds an assistant `compact_conversation` tool call attributed to the
         model node, then advances the graph so the agent's own `ToolNode`
-        executes the tool. The tool is HITL-gated, so `astream(None)` surfaces
-        an approval interrupt; only the first forced `compact_conversation`
-        request is approved here (this is an explicit user-initiated
-        `/offload`). The runtime context carries the seeded call ID so the
+        executes the tool. The tool is HITL-gated; on the normal path the
+        Auto-mode HITL middleware recognizes the seeded call from the run
+        context and lets it through without interrupting. If an approval
+        interrupt still surfaces (e.g. a graph without that middleware),
+        only the first forced `compact_conversation` request is approved
+        here (this is an explicit user-initiated `/offload`). The runtime
+        context carries the seeded call ID so the
         compaction middleware can reject every other tool independently of
         HITL configuration, including tools requested by the trailing model
         turn.

libs/code/deepagents_code/client/launch/server.py

@@ -181,7 +181,10 @@ def generate_langgraph_json(
     """
     config: dict[str, Any] = {
         "dependencies": ["."],
-        "graphs": {"agent": graph_ref},
+        "graphs": {
+            "agent": graph_ref,
+            "offload": "deepagents_code.server_graph:make_offload_graph",
+        },
     }
     if env_file:
         config["env"] = env_file

libs/code/deepagents_code/client/remote_client.py

@@ -154,6 +154,22 @@ def _get_graph(self) -> Any:  # noqa: ANN401
             )
         return self._graph
 
+    def for_graph(self, graph_name: str) -> RemoteAgent:
+        """Return a client for another graph served by the same runtime.
+
+        Args:
+            graph_name: Registered LangGraph graph name.
+
+        Returns:
+            A fresh client that preserves this connection's URL and credentials.
+        """
+        return RemoteAgent(
+            self._url,
+            graph_name=graph_name,
+            api_key=self._api_key,
+            headers=self._headers,
+        )
+
     async def astream(
         self,
         input: dict | Any,  # noqa: A002, ANN401