mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-21 17:25:26 -04:00
fix(code): run /offload server-side (#4696)
`/offload` now stores archived conversation history through the agent's backend. --- `/offload` previously summarized and saved conversation history in the client process. In server and sandbox modes, that process does not own the backend used by the agent: persistence could fail against a read-only filesystem, and even a successfully written archive would not be available to the agent through `read_file`. This changes Deep Agents Code to run the existing `compact_conversation` tool through the active agent instead. The command seeds a tool call into the thread, approves the expected human-in-the-loop interrupt because the user explicitly requested `/offload`, resumes the graph, and reads the persisted summarization event back from server state. The archive is therefore written through the agent's composite backend and remains readable by the agent in local, server, and sandbox runs. The old client-side `perform_offload` helper (and its `OffloadResult` / `OffloadThresholdNotMet` / `OffloadModelError` result types) is removed, as summarization and persistence now run through the agent. In local mode, the `conversation_history` backend now roots under `~/.deepagents` (via `_offload_fallback_root`, with a hardened private-temp fallback when the home directory is not writable) instead of a throwaway `tempfile.mkdtemp` directory, so offloaded history persists across sessions. The SDK's `compact_conversation` API is unchanged; this PR is confined to Deep Agents Code. Made by [Open SWE](https://openswe.vercel.app/agents/1cbc308e-b411-2a09-bd6b-541e606ca4c5) --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
@@ -35,6 +35,10 @@ class CLIContextSchema:
|
||||
|
||||
model_params: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
profile_overrides: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
model_context_limit: int | None = None
|
||||
|
||||
auto_approve: bool = False
|
||||
|
||||
approval_mode_key: str | None = None
|
||||
@@ -43,6 +47,8 @@ class CLIContextSchema:
|
||||
|
||||
blocked_goal_retry_context: str | None = None
|
||||
|
||||
offload_tool_call_id: str | None = None
|
||||
|
||||
|
||||
class CLIContext(TypedDict, total=False):
|
||||
"""Client-facing builder for the per-run graph context payload.
|
||||
@@ -61,6 +67,12 @@ class CLIContext(TypedDict, total=False):
|
||||
"""Invocation params (e.g. `temperature`, `max_tokens`) to merge
|
||||
into `model_settings`."""
|
||||
|
||||
profile_overrides: dict[str, Any]
|
||||
"""Model profile metadata supplied by `--profile-override`."""
|
||||
|
||||
model_context_limit: int | None
|
||||
"""Effective context-window limit for profile-aware middleware."""
|
||||
|
||||
auto_approve: bool
|
||||
"""Whether gated tool calls should skip the human-approval interrupt.
|
||||
|
||||
@@ -94,3 +106,10 @@ class CLIContext(TypedDict, total=False):
|
||||
message so it is not parsed as a file mention or checkpointed as human
|
||||
input.
|
||||
"""
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
@@ -86,6 +86,8 @@ from deepagents_code.local_context import (
|
||||
_AsyncExecutableBackend,
|
||||
_ExecutableBackend,
|
||||
)
|
||||
from deepagents_code.offload import _offload_fallback_root
|
||||
from deepagents_code.offload_middleware import _create_cli_compaction_middleware
|
||||
from deepagents_code.project_utils import ProjectContext, get_server_project_context
|
||||
from deepagents_code.reliable_rubric import ReliableRubricMiddleware
|
||||
from deepagents_code.subagents import list_subagents
|
||||
@@ -1916,7 +1918,7 @@ def create_cli_agent(
|
||||
virtual_mode=True,
|
||||
)
|
||||
conversation_history_backend = FilesystemBackend(
|
||||
root_dir=tempfile.mkdtemp(prefix="deepagents_conversation_history_"),
|
||||
root_dir=_offload_fallback_root() / "conversation_history",
|
||||
virtual_mode=True,
|
||||
)
|
||||
composite_backend = CompositeBackend(
|
||||
@@ -1933,11 +1935,7 @@ def create_cli_agent(
|
||||
routes={},
|
||||
)
|
||||
|
||||
from deepagents.middleware.summarization import create_summarization_tool_middleware
|
||||
|
||||
agent_middleware.append(
|
||||
create_summarization_tool_middleware(model, composite_backend)
|
||||
)
|
||||
agent_middleware.append(_create_cli_compaction_middleware(model, composite_backend))
|
||||
|
||||
# Rubric-driven self-evaluation. The middleware is a no-op until a
|
||||
# `rubric` is supplied on invocation state, so installing it is safe.
|
||||
|
||||
@@ -288,6 +288,170 @@ def _warn_discarded_goal_channels(state_values: dict[str, Any]) -> list[str]:
|
||||
return discarded
|
||||
|
||||
|
||||
_OFFLOAD_WEDGE_WARNING = (
|
||||
"Offload failed and the conversation may be left in an inconsistent state "
|
||||
"(a compaction request could not be cleaned up). If your next message "
|
||||
"errors, start a new thread."
|
||||
)
|
||||
"""Shown when a failed `/offload` could not remove its unanswered seed.
|
||||
|
||||
A dangling `compact_conversation` tool call the model API later rejects would
|
||||
otherwise wedge the thread with only a log warning; surfacing this tells the
|
||||
user why an unrelated next turn might fail and how to recover.
|
||||
"""
|
||||
|
||||
|
||||
def _summarization_cutoff(event: Any) -> int: # noqa: ANN401
|
||||
"""Return the absolute cutoff index of a `_summarization_event`.
|
||||
|
||||
Args:
|
||||
event: A `_summarization_event` mapping (as persisted in state), or
|
||||
`None`.
|
||||
|
||||
Returns:
|
||||
The `cutoff_index`, or `0` when the event is missing or malformed.
|
||||
"""
|
||||
if isinstance(event, dict):
|
||||
cutoff = event.get("cutoff_index")
|
||||
if isinstance(cutoff, int):
|
||||
return cutoff
|
||||
return 0
|
||||
|
||||
|
||||
def _effective_conversation(messages: list[Any], event: Any) -> list[Any]: # noqa: ANN401
|
||||
"""Reconstruct the effective conversation the model would see.
|
||||
|
||||
A hardened local variant of
|
||||
`SummarizationMiddleware._apply_event_to_messages`, kept in the client
|
||||
because it runs against possibly-malformed remote-snapshot dicts and must
|
||||
degrade gracefully (a `None` summary or non-int cutoff returns the full
|
||||
list) rather than raise or emit a `None`-led list. Like the SDK method,
|
||||
when a prior summarization event exists the effective conversation is the
|
||||
summary message followed by the messages from `cutoff_index` onward, and it
|
||||
works on both LangChain message objects and serialized dicts since it only
|
||||
slices and prepends.
|
||||
|
||||
Args:
|
||||
messages: Full message list from state.
|
||||
event: The `_summarization_event` mapping, or `None`.
|
||||
|
||||
Returns:
|
||||
The effective message list.
|
||||
"""
|
||||
if not isinstance(event, dict):
|
||||
return list(messages)
|
||||
summary = event.get("summary_message")
|
||||
cutoff = event.get("cutoff_index")
|
||||
if summary is None or not isinstance(cutoff, int):
|
||||
return list(messages)
|
||||
if cutoff > len(messages):
|
||||
return [summary]
|
||||
return [summary, *messages[cutoff:]]
|
||||
|
||||
|
||||
def _message_text(msg: Any) -> str: # noqa: ANN401
|
||||
"""Extract the text content of a message object or serialized dict.
|
||||
|
||||
Handles the shapes `/offload` sees across the LangGraph server boundary:
|
||||
a message object with `.content`, or a serialized dict with `"content"`.
|
||||
A string content is returned as-is; a list of content blocks has its text
|
||||
parts concatenated (so a `ToolMessage` whose content is a block list is not
|
||||
stringified to `"[{...}]"`, which would defeat prefix matching).
|
||||
|
||||
Args:
|
||||
msg: A message object or serialized message dict.
|
||||
|
||||
Returns:
|
||||
The concatenated text content, or an empty string when there is none.
|
||||
"""
|
||||
content = (
|
||||
msg.get("content") if isinstance(msg, dict) else getattr(msg, "content", "")
|
||||
)
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list):
|
||||
parts: list[str] = []
|
||||
for block in content:
|
||||
if isinstance(block, str):
|
||||
parts.append(block)
|
||||
elif isinstance(block, dict) and isinstance(block.get("text"), str):
|
||||
parts.append(block["text"])
|
||||
return "".join(parts)
|
||||
return "" if content is None else str(content)
|
||||
|
||||
|
||||
def _is_tool_message(msg: Any) -> bool: # noqa: ANN401
|
||||
"""Return whether `msg` is a tool message in object or serialized form."""
|
||||
if isinstance(msg, dict):
|
||||
return msg.get("type") == "tool" or msg.get("role") == "tool"
|
||||
from langchain_core.messages import ToolMessage
|
||||
|
||||
# `isinstance` (not a by-name check) so `ToolMessage` subclasses still match.
|
||||
return isinstance(msg, ToolMessage)
|
||||
|
||||
|
||||
def _find_compaction_failure(messages: list[Any]) -> str | None:
|
||||
"""Return a persisted forced-compaction failure message, if present.
|
||||
|
||||
`/offload` primarily detects tool failures from the live message stream,
|
||||
but a stream hiccup (or an update-injected `ToolMessage` that never surfaces
|
||||
on the `messages` stream) can drop that signal even though the failure
|
||||
`ToolMessage` still lands in durable state. Scanning committed state closes
|
||||
that gap so a genuine failure is not misreported as "nothing to offload".
|
||||
|
||||
The caller passes only the messages produced by the *current* `/offload`
|
||||
attempt (the tail after the pre-seed prefix). This matters because the
|
||||
failure prefix is shared with the SDK's own compaction-failure wording, so
|
||||
an unbounded scan could match a stale failure from an unrelated prior turn;
|
||||
slicing to the current attempt keeps detection specific to this run.
|
||||
|
||||
Args:
|
||||
messages: The messages produced by this `/offload` attempt (objects or
|
||||
serialized dicts), i.e. committed state beyond the pre-seed prefix.
|
||||
|
||||
Returns:
|
||||
The failure message text, or `None` if no failure marker is found.
|
||||
"""
|
||||
from deepagents_code.offload_middleware import COMPACTION_FAILURE_PREFIX
|
||||
|
||||
for msg in reversed(messages):
|
||||
if not _is_tool_message(msg):
|
||||
continue
|
||||
text = _message_text(msg)
|
||||
if text.startswith(COMPACTION_FAILURE_PREFIX):
|
||||
return text
|
||||
return None
|
||||
|
||||
|
||||
def _message_id(msg: Any) -> str | None: # noqa: ANN401
|
||||
"""Return a message's id from object or serialized-dict form."""
|
||||
return msg.get("id") if isinstance(msg, dict) else getattr(msg, "id", None)
|
||||
|
||||
|
||||
def _message_tool_call_id(msg: Any) -> str | None: # noqa: ANN401
|
||||
"""Return the `tool_call_id` a tool message answers, if any."""
|
||||
return (
|
||||
msg.get("tool_call_id")
|
||||
if isinstance(msg, dict)
|
||||
else getattr(msg, "tool_call_id", None)
|
||||
)
|
||||
|
||||
|
||||
def _message_tool_call_ids(msg: Any) -> list[str]: # noqa: ANN401
|
||||
"""Return the ids of tool calls requested by a message (object or dict)."""
|
||||
tool_calls = (
|
||||
msg.get("tool_calls")
|
||||
if isinstance(msg, dict)
|
||||
else getattr(msg, "tool_calls", None)
|
||||
)
|
||||
ids: list[str] = []
|
||||
for call in tool_calls or []:
|
||||
cid = call.get("id") if isinstance(call, dict) else getattr(call, "id", None)
|
||||
if isinstance(cid, str):
|
||||
ids.append(cid)
|
||||
return ids
|
||||
|
||||
|
||||
def _create_model_with_deepagents_import_lock(
|
||||
model_spec: str | None = None,
|
||||
*,
|
||||
@@ -11425,13 +11589,17 @@ class DeepAgentsApp(App):
|
||||
return None
|
||||
|
||||
async def _handle_offload(self) -> None:
|
||||
"""Offload older messages to free context window space."""
|
||||
from deepagents_code.config import settings
|
||||
from deepagents_code.offload import (
|
||||
OffloadModelError,
|
||||
OffloadThresholdNotMet,
|
||||
perform_offload,
|
||||
)
|
||||
"""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`.
|
||||
"""
|
||||
from langchain_core.messages.utils import count_tokens_approximately
|
||||
|
||||
if not self._agent or not self._lc_thread_id:
|
||||
await self._mount_message(
|
||||
@@ -11469,83 +11637,173 @@ class DeepAgentsApp(App):
|
||||
await dispatch_hook("context.compact", {})
|
||||
await self._set_spinner("Offloading")
|
||||
|
||||
result = await perform_offload(
|
||||
messages=state_values.get("messages", []),
|
||||
prior_event=state_values.get("_summarization_event"),
|
||||
thread_id=self._lc_thread_id,
|
||||
model_spec=(f"{settings.model_provider}:{settings.model_name}"),
|
||||
profile_overrides=self._profile_override,
|
||||
context_limit=settings.model_context_limit,
|
||||
total_context_tokens=self._context_tokens,
|
||||
backend=self._backend,
|
||||
prior_event = state_values.get("_summarization_event")
|
||||
before_messages = state_values.get("messages", [])
|
||||
prior_cutoff = _summarization_cutoff(prior_event)
|
||||
tokens_before = count_tokens_approximately(
|
||||
_effective_conversation(before_messages, prior_event)
|
||||
)
|
||||
|
||||
if isinstance(result, OffloadThresholdNotMet):
|
||||
conv_str = format_token_count(result.conversation_tokens)
|
||||
if (
|
||||
result.total_context_tokens > 0
|
||||
and result.context_limit is not None
|
||||
and result.total_context_tokens > result.context_limit
|
||||
):
|
||||
total_str = format_token_count(
|
||||
result.total_context_tokens,
|
||||
)
|
||||
await self._mount_message(
|
||||
AppMessage(
|
||||
f"Offload threshold not met \u2014 conversation "
|
||||
f"is only ~{conv_str} tokens.\n\n"
|
||||
f"The remaining context "
|
||||
f"({total_str} tokens) is system overhead "
|
||||
f"that can't be offloaded.\n\n"
|
||||
f"Use /tokens for a full breakdown.",
|
||||
),
|
||||
)
|
||||
else:
|
||||
await self._mount_message(
|
||||
AppMessage(
|
||||
f"Offload threshold not met \u2014 conversation "
|
||||
f"(~{conv_str} tokens) is within the "
|
||||
f"retention budget "
|
||||
f"({result.budget_str}).\n\n"
|
||||
f"Use /tokens for a full breakdown.",
|
||||
),
|
||||
# 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())
|
||||
|
||||
try:
|
||||
tool_error = await self._drive_server_side_compaction(
|
||||
config, seed_tool_call_id
|
||||
)
|
||||
except Exception as stream_error:
|
||||
# A server graph can checkpoint the tool-node update before a
|
||||
# later stream transport failure reaches this client. Reconcile
|
||||
# the durable event before reporting the operation as failed.
|
||||
logger.warning(
|
||||
"Offload stream failed; checking for committed compaction state",
|
||||
exc_info=True,
|
||||
)
|
||||
try:
|
||||
new_state = await self._get_thread_state_values(self._lc_thread_id)
|
||||
except Exception as state_error:
|
||||
logger.warning(
|
||||
"Failed to reconcile state after offload stream error",
|
||||
exc_info=True,
|
||||
)
|
||||
if 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
|
||||
# `tool_use` that the model API rejects on the next turn.
|
||||
if not await self._remove_unanswered_offload_seed(
|
||||
config, seed_tool_call_id
|
||||
):
|
||||
await self._mount_message(ErrorMessage(_OFFLOAD_WEDGE_WARNING))
|
||||
raise
|
||||
else:
|
||||
if tool_error is not None:
|
||||
await self._mount_message(ErrorMessage(tool_error))
|
||||
return
|
||||
|
||||
# Read the persisted result back so the UI reflects server state
|
||||
# (the archive now lives in the agent's own backend, not a
|
||||
# client-local directory the server can never read).
|
||||
new_state = await self._get_thread_state_values(self._lc_thread_id)
|
||||
new_event = new_state.get("_summarization_event")
|
||||
new_cutoff = _summarization_cutoff(new_event)
|
||||
|
||||
if new_event is None or new_cutoff <= prior_cutoff:
|
||||
# A failure and a genuine no-op both leave `_summarization_event`
|
||||
# unchanged. Stream-based detection can miss the failure
|
||||
# `ToolMessage` (e.g. an update-injected message that never
|
||||
# surfaces on the `messages` stream), so cross-check committed
|
||||
# state before concluding there was nothing to do.
|
||||
current_messages = new_state.get("messages", [])[len(before_messages) :]
|
||||
failure = _find_compaction_failure(current_messages)
|
||||
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
|
||||
)
|
||||
# `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
|
||||
# degenerate chained case where only the prior summary would be
|
||||
# re-summarized (effective cutoff 1 -> new_cutoff == prior_cutoff
|
||||
# via `_compute_state_cutoff`): a fresh event may commit but the
|
||||
# absolute cutoff does not advance, so "nothing to offload" is
|
||||
# the correct, if conservative, report.
|
||||
await self._mount_message(
|
||||
AppMessage(
|
||||
"Nothing to offload \u2014 the conversation is already "
|
||||
"compact.",
|
||||
),
|
||||
)
|
||||
return
|
||||
|
||||
# OffloadResult — success
|
||||
if result.offload_warning:
|
||||
await self._mount_message(ErrorMessage(result.offload_warning))
|
||||
|
||||
# Intentionally traced: the summarization event is a meaningful state
|
||||
# transition that should surface in LangSmith alongside real agent turns.
|
||||
# The new `_context_tokens` count rides along on the same update so it
|
||||
# shares a checkpoint with the offload and doesn't create a separate
|
||||
# `UpdateState` run.
|
||||
await self._agent.aupdate_state(
|
||||
config,
|
||||
{
|
||||
"_summarization_event": result.new_event,
|
||||
"_context_tokens": result.tokens_after,
|
||||
},
|
||||
archive_path = (
|
||||
new_event.get("file_path")
|
||||
if isinstance(new_event, dict)
|
||||
else getattr(new_event, "file_path", None)
|
||||
)
|
||||
# Recompute the post-offload size from the ORIGINAL pre-seed
|
||||
# messages plus the new event. `_effective_conversation` yields
|
||||
# `[summary, *before_messages[new_cutoff:]]` — the compacted
|
||||
# conversation without the tool's own machinery (the seeded tool
|
||||
# call, the tool result, and the trailing model turn), all of which
|
||||
# land in `new_state["messages"]` at/after `new_cutoff`. Counting
|
||||
# `before_messages` keeps this token figure consistent with the
|
||||
# message counts below and avoids understating the reduction.
|
||||
#
|
||||
# This is a client-side approximation for the status bar and is
|
||||
# deliberately not the persisted `_context_tokens` (refreshed from
|
||||
# the trailing turn's real provider usage, which includes
|
||||
# system/tool overhead and the machinery messages). The two can
|
||||
# differ, and if the trailing turn failed `_context_tokens` keeps
|
||||
# its pre-offload value.
|
||||
tokens_after = count_tokens_approximately(
|
||||
_effective_conversation(before_messages, new_event)
|
||||
)
|
||||
# Message counts are likewise derived purely from the absolute
|
||||
# cutoffs, so those same machinery artifacts are never mistaken for
|
||||
# kept conversation.
|
||||
messages_offloaded = max(0, new_cutoff - prior_cutoff)
|
||||
messages_kept = max(0, len(before_messages) - new_cutoff)
|
||||
pct = (
|
||||
round((tokens_before - tokens_after) / tokens_before * 100)
|
||||
if tokens_before > 0
|
||||
else 0
|
||||
)
|
||||
|
||||
before = format_token_count(result.tokens_before)
|
||||
after = format_token_count(result.tokens_after)
|
||||
await self._mount_message(
|
||||
AppMessage(
|
||||
f"Offloaded {result.messages_offloaded} older messages, "
|
||||
f"freeing up context window space.\n"
|
||||
f"Context: {before} \u2192 {after} tokens "
|
||||
f"({result.pct_decrease}% decrease), "
|
||||
f"{result.messages_kept} messages kept.",
|
||||
),
|
||||
before = format_token_count(tokens_before)
|
||||
after = format_token_count(tokens_after)
|
||||
stats_line = (
|
||||
f"Context: {before} → {after} tokens "
|
||||
f"({pct}% decrease), {messages_kept} messages kept."
|
||||
)
|
||||
if archive_path:
|
||||
from deepagents_code.offload import offload_storage_is_ephemeral
|
||||
|
||||
self._on_tokens_update(result.tokens_after)
|
||||
# In local mode the archive may have landed in a temp fallback
|
||||
# directory (persistent `~/.deepagents` was unwritable). The
|
||||
# write succeeded, so context was freed and history is readable
|
||||
# now, but it may not survive a restart -- say so rather than
|
||||
# imply durable storage.
|
||||
caveat = (
|
||||
"\nNote: history was saved to temporary storage and may not "
|
||||
"survive a restart."
|
||||
if offload_storage_is_ephemeral()
|
||||
else ""
|
||||
)
|
||||
await self._mount_message(
|
||||
AppMessage(
|
||||
f"Offloaded {messages_offloaded} older messages, "
|
||||
f"freeing up context window space.\n{stats_line}{caveat}",
|
||||
),
|
||||
)
|
||||
else:
|
||||
# Context was still freed (the summary is in-context), but the
|
||||
# archive write failed, so the offloaded messages are not
|
||||
# recoverable. Surface both facts in one message rather than a
|
||||
# separate warning immediately followed by a success line.
|
||||
await self._mount_message(
|
||||
ErrorMessage(
|
||||
f"Offloaded {messages_offloaded} older messages and "
|
||||
"freed context, but the conversation history could not "
|
||||
"be saved to storage, so those messages are not "
|
||||
f"recoverable. Check logs for details.\n{stats_line}",
|
||||
)
|
||||
)
|
||||
|
||||
self._on_tokens_update(tokens_after)
|
||||
|
||||
except OffloadModelError as exc:
|
||||
logger.warning("Offload model creation failed: %s", exc, exc_info=True)
|
||||
await self._mount_message(ErrorMessage(str(exc)))
|
||||
except Exception as exc: # surface offload errors to user
|
||||
logger.exception("Offload failed")
|
||||
await self._mount_message(ErrorMessage(f"Offload failed: {exc}"))
|
||||
@@ -11556,6 +11814,337 @@ class DeepAgentsApp(App):
|
||||
except Exception: # best-effort spinner cleanup
|
||||
logger.exception("Failed to dismiss spinner after offload")
|
||||
|
||||
async def _drive_server_side_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
|
||||
compaction middleware can reject every other tool independently of
|
||||
HITL configuration, including tools requested by the trailing model
|
||||
turn.
|
||||
|
||||
A first-turn `Command(update=..., goto=...)` is intentionally avoided:
|
||||
the LangGraph API server rebuilds it with `goto=None` and crashes
|
||||
`_control_branch`. The `aupdate_state(as_node="model")` + `astream`
|
||||
continuation is the stable path.
|
||||
|
||||
Args:
|
||||
config: Config with `configurable.thread_id`.
|
||||
seed_tool_call_id: Id for the seeded tool call. Supplied by
|
||||
`_handle_offload` so it can remove the seed if the run fails;
|
||||
a fresh id is generated when omitted (e.g. direct callers).
|
||||
|
||||
Returns:
|
||||
An error string when the tool reported a compaction failure, or
|
||||
`None` when the run completed (whether it compacted or was a
|
||||
no-op — the caller distinguishes those from persisted state).
|
||||
Note the `None` return also covers the bounded-drain-exceeded
|
||||
path, which has already mounted its own user-facing message
|
||||
before returning.
|
||||
"""
|
||||
from langchain.agents.middleware.human_in_the_loop import (
|
||||
ApproveDecision,
|
||||
RejectDecision,
|
||||
)
|
||||
from langchain_core.messages import AIMessage
|
||||
from langgraph.types import Command
|
||||
|
||||
from deepagents_code.config import settings
|
||||
from deepagents_code.offload_middleware import (
|
||||
COMPACTION_FAILURE_PREFIX,
|
||||
_offload_seed_message_id,
|
||||
)
|
||||
|
||||
agent = self._agent
|
||||
if agent is None:
|
||||
return None
|
||||
|
||||
tool_call_id = seed_tool_call_id or str(uuid.uuid4())
|
||||
# Stable message id so a failed run can address the seed for removal.
|
||||
seed = AIMessage(
|
||||
content="",
|
||||
id=_offload_seed_message_id(tool_call_id),
|
||||
tool_calls=[
|
||||
{
|
||||
"name": "compact_conversation",
|
||||
"args": {"force": True},
|
||||
"id": tool_call_id,
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
# Remote dev servers separate checkpoint persistence from HTTP thread
|
||||
# registration; register before mutating state so the write lands.
|
||||
if remote := self._remote_agent():
|
||||
await remote.aensure_thread(
|
||||
{"configurable": {"thread_id": self._lc_thread_id}}
|
||||
)
|
||||
await agent.aupdate_state(config, {"messages": [seed]}, as_node="model")
|
||||
|
||||
tool_error: str | None = None
|
||||
# `self._agent` includes local graphs whose generic context defaults to
|
||||
# `None`, but the graph is built with `CLIContextSchema` at runtime.
|
||||
streaming_agent = cast("Any", agent)
|
||||
|
||||
seeded_compaction_approved = False
|
||||
|
||||
def _decisions_for_interrupt(interrupt_obj: Any) -> list[Any]: # noqa: ANN401
|
||||
"""Approve the forced compaction; reject any other gated tool call.
|
||||
|
||||
HITL action requests do not expose tool-call IDs, so the seeded
|
||||
request is identified by its exact forced arguments and approved
|
||||
at most once. Any repeated compaction request fails closed.
|
||||
|
||||
Args:
|
||||
interrupt_obj: The interrupt surfaced by the HITL middleware.
|
||||
|
||||
Returns:
|
||||
One decision per `action_request`, in order, as the HITL
|
||||
middleware requires.
|
||||
"""
|
||||
nonlocal seeded_compaction_approved
|
||||
value = getattr(interrupt_obj, "value", None)
|
||||
action_requests = (
|
||||
value.get("action_requests") if isinstance(value, dict) else None
|
||||
)
|
||||
if not action_requests:
|
||||
# Without an identifiable action, approving could execute a
|
||||
# different gated tool. A singleton rejection safely answers
|
||||
# the surfaced interrupt.
|
||||
return [
|
||||
RejectDecision(
|
||||
type="reject",
|
||||
message=(
|
||||
"Not executed: /offload could not identify the "
|
||||
"requested action."
|
||||
),
|
||||
)
|
||||
]
|
||||
decisions: list[Any] = []
|
||||
for req in action_requests:
|
||||
name = req.get("name") if isinstance(req, dict) else None
|
||||
args = req.get("args") if isinstance(req, dict) else None
|
||||
is_seeded_request = (
|
||||
not seeded_compaction_approved
|
||||
and name == "compact_conversation"
|
||||
and isinstance(args, dict)
|
||||
and args.get("force") is True
|
||||
)
|
||||
if is_seeded_request:
|
||||
decisions.append(ApproveDecision(type="approve"))
|
||||
seeded_compaction_approved = True
|
||||
else:
|
||||
decisions.append(
|
||||
RejectDecision(
|
||||
type="reject",
|
||||
message=(
|
||||
"Not executed: /offload only performs "
|
||||
"conversation compaction."
|
||||
),
|
||||
)
|
||||
)
|
||||
return decisions
|
||||
|
||||
async def _drain(stream_input: Any) -> list[tuple[str, dict[str, Any]]]: # noqa: ANN401
|
||||
"""Advance the graph, collecting interrupts that need a resume.
|
||||
|
||||
Sets `tool_error` if the compaction tool reported a failure.
|
||||
|
||||
Args:
|
||||
stream_input: `None` to advance, or a `Command(resume=...)` to
|
||||
answer pending interrupts.
|
||||
|
||||
Returns:
|
||||
`(interrupt_id, resume_value)` pairs for every interrupt
|
||||
surfaced during this stream.
|
||||
"""
|
||||
nonlocal tool_error
|
||||
pending: list[tuple[str, dict[str, Any]]] = []
|
||||
async for chunk in streaming_agent.astream(
|
||||
stream_input,
|
||||
stream_mode=["messages", "updates"],
|
||||
subgraphs=True,
|
||||
config=config,
|
||||
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,
|
||||
offload_tool_call_id=tool_call_id,
|
||||
),
|
||||
durability="exit",
|
||||
):
|
||||
if not isinstance(chunk, tuple) or len(chunk) != 3: # noqa: PLR2004 # (namespace, mode, data)
|
||||
continue
|
||||
_namespace, mode, data = chunk
|
||||
if mode == "updates" and isinstance(data, dict):
|
||||
for interrupt_obj in data.get("__interrupt__") or []:
|
||||
iid = getattr(interrupt_obj, "id", None)
|
||||
if iid:
|
||||
decisions = _decisions_for_interrupt(interrupt_obj)
|
||||
pending.append((iid, {"decisions": decisions}))
|
||||
elif mode == "messages" and isinstance(data, tuple):
|
||||
msg = data[0]
|
||||
if _is_tool_message(msg):
|
||||
text = _message_text(msg)
|
||||
if text.startswith(COMPACTION_FAILURE_PREFIX):
|
||||
tool_error = text
|
||||
return pending
|
||||
|
||||
# Bound the resume loop: after compaction the model runs again, and a
|
||||
# rejected gated call could prompt another. The middleware blocks
|
||||
# execution even when HITL is disabled; this bound handles HITL retries.
|
||||
max_resume_rounds = 10
|
||||
pending = await _drain(None)
|
||||
rounds = 0
|
||||
while pending:
|
||||
rounds += 1
|
||||
if rounds > max_resume_rounds:
|
||||
logger.warning(
|
||||
"Offload exceeded %d resume rounds; leaving %d interrupt(s) "
|
||||
"unresolved",
|
||||
max_resume_rounds,
|
||||
len(pending),
|
||||
)
|
||||
# Compaction itself already committed in round 1, so the caller
|
||||
# still reports the offload. Surface the abandoned drain so the
|
||||
# user knows the thread was left paused mid-run and may need a
|
||||
# fresh message to reset. Skip this when a tool failure is
|
||||
# already pending, so the caller shows that error instead of
|
||||
# the user seeing two conflicting messages.
|
||||
if tool_error is None:
|
||||
await self._mount_message(
|
||||
ErrorMessage(
|
||||
"Offload completed, but the agent kept requesting "
|
||||
"tools afterward and the run could not be fully "
|
||||
"drained. Send a new message to continue; the "
|
||||
"thread may need to reset."
|
||||
)
|
||||
)
|
||||
break
|
||||
resume_payload = dict(pending)
|
||||
pending = await _drain(Command(resume=resume_payload))
|
||||
|
||||
return tool_error
|
||||
|
||||
async def _remove_offload_artifacts(
|
||||
self,
|
||||
config: RunnableConfig,
|
||||
messages: list[Any],
|
||||
prior_event: object,
|
||||
) -> None:
|
||||
"""Restore state changed by a no-op `/offload` graph run.
|
||||
|
||||
Best-effort: a failed restoration is logged and swallowed rather than
|
||||
raised. The no-op path answers the seed with a valid tool result, so the
|
||||
committed seed/result pair left behind is harmless (unlike an unanswered
|
||||
seed); letting the write raise here would misreport a working offload as
|
||||
"Offload failed" via the caller's outer handler.
|
||||
|
||||
Args:
|
||||
config: Config with `configurable.thread_id`.
|
||||
messages: Messages appended after the pre-run state snapshot.
|
||||
prior_event: Summarization event from the pre-run state snapshot.
|
||||
"""
|
||||
from langchain_core.messages import RemoveMessage
|
||||
|
||||
agent = self._agent
|
||||
if agent is None:
|
||||
return
|
||||
removals = [
|
||||
RemoveMessage(id=message_id)
|
||||
for message in messages
|
||||
if (message_id := _message_id(message)) is not None
|
||||
]
|
||||
try:
|
||||
await agent.aupdate_state(
|
||||
config,
|
||||
{
|
||||
"messages": removals,
|
||||
"_summarization_event": prior_event,
|
||||
},
|
||||
as_node="model",
|
||||
)
|
||||
except Exception: # best-effort restoration; keep the no-op report
|
||||
logger.warning(
|
||||
"Failed to restore state after a no-op offload run", exc_info=True
|
||||
)
|
||||
|
||||
async def _remove_unanswered_offload_seed(
|
||||
self, config: RunnableConfig, seed_tool_call_id: str
|
||||
) -> bool:
|
||||
"""Remove a committed `/offload` seed whose tool call was never answered.
|
||||
|
||||
The seed `AIMessage` carrying the forced `compact_conversation` call is
|
||||
committed via `aupdate_state` before the run advances — independently of
|
||||
the stream's durability. If the run then fails before the tool produces
|
||||
a `ToolMessage`, the seed is left as an unanswered `tool_use` in
|
||||
committed state, which the model API rejects on the next turn
|
||||
("tool_use ids ... without tool_result"), potentially wedging the
|
||||
thread. This best-effort removes that seed so a failed `/offload` leaves
|
||||
a valid conversation.
|
||||
|
||||
If the tool *did* run (a `ToolMessage` answers the call), the seed and
|
||||
its result form a valid pair and are left untouched — removing the seed
|
||||
alone would orphan the `ToolMessage`.
|
||||
|
||||
Args:
|
||||
config: Config with `configurable.thread_id`.
|
||||
seed_tool_call_id: The id of the seeded `compact_conversation` call.
|
||||
|
||||
Returns:
|
||||
True if the thread is known to be free of a dangling seed (removed,
|
||||
validly answered, or absent). False if a dangling seed may
|
||||
remain because the state read or the removal write failed — the
|
||||
caller should warn the user the thread may be inconsistent.
|
||||
"""
|
||||
from langchain_core.messages import RemoveMessage
|
||||
|
||||
agent = self._agent
|
||||
if agent is None or not self._lc_thread_id:
|
||||
return True
|
||||
try:
|
||||
state = await self._get_thread_state_values(self._lc_thread_id)
|
||||
except Exception: # best-effort cleanup; keep the original error
|
||||
logger.warning(
|
||||
"Could not read state to clean up offload seed", exc_info=True
|
||||
)
|
||||
return False
|
||||
|
||||
messages = state.get("messages", [])
|
||||
# An answering ToolMessage means the tool ran; the pair is valid.
|
||||
if any(
|
||||
_is_tool_message(msg) and _message_tool_call_id(msg) == seed_tool_call_id
|
||||
for msg in messages
|
||||
):
|
||||
return True
|
||||
|
||||
seed_id = next(
|
||||
(
|
||||
_message_id(msg)
|
||||
for msg in messages
|
||||
if seed_tool_call_id in _message_tool_call_ids(msg)
|
||||
),
|
||||
None,
|
||||
)
|
||||
if not seed_id:
|
||||
return True
|
||||
try:
|
||||
await agent.aupdate_state(
|
||||
config, {"messages": [RemoveMessage(id=seed_id)]}, as_node="model"
|
||||
)
|
||||
except Exception: # best-effort cleanup; keep the original error
|
||||
logger.warning("Failed to remove dangling offload seed", exc_info=True)
|
||||
return False
|
||||
return True
|
||||
|
||||
async def _handle_user_message(self, message: str) -> None:
|
||||
"""Handle a user message to send to the agent.
|
||||
|
||||
|
||||
@@ -264,6 +264,8 @@ def _get_context(request: ModelRequest) -> CLIContextSchema | None:
|
||||
return CLIContextSchema(
|
||||
model=ctx.get("model"),
|
||||
model_params=ctx.get("model_params") or {},
|
||||
profile_overrides=ctx.get("profile_overrides") or {},
|
||||
model_context_limit=ctx.get("model_context_limit"),
|
||||
auto_approve=bool(ctx.get("auto_approve", False)),
|
||||
approval_mode_key=raw_key if isinstance(raw_key, str) else None,
|
||||
thread_id=raw_thread_id if isinstance(raw_thread_id, str) else None,
|
||||
|
||||
@@ -1,402 +1,146 @@
|
||||
"""Business logic for the `/offload` command.
|
||||
|
||||
Extracts the core offload workflow from the UI layer so it can be
|
||||
tested independently of the Textual app.
|
||||
"""
|
||||
"""Storage paths for offloaded conversation history."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
|
||||
from langchain_core.messages import get_buffer_string
|
||||
from langchain_core.messages.utils import count_tokens_approximately
|
||||
|
||||
from deepagents_code._session_stats import format_token_count
|
||||
from deepagents_code.config import create_model
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from deepagents.backends.protocol import BackendProtocol
|
||||
from deepagents.middleware.summarization import (
|
||||
SummarizationEvent,
|
||||
SummarizationMiddleware,
|
||||
)
|
||||
from langchain_core.messages import AnyMessage, HumanMessage
|
||||
import os
|
||||
import stat
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Result types
|
||||
# ---------------------------------------------------------------------------
|
||||
_EPHEMERAL_OFFLOAD_STORAGE = False
|
||||
"""Whether the most recent `_offload_fallback_root` fell back to temp storage."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OffloadResult:
|
||||
"""Successful offload result."""
|
||||
def offload_storage_is_ephemeral() -> bool:
|
||||
"""Return whether offload history is routed to non-persistent storage.
|
||||
|
||||
new_event: SummarizationEvent
|
||||
"""The summarization event to write into agent state."""
|
||||
|
||||
messages_offloaded: int
|
||||
"""Number of older messages that were offloaded."""
|
||||
|
||||
messages_kept: int
|
||||
"""Number of recent messages retained in context."""
|
||||
|
||||
tokens_before: int
|
||||
"""Approximate token count of the conversation before offloading."""
|
||||
|
||||
tokens_after: int
|
||||
"""Approximate token count of the conversation after offloading."""
|
||||
|
||||
pct_decrease: int
|
||||
"""Percentage decrease in token usage."""
|
||||
|
||||
offload_warning: str | None
|
||||
"""Non-`None` when the backend write failed (non-fatal)."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OffloadThresholdNotMet:
|
||||
"""Offload was a no-op — conversation is within the retention budget."""
|
||||
|
||||
conversation_tokens: int
|
||||
"""Approximate token count of the conversation messages alone."""
|
||||
|
||||
total_context_tokens: int
|
||||
"""Total context token count including system overhead, or `0` when no
|
||||
token tracker is available."""
|
||||
|
||||
context_limit: int | None
|
||||
"""Model context window limit, if available."""
|
||||
|
||||
budget_str: str
|
||||
"""Human-readable retention budget (e.g. "20.0K tokens")."""
|
||||
|
||||
|
||||
class OffloadModelError(Exception):
|
||||
"""Raised when the model cannot be created for offloading."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def format_offload_limit(
|
||||
keep: tuple[str, int | float], context_limit: int | None
|
||||
) -> str:
|
||||
"""Format offload retention settings into a human-readable limit string.
|
||||
|
||||
Args:
|
||||
keep: Retention policy tuple `(type, value)` from summarization
|
||||
defaults, where `type` is one of `"messages"`, `"tokens"`, or
|
||||
`"fraction"`.
|
||||
context_limit: Model context limit when available.
|
||||
`True` when the persistent `~/.deepagents` location was unwritable and the
|
||||
most recent `_offload_fallback_root` fell back to a temporary directory that
|
||||
may not survive a restart. Only meaningful in local mode, where
|
||||
`_offload_fallback_root` runs in the same process as the UI; in
|
||||
server/sandbox mode persistence is owned by the server backend and this flag
|
||||
stays `False` client-side.
|
||||
|
||||
Returns:
|
||||
A short display string describing the offload retention limit.
|
||||
`True` if the local offload root is a temporary, non-persistent
|
||||
directory; `False` when it is the persistent per-user location (or was
|
||||
never resolved in this process).
|
||||
"""
|
||||
keep_type, keep_value = keep
|
||||
|
||||
if keep_type == "messages":
|
||||
count = int(keep_value)
|
||||
noun = "message" if count == 1 else "messages"
|
||||
return f"last {count} {noun}"
|
||||
|
||||
if keep_type == "tokens":
|
||||
return f"{format_token_count(int(keep_value))} tokens"
|
||||
|
||||
if keep_type == "fraction":
|
||||
percent = float(keep_value) * 100
|
||||
if context_limit is not None:
|
||||
token_limit = max(1, int(context_limit * float(keep_value)))
|
||||
return f"{format_token_count(token_limit)} tokens"
|
||||
return f"{percent:.0f}% of context window"
|
||||
|
||||
return "current retention threshold"
|
||||
return _EPHEMERAL_OFFLOAD_STORAGE
|
||||
|
||||
|
||||
async def offload_messages_to_backend(
|
||||
messages: list[Any],
|
||||
middleware: SummarizationMiddleware,
|
||||
*,
|
||||
thread_id: str,
|
||||
backend: BackendProtocol,
|
||||
) -> str | None:
|
||||
"""Write messages to backend storage before offloading.
|
||||
def _offload_fallback_root() -> Path:
|
||||
"""Return a writable base directory for offloaded conversation history.
|
||||
|
||||
Appends messages as a timestamped markdown section to the conversation
|
||||
history file, matching the `SummarizationMiddleware` offload pattern.
|
||||
Prefers the persistent per-user `~/.deepagents` directory so offloaded
|
||||
history survives across sessions and is easy to locate; falls back to a
|
||||
private temporary directory when the home directory cannot be resolved or
|
||||
written. This is the live root for the local-mode `conversation_history`
|
||||
backend in `agent.py`.
|
||||
|
||||
Filters out prior summary messages using the middleware's
|
||||
`_filter_summary_messages` to avoid storing summaries-of-summaries.
|
||||
Archives always live in the `conversation_history` subdirectory of the
|
||||
returned root. The `0o700` hardening therefore targets that subdirectory,
|
||||
never the shared `~/.deepagents` config root -- which also houses
|
||||
`config.toml`, `hooks.json`, `.env`, and `.state/`, whose permissions this
|
||||
must not disturb. A temporary fallback root is created solely for offload,
|
||||
so the whole directory is hardened in that case.
|
||||
|
||||
Args:
|
||||
messages: Messages to offload.
|
||||
middleware: `SummarizationMiddleware` instance for filtering.
|
||||
thread_id: Thread identifier used to derive the storage path.
|
||||
backend: Backend to persist conversation history to.
|
||||
Note: the `S_ISDIR` check below (which uses `lstat`, deliberately not
|
||||
following the link) guards the paths it is applied to -- the
|
||||
`conversation_history` subdirectory and, in the fallback case, the temp
|
||||
root -- not `~/.deepagents` itself, which is created with a plain `mkdir`.
|
||||
So a `conversation_history` (or temp root) that is itself a symlink is
|
||||
rejected, whereas a symlinked `~/.deepagents` pointing at a directory the
|
||||
current user owns is followed transparently and archives persist normally.
|
||||
(A dangling `~/.deepagents` symlink still falls through to temporary
|
||||
storage, but via `mkdir` raising, not via this check.)
|
||||
|
||||
Returns:
|
||||
File path where history was stored, `""` (empty string) if there were no
|
||||
non-summary messages to offload (not an error), or `None` if the
|
||||
write failed.
|
||||
A directory whose `conversation_history` subdirectory is private and
|
||||
writable.
|
||||
"""
|
||||
path = f"/conversation_history/{thread_id}.md"
|
||||
|
||||
# Exclude prior summaries so the offloaded history contains only
|
||||
# original messages
|
||||
filtered = middleware._filter_summary_messages(messages)
|
||||
if not filtered:
|
||||
return ""
|
||||
def _harden(path: Path) -> None:
|
||||
"""Create `path` if needed and restrict it to the current user.
|
||||
|
||||
timestamp = datetime.now(UTC).isoformat()
|
||||
buf = get_buffer_string(filtered)
|
||||
new_section = f"## Offloaded at {timestamp}\n\n{buf}\n\n"
|
||||
Only ever called on directories owned by offload (a fresh temp dir or
|
||||
the dedicated `conversation_history` subdirectory), never on the shared
|
||||
`~/.deepagents` config root.
|
||||
|
||||
existing_content = ""
|
||||
Raises:
|
||||
OSError: If the path exists but is not a directory, or the
|
||||
directory cannot be created or its mode changed (e.g. a
|
||||
read-only mount).
|
||||
PermissionError: If the existing directory is owned by another
|
||||
local user.
|
||||
"""
|
||||
path.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
info = path.lstat()
|
||||
if not stat.S_ISDIR(info.st_mode):
|
||||
msg = f"Offload path is not a directory: {path}"
|
||||
raise OSError(msg)
|
||||
getuid = getattr(os, "getuid", None)
|
||||
if getuid is not None and info.st_uid != getuid():
|
||||
msg = f"Offload directory is owned by another user: {path}"
|
||||
raise PermissionError(msg)
|
||||
# `mkdir(mode=...)` does not tighten an existing directory. Archives
|
||||
# contain conversation data, so the offload directory must remain
|
||||
# inaccessible to other local accounts regardless of the process umask.
|
||||
path.chmod(0o700)
|
||||
|
||||
def _probe_writable(path: Path) -> None:
|
||||
# Creating the directory is insufficient when it already exists on a
|
||||
# read-only mount. A temporary file proves archive writes can succeed.
|
||||
with tempfile.NamedTemporaryFile(dir=path, prefix=".write-test-"):
|
||||
pass
|
||||
|
||||
def _prepare_user_dir() -> Path:
|
||||
base = Path.home() / ".deepagents"
|
||||
# Ensure the shared config root exists and is usable, but leave its
|
||||
# permissions untouched -- hardening belongs on the archive subdir only.
|
||||
base.mkdir(parents=True, exist_ok=True)
|
||||
archive_dir = base / "conversation_history"
|
||||
_harden(archive_dir)
|
||||
_probe_writable(archive_dir)
|
||||
return base
|
||||
|
||||
def _prepare_temp_dir(path: Path) -> Path:
|
||||
# A temp dir is created solely for offload and is not shared config, so
|
||||
# hardening the whole directory (which protects its archive subdir) is
|
||||
# both safe and necessary in world-writable temp locations.
|
||||
_harden(path)
|
||||
_probe_writable(path)
|
||||
return path
|
||||
|
||||
global _EPHEMERAL_OFFLOAD_STORAGE # noqa: PLW0603
|
||||
try:
|
||||
responses = await backend.adownload_files([path])
|
||||
resp = responses[0] if responses else None
|
||||
if resp and resp.content is not None and resp.error is None:
|
||||
existing_content = resp.content.decode("utf-8")
|
||||
except Exception as exc: # abort write on read failure
|
||||
root = _prepare_user_dir()
|
||||
except (RuntimeError, OSError):
|
||||
logger.warning(
|
||||
"Failed to read existing history at %s; aborting offload to "
|
||||
"avoid overwriting prior history: %s",
|
||||
path,
|
||||
exc,
|
||||
"User data directory is not writable; falling back to temporary "
|
||||
"offload storage, which may not persist across restarts",
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
|
||||
combined = existing_content + new_section
|
||||
|
||||
else:
|
||||
_EPHEMERAL_OFFLOAD_STORAGE = False
|
||||
return root
|
||||
# Only reached on the fallback path: every root produced below is temporary
|
||||
# and may not survive a restart.
|
||||
_EPHEMERAL_OFFLOAD_STORAGE = True
|
||||
getuid = getattr(os, "getuid", None)
|
||||
suffix = str(getuid()) if getuid is not None else str(os.getpid())
|
||||
temp_root = Path(tempfile.gettempdir())
|
||||
path = temp_root / f"deepagents-{suffix}"
|
||||
try:
|
||||
result = (
|
||||
await backend.aedit(path, existing_content, combined)
|
||||
if existing_content
|
||||
else await backend.awrite(path, combined)
|
||||
)
|
||||
if result is None or result.error:
|
||||
error_detail = result.error if result else "backend returned None"
|
||||
logger.warning(
|
||||
"Failed to offload conversation history to %s: %s",
|
||||
path,
|
||||
error_detail,
|
||||
)
|
||||
return None
|
||||
except Exception as exc: # defensive: surface write failures gracefully
|
||||
return _prepare_temp_dir(path)
|
||||
except (OSError, RuntimeError):
|
||||
logger.warning(
|
||||
"Exception offloading conversation history to %s: %s",
|
||||
path,
|
||||
exc,
|
||||
"Per-user temporary offload directory is unavailable; creating "
|
||||
"a private unique directory",
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
|
||||
logger.debug("Offloaded %d messages to %s", len(filtered), path)
|
||||
return path
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Core offload workflow
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def perform_offload(
|
||||
*,
|
||||
messages: list[Any],
|
||||
prior_event: SummarizationEvent | None,
|
||||
thread_id: str,
|
||||
model_spec: str,
|
||||
profile_overrides: dict[str, Any] | None,
|
||||
context_limit: int | None,
|
||||
total_context_tokens: int,
|
||||
backend: BackendProtocol | None,
|
||||
) -> OffloadResult | OffloadThresholdNotMet:
|
||||
"""Execute the offload workflow: summarize old messages and free context.
|
||||
|
||||
Args:
|
||||
messages: Current conversation messages from agent state.
|
||||
|
||||
May be LangChain message objects or serialized dicts (the latter
|
||||
when read from a remote HTTP state snapshot).
|
||||
prior_event: Existing `_summarization_event` if any.
|
||||
|
||||
In server mode `summary_message` may be a serialized message dict.
|
||||
thread_id: Thread identifier for backend storage.
|
||||
model_spec: Model specification string (e.g. "openai:gpt-4").
|
||||
profile_overrides: Optional profile overrides from CLI flags.
|
||||
context_limit: Model context limit from settings.
|
||||
total_context_tokens: Current total context token count, or `0` when
|
||||
no token tracker is available.
|
||||
backend: Backend for persisting offloaded history.
|
||||
|
||||
Returns:
|
||||
`OffloadResult` on success, `OffloadThresholdNotMet` when the
|
||||
conversation is within the retention budget.
|
||||
|
||||
Raises:
|
||||
OffloadModelError: If the model cannot be created.
|
||||
"""
|
||||
from deepagents.middleware.summarization import (
|
||||
SummarizationMiddleware,
|
||||
compute_summarization_defaults,
|
||||
)
|
||||
|
||||
# Remote HTTP state snapshots may surface serialized message dicts instead
|
||||
# of LangChain message objects. Normalize them before passing state into
|
||||
# summarization middleware helpers. `any(...)` rather than checking index
|
||||
# 0 guards against heterogeneous lists (e.g. a snapshot with a streamed
|
||||
# append).
|
||||
needs_message_conversion = any(isinstance(m, dict) for m in messages)
|
||||
needs_summary_conversion = prior_event is not None and isinstance(
|
||||
prior_event.get("summary_message"), dict
|
||||
)
|
||||
if needs_message_conversion or needs_summary_conversion:
|
||||
from langchain_core.messages.utils import convert_to_messages
|
||||
|
||||
if needs_message_conversion:
|
||||
messages = cast("list[AnyMessage]", convert_to_messages(messages))
|
||||
if needs_summary_conversion and prior_event is not None:
|
||||
converted_summary = cast(
|
||||
"HumanMessage",
|
||||
convert_to_messages([prior_event["summary_message"]])[0],
|
||||
)
|
||||
prior_event = {
|
||||
"cutoff_index": prior_event["cutoff_index"],
|
||||
"summary_message": converted_summary,
|
||||
"file_path": prior_event["file_path"],
|
||||
}
|
||||
|
||||
try:
|
||||
result = create_model(model_spec, profile_overrides=profile_overrides)
|
||||
model = result.model
|
||||
except Exception as exc:
|
||||
msg = f"Offload requires a working model configuration: {exc}"
|
||||
raise OffloadModelError(msg) from exc
|
||||
|
||||
# Patch context limit into model profile when it differs from the native
|
||||
# value (e.g. set via --profile-override or runtime config).
|
||||
if context_limit is not None:
|
||||
profile = getattr(model, "profile", None)
|
||||
native = profile.get("max_input_tokens") if isinstance(profile, dict) else None
|
||||
if native != context_limit:
|
||||
merged = (
|
||||
{**profile, "max_input_tokens": context_limit}
|
||||
if isinstance(profile, dict)
|
||||
else {"max_input_tokens": context_limit}
|
||||
)
|
||||
try:
|
||||
model.profile = merged # ty: ignore[invalid-assignment]
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
logger.warning(
|
||||
"Could not patch context limit (%d) into model profile; "
|
||||
"offload budget will use the model's native context window",
|
||||
context_limit,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
defaults = compute_summarization_defaults(model)
|
||||
offload_backend = backend
|
||||
if offload_backend is None:
|
||||
from deepagents.backends.filesystem import FilesystemBackend
|
||||
|
||||
offload_backend = FilesystemBackend(virtual_mode=False)
|
||||
logger.info("Using local FilesystemBackend for offload")
|
||||
|
||||
middleware = SummarizationMiddleware(
|
||||
model=model,
|
||||
backend=offload_backend,
|
||||
keep=defaults["keep"],
|
||||
trim_tokens_to_summarize=None,
|
||||
)
|
||||
|
||||
# Rebuild the message list the model would see, accounting for
|
||||
# any prior offload
|
||||
effective = middleware._apply_event_to_messages(messages, prior_event)
|
||||
cutoff = middleware._determine_cutoff_index(effective)
|
||||
budget_str = format_offload_limit(defaults["keep"], context_limit)
|
||||
|
||||
if cutoff == 0:
|
||||
return OffloadThresholdNotMet(
|
||||
conversation_tokens=count_tokens_approximately(effective),
|
||||
total_context_tokens=total_context_tokens,
|
||||
context_limit=context_limit,
|
||||
budget_str=budget_str,
|
||||
)
|
||||
|
||||
to_summarize, to_keep = middleware._partition_messages(effective, cutoff)
|
||||
|
||||
tokens_summarized = count_tokens_approximately(to_summarize)
|
||||
tokens_kept = count_tokens_approximately(to_keep)
|
||||
tokens_before = tokens_summarized + tokens_kept
|
||||
|
||||
# Generate summary first so no side effects occur if the LLM fails
|
||||
summary = await middleware._acreate_summary(to_summarize)
|
||||
|
||||
backend_path = await offload_messages_to_backend(
|
||||
to_summarize,
|
||||
middleware,
|
||||
thread_id=thread_id,
|
||||
backend=offload_backend,
|
||||
)
|
||||
offload_warning: str | None = None
|
||||
if backend_path is None:
|
||||
offload_warning = (
|
||||
"Warning: conversation history could not be saved to "
|
||||
"storage. Older messages will not be recoverable. "
|
||||
"Check logs for details."
|
||||
)
|
||||
logger.error(
|
||||
"Backend write failed for thread %s; offloading will proceed "
|
||||
"but messages are not recoverable",
|
||||
thread_id,
|
||||
)
|
||||
file_path = backend_path or None
|
||||
|
||||
summary_msg = middleware._build_new_messages_with_path(summary, file_path)[0]
|
||||
|
||||
# Append token savings note so the model is aware of how much context
|
||||
# was reclaimed.
|
||||
tokens_summary = count_tokens_approximately([summary_msg])
|
||||
tokens_after = tokens_summary + tokens_kept
|
||||
pct = (
|
||||
round((tokens_before - tokens_after) / tokens_before * 100)
|
||||
if tokens_before > 0
|
||||
else 0
|
||||
)
|
||||
summarized_before = format_token_count(tokens_summarized)
|
||||
summarized_after = format_token_count(tokens_summary)
|
||||
savings_note = (
|
||||
f"\n\n{len(to_summarize)} messages were offloaded "
|
||||
f"({summarized_before} \u2192 {summarized_after} tokens). "
|
||||
f"Total context: {format_token_count(tokens_before)} \u2192 "
|
||||
f"{format_token_count(tokens_after)} tokens "
|
||||
f"({pct}% decrease), "
|
||||
f"{len(to_keep)} messages unchanged."
|
||||
)
|
||||
summary_msg.content += savings_note
|
||||
|
||||
state_cutoff = middleware._compute_state_cutoff(prior_event, cutoff)
|
||||
|
||||
new_event: SummarizationEvent = {
|
||||
"cutoff_index": state_cutoff,
|
||||
"summary_message": summary_msg, # ty: ignore[invalid-argument-type]
|
||||
"file_path": file_path,
|
||||
}
|
||||
|
||||
return OffloadResult(
|
||||
new_event=new_event,
|
||||
messages_offloaded=len(to_summarize),
|
||||
messages_kept=len(to_keep),
|
||||
tokens_before=tokens_before,
|
||||
tokens_after=tokens_after,
|
||||
pct_decrease=pct,
|
||||
offload_warning=offload_warning,
|
||||
)
|
||||
unique = Path(tempfile.mkdtemp(prefix=f"deepagents-{suffix}-", dir=temp_root))
|
||||
return _prepare_temp_dir(unique)
|
||||
|
||||
@@ -0,0 +1,643 @@
|
||||
"""CLI-specific conversation compaction middleware."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Annotated, Any, NamedTuple, cast
|
||||
|
||||
from deepagents.backends.protocol import FILE_NOT_FOUND
|
||||
from deepagents.middleware.summarization import (
|
||||
SummarizationToolMiddleware,
|
||||
create_summarization_middleware,
|
||||
create_summarization_tool_middleware,
|
||||
)
|
||||
from langchain.tools import (
|
||||
ToolRuntime, # noqa: TC002 # inspected for runtime injection
|
||||
)
|
||||
from langchain_core.messages import ToolMessage
|
||||
from langchain_core.tools import InjectedToolArg, StructuredTool
|
||||
from langgraph.types import Command
|
||||
|
||||
from deepagents_code._cli_context import CLIContextSchema
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from deepagents.backends.protocol import (
|
||||
BACKEND_TYPES,
|
||||
BackendProtocol,
|
||||
EditResult,
|
||||
FileDownloadResponse,
|
||||
WriteResult,
|
||||
)
|
||||
from deepagents.middleware.summarization import SummarizationMiddleware
|
||||
from langchain.chat_models import BaseChatModel
|
||||
from langgraph.prebuilt.tool_node import ToolCallRequest
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
COMPACTION_FAILURE_PREFIX = "Compaction failed"
|
||||
"""Stable prefix for forced-compaction failure tool messages.
|
||||
|
||||
`/offload` drives the tool server-side and can only observe the resulting
|
||||
`ToolMessage` text across the LangGraph server boundary, so it keys failure
|
||||
detection on this prefix. Owning the literal here means the producer
|
||||
(`_forced_compact_error`) and both consumers (`app._drive_server_side_compaction`
|
||||
live-stream detection and `app._find_compaction_failure` committed-state scan)
|
||||
reference one constant instead of re-hardcoding the wording independently.
|
||||
|
||||
Note: this value is deliberately identical to the leading text of the SDK's own
|
||||
model-initiated compaction-failure message, so a failure emitted by either path
|
||||
is recognized. Because the scan is bounded to messages produced by the current
|
||||
`/offload` attempt, a stale failure from an unrelated prior turn is not matched.
|
||||
Only the *prefix position* is load-bearing; wording after it is free to change.
|
||||
"""
|
||||
|
||||
_OFFLOAD_SEED_ID_PREFIX = "offload-seed-"
|
||||
|
||||
|
||||
def _offload_seed_message_id(tool_call_id: str) -> str:
|
||||
"""Return the stable message ID for a forced `/offload` tool call.
|
||||
|
||||
Args:
|
||||
tool_call_id: The seeded `compact_conversation` tool call ID.
|
||||
|
||||
Returns:
|
||||
The synthetic assistant message ID associated with the tool call.
|
||||
"""
|
||||
return f"{_OFFLOAD_SEED_ID_PREFIX}{tool_call_id}"
|
||||
|
||||
|
||||
def _without_offload_seed(messages: list[Any], tool_call_id: str) -> list[Any]:
|
||||
"""Exclude the synthetic `/offload` seed from retention calculations.
|
||||
|
||||
Args:
|
||||
messages: Effective conversation messages including the forced tool call.
|
||||
tool_call_id: The seeded `compact_conversation` tool call ID.
|
||||
|
||||
Returns:
|
||||
Conversation messages without the matching synthetic assistant message.
|
||||
"""
|
||||
if not tool_call_id:
|
||||
return messages
|
||||
seed_id = _offload_seed_message_id(tool_call_id)
|
||||
return [
|
||||
message
|
||||
for message in messages
|
||||
if (
|
||||
message.get("id")
|
||||
if isinstance(message, dict)
|
||||
else getattr(message, "id", None)
|
||||
)
|
||||
!= seed_id
|
||||
]
|
||||
|
||||
|
||||
class RuntimeModelConfig(NamedTuple):
|
||||
"""Active model configuration read from a tool runtime.
|
||||
|
||||
A named tuple rather than a bare 4-tuple so the two structurally identical
|
||||
`dict` slots (`model_params`, `profile_overrides`) are addressed by name at
|
||||
both the construction sites (keyword args) and the read site (attribute
|
||||
access) — a silent positional transposition the type checker would not catch
|
||||
is thereby avoided. Positional construction/unpacking is still possible and
|
||||
would defeat this, so call sites must keep using names.
|
||||
"""
|
||||
|
||||
model_spec: str | None
|
||||
model_params: dict[str, Any]
|
||||
profile_overrides: dict[str, Any]
|
||||
context_limit: int | None
|
||||
|
||||
|
||||
def _runtime_model_config(runtime: ToolRuntime) -> RuntimeModelConfig:
|
||||
"""Read the active model configuration from a tool runtime.
|
||||
|
||||
Args:
|
||||
runtime: Runtime injected into the compaction tool.
|
||||
|
||||
Returns:
|
||||
The active model specification, invocation parameters, profile
|
||||
overrides, and effective context-window limit.
|
||||
"""
|
||||
context = runtime.context
|
||||
if isinstance(context, CLIContextSchema):
|
||||
return RuntimeModelConfig(
|
||||
model_spec=context.model,
|
||||
model_params=context.model_params,
|
||||
profile_overrides=context.profile_overrides,
|
||||
context_limit=context.model_context_limit,
|
||||
)
|
||||
if isinstance(context, dict):
|
||||
model = context.get("model")
|
||||
params = context.get("model_params")
|
||||
profile_overrides = context.get("profile_overrides")
|
||||
context_limit = context.get("model_context_limit")
|
||||
return RuntimeModelConfig(
|
||||
model_spec=model if isinstance(model, str) else None,
|
||||
model_params=dict(params) if isinstance(params, dict) else {},
|
||||
profile_overrides=(
|
||||
dict(profile_overrides) if isinstance(profile_overrides, dict) else {}
|
||||
),
|
||||
context_limit=context_limit if isinstance(context_limit, int) else None,
|
||||
)
|
||||
return RuntimeModelConfig(
|
||||
model_spec=None, model_params={}, profile_overrides={}, context_limit=None
|
||||
)
|
||||
|
||||
|
||||
def _offload_tool_call_id(context: object) -> str | None:
|
||||
"""Read the sole tool-call ID authorized for an `/offload` run.
|
||||
|
||||
Args:
|
||||
context: Runtime context supplied to the agent graph.
|
||||
|
||||
Returns:
|
||||
The authorized tool-call ID, or `None` during an ordinary agent run.
|
||||
"""
|
||||
value = (
|
||||
context.offload_tool_call_id
|
||||
if isinstance(context, CLIContextSchema)
|
||||
else context.get("offload_tool_call_id")
|
||||
if isinstance(context, dict)
|
||||
else None
|
||||
)
|
||||
return value if isinstance(value, str) and value else None
|
||||
|
||||
|
||||
class _ArchiveReadGuard:
|
||||
"""Prevent an archive write after its prerequisite read fails.
|
||||
|
||||
The SDK archive helper treats any unsuccessful read like a missing file and
|
||||
follows it with a truncating `write`. This narrow backend adapter preserves
|
||||
the SDK formatting and append behavior while making that fallback fail closed.
|
||||
"""
|
||||
|
||||
def __init__(self, backend: BackendProtocol) -> None:
|
||||
self._backend = backend
|
||||
self._read_failed = False
|
||||
|
||||
def _record_response_errors(
|
||||
self, responses: list[FileDownloadResponse]
|
||||
) -> list[FileDownloadResponse]:
|
||||
"""Record read errors other than an expected missing archive.
|
||||
|
||||
Args:
|
||||
responses: Backend download responses to inspect.
|
||||
|
||||
Returns:
|
||||
The unchanged backend download responses.
|
||||
"""
|
||||
if any(
|
||||
response.error is not None and response.error != FILE_NOT_FOUND
|
||||
for response in responses
|
||||
):
|
||||
self._read_failed = True
|
||||
return responses
|
||||
|
||||
def _ensure_read_succeeded(self) -> None:
|
||||
"""Raise when a prior archive read failed in this operation.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If the prerequisite archive read failed.
|
||||
"""
|
||||
if self._read_failed:
|
||||
msg = "archive read failed; refusing to overwrite existing history"
|
||||
raise RuntimeError(msg)
|
||||
|
||||
def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
|
||||
"""Delegate a synchronous read while recording failures.
|
||||
|
||||
Args:
|
||||
paths: Backend paths to read.
|
||||
|
||||
Returns:
|
||||
The backend download responses.
|
||||
"""
|
||||
try:
|
||||
responses = self._backend.download_files(paths)
|
||||
except Exception:
|
||||
self._read_failed = True
|
||||
raise
|
||||
return self._record_response_errors(responses)
|
||||
|
||||
async def adownload_files(self, paths: list[str]) -> list[FileDownloadResponse]:
|
||||
"""Delegate an asynchronous read while recording failures.
|
||||
|
||||
Args:
|
||||
paths: Backend paths to read.
|
||||
|
||||
Returns:
|
||||
The backend download responses.
|
||||
"""
|
||||
try:
|
||||
responses = await self._backend.adownload_files(paths)
|
||||
except Exception:
|
||||
self._read_failed = True
|
||||
raise
|
||||
return self._record_response_errors(responses)
|
||||
|
||||
def write(self, file_path: str, content: str) -> WriteResult:
|
||||
"""Write only when the prerequisite archive read succeeded.
|
||||
|
||||
Args:
|
||||
file_path: Backend path to write.
|
||||
content: Complete archive content.
|
||||
|
||||
Returns:
|
||||
The backend write result.
|
||||
"""
|
||||
self._ensure_read_succeeded()
|
||||
return self._backend.write(file_path, content)
|
||||
|
||||
async def awrite(self, file_path: str, content: str) -> WriteResult:
|
||||
"""Asynchronously write only after a successful archive read.
|
||||
|
||||
Args:
|
||||
file_path: Backend path to write.
|
||||
content: Complete archive content.
|
||||
|
||||
Returns:
|
||||
The backend write result.
|
||||
"""
|
||||
self._ensure_read_succeeded()
|
||||
return await self._backend.awrite(file_path, content)
|
||||
|
||||
def edit(
|
||||
self,
|
||||
file_path: str,
|
||||
old_string: str,
|
||||
new_string: str,
|
||||
replace_all: bool = False,
|
||||
) -> EditResult:
|
||||
"""Edit only when the prerequisite archive read did not raise.
|
||||
|
||||
Args:
|
||||
file_path: Backend path to edit.
|
||||
old_string: Existing archive content.
|
||||
new_string: Archive content with the new section appended.
|
||||
replace_all: Whether to replace every match.
|
||||
|
||||
Returns:
|
||||
The backend edit result.
|
||||
"""
|
||||
self._ensure_read_succeeded()
|
||||
return self._backend.edit(
|
||||
file_path, old_string, new_string, replace_all=replace_all
|
||||
)
|
||||
|
||||
async def aedit(
|
||||
self,
|
||||
file_path: str,
|
||||
old_string: str,
|
||||
new_string: str,
|
||||
replace_all: bool = False,
|
||||
) -> EditResult:
|
||||
"""Asynchronously edit only after a successful archive read.
|
||||
|
||||
Args:
|
||||
file_path: Backend path to edit.
|
||||
old_string: Existing archive content.
|
||||
new_string: Archive content with the new section appended.
|
||||
replace_all: Whether to replace every match.
|
||||
|
||||
Returns:
|
||||
The backend edit result.
|
||||
"""
|
||||
self._ensure_read_succeeded()
|
||||
return await self._backend.aedit(
|
||||
file_path, old_string, new_string, replace_all=replace_all
|
||||
)
|
||||
|
||||
|
||||
class CLICompactionMiddleware(SummarizationToolMiddleware):
|
||||
"""Add explicit forced compaction and runtime model selection for dcode.
|
||||
|
||||
The SDK tool's normal, model-initiated behavior remains unchanged. The
|
||||
private `force` input is used only by the user-initiated `/offload` path,
|
||||
which must compact whenever messages exceed the retention window even when
|
||||
the conversation has not reached the SDK's proactive eligibility gate.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _offload_rejection(request: ToolCallRequest) -> ToolMessage | None:
|
||||
"""Reject every tool except the exact call seeded by `/offload`.
|
||||
|
||||
Args:
|
||||
request: Tool call about to be executed by the graph's tool node.
|
||||
|
||||
Returns:
|
||||
An error result for an unauthorized `/offload` tool call, otherwise
|
||||
`None` for an ordinary run or the exact seeded compaction call.
|
||||
"""
|
||||
expected_id = _offload_tool_call_id(request.runtime.context)
|
||||
if expected_id is None:
|
||||
return None
|
||||
|
||||
tool_call = request.tool_call
|
||||
args = tool_call.get("args")
|
||||
messages = request.state.get("messages", [])
|
||||
last_message = messages[-1] if messages else None
|
||||
last_message_id = (
|
||||
last_message.get("id")
|
||||
if isinstance(last_message, dict)
|
||||
else getattr(last_message, "id", None)
|
||||
)
|
||||
is_seeded_compaction = (
|
||||
tool_call.get("id") == expected_id
|
||||
and tool_call.get("name") == "compact_conversation"
|
||||
and isinstance(args, dict)
|
||||
and args.get("force") is True
|
||||
and last_message_id == _offload_seed_message_id(expected_id)
|
||||
)
|
||||
if is_seeded_compaction:
|
||||
return None
|
||||
|
||||
return ToolMessage(
|
||||
content=(
|
||||
"Not executed: /offload only authorizes its seeded "
|
||||
"conversation compaction call."
|
||||
),
|
||||
name=tool_call.get("name"),
|
||||
tool_call_id=tool_call["id"],
|
||||
status="error",
|
||||
)
|
||||
|
||||
def wrap_tool_call(
|
||||
self,
|
||||
request: ToolCallRequest,
|
||||
handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
|
||||
) -> ToolMessage | Command[Any]:
|
||||
"""Apply the `/offload` per-run tool guard before synchronous tools.
|
||||
|
||||
Args:
|
||||
request: Tool call about to be executed.
|
||||
handler: The remaining middleware/tool execution chain.
|
||||
|
||||
Returns:
|
||||
The guarded rejection or the downstream tool result.
|
||||
"""
|
||||
if (rejection := self._offload_rejection(request)) is not None:
|
||||
return rejection
|
||||
return handler(request)
|
||||
|
||||
async def awrap_tool_call(
|
||||
self,
|
||||
request: ToolCallRequest,
|
||||
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],
|
||||
) -> ToolMessage | Command[Any]:
|
||||
"""Apply the `/offload` per-run tool guard before asynchronous tools.
|
||||
|
||||
Args:
|
||||
request: Tool call about to be executed.
|
||||
handler: The remaining middleware/tool execution chain.
|
||||
|
||||
Returns:
|
||||
The guarded rejection or the downstream tool result.
|
||||
"""
|
||||
if (rejection := self._offload_rejection(request)) is not None:
|
||||
return rejection
|
||||
return await handler(request)
|
||||
|
||||
def _create_compact_tool(self) -> StructuredTool:
|
||||
"""Create the CLI variant of `compact_conversation`.
|
||||
|
||||
Returns:
|
||||
A tool that accepts the `/offload`-only `force` flag.
|
||||
"""
|
||||
middleware = self
|
||||
|
||||
# `force` is annotated `InjectedToolArg` so it is stripped from the
|
||||
# schema the model sees. ToolNode also strips the seeded value before
|
||||
# invocation, so forced mode is selected from the trusted runtime
|
||||
# context after `_offload_rejection` validates the raw tool call.
|
||||
def sync_compact(
|
||||
runtime: ToolRuntime,
|
||||
force: Annotated[bool, InjectedToolArg] = False,
|
||||
) -> Command:
|
||||
del force
|
||||
if _offload_tool_call_id(runtime.context) != runtime.tool_call_id:
|
||||
return middleware._run_compact(runtime)
|
||||
return middleware._run_forced_compact(runtime)
|
||||
|
||||
async def async_compact(
|
||||
runtime: ToolRuntime,
|
||||
force: Annotated[bool, InjectedToolArg] = False,
|
||||
) -> Command:
|
||||
del force
|
||||
if _offload_tool_call_id(runtime.context) != runtime.tool_call_id:
|
||||
return await middleware._arun_compact(runtime)
|
||||
return await middleware._arun_forced_compact(runtime)
|
||||
|
||||
return StructuredTool.from_function(
|
||||
name="compact_conversation",
|
||||
description=(
|
||||
"Compact the conversation by summarizing older messages into "
|
||||
"a concise summary. Use this proactively when the conversation "
|
||||
"is getting long to free up context window space."
|
||||
),
|
||||
func=sync_compact,
|
||||
coroutine=async_compact,
|
||||
)
|
||||
|
||||
def _resolve_backend(self, runtime: ToolRuntime) -> BackendProtocol:
|
||||
"""Resolve the backend with fail-closed archive append behavior.
|
||||
|
||||
Args:
|
||||
runtime: Runtime used to resolve backend factories.
|
||||
|
||||
Returns:
|
||||
A backend adapter that refuses writes after raised archive reads.
|
||||
"""
|
||||
backend = super()._resolve_backend(runtime)
|
||||
return cast("BackendProtocol", _ArchiveReadGuard(backend))
|
||||
|
||||
def _summarization_for_runtime(
|
||||
self, runtime: ToolRuntime
|
||||
) -> SummarizationMiddleware:
|
||||
"""Build a summarizer for the active runtime model when overridden.
|
||||
|
||||
Args:
|
||||
runtime: Runtime carrying the current `CLIContext`.
|
||||
|
||||
Returns:
|
||||
The startup summarizer when no runtime model is selected, otherwise
|
||||
a model-aware summarizer using the same resolved backend.
|
||||
"""
|
||||
config = _runtime_model_config(runtime)
|
||||
if not config.model_spec:
|
||||
return self._summarization
|
||||
|
||||
from deepagents_code.config import create_model
|
||||
|
||||
model = create_model(
|
||||
config.model_spec,
|
||||
extra_kwargs=config.model_params or None,
|
||||
profile_overrides=config.profile_overrides or None,
|
||||
).model
|
||||
context_limit = config.context_limit
|
||||
if context_limit is not None:
|
||||
profile = getattr(model, "profile", None)
|
||||
native = (
|
||||
profile.get("max_input_tokens") if isinstance(profile, dict) else None
|
||||
)
|
||||
if native != context_limit:
|
||||
merged = (
|
||||
{**profile, "max_input_tokens": context_limit}
|
||||
if isinstance(profile, dict)
|
||||
else {"max_input_tokens": context_limit}
|
||||
)
|
||||
try:
|
||||
model.profile = merged # ty: ignore[invalid-assignment]
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
logger.warning(
|
||||
"Could not apply runtime context limit %d to the offload "
|
||||
"model profile; using its resolved profile",
|
||||
context_limit,
|
||||
exc_info=True,
|
||||
)
|
||||
backend = self._resolve_backend(runtime)
|
||||
return create_summarization_middleware(model, backend)
|
||||
|
||||
def _run_forced_compact(self, runtime: ToolRuntime) -> Command:
|
||||
"""Synchronously compact without the SDK eligibility gate.
|
||||
|
||||
This deliberately mirrors the SDK's own `_run_compact` step sequence
|
||||
(apply prior event, determine cutoff, partition, summarize, offload,
|
||||
build result) minus the eligibility gate. Because it is a fork rather
|
||||
than an override, it must be kept in parity when the SDK's compaction
|
||||
flow changes; the closest-fitting SDK-side fix (a `force=` seam on
|
||||
`_run_compact`) is out of scope for this PR, which is confined to
|
||||
Deep Agents Code. `test_forced_compact_matches_sdk_summarizer_calls`
|
||||
guards the summarizer-method call set against drift, but only by
|
||||
*existence*: it catches a renamed or removed dependency, not a changed
|
||||
signature nor a new step added to `_run_compact` (e.g. if the SDK later
|
||||
moved inline-media offload into the gated path). Two known consequences
|
||||
of that today: this fork does not call `_offload_inline_media` (only the
|
||||
auto `wrap_model_call` path does), so inline base64 media in compacted
|
||||
messages is not offloaded to referenceable paths and is dropped from the
|
||||
XML archive -- pre-existing SDK tool-path behavior, not introduced here.
|
||||
|
||||
Returns:
|
||||
The compaction state update or an error tool message.
|
||||
"""
|
||||
tool_call_id = runtime.tool_call_id or ""
|
||||
try:
|
||||
summarization = self._summarization_for_runtime(runtime)
|
||||
messages = runtime.state.get("messages", [])
|
||||
event = runtime.state.get("_summarization_event")
|
||||
effective = summarization._apply_event_to_messages(messages, event)
|
||||
effective = _without_offload_seed(effective, tool_call_id)
|
||||
cutoff = summarization._determine_cutoff_index(effective)
|
||||
if cutoff == 0:
|
||||
return self._nothing_to_compact(tool_call_id)
|
||||
|
||||
to_summarize, _ = summarization._partition_messages(effective, cutoff)
|
||||
summary = summarization._create_summary(to_summarize)
|
||||
backend = self._resolve_backend(runtime)
|
||||
file_path = summarization._offload_to_backend(backend, to_summarize)
|
||||
# The inherited `_build_compact_result` produces the same event and
|
||||
# tool message as the SDK's gated path via model-independent helpers
|
||||
# (string formatting + a staticmethod), so the runtime-selected
|
||||
# summarizer is not needed to build it. Kept inside the `try` so a
|
||||
# failure here still returns a ToolMessage rather than raising.
|
||||
return self._build_compact_result(
|
||||
runtime, to_summarize, summary, file_path, event, cutoff
|
||||
)
|
||||
except Exception as exc: # tool errors must surface as ToolMessages
|
||||
logger.exception("forced compact_conversation failed")
|
||||
return self._forced_compact_error(tool_call_id, exc)
|
||||
|
||||
async def _arun_forced_compact(self, runtime: ToolRuntime) -> Command:
|
||||
"""Asynchronously compact without the SDK eligibility gate.
|
||||
|
||||
Returns:
|
||||
The compaction state update or an error tool message.
|
||||
"""
|
||||
tool_call_id = runtime.tool_call_id or ""
|
||||
try:
|
||||
summarization = await asyncio.to_thread(
|
||||
self._summarization_for_runtime, runtime
|
||||
)
|
||||
messages = runtime.state.get("messages", [])
|
||||
event = runtime.state.get("_summarization_event")
|
||||
effective = summarization._apply_event_to_messages(messages, event)
|
||||
effective = _without_offload_seed(effective, tool_call_id)
|
||||
cutoff = summarization._determine_cutoff_index(effective)
|
||||
if cutoff == 0:
|
||||
return self._nothing_to_compact(tool_call_id)
|
||||
|
||||
to_summarize, _ = summarization._partition_messages(effective, cutoff)
|
||||
summary = await summarization._acreate_summary(to_summarize)
|
||||
backend = self._resolve_backend(runtime)
|
||||
file_path = await summarization._aoffload_to_backend(backend, to_summarize)
|
||||
# See `_run_forced_compact` for why the inherited builder is reused
|
||||
# and why it stays inside the `try`.
|
||||
return self._build_compact_result(
|
||||
runtime, to_summarize, summary, file_path, event, cutoff
|
||||
)
|
||||
except Exception as exc: # tool errors must surface as ToolMessages
|
||||
logger.exception("forced compact_conversation failed")
|
||||
return self._forced_compact_error(tool_call_id, exc)
|
||||
|
||||
@staticmethod
|
||||
def _forced_compact_error(tool_call_id: str, exc: Exception) -> Command:
|
||||
"""Build a forced-compaction failure result with a stable prefix.
|
||||
|
||||
Owned by dcode so the `/offload` client can detect failures via
|
||||
`COMPACTION_FAILURE_PREFIX`. The tool must return a `ToolMessage` rather
|
||||
than raise, so the model (and the client) see the failure as ordinary
|
||||
tool output.
|
||||
|
||||
The message is intentionally generic about *where* the failure occurred:
|
||||
the guarded body spans cutoff determination, summary generation, the
|
||||
archive write, and result building, so it does not assert a specific
|
||||
stage (and does not claim nothing was written — an archive may have been
|
||||
persisted before a later step failed). It states only what is always
|
||||
true on this path: the summarization event was not committed, so the
|
||||
effective conversation is unchanged.
|
||||
|
||||
Args:
|
||||
tool_call_id: The originating tool call ID.
|
||||
exc: The exception raised while compacting.
|
||||
|
||||
Returns:
|
||||
A `Command` whose `ToolMessage` content starts with
|
||||
`COMPACTION_FAILURE_PREFIX`.
|
||||
"""
|
||||
return Command(
|
||||
update={
|
||||
"messages": [
|
||||
ToolMessage(
|
||||
content=(
|
||||
f"{COMPACTION_FAILURE_PREFIX}: an error occurred "
|
||||
f"during compaction ({type(exc).__name__}: {exc}). "
|
||||
"Your conversation is unchanged."
|
||||
),
|
||||
tool_call_id=tool_call_id,
|
||||
)
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _create_cli_compaction_middleware(
|
||||
model: str | BaseChatModel,
|
||||
backend: BACKEND_TYPES,
|
||||
) -> CLICompactionMiddleware:
|
||||
"""Create the dcode compaction middleware from the SDK configuration.
|
||||
|
||||
Args:
|
||||
model: Startup model or model specification.
|
||||
backend: Agent backend used for archive persistence.
|
||||
|
||||
Returns:
|
||||
CLI compaction middleware with the SDK's model-aware defaults.
|
||||
"""
|
||||
sdk_middleware = create_summarization_tool_middleware(model, backend)
|
||||
return CLICompactionMiddleware(
|
||||
sdk_middleware._summarization,
|
||||
system_prompt=sdk_middleware.system_prompt,
|
||||
)
|
||||
@@ -56,6 +56,66 @@ def _event_field(event: object, key: str) -> object | None:
|
||||
return getattr(event, key, None)
|
||||
|
||||
|
||||
async def _read_file_through_agent(agent, *, thread_id: str, file_path: str) -> str:
|
||||
"""Read `file_path` via the running agent's own `read_file` tool.
|
||||
|
||||
Seeds a `read_file` tool call attributed to the model node and advances the
|
||||
graph so the agent's `ToolNode` executes the read against its own backend,
|
||||
proving the offloaded archive exists server-side (not in a client dir).
|
||||
Auto-approves any HITL interrupt the read raises.
|
||||
"""
|
||||
import uuid
|
||||
|
||||
from langchain.agents.middleware.human_in_the_loop import ApproveDecision
|
||||
from langchain_core.messages import AIMessage
|
||||
from langgraph.types import Command
|
||||
|
||||
config = {"configurable": {"thread_id": thread_id}}
|
||||
tool_call_id = str(uuid.uuid4())
|
||||
seed = AIMessage(
|
||||
content="",
|
||||
tool_calls=[
|
||||
{"name": "read_file", "args": {"file_path": file_path}, "id": tool_call_id}
|
||||
],
|
||||
)
|
||||
await agent.aensure_thread(config)
|
||||
await agent.aupdate_state(config, {"messages": [seed]}, as_node="model")
|
||||
|
||||
interrupt_ids: list[str] = []
|
||||
tool_contents: list[str] = []
|
||||
|
||||
async def _drain(stream_input) -> None:
|
||||
async for chunk in agent.astream(
|
||||
stream_input,
|
||||
stream_mode=["messages", "updates"],
|
||||
subgraphs=True,
|
||||
config=config,
|
||||
durability="exit",
|
||||
):
|
||||
if not isinstance(chunk, tuple) or len(chunk) != 3:
|
||||
continue
|
||||
_ns, mode, data = chunk
|
||||
if mode == "updates" and isinstance(data, dict):
|
||||
for interrupt_obj in data.get("__interrupt__", []) or []:
|
||||
iid = getattr(interrupt_obj, "id", None)
|
||||
if iid:
|
||||
interrupt_ids.append(iid)
|
||||
elif mode == "messages" and isinstance(data, tuple):
|
||||
msg = data[0]
|
||||
if type(msg).__name__ == "ToolMessage":
|
||||
tool_contents.append(str(getattr(msg, "content", "")))
|
||||
|
||||
await _drain(None)
|
||||
if interrupt_ids:
|
||||
resume = {
|
||||
iid: {"decisions": [ApproveDecision(type="approve")]}
|
||||
for iid in interrupt_ids
|
||||
}
|
||||
await _drain(Command(resume=resume))
|
||||
|
||||
return "\n".join(tool_contents)
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
async def test_compact_resumed_thread_uses_persisted_history(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
@@ -63,17 +123,16 @@ async def test_compact_resumed_thread_uses_persisted_history(
|
||||
"""Offloads a resumed thread after restart using remote server state.
|
||||
|
||||
The test seeds a real persisted thread on one server instance, restarts the
|
||||
server, resumes that thread in a fresh `DeepAgentsApp`, and verifies that
|
||||
`/offload` succeeds after the app registers the thread with the server.
|
||||
server, resumes that thread in a fresh `DeepAgentsApp` constructed the
|
||||
PRODUCTION way (`backend=None`), and verifies that `/offload` succeeds
|
||||
server-side and the archive stays readable through the agent's own backend.
|
||||
"""
|
||||
home_dir = tmp_path / "home"
|
||||
project_dir = tmp_path / "project"
|
||||
backend_root = tmp_path / "compact_backend"
|
||||
assistant_id = "itest-compact"
|
||||
|
||||
home_dir.mkdir()
|
||||
project_dir.mkdir()
|
||||
backend_root.mkdir()
|
||||
|
||||
# Keep config and the global sessions DB fully test-local.
|
||||
monkeypatch.setenv("HOME", str(home_dir))
|
||||
@@ -82,9 +141,6 @@ async def test_compact_resumed_thread_uses_persisted_history(
|
||||
|
||||
_write_model_config(home_dir)
|
||||
|
||||
from deepagents.backends.composite import CompositeBackend
|
||||
from deepagents.backends.filesystem import FilesystemBackend
|
||||
|
||||
from deepagents_code import model_config
|
||||
from deepagents_code.app import DeepAgentsApp
|
||||
from deepagents_code.client.launch.server_manager import server_session
|
||||
@@ -121,11 +177,6 @@ async def test_compact_resumed_thread_uses_persisted_history(
|
||||
prompt=_build_long_prompt(turn),
|
||||
)
|
||||
|
||||
compact_backend = CompositeBackend(
|
||||
default=FilesystemBackend(root_dir=backend_root, virtual_mode=True),
|
||||
routes={},
|
||||
)
|
||||
|
||||
# Server 2: same SQLite DB, but a fresh server process.
|
||||
async with server_session(
|
||||
assistant_id=assistant_id,
|
||||
@@ -137,10 +188,12 @@ async def test_compact_resumed_thread_uses_persisted_history(
|
||||
) as (agent, _server_proc):
|
||||
config = {"configurable": {"thread_id": thread_id}}
|
||||
|
||||
# Production construction: no client-owned backend. Offload runs
|
||||
# server-side through the agent's own `compact_conversation` tool.
|
||||
app = DeepAgentsApp(
|
||||
agent=agent, # ty: ignore
|
||||
assistant_id=assistant_id,
|
||||
backend=compact_backend,
|
||||
backend=None,
|
||||
cwd=project_dir,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
@@ -188,17 +241,32 @@ async def test_compact_resumed_thread_uses_persisted_history(
|
||||
cutoff = _event_field(summarization_event, "cutoff_index")
|
||||
assert isinstance(cutoff, int)
|
||||
assert cutoff > 0
|
||||
assert (
|
||||
_event_field(summarization_event, "file_path")
|
||||
== f"/conversation_history/{thread_id}.md"
|
||||
)
|
||||
archive_path = f"/conversation_history/{thread_id}.md"
|
||||
assert _event_field(summarization_event, "file_path") == archive_path
|
||||
|
||||
# The offloaded archive should land in the explicit temp-backed backend,
|
||||
# not the host filesystem root.
|
||||
archive_path = backend_root / "conversation_history" / f"{thread_id}.md"
|
||||
assert archive_path.exists()
|
||||
archive_text = archive_path.read_text()
|
||||
assert "Offloaded at" in archive_text
|
||||
assert "keeps enough unique detail" in archive_text
|
||||
# The archive must be readable THROUGH THE AGENT, proving the bytes
|
||||
# live in the agent's own composite backend server-side rather than
|
||||
# in a client-local directory the server can never read.
|
||||
read_back = await _read_file_through_agent(
|
||||
agent, thread_id=thread_id, file_path=archive_path
|
||||
)
|
||||
assert "keeps enough unique detail" in read_back
|
||||
assert "Summarized at" in read_back
|
||||
|
||||
# Server 3: the event and archive path must remain usable after the
|
||||
# process that performed the offload has exited.
|
||||
async with server_session(
|
||||
assistant_id=assistant_id,
|
||||
model_name="itest:fake",
|
||||
no_mcp=True,
|
||||
enable_shell=False,
|
||||
interactive=True,
|
||||
sandbox_type="none",
|
||||
) as (agent, _server_proc):
|
||||
read_back = await _read_file_through_agent(
|
||||
agent, thread_id=thread_id, file_path=archive_path
|
||||
)
|
||||
assert "keeps enough unique detail" in read_back
|
||||
assert "Summarized at" in read_back
|
||||
finally:
|
||||
model_config.clear_caches()
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
"""Integration coverage for the server-side `/offload` path.
|
||||
|
||||
`/offload` drives the agent's own `compact_conversation` tool (with
|
||||
`force=True`) server-side, so the offloaded archive lands in the agent's
|
||||
composite backend and is readable via `read_file` in every run mode — not in a
|
||||
client-local directory the server can never read. These tests construct the app
|
||||
the PRODUCTION way (`backend=None`) and prove the archive is readable *through
|
||||
the agent*.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _write_model_config(home_dir: Path) -> None:
|
||||
"""Write a temp config that points the server subprocess at the test model."""
|
||||
config_dir = home_dir / ".deepagents"
|
||||
config_dir.mkdir(parents=True, exist_ok=True)
|
||||
(config_dir / "config.toml").write_text(
|
||||
"""
|
||||
[models.providers.itest]
|
||||
class_path = "deepagents_code._testing_models:DeterministicIntegrationChatModel"
|
||||
models = ["fake"]
|
||||
""".strip()
|
||||
+ "\n"
|
||||
)
|
||||
|
||||
|
||||
def _build_long_prompt(turn: int) -> str:
|
||||
"""Build a long user message so the seeded thread is worth compacting."""
|
||||
sentence = (
|
||||
f"Turn {turn} keeps enough unique detail to make resume-compaction meaningful. "
|
||||
"The quick brown fox documents repeatable integration behavior for the CLI. "
|
||||
)
|
||||
return sentence * 30
|
||||
|
||||
|
||||
async def _run_turn(agent, *, thread_id: str, assistant_id: str, prompt: str) -> None:
|
||||
"""Execute one real remote agent turn and drain the stream to completion."""
|
||||
from deepagents_code.config import build_stream_config
|
||||
|
||||
config = build_stream_config(thread_id, assistant_id)
|
||||
stream_input = {"messages": [{"role": "user", "content": prompt}]}
|
||||
async for _chunk in agent.astream(
|
||||
stream_input,
|
||||
stream_mode=["messages", "updates"],
|
||||
subgraphs=True,
|
||||
config=config,
|
||||
durability="exit",
|
||||
):
|
||||
pass
|
||||
|
||||
|
||||
def _event_field(event: object, key: str) -> object | None:
|
||||
"""Read a summarization-event field from either dict or object form."""
|
||||
if isinstance(event, dict):
|
||||
return event.get(key) # ty: ignore
|
||||
return getattr(event, key, None)
|
||||
|
||||
|
||||
async def _read_file_through_agent(agent, *, thread_id: str, file_path: str) -> str:
|
||||
"""Read `file_path` via the running agent's own `read_file` tool.
|
||||
|
||||
Seeds a `read_file` tool call attributed to the model node and advances the
|
||||
graph so the agent's `ToolNode` executes the read against its own backend.
|
||||
This proves the offloaded archive exists server-side (not merely in a
|
||||
client-local directory). Auto-approves any HITL interrupt the read raises.
|
||||
|
||||
Returns:
|
||||
The concatenated content of every `ToolMessage` produced by the run.
|
||||
"""
|
||||
from langchain.agents.middleware.human_in_the_loop import ApproveDecision
|
||||
from langchain_core.messages import AIMessage
|
||||
from langgraph.types import Command
|
||||
|
||||
config = {"configurable": {"thread_id": thread_id}}
|
||||
tool_call_id = str(uuid.uuid4())
|
||||
seed = AIMessage(
|
||||
content="",
|
||||
tool_calls=[
|
||||
{"name": "read_file", "args": {"file_path": file_path}, "id": tool_call_id}
|
||||
],
|
||||
)
|
||||
await agent.aensure_thread(config)
|
||||
await agent.aupdate_state(config, {"messages": [seed]}, as_node="model")
|
||||
|
||||
interrupt_ids: list[str] = []
|
||||
tool_contents: list[str] = []
|
||||
|
||||
async def _drain(stream_input) -> None:
|
||||
async for chunk in agent.astream(
|
||||
stream_input,
|
||||
stream_mode=["messages", "updates"],
|
||||
subgraphs=True,
|
||||
config=config,
|
||||
durability="exit",
|
||||
):
|
||||
if not isinstance(chunk, tuple) or len(chunk) != 3:
|
||||
continue
|
||||
_ns, mode, data = chunk
|
||||
if mode == "updates" and isinstance(data, dict):
|
||||
for interrupt_obj in data.get("__interrupt__", []) or []:
|
||||
iid = getattr(interrupt_obj, "id", None)
|
||||
if iid:
|
||||
interrupt_ids.append(iid)
|
||||
elif mode == "messages" and isinstance(data, tuple):
|
||||
msg = data[0]
|
||||
if type(msg).__name__ == "ToolMessage":
|
||||
tool_contents.append(str(getattr(msg, "content", "")))
|
||||
|
||||
await _drain(None)
|
||||
if interrupt_ids:
|
||||
resume = {
|
||||
iid: {"decisions": [ApproveDecision(type="approve")]}
|
||||
for iid in interrupt_ids
|
||||
}
|
||||
await _drain(Command(resume=resume))
|
||||
|
||||
return "\n".join(tool_contents)
|
||||
|
||||
|
||||
@pytest.mark.timeout(240)
|
||||
async def test_offload_runs_server_side_and_is_agent_readable(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""`/offload` compacts server-side with `backend=None` and stays readable.
|
||||
|
||||
Constructs the app the production way (`backend=None`), seeds a thread with
|
||||
enough content, runs `/offload`, and asserts:
|
||||
|
||||
- no `ErrorMessage` and an "Offloaded " success message,
|
||||
- a persisted `_summarization_event` with `cutoff > 0` and
|
||||
`file_path == /conversation_history/<thread>.md`,
|
||||
- the archive is readable THROUGH THE AGENT (via its own `read_file` tool),
|
||||
proving the bytes live in the agent's backend server-side, and
|
||||
- local archives land in the persistent per-user history directory.
|
||||
"""
|
||||
home_dir = tmp_path / "home"
|
||||
project_dir = tmp_path / "project"
|
||||
assistant_id = "itest-offload"
|
||||
|
||||
home_dir.mkdir()
|
||||
project_dir.mkdir()
|
||||
|
||||
monkeypatch.setenv("HOME", str(home_dir))
|
||||
monkeypatch.setenv("DEEPAGENTS_CODE_NO_UPDATE_CHECK", "1")
|
||||
monkeypatch.chdir(project_dir)
|
||||
|
||||
_write_model_config(home_dir)
|
||||
|
||||
from deepagents_code import model_config
|
||||
from deepagents_code.app import DeepAgentsApp
|
||||
from deepagents_code.client.launch.server_manager import server_session
|
||||
from deepagents_code.config import create_model
|
||||
from deepagents_code.sessions import generate_thread_id
|
||||
from deepagents_code.tui.widgets.messages import AppMessage, ErrorMessage
|
||||
|
||||
config_path = home_dir / ".deepagents" / "config.toml"
|
||||
monkeypatch.setattr(model_config, "DEFAULT_CONFIG_DIR", config_path.parent)
|
||||
monkeypatch.setattr(model_config, "DEFAULT_CONFIG_PATH", config_path)
|
||||
|
||||
model_config.clear_caches()
|
||||
try:
|
||||
create_model("itest:fake").apply_to_settings()
|
||||
thread_id = generate_thread_id()
|
||||
|
||||
async with server_session(
|
||||
assistant_id=assistant_id,
|
||||
model_name="itest:fake",
|
||||
no_mcp=True,
|
||||
enable_shell=False,
|
||||
interactive=True,
|
||||
sandbox_type="none",
|
||||
) as (agent, _server_proc):
|
||||
for turn in range(1, 5):
|
||||
await _run_turn(
|
||||
agent,
|
||||
thread_id=thread_id,
|
||||
assistant_id=assistant_id,
|
||||
prompt=_build_long_prompt(turn),
|
||||
)
|
||||
|
||||
config = {"configurable": {"thread_id": thread_id}}
|
||||
|
||||
# Production construction: no client-owned backend.
|
||||
app = DeepAgentsApp(
|
||||
agent=agent, # ty: ignore
|
||||
assistant_id=assistant_id,
|
||||
backend=None,
|
||||
cwd=project_dir,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
for _ in range(120):
|
||||
await pilot.pause(0.1)
|
||||
if app._message_store.total_count > 0:
|
||||
break
|
||||
|
||||
assert app._message_store.total_count > 0
|
||||
|
||||
await app._handle_offload()
|
||||
|
||||
for _ in range(120):
|
||||
await pilot.pause(0.1)
|
||||
if any(
|
||||
"Offloaded " in str(widget._content)
|
||||
for widget in app.query(AppMessage)
|
||||
):
|
||||
break
|
||||
|
||||
app_messages = [
|
||||
str(widget._content) for widget in app.query(AppMessage)
|
||||
]
|
||||
error_messages = [
|
||||
str(widget._content) for widget in app.query(ErrorMessage)
|
||||
]
|
||||
|
||||
assert not error_messages
|
||||
assert "Nothing to offload" not in "\n".join(app_messages)
|
||||
assert any("Offloaded " in content for content in app_messages)
|
||||
|
||||
# The summarization event must be visible through server state.
|
||||
state = await agent.aget_state(config)
|
||||
values = getattr(state, "values", None) or {}
|
||||
summarization_event = values.get("_summarization_event")
|
||||
assert summarization_event is not None
|
||||
cutoff = _event_field(summarization_event, "cutoff_index")
|
||||
assert isinstance(cutoff, int)
|
||||
assert cutoff > 0
|
||||
archive_path = f"/conversation_history/{thread_id}.md"
|
||||
assert _event_field(summarization_event, "file_path") == archive_path
|
||||
|
||||
# CRUCIAL: the archive must be readable THROUGH THE AGENT, proving
|
||||
# the bytes exist in the agent's own backend server-side.
|
||||
read_back = await _read_file_through_agent(
|
||||
agent, thread_id=thread_id, file_path=archive_path
|
||||
)
|
||||
assert "keeps enough unique detail" in read_back
|
||||
# The SDK middleware writes a "## Summarized at" archive header.
|
||||
assert "Summarized at" in read_back
|
||||
|
||||
persistent_archive = (
|
||||
home_dir / ".deepagents" / "conversation_history" / f"{thread_id}.md"
|
||||
)
|
||||
assert persistent_archive.exists()
|
||||
assert "keeps enough unique detail" in persistent_archive.read_text()
|
||||
finally:
|
||||
model_config.clear_caches()
|
||||
@@ -340,7 +340,7 @@ Some paths returned by the file tools are virtual mounts:
|
||||
Do not assume that a path returned by a file tool can be used directly in a shell command.
|
||||
|
||||
Host path mappings:
|
||||
- `/conversation_history/` -> `<tmp_path>/deepagents_conversation_history/` (e.g. `/conversation_history/dir/x.py` -> `<tmp_path>/deepagents_conversation_history/dir/x.py`)
|
||||
- `/conversation_history/` -> `<tmp_path>/.deepagents/conversation_history/` (e.g. `/conversation_history/dir/x.py` -> `<tmp_path>/.deepagents/conversation_history/dir/x.py`)
|
||||
- `/large_tool_results/` -> `<tmp_path>/deepagents_large_results/` (e.g. `/large_tool_results/dir/x.py` -> `<tmp_path>/deepagents_large_results/dir/x.py`)
|
||||
|
||||
## `task` (subagent spawner)
|
||||
|
||||
@@ -132,13 +132,18 @@ def _mock_settings(tmp_path: Path) -> Generator[None, None, None]:
|
||||
mock_s.user_langchain_project = None
|
||||
mock_s.shell_allow_list = None
|
||||
|
||||
# Patch tempfile.mkdtemp inside agent.py so the conversation-history
|
||||
# and large-results temp directories are deterministic across runs
|
||||
# (the real mkdtemp appends a random suffix that breaks snapshots).
|
||||
# Patch generated backend roots so host-path mappings are deterministic
|
||||
# across machines and runs.
|
||||
def _fake_mkdtemp(prefix: str = "", **_kw: Any) -> str:
|
||||
return str(tmp_path / prefix.rstrip("_"))
|
||||
|
||||
with patch("deepagents_code.agent.tempfile.mkdtemp", _fake_mkdtemp):
|
||||
with (
|
||||
patch("deepagents_code.agent.tempfile.mkdtemp", _fake_mkdtemp),
|
||||
patch(
|
||||
"deepagents_code.agent._offload_fallback_root",
|
||||
return_value=tmp_path / ".deepagents",
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
|
||||
@@ -107,6 +107,32 @@ def test_add_interrupt_on_attaches_auto_approve_predicate() -> None:
|
||||
assert config.get("when") is _should_interrupt_tool_call
|
||||
|
||||
|
||||
def test_local_conversation_history_route_is_persistent(tmp_path: Path) -> None:
|
||||
"""Local archives use the stable user data directory across server restarts."""
|
||||
history_root = tmp_path / ".deepagents"
|
||||
model = _make_fake_chat_model()
|
||||
|
||||
with patch(
|
||||
"deepagents_code.agent._offload_fallback_root", return_value=history_root
|
||||
):
|
||||
_agent, backend = create_cli_agent(
|
||||
model=model,
|
||||
assistant_id="test-agent",
|
||||
enable_memory=False,
|
||||
enable_skills=False,
|
||||
enable_shell=False,
|
||||
system_prompt="test prompt",
|
||||
cwd=tmp_path,
|
||||
)
|
||||
|
||||
result = backend.write("/conversation_history/thread.md", "archived")
|
||||
|
||||
assert result.error is None
|
||||
assert (
|
||||
history_root / "conversation_history" / "thread.md"
|
||||
).read_text() == "archived"
|
||||
|
||||
|
||||
def _request_with_context(
|
||||
context: object,
|
||||
*,
|
||||
@@ -320,6 +346,33 @@ def test_cli_context_field_parity() -> None:
|
||||
assert typed_dict_keys == dataclass_keys
|
||||
|
||||
|
||||
def test_get_context_preserves_compaction_fields_from_dict() -> None:
|
||||
"""`_get_context` must carry the compaction fields across the dict boundary.
|
||||
|
||||
On the RemoteGraph path the run context arrives as a dict and
|
||||
`_get_context` reconstructs `CLIContextSchema` field-by-field. The field
|
||||
parity test only guards the *declarations*; this guards the coercion, so
|
||||
`profile_overrides`/`model_context_limit` (which `/offload` reads via the
|
||||
compaction middleware) are not silently dropped on that path.
|
||||
"""
|
||||
from deepagents_code.configurable_model import _get_context
|
||||
|
||||
ctx = {
|
||||
"model": "anthropic:claude-haiku-4-5-20251001",
|
||||
"model_params": {"temperature": 0.5},
|
||||
"profile_overrides": {"max_input_tokens": 12345},
|
||||
"model_context_limit": 4096,
|
||||
}
|
||||
request = cast("Any", SimpleNamespace(runtime=SimpleNamespace(context=ctx)))
|
||||
|
||||
resolved = _get_context(request)
|
||||
|
||||
assert resolved is not None
|
||||
assert resolved.profile_overrides == {"max_input_tokens": 12345}
|
||||
assert resolved.model_context_limit == 4096
|
||||
assert resolved.model_params == {"temperature": 0.5}
|
||||
|
||||
|
||||
def test_sanitize_agent_message_name_replaces_provider_unsafe_chars() -> None:
|
||||
"""Agent display names with spaces must become valid message names."""
|
||||
assert _sanitize_agent_message_name("my agent") == "my_agent"
|
||||
|
||||
@@ -6,10 +6,27 @@ Core compact tool logic tests live in the SDK at
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from deepagents.backends.protocol import FileDownloadResponse, WriteResult
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
|
||||
from deepagents_code._cli_context import CLIContextSchema
|
||||
from deepagents_code.offload_middleware import (
|
||||
COMPACTION_FAILURE_PREFIX,
|
||||
CLICompactionMiddleware,
|
||||
_ArchiveReadGuard,
|
||||
_runtime_model_config,
|
||||
)
|
||||
from deepagents_code.tool_display import format_tool_display
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from deepagents.backends.protocol import BackendProtocol
|
||||
from langchain_core.messages import AnyMessage
|
||||
|
||||
|
||||
class TestHITLGating:
|
||||
"""Test that compact_conversation HITL gating respects the constant."""
|
||||
@@ -38,3 +55,606 @@ class TestDisplayFormatting:
|
||||
"""format_tool_display should return the expected string."""
|
||||
result = format_tool_display("compact_conversation", {})
|
||||
assert "compact_conversation()" in result
|
||||
|
||||
|
||||
class TestArchiveReadGuard:
|
||||
"""Cover fail-closed archive writes after backend read errors."""
|
||||
|
||||
def test_sync_error_response_blocks_write(self) -> None:
|
||||
"""A synchronous error response must not permit a truncating write."""
|
||||
response = FileDownloadResponse(
|
||||
path="/conversation_history/thread.md",
|
||||
error="permission_denied",
|
||||
)
|
||||
backend = MagicMock()
|
||||
backend.download_files.return_value = [response]
|
||||
backend.write.return_value = WriteResult(path=response.path)
|
||||
guard = _ArchiveReadGuard(backend)
|
||||
|
||||
assert guard.download_files([response.path]) == [response]
|
||||
with pytest.raises(RuntimeError, match="refusing to overwrite"):
|
||||
guard.write(response.path, "new history")
|
||||
|
||||
backend.write.assert_not_called()
|
||||
|
||||
async def test_async_error_response_blocks_write(self) -> None:
|
||||
"""An asynchronous error response must not permit a truncating write."""
|
||||
response = FileDownloadResponse(
|
||||
path="/conversation_history/thread.md",
|
||||
error="transient backend error",
|
||||
)
|
||||
backend = MagicMock()
|
||||
backend.adownload_files = AsyncMock(return_value=[response])
|
||||
backend.awrite = AsyncMock(return_value=WriteResult(path=response.path))
|
||||
guard = _ArchiveReadGuard(backend)
|
||||
|
||||
assert await guard.adownload_files([response.path]) == [response]
|
||||
with pytest.raises(RuntimeError, match="refusing to overwrite"):
|
||||
await guard.awrite(response.path, "new history")
|
||||
|
||||
backend.awrite.assert_not_awaited()
|
||||
|
||||
def test_missing_archive_allows_create(self) -> None:
|
||||
"""A missing archive remains the expected first-write path."""
|
||||
response = FileDownloadResponse(
|
||||
path="/conversation_history/thread.md",
|
||||
error="file_not_found",
|
||||
)
|
||||
backend = MagicMock()
|
||||
backend.download_files.return_value = [response]
|
||||
expected = WriteResult(path=response.path)
|
||||
backend.write.return_value = expected
|
||||
guard = _ArchiveReadGuard(backend)
|
||||
|
||||
assert guard.download_files([response.path]) == [response]
|
||||
assert guard.write(response.path, "new history") == expected
|
||||
|
||||
|
||||
class TestCLICompactionMiddleware:
|
||||
"""Cover dcode's explicit `/offload` behavior layered over the SDK tool."""
|
||||
|
||||
@staticmethod
|
||||
def _summarization() -> MagicMock:
|
||||
summarization = MagicMock()
|
||||
summarization._backend = object()
|
||||
summarization._apply_event_to_messages.side_effect = lambda messages, _event: (
|
||||
messages
|
||||
)
|
||||
summarization._determine_cutoff_index.return_value = 2
|
||||
summarization._partition_messages.side_effect = lambda messages, cutoff: (
|
||||
messages[:cutoff],
|
||||
messages[cutoff:],
|
||||
)
|
||||
summarization._acreate_summary = AsyncMock(return_value="Summary")
|
||||
summarization._aoffload_to_backend = AsyncMock(
|
||||
return_value="/conversation_history/thread.md"
|
||||
)
|
||||
summarization._build_new_messages_with_path.return_value = [
|
||||
HumanMessage(content="Summary")
|
||||
]
|
||||
summarization._compute_state_cutoff.return_value = 2
|
||||
return summarization
|
||||
|
||||
async def test_force_bypasses_sdk_eligibility_gate(self) -> None:
|
||||
"""Forced compaction partitions directly even below the proactive gate."""
|
||||
summarization = self._summarization()
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
runtime = MagicMock()
|
||||
runtime.context = None
|
||||
runtime.state = {"messages": [HumanMessage("one"), HumanMessage("two")]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = await middleware._arun_forced_compact(runtime)
|
||||
|
||||
summarization._is_eligible_for_compaction.assert_not_called()
|
||||
summarization._acreate_summary.assert_awaited_once()
|
||||
assert result.update is not None
|
||||
assert result.update["_summarization_event"]["cutoff_index"] == 2
|
||||
|
||||
def test_runtime_model_builds_matching_summarizer(self) -> None:
|
||||
"""A `/model` override selects the summarizer used by `/offload`."""
|
||||
startup = self._summarization()
|
||||
middleware = CLICompactionMiddleware(startup)
|
||||
runtime = MagicMock()
|
||||
runtime.context = {
|
||||
"model": "provider:active-model",
|
||||
"model_params": {"temperature": 0},
|
||||
}
|
||||
active_model = object()
|
||||
result = SimpleNamespace(model=active_model)
|
||||
selected = MagicMock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"deepagents_code.config.create_model", return_value=result
|
||||
) as create_model,
|
||||
patch(
|
||||
"deepagents_code.offload_middleware.create_summarization_middleware",
|
||||
return_value=selected,
|
||||
) as create_summarization,
|
||||
):
|
||||
actual = middleware._summarization_for_runtime(runtime)
|
||||
|
||||
assert actual is selected
|
||||
create_model.assert_called_once_with(
|
||||
"provider:active-model",
|
||||
extra_kwargs={"temperature": 0},
|
||||
profile_overrides=None,
|
||||
)
|
||||
create_summarization.assert_called_once()
|
||||
assert create_summarization.call_args.args[0] is active_model
|
||||
guarded_backend = create_summarization.call_args.args[1]
|
||||
assert guarded_backend._backend is startup._backend
|
||||
|
||||
def test_runtime_profile_overrides_and_context_limit_are_applied(self) -> None:
|
||||
"""Server-side offload uses the CLI's effective model profile."""
|
||||
startup = self._summarization()
|
||||
middleware = CLICompactionMiddleware(startup)
|
||||
runtime = MagicMock()
|
||||
runtime.context = {
|
||||
"model": "provider:active-model",
|
||||
"model_params": {},
|
||||
"profile_overrides": {"max_input_tokens": 32_000},
|
||||
"model_context_limit": 24_000,
|
||||
}
|
||||
active_model = SimpleNamespace(profile={"max_input_tokens": 200_000})
|
||||
result = SimpleNamespace(model=active_model)
|
||||
selected = MagicMock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"deepagents_code.config.create_model", return_value=result
|
||||
) as create_model,
|
||||
patch(
|
||||
"deepagents_code.offload_middleware.create_summarization_middleware",
|
||||
return_value=selected,
|
||||
) as create_summarization,
|
||||
):
|
||||
actual = middleware._summarization_for_runtime(runtime)
|
||||
|
||||
assert actual is selected
|
||||
create_model.assert_called_once_with(
|
||||
"provider:active-model",
|
||||
extra_kwargs=None,
|
||||
profile_overrides={"max_input_tokens": 32_000},
|
||||
)
|
||||
assert active_model.profile["max_input_tokens"] == 24_000
|
||||
create_summarization.assert_called_once()
|
||||
assert create_summarization.call_args.args[0] is active_model
|
||||
guarded_backend = create_summarization.call_args.args[1]
|
||||
assert guarded_backend._backend is startup._backend
|
||||
|
||||
async def test_force_noops_when_nothing_old_enough(self) -> None:
|
||||
"""Forced compaction still no-ops at cutoff 0 (bypasses only the gate)."""
|
||||
summarization = self._summarization()
|
||||
summarization._determine_cutoff_index.return_value = 0
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
runtime = MagicMock()
|
||||
runtime.context = None
|
||||
runtime.state = {"messages": [HumanMessage("one")]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = await middleware._arun_forced_compact(runtime)
|
||||
|
||||
assert result.update is not None
|
||||
assert "_summarization_event" not in result.update
|
||||
summarization._acreate_summary.assert_not_awaited()
|
||||
assert "Nothing to compact" in result.update["messages"][0].content
|
||||
|
||||
async def test_async_force_excludes_seed_from_retention_cutoff(self) -> None:
|
||||
"""The async cutoff is calculated from the pre-seed conversation."""
|
||||
summarization = self._summarization()
|
||||
summarization._determine_cutoff_index.side_effect = lambda messages: (
|
||||
0 if len(messages) == 6 else 1
|
||||
)
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
conversation = [HumanMessage(str(index)) for index in range(6)]
|
||||
seed = AIMessage(
|
||||
content="",
|
||||
id="offload-seed-tool-call",
|
||||
tool_calls=[
|
||||
{
|
||||
"name": "compact_conversation",
|
||||
"args": {"force": True},
|
||||
"id": "tool-call",
|
||||
}
|
||||
],
|
||||
)
|
||||
runtime = MagicMock()
|
||||
runtime.context = None
|
||||
runtime.state = {"messages": [*conversation, seed]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = await middleware._arun_forced_compact(runtime)
|
||||
|
||||
assert result.update is not None
|
||||
assert "_summarization_event" not in result.update
|
||||
summarization._determine_cutoff_index.assert_called_once_with(conversation)
|
||||
summarization._partition_messages.assert_not_called()
|
||||
|
||||
def test_sync_force_excludes_serialized_seed_from_retention_cutoff(self) -> None:
|
||||
"""The sync cutoff also ignores a serialized synthetic seed."""
|
||||
summarization = self._summarization()
|
||||
summarization._determine_cutoff_index.side_effect = lambda messages: (
|
||||
0 if len(messages) == 6 else 1
|
||||
)
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
conversation = [HumanMessage(str(index)) for index in range(6)]
|
||||
seed = {"id": "offload-seed-tool-call", "type": "ai", "content": ""}
|
||||
runtime = MagicMock()
|
||||
runtime.context = None
|
||||
runtime.state = {"messages": [*conversation, seed]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = middleware._run_forced_compact(runtime)
|
||||
|
||||
assert result.update is not None
|
||||
assert "_summarization_event" not in result.update
|
||||
summarization._determine_cutoff_index.assert_called_once_with(conversation)
|
||||
summarization._partition_messages.assert_not_called()
|
||||
|
||||
async def test_forced_compact_error_when_summary_fails(self) -> None:
|
||||
"""A summary failure returns the failure prefix and does not compact."""
|
||||
summarization = self._summarization()
|
||||
summarization._acreate_summary = AsyncMock(side_effect=RuntimeError("boom"))
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
runtime = MagicMock()
|
||||
runtime.context = None
|
||||
runtime.state = {"messages": [HumanMessage("one"), HumanMessage("two")]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = await middleware._arun_forced_compact(runtime)
|
||||
|
||||
# The failure must NOT persist an event, and must carry the stable
|
||||
# prefix the `/offload` client keys on.
|
||||
assert result.update is not None
|
||||
assert "_summarization_event" not in result.update
|
||||
content = result.update["messages"][0].content
|
||||
assert content.startswith(COMPACTION_FAILURE_PREFIX)
|
||||
assert "RuntimeError" in content
|
||||
|
||||
def test_sync_forced_compact_compacts(self) -> None:
|
||||
"""The synchronous forced path mirrors the async one."""
|
||||
summarization = self._summarization()
|
||||
summarization._create_summary.return_value = "Summary"
|
||||
summarization._offload_to_backend.return_value = (
|
||||
"/conversation_history/thread.md"
|
||||
)
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
runtime = MagicMock()
|
||||
runtime.context = None
|
||||
runtime.state = {"messages": [HumanMessage("one"), HumanMessage("two")]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = middleware._run_forced_compact(runtime)
|
||||
|
||||
summarization._create_summary.assert_called_once()
|
||||
assert result.update is not None
|
||||
assert result.update["_summarization_event"]["cutoff_index"] == 2
|
||||
|
||||
def test_force_is_hidden_from_model_schema(self) -> None:
|
||||
"""`force` must not appear in the schema the model sees."""
|
||||
middleware = CLICompactionMiddleware(self._summarization())
|
||||
tool = middleware.tools[0]
|
||||
# `tool_call_schema` is a pydantic model (or, rarely, a dict); either
|
||||
# way the model-facing property set must not expose `force`.
|
||||
schema: Any = tool.tool_call_schema
|
||||
props = (
|
||||
schema.get("properties", {})
|
||||
if isinstance(schema, dict)
|
||||
else schema.model_json_schema().get("properties", {})
|
||||
)
|
||||
assert "force" not in props
|
||||
|
||||
def test_ordinary_context_delegates_to_gated_path(self) -> None:
|
||||
"""Caller-supplied `force` cannot bypass the trusted runtime context."""
|
||||
middleware = CLICompactionMiddleware(self._summarization())
|
||||
tool: Any = middleware.tools[0]
|
||||
runtime = MagicMock()
|
||||
runtime.context = {}
|
||||
runtime.tool_call_id = "model-call"
|
||||
with (
|
||||
patch.object(middleware, "_run_compact", return_value="gated") as gated,
|
||||
patch.object(
|
||||
middleware, "_run_forced_compact", return_value="forced"
|
||||
) as forced,
|
||||
):
|
||||
assert tool.func(runtime, force=False) == "gated"
|
||||
assert tool.func(runtime, force=True) == "gated"
|
||||
assert gated.call_count == 2
|
||||
forced.assert_not_called()
|
||||
|
||||
async def test_offload_context_delegates_to_forced_path_async(self) -> None:
|
||||
"""The authorized call ID in runtime context selects forced mode."""
|
||||
middleware = CLICompactionMiddleware(self._summarization())
|
||||
tool: Any = middleware.tools[0]
|
||||
runtime = MagicMock()
|
||||
runtime.context = {"offload_tool_call_id": "offload-call"}
|
||||
runtime.tool_call_id = "offload-call"
|
||||
with (
|
||||
patch.object(
|
||||
middleware,
|
||||
"_arun_compact",
|
||||
new_callable=AsyncMock,
|
||||
return_value="gated",
|
||||
) as gated,
|
||||
patch.object(
|
||||
middleware,
|
||||
"_arun_forced_compact",
|
||||
new_callable=AsyncMock,
|
||||
return_value="forced",
|
||||
) as forced,
|
||||
):
|
||||
# ToolNode replaces the seeded `force=True` with this default.
|
||||
assert await tool.coroutine(runtime, force=False) == "forced"
|
||||
gated.assert_not_awaited()
|
||||
forced.assert_awaited_once_with(runtime)
|
||||
|
||||
async def test_tool_node_preserves_forced_mode_via_runtime_context(self) -> None:
|
||||
"""A real ToolNode strips `force` but still reaches forced compaction."""
|
||||
from langchain_core.messages import ToolMessage
|
||||
from langgraph.graph import END, START, StateGraph
|
||||
from langgraph.prebuilt import ToolNode
|
||||
from langgraph.types import Command
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
class ToolState(TypedDict):
|
||||
messages: list[object]
|
||||
|
||||
middleware = CLICompactionMiddleware(self._summarization())
|
||||
# LangGraph accepts these runtime schemas, but its generic bound is not
|
||||
# recognized by ty on Python 3.14.
|
||||
builder = StateGraph(
|
||||
ToolState, # ty: ignore[invalid-argument-type]
|
||||
context_schema=CLIContextSchema,
|
||||
)
|
||||
builder.add_node("tools", ToolNode(middleware.tools))
|
||||
builder.add_edge(START, "tools")
|
||||
builder.add_edge("tools", END)
|
||||
graph = builder.compile()
|
||||
tool_call_id = "offload-call"
|
||||
seed = AIMessage(
|
||||
content="",
|
||||
id=f"offload-seed-{tool_call_id}",
|
||||
tool_calls=[
|
||||
{
|
||||
"name": "compact_conversation",
|
||||
"args": {"force": True},
|
||||
"id": tool_call_id,
|
||||
}
|
||||
],
|
||||
)
|
||||
command = Command(
|
||||
update={
|
||||
"messages": [
|
||||
ToolMessage(content="compacted", tool_call_id=tool_call_id)
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
middleware,
|
||||
"_arun_compact",
|
||||
new_callable=AsyncMock,
|
||||
return_value=command,
|
||||
) as gated,
|
||||
patch.object(
|
||||
middleware,
|
||||
"_arun_forced_compact",
|
||||
new_callable=AsyncMock,
|
||||
return_value=command,
|
||||
) as forced,
|
||||
):
|
||||
await graph.ainvoke(
|
||||
ToolState(messages=[seed]), # ty: ignore[invalid-argument-type]
|
||||
context=CLIContextSchema( # ty: ignore[invalid-argument-type]
|
||||
offload_tool_call_id=tool_call_id
|
||||
),
|
||||
)
|
||||
|
||||
gated.assert_not_awaited()
|
||||
forced.assert_awaited_once()
|
||||
|
||||
async def test_read_failure_never_reaches_truncating_archive_write(self) -> None:
|
||||
"""A transient archive read failure aborts the SDK write fallback."""
|
||||
from deepagents.middleware.summarization import SummarizationMiddleware
|
||||
|
||||
summarization = self._summarization()
|
||||
backend = MagicMock()
|
||||
backend.adownload_files = AsyncMock(side_effect=RuntimeError("read failed"))
|
||||
backend.awrite = AsyncMock()
|
||||
backend.aedit = AsyncMock()
|
||||
summarization._backend = backend
|
||||
summarization._get_history_path.return_value = "/conversation_history/thread.md"
|
||||
summarization._filter_summary_messages.side_effect = lambda messages: messages
|
||||
|
||||
async def sdk_offload(
|
||||
guarded: BackendProtocol, messages: list[AnyMessage]
|
||||
) -> str | None:
|
||||
return await SummarizationMiddleware._aoffload_to_backend(
|
||||
summarization, guarded, messages
|
||||
)
|
||||
|
||||
summarization._aoffload_to_backend = AsyncMock(side_effect=sdk_offload)
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
runtime = MagicMock()
|
||||
runtime.context = {"offload_tool_call_id": "tool-call"}
|
||||
runtime.state = {"messages": [HumanMessage("one"), HumanMessage("two")]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = await middleware._arun_forced_compact(runtime)
|
||||
|
||||
backend.awrite.assert_not_awaited()
|
||||
backend.aedit.assert_not_awaited()
|
||||
assert result.update is not None
|
||||
assert result.update["_summarization_event"]["file_path"] is None
|
||||
|
||||
def test_sync_forced_compact_noops_when_nothing_old_enough(self) -> None:
|
||||
"""The sync forced path also no-ops at cutoff 0 (mirrors the async one)."""
|
||||
summarization = self._summarization()
|
||||
summarization._determine_cutoff_index.return_value = 0
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
runtime = MagicMock()
|
||||
runtime.context = None
|
||||
runtime.state = {"messages": [HumanMessage("one")]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = middleware._run_forced_compact(runtime)
|
||||
|
||||
assert result.update is not None
|
||||
assert "_summarization_event" not in result.update
|
||||
summarization._create_summary.assert_not_called()
|
||||
assert "Nothing to compact" in result.update["messages"][0].content
|
||||
|
||||
def test_sync_forced_compact_error_when_summary_fails(self) -> None:
|
||||
"""A sync summary failure returns the failure prefix and does not compact."""
|
||||
summarization = self._summarization()
|
||||
summarization._create_summary = MagicMock(side_effect=RuntimeError("boom"))
|
||||
middleware = CLICompactionMiddleware(summarization)
|
||||
runtime = MagicMock()
|
||||
runtime.context = None
|
||||
runtime.state = {"messages": [HumanMessage("one"), HumanMessage("two")]}
|
||||
runtime.tool_call_id = "tool-call"
|
||||
|
||||
result = middleware._run_forced_compact(runtime)
|
||||
|
||||
assert result.update is not None
|
||||
assert "_summarization_event" not in result.update
|
||||
content = result.update["messages"][0].content
|
||||
assert content.startswith(COMPACTION_FAILURE_PREFIX)
|
||||
assert "RuntimeError" in content
|
||||
|
||||
def test_forced_compact_error_starts_with_prefix(self) -> None:
|
||||
"""The prefix position is the load-bearing failure-detection contract."""
|
||||
command = CLICompactionMiddleware._forced_compact_error(
|
||||
"call-1", RuntimeError("boom")
|
||||
)
|
||||
assert command.update is not None
|
||||
(message,) = command.update["messages"]
|
||||
assert message.content.startswith(COMPACTION_FAILURE_PREFIX)
|
||||
assert message.tool_call_id == "call-1"
|
||||
assert "RuntimeError" in message.content
|
||||
|
||||
def test_factory_builds_cli_middleware_threading_system_prompt(self) -> None:
|
||||
"""The factory returns a CLI middleware carrying the SDK's config."""
|
||||
from deepagents_code import offload_middleware as om
|
||||
|
||||
sdk = MagicMock()
|
||||
sdk._summarization = MagicMock()
|
||||
sdk.system_prompt = "SYSTEM PROMPT"
|
||||
backend: Any = object()
|
||||
with patch.object(
|
||||
om, "create_summarization_tool_middleware", return_value=sdk
|
||||
) as factory:
|
||||
result = om._create_cli_compaction_middleware("provider:model", backend)
|
||||
|
||||
factory.assert_called_once()
|
||||
assert isinstance(result, om.CLICompactionMiddleware)
|
||||
assert result.system_prompt == "SYSTEM PROMPT"
|
||||
assert result._summarization is sdk._summarization
|
||||
|
||||
|
||||
class TestRuntimeModelConfig:
|
||||
"""Cover the three context shapes `_runtime_model_config` accepts."""
|
||||
|
||||
@staticmethod
|
||||
def _runtime(context: object) -> MagicMock:
|
||||
runtime = MagicMock()
|
||||
runtime.context = context
|
||||
return runtime
|
||||
|
||||
def test_schema_instance(self) -> None:
|
||||
ctx = CLIContextSchema(model="p:m", model_params={"temperature": 0})
|
||||
assert _runtime_model_config(self._runtime(ctx)) == (
|
||||
"p:m",
|
||||
{"temperature": 0},
|
||||
{},
|
||||
None,
|
||||
)
|
||||
|
||||
def test_serialized_dict(self) -> None:
|
||||
ctx = {"model": "p:m2", "model_params": {"x": 1}}
|
||||
assert _runtime_model_config(self._runtime(ctx)) == (
|
||||
"p:m2",
|
||||
{"x": 1},
|
||||
{},
|
||||
None,
|
||||
)
|
||||
|
||||
def test_dict_with_bad_types_normalizes(self) -> None:
|
||||
ctx = {"model": 123, "model_params": None}
|
||||
assert _runtime_model_config(self._runtime(ctx)) == (None, {}, {}, None)
|
||||
|
||||
def test_unknown_shape(self) -> None:
|
||||
assert _runtime_model_config(self._runtime(object())) == (None, {}, {}, None)
|
||||
|
||||
def test_named_fields_disambiguate_the_two_dict_slots(self) -> None:
|
||||
"""The two `dict` slots are addressable by name, not just position.
|
||||
|
||||
`model_params` and `profile_overrides` are structurally identical, so a
|
||||
positional swap would be invisible; named-field access pins each to the
|
||||
right source value.
|
||||
"""
|
||||
ctx = CLIContextSchema(
|
||||
model="p:m",
|
||||
model_params={"temperature": 0},
|
||||
profile_overrides={"max_input_tokens": 99},
|
||||
model_context_limit=7,
|
||||
)
|
||||
config = _runtime_model_config(self._runtime(ctx))
|
||||
assert config.model_params == {"temperature": 0}
|
||||
assert config.profile_overrides == {"max_input_tokens": 99}
|
||||
assert config.context_limit == 7
|
||||
|
||||
|
||||
class TestSdkContractGuards:
|
||||
"""Guard the SDK seams the forced-compaction fork depends on.
|
||||
|
||||
`CLICompactionMiddleware` forks the SDK's gated compaction flow and keys
|
||||
failure detection on a shared message prefix. These tests fail loudly in CI
|
||||
if a coordinated SDK bump renames a depended-on private method or changes
|
||||
the failure wording, instead of the fork silently drifting out of parity.
|
||||
"""
|
||||
|
||||
def test_forced_compact_matches_sdk_summarizer_calls(self) -> None:
|
||||
"""Every SDK method the fork invokes must still exist."""
|
||||
from deepagents.middleware.summarization import (
|
||||
SummarizationMiddleware,
|
||||
SummarizationToolMiddleware,
|
||||
)
|
||||
|
||||
# Called on `self._summarization` (a SummarizationMiddleware).
|
||||
for name in (
|
||||
"_apply_event_to_messages",
|
||||
"_determine_cutoff_index",
|
||||
"_partition_messages",
|
||||
"_create_summary",
|
||||
"_acreate_summary",
|
||||
"_offload_to_backend",
|
||||
"_aoffload_to_backend",
|
||||
):
|
||||
assert callable(getattr(SummarizationMiddleware, name, None)), name
|
||||
|
||||
# Called on the tool-middleware subclass itself.
|
||||
for name in (
|
||||
"_resolve_backend",
|
||||
"_build_compact_result",
|
||||
"_nothing_to_compact",
|
||||
):
|
||||
assert callable(getattr(SummarizationToolMiddleware, name, None)), name
|
||||
|
||||
def test_failure_prefix_matches_sdk_failure_message(self) -> None:
|
||||
"""Dcode's prefix must match the SDK's own compaction-failure wording.
|
||||
|
||||
`/offload` detects failures from either path by this prefix, so the
|
||||
SDK's `_compact_error` message must keep starting with it.
|
||||
"""
|
||||
from deepagents.middleware.summarization import SummarizationToolMiddleware
|
||||
|
||||
command = SummarizationToolMiddleware._compact_error(
|
||||
"call-1", RuntimeError("boom")
|
||||
)
|
||||
assert command.update is not None
|
||||
(message,) = command.update["messages"]
|
||||
assert message.content.startswith(COMPACTION_FAILURE_PREFIX)
|
||||
|
||||
+1616
-1064
File diff suppressed because it is too large
Load Diff
@@ -1,335 +0,0 @@
|
||||
"""Tests for #2741 fix: `perform_offload()` handles serialized state payloads.
|
||||
|
||||
Remote HTTP state snapshots may surface messages as raw dicts rather than
|
||||
`BaseMessage` objects. `perform_offload()` must
|
||||
normalize these before passing them to middleware helpers like
|
||||
`get_buffer_string()`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from langchain_core.messages import BaseMessage
|
||||
|
||||
from deepagents_code.offload import OffloadResult, perform_offload
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from deepagents.middleware.summarization import SummarizationEvent
|
||||
|
||||
# Raw dict messages — what remote state snapshots may return before deserialization.
|
||||
_DICT_MESSAGES: list[dict[str, Any]] = [
|
||||
{
|
||||
"content": "Hi!",
|
||||
"additional_kwargs": {},
|
||||
"response_metadata": {},
|
||||
"type": "human",
|
||||
"name": None,
|
||||
"id": "human-1",
|
||||
},
|
||||
{
|
||||
"content": "Hello! How can I help?",
|
||||
"additional_kwargs": {},
|
||||
"response_metadata": {},
|
||||
"type": "ai",
|
||||
"name": None,
|
||||
"id": "ai-1",
|
||||
},
|
||||
{
|
||||
"content": "Tell me a joke",
|
||||
"additional_kwargs": {},
|
||||
"response_metadata": {},
|
||||
"type": "human",
|
||||
"name": None,
|
||||
"id": "human-2",
|
||||
},
|
||||
{
|
||||
"content": "Why did the chicken cross the road?",
|
||||
"additional_kwargs": {},
|
||||
"response_metadata": {},
|
||||
"type": "ai",
|
||||
"name": None,
|
||||
"id": "ai-2",
|
||||
},
|
||||
{
|
||||
"content": "Why?",
|
||||
"additional_kwargs": {},
|
||||
"response_metadata": {},
|
||||
"type": "human",
|
||||
"name": None,
|
||||
"id": "human-3",
|
||||
},
|
||||
{
|
||||
"content": "To get to the other side!",
|
||||
"additional_kwargs": {},
|
||||
"response_metadata": {},
|
||||
"type": "ai",
|
||||
"name": None,
|
||||
"id": "ai-3",
|
||||
},
|
||||
]
|
||||
|
||||
# Patch targets
|
||||
_CREATE_MODEL_PATH = "deepagents_code.offload.create_model"
|
||||
_COMPUTE_DEFAULTS_PATH = (
|
||||
"deepagents.middleware.summarization.compute_summarization_defaults"
|
||||
)
|
||||
_MW_CLASS_PATH = "deepagents.middleware.summarization.SummarizationMiddleware"
|
||||
_TOKEN_COUNT_PATH = "deepagents_code.offload.count_tokens_approximately"
|
||||
_OFFLOAD_BACKEND_PATH = "deepagents_code.offload.offload_messages_to_backend"
|
||||
|
||||
|
||||
def _mock_perform_deps(*, cutoff: int = 3) -> tuple[MagicMock, MagicMock]:
|
||||
"""Return (mock_model_result, mock_middleware) for perform_offload tests."""
|
||||
mock_model = MagicMock()
|
||||
mock_model.profile = {"max_input_tokens": 200_000}
|
||||
mock_result = MagicMock()
|
||||
mock_result.model = mock_model
|
||||
|
||||
mock_mw = MagicMock()
|
||||
mock_mw._apply_event_to_messages.side_effect = lambda msgs, _ev: list(msgs)
|
||||
mock_mw._determine_cutoff_index.return_value = cutoff
|
||||
mock_mw._partition_messages.side_effect = lambda msgs, idx: (
|
||||
msgs[:idx],
|
||||
msgs[idx:],
|
||||
)
|
||||
mock_mw._acreate_summary = AsyncMock(return_value="Summary of conversation.")
|
||||
|
||||
summary_msg = MagicMock()
|
||||
summary_msg.content = "Summary of conversation."
|
||||
summary_msg.additional_kwargs = {"lc_source": "summarization"}
|
||||
mock_mw._build_new_messages_with_path.return_value = [summary_msg]
|
||||
mock_mw._compute_state_cutoff.side_effect = lambda _ev, c: c
|
||||
mock_mw._filter_summary_messages.side_effect = lambda msgs: msgs
|
||||
|
||||
return mock_result, mock_mw
|
||||
|
||||
|
||||
class TestDictMessageNormalization:
|
||||
"""Verify `perform_offload()` normalizes serialized state payloads."""
|
||||
|
||||
async def test_dict_messages_converted_before_middleware(self) -> None:
|
||||
"""Dict messages are converted to BaseMessage before reaching middleware."""
|
||||
model_result, mock_mw = _mock_perform_deps(cutoff=3)
|
||||
|
||||
with (
|
||||
patch(_CREATE_MODEL_PATH, return_value=model_result),
|
||||
patch(_COMPUTE_DEFAULTS_PATH, return_value={"keep": ("fraction", 0.1)}),
|
||||
patch(_MW_CLASS_PATH, return_value=mock_mw),
|
||||
patch(_TOKEN_COUNT_PATH, return_value=100),
|
||||
patch(_OFFLOAD_BACKEND_PATH, new_callable=AsyncMock, return_value="/p.md"),
|
||||
):
|
||||
result = await perform_offload(
|
||||
messages=list(_DICT_MESSAGES),
|
||||
prior_event=None,
|
||||
thread_id="test-thread",
|
||||
model_spec="anthropic:claude-sonnet-4-20250514",
|
||||
profile_overrides=None,
|
||||
context_limit=None,
|
||||
total_context_tokens=0,
|
||||
backend=MagicMock(),
|
||||
)
|
||||
|
||||
assert isinstance(result, OffloadResult)
|
||||
assert result.messages_offloaded == 3
|
||||
assert result.messages_kept == 3
|
||||
|
||||
passed_msgs = mock_mw._apply_event_to_messages.call_args[0][0]
|
||||
assert all(isinstance(m, BaseMessage) for m in passed_msgs)
|
||||
|
||||
async def test_dict_prior_event_summary_message_converted(self) -> None:
|
||||
"""A prior_event with a dict summary_message is normalized."""
|
||||
model_result, mock_mw = _mock_perform_deps(cutoff=0)
|
||||
|
||||
dict_summary = {
|
||||
"content": "Previous summary.",
|
||||
"additional_kwargs": {"lc_source": "summarization"},
|
||||
"response_metadata": {},
|
||||
"type": "human",
|
||||
"name": None,
|
||||
"id": "summary-1",
|
||||
}
|
||||
prior_event = cast(
|
||||
"SummarizationEvent",
|
||||
{
|
||||
"cutoff_index": 2,
|
||||
"summary_message": dict_summary,
|
||||
"file_path": "/old.md",
|
||||
},
|
||||
)
|
||||
|
||||
with (
|
||||
patch(_CREATE_MODEL_PATH, return_value=model_result),
|
||||
patch(_COMPUTE_DEFAULTS_PATH, return_value={"keep": ("fraction", 0.1)}),
|
||||
patch(_MW_CLASS_PATH, return_value=mock_mw),
|
||||
patch(_TOKEN_COUNT_PATH, return_value=50),
|
||||
):
|
||||
from deepagents_code.offload import OffloadThresholdNotMet
|
||||
|
||||
result = await perform_offload(
|
||||
messages=list(_DICT_MESSAGES),
|
||||
prior_event=prior_event,
|
||||
thread_id="test-thread",
|
||||
model_spec="anthropic:claude-sonnet-4-20250514",
|
||||
profile_overrides=None,
|
||||
context_limit=None,
|
||||
total_context_tokens=0,
|
||||
backend=MagicMock(),
|
||||
)
|
||||
|
||||
assert isinstance(result, OffloadThresholdNotMet)
|
||||
|
||||
passed_event = mock_mw._apply_event_to_messages.call_args[0][1]
|
||||
assert isinstance(passed_event["summary_message"], BaseMessage)
|
||||
|
||||
async def test_base_message_inputs_unchanged(self) -> None:
|
||||
"""Already-proper BaseMessage objects pass through without issue."""
|
||||
from langchain_core.messages.utils import convert_to_messages
|
||||
|
||||
model_result, mock_mw = _mock_perform_deps(cutoff=3)
|
||||
proper_messages = convert_to_messages(_DICT_MESSAGES)
|
||||
|
||||
with (
|
||||
patch(_CREATE_MODEL_PATH, return_value=model_result),
|
||||
patch(_COMPUTE_DEFAULTS_PATH, return_value={"keep": ("fraction", 0.1)}),
|
||||
patch(_MW_CLASS_PATH, return_value=mock_mw),
|
||||
patch(_TOKEN_COUNT_PATH, return_value=100),
|
||||
patch(_OFFLOAD_BACKEND_PATH, new_callable=AsyncMock, return_value="/p.md"),
|
||||
):
|
||||
result = await perform_offload(
|
||||
messages=proper_messages,
|
||||
prior_event=None,
|
||||
thread_id="test-thread",
|
||||
model_spec="anthropic:claude-sonnet-4-20250514",
|
||||
profile_overrides=None,
|
||||
context_limit=None,
|
||||
total_context_tokens=0,
|
||||
backend=MagicMock(),
|
||||
)
|
||||
|
||||
assert isinstance(result, OffloadResult)
|
||||
|
||||
async def test_dict_messages_and_dict_summary_normalized_together(self) -> None:
|
||||
"""Both flags at once: dicts in `messages` and in `prior_event`."""
|
||||
model_result, mock_mw = _mock_perform_deps(cutoff=3)
|
||||
|
||||
dict_summary = {
|
||||
"content": "Previous summary.",
|
||||
"additional_kwargs": {"lc_source": "summarization"},
|
||||
"response_metadata": {},
|
||||
"type": "human",
|
||||
"name": None,
|
||||
"id": "summary-1",
|
||||
}
|
||||
prior_event = cast(
|
||||
"SummarizationEvent",
|
||||
{
|
||||
"cutoff_index": 1,
|
||||
"summary_message": dict_summary,
|
||||
"file_path": "/old.md",
|
||||
},
|
||||
)
|
||||
|
||||
with (
|
||||
patch(_CREATE_MODEL_PATH, return_value=model_result),
|
||||
patch(_COMPUTE_DEFAULTS_PATH, return_value={"keep": ("fraction", 0.1)}),
|
||||
patch(_MW_CLASS_PATH, return_value=mock_mw),
|
||||
patch(_TOKEN_COUNT_PATH, return_value=100),
|
||||
patch(_OFFLOAD_BACKEND_PATH, new_callable=AsyncMock, return_value="/p.md"),
|
||||
):
|
||||
result = await perform_offload(
|
||||
messages=list(_DICT_MESSAGES),
|
||||
prior_event=prior_event,
|
||||
thread_id="test-thread",
|
||||
model_spec="anthropic:claude-sonnet-4-20250514",
|
||||
profile_overrides=None,
|
||||
context_limit=None,
|
||||
total_context_tokens=0,
|
||||
backend=MagicMock(),
|
||||
)
|
||||
|
||||
assert isinstance(result, OffloadResult)
|
||||
|
||||
passed_msgs = mock_mw._apply_event_to_messages.call_args[0][0]
|
||||
passed_event = mock_mw._apply_event_to_messages.call_args[0][1]
|
||||
assert all(isinstance(m, BaseMessage) for m in passed_msgs)
|
||||
assert isinstance(passed_event["summary_message"], BaseMessage)
|
||||
|
||||
async def test_heterogeneous_messages_fully_normalized(self) -> None:
|
||||
"""A list mixing BaseMessage and dict entries is normalized end-to-end."""
|
||||
from langchain_core.messages.utils import convert_to_messages
|
||||
|
||||
model_result, mock_mw = _mock_perform_deps(cutoff=3)
|
||||
# First element is a BaseMessage; rest are dicts. The [0]-only
|
||||
# heuristic would have missed this; `any(...)` catches it.
|
||||
head = convert_to_messages([_DICT_MESSAGES[0]])
|
||||
tail: list[Any] = list(_DICT_MESSAGES[1:])
|
||||
mixed: list[Any] = [*head, *tail]
|
||||
|
||||
with (
|
||||
patch(_CREATE_MODEL_PATH, return_value=model_result),
|
||||
patch(_COMPUTE_DEFAULTS_PATH, return_value={"keep": ("fraction", 0.1)}),
|
||||
patch(_MW_CLASS_PATH, return_value=mock_mw),
|
||||
patch(_TOKEN_COUNT_PATH, return_value=100),
|
||||
patch(_OFFLOAD_BACKEND_PATH, new_callable=AsyncMock, return_value="/p.md"),
|
||||
):
|
||||
result = await perform_offload(
|
||||
messages=mixed,
|
||||
prior_event=None,
|
||||
thread_id="test-thread",
|
||||
model_spec="anthropic:claude-sonnet-4-20250514",
|
||||
profile_overrides=None,
|
||||
context_limit=None,
|
||||
total_context_tokens=0,
|
||||
backend=MagicMock(),
|
||||
)
|
||||
|
||||
assert isinstance(result, OffloadResult)
|
||||
|
||||
passed_msgs = mock_mw._apply_event_to_messages.call_args[0][0]
|
||||
assert all(isinstance(m, BaseMessage) for m in passed_msgs)
|
||||
|
||||
async def test_empty_messages_with_dict_summary_still_normalizes(self) -> None:
|
||||
"""Empty `messages=[]` does not crash the guard; summary still normalized."""
|
||||
model_result, mock_mw = _mock_perform_deps(cutoff=0)
|
||||
|
||||
dict_summary = {
|
||||
"content": "Previous summary.",
|
||||
"additional_kwargs": {"lc_source": "summarization"},
|
||||
"response_metadata": {},
|
||||
"type": "human",
|
||||
"name": None,
|
||||
"id": "summary-1",
|
||||
}
|
||||
prior_event = cast(
|
||||
"SummarizationEvent",
|
||||
{
|
||||
"cutoff_index": 0,
|
||||
"summary_message": dict_summary,
|
||||
"file_path": "/old.md",
|
||||
},
|
||||
)
|
||||
|
||||
with (
|
||||
patch(_CREATE_MODEL_PATH, return_value=model_result),
|
||||
patch(_COMPUTE_DEFAULTS_PATH, return_value={"keep": ("fraction", 0.1)}),
|
||||
patch(_MW_CLASS_PATH, return_value=mock_mw),
|
||||
patch(_TOKEN_COUNT_PATH, return_value=0),
|
||||
):
|
||||
from deepagents_code.offload import OffloadThresholdNotMet
|
||||
|
||||
result = await perform_offload(
|
||||
messages=[],
|
||||
prior_event=prior_event,
|
||||
thread_id="test-thread",
|
||||
model_spec="anthropic:claude-sonnet-4-20250514",
|
||||
profile_overrides=None,
|
||||
context_limit=None,
|
||||
total_context_tokens=0,
|
||||
backend=MagicMock(),
|
||||
)
|
||||
|
||||
assert isinstance(result, OffloadThresholdNotMet)
|
||||
passed_event = mock_mw._apply_event_to_messages.call_args[0][1]
|
||||
assert isinstance(passed_event["summary_message"], BaseMessage)
|
||||
Reference in New Issue
Block a user