From 206f74baac0736d370f0e97ec11d3544290880cd Mon Sep 17 00:00:00 2001
From: kshitij <82637225+kshitijk4poor@users.noreply.github.com>
Date: Sat, 8 Aug 2026 23:09:43 +0530
Subject: [PATCH 1/2] Revert "feat: add DCP context engine"
This reverts commit d7072ab914de0bd3c98cd5cf006fd193a5ee752d.
---
agent/dcp_config.py | 256 -------
agent/dcp_context_engine.py | 656 ------------------
agent/dcp_state.py | 70 --
tests/agent/test_dcp_config.py | 63 --
tests/agent/test_dcp_context_engine.py | 331 ---------
website/docs/developer-guide/agent-loop.md | 24 +-
website/docs/developer-guide/architecture.md | 12 +-
.../context-compression-and-caching.md | 82 ---
8 files changed, 5 insertions(+), 1489 deletions(-)
delete mode 100644 agent/dcp_config.py
delete mode 100644 agent/dcp_context_engine.py
delete mode 100644 agent/dcp_state.py
delete mode 100644 tests/agent/test_dcp_config.py
delete mode 100644 tests/agent/test_dcp_context_engine.py
diff --git a/agent/dcp_config.py b/agent/dcp_config.py
deleted file mode 100644
index aac674907863b..0000000000000
--- a/agent/dcp_config.py
+++ /dev/null
@@ -1,256 +0,0 @@
-"""Configuration helpers for the DCP context engine."""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from typing import Any, Literal
-
-
-DCP_DEFAULT_PROTECTED_TOOLS = {
- "delegate_task",
- "todo",
- "memory",
- "skill_view",
- "skill_manage",
- "write_file",
- "patch",
- "clarify",
- "cronjob",
- "compress",
-}
-
-
-@dataclass(slots=True)
-class DCPStrategyConfig:
- enabled: bool = True
- protected_tools: set[str] = field(default_factory=set)
-
-
-@dataclass(slots=True)
-class DCPPurgeErrorsConfig(DCPStrategyConfig):
- turns: int = 4
-
-
-@dataclass(slots=True)
-class DCPCompressConfig:
- mode: Literal["range", "message"] = "range"
- permission: Literal["allow", "ask", "deny"] = "allow"
- show_compression: bool = False
- summary_buffer: bool = True
- max_context_limit: int | str = 100000
- min_context_limit: int | str = 50000
- model_max_limits: dict[str, int | str] = field(default_factory=dict)
- model_min_limits: dict[str, int | str] = field(default_factory=dict)
- nudge_frequency: int = 5
- iteration_nudge_threshold: int = 15
- nudge_force: Literal["soft", "strong"] = "soft"
- protected_tools: set[str] = field(default_factory=set)
- protect_user_messages: bool = False
-
-
-@dataclass(slots=True)
-class DCPTurnProtectionConfig:
- enabled: bool = False
- turns: int = 4
-
-
-@dataclass(slots=True)
-class DCPManualModeConfig:
- enabled: bool = False
- automatic_strategies: bool = True
-
-
-@dataclass(slots=True)
-class DCPCommandsConfig:
- enabled: bool = True
- protected_tools: set[str] = field(default_factory=set)
-
-
-@dataclass(slots=True)
-class DCPExperimentalConfig:
- allow_subagents: bool = False
- custom_prompts: bool = False
-
-
-@dataclass(slots=True)
-class DCPConfig:
- enabled: bool = True
- debug: bool = False
- prune_notification: Literal["off", "minimal", "detailed"] = "detailed"
- prune_notification_type: Literal["chat", "toast"] = "chat"
- commands: DCPCommandsConfig = field(default_factory=DCPCommandsConfig)
- manual_mode: DCPManualModeConfig = field(default_factory=DCPManualModeConfig)
- turn_protection: DCPTurnProtectionConfig = field(default_factory=DCPTurnProtectionConfig)
- experimental: DCPExperimentalConfig = field(default_factory=DCPExperimentalConfig)
- protected_file_patterns: list[str] = field(default_factory=list)
- compress: DCPCompressConfig = field(default_factory=DCPCompressConfig)
- deduplication: DCPStrategyConfig = field(default_factory=DCPStrategyConfig)
- purge_errors: DCPPurgeErrorsConfig = field(default_factory=DCPPurgeErrorsConfig)
-
-
-def _as_bool(value: Any, default: bool) -> bool:
- if isinstance(value, bool):
- return value
- return default
-
-
-def _as_int(value: Any, default: int, *, minimum: int | None = None) -> int:
- try:
- parsed = int(value)
- except (TypeError, ValueError):
- return default
- if minimum is not None:
- parsed = max(minimum, parsed)
- return parsed
-
-
-def _as_choice(value: Any, choices: set[str], default: str) -> str:
- if isinstance(value, str) and value in choices:
- return value
- return default
-
-
-def _as_limit(value: Any, default: int | str) -> int | str:
- if isinstance(value, int) and value > 0:
- return value
- if isinstance(value, str):
- raw = value.strip()
- if raw.endswith("%"):
- try:
- pct = float(raw[:-1])
- except ValueError:
- return default
- if pct > 0:
- return raw
- else:
- try:
- parsed = int(raw)
- except ValueError:
- return default
- if parsed > 0:
- return parsed
- return default
-
-
-def _as_limit_map(value: Any) -> dict[str, int | str]:
- if not isinstance(value, dict):
- return {}
- out: dict[str, int | str] = {}
- for key, limit in value.items():
- if not isinstance(key, str):
- continue
- parsed = _as_limit(limit, 0)
- if parsed:
- out[key] = parsed
- return out
-
-
-def _tool_set(value: Any) -> set[str]:
- if not isinstance(value, list):
- return set()
- return {item for item in value if isinstance(item, str) and item.strip()}
-
-
-def _str_list(value: Any) -> list[str]:
- if not isinstance(value, list):
- return []
- return [item for item in value if isinstance(item, str)]
-
-
-def parse_dcp_config(config: dict[str, Any] | None) -> DCPConfig:
- """Parse ``context.dcp`` config into typed defaults."""
- raw = config if isinstance(config, dict) else {}
-
- commands_raw = raw.get("commands", {}) if isinstance(raw.get("commands", {}), dict) else {}
- manual_raw = raw.get("manualMode", {}) if isinstance(raw.get("manualMode", {}), dict) else {}
- turn_raw = raw.get("turnProtection", {}) if isinstance(raw.get("turnProtection", {}), dict) else {}
- exp_raw = raw.get("experimental", {}) if isinstance(raw.get("experimental", {}), dict) else {}
- compress_raw = raw.get("compress", {}) if isinstance(raw.get("compress", {}), dict) else {}
- strategies_raw = raw.get("strategies", {}) if isinstance(raw.get("strategies", {}), dict) else {}
- dedup_raw = strategies_raw.get("deduplication", {}) if isinstance(strategies_raw.get("deduplication", {}), dict) else {}
- purge_raw = strategies_raw.get("purgeErrors", {}) if isinstance(strategies_raw.get("purgeErrors", {}), dict) else {}
-
- return DCPConfig(
- enabled=_as_bool(raw.get("enabled"), True),
- debug=_as_bool(raw.get("debug"), False),
- prune_notification=_as_choice(raw.get("pruneNotification"), {"off", "minimal", "detailed"}, "detailed"), # type: ignore[arg-type]
- prune_notification_type=_as_choice(raw.get("pruneNotificationType"), {"chat", "toast"}, "chat"), # type: ignore[arg-type]
- commands=DCPCommandsConfig(
- enabled=_as_bool(commands_raw.get("enabled"), True),
- protected_tools=_tool_set(commands_raw.get("protectedTools")),
- ),
- manual_mode=DCPManualModeConfig(
- enabled=_as_bool(manual_raw.get("enabled"), False),
- automatic_strategies=_as_bool(manual_raw.get("automaticStrategies"), True),
- ),
- turn_protection=DCPTurnProtectionConfig(
- enabled=_as_bool(turn_raw.get("enabled"), False),
- turns=_as_int(turn_raw.get("turns"), 4, minimum=0),
- ),
- experimental=DCPExperimentalConfig(
- allow_subagents=_as_bool(exp_raw.get("allowSubAgents"), False),
- custom_prompts=_as_bool(exp_raw.get("customPrompts"), False),
- ),
- protected_file_patterns=_str_list(raw.get("protectedFilePatterns")),
- compress=DCPCompressConfig(
- mode=_as_choice(compress_raw.get("mode"), {"range", "message"}, "range"), # type: ignore[arg-type]
- permission=_as_choice(compress_raw.get("permission"), {"allow", "ask", "deny"}, "allow"), # type: ignore[arg-type]
- show_compression=_as_bool(compress_raw.get("showCompression"), False),
- summary_buffer=_as_bool(compress_raw.get("summaryBuffer"), True),
- max_context_limit=_as_limit(compress_raw.get("maxContextLimit"), 100000),
- min_context_limit=_as_limit(compress_raw.get("minContextLimit"), 50000),
- model_max_limits=_as_limit_map(compress_raw.get("modelMaxLimits")),
- model_min_limits=_as_limit_map(compress_raw.get("modelMinLimits")),
- nudge_frequency=_as_int(compress_raw.get("nudgeFrequency"), 5, minimum=1),
- iteration_nudge_threshold=_as_int(compress_raw.get("iterationNudgeThreshold"), 15, minimum=1),
- nudge_force=_as_choice(compress_raw.get("nudgeForce"), {"soft", "strong"}, "soft"), # type: ignore[arg-type]
- protected_tools=_tool_set(compress_raw.get("protectedTools")),
- protect_user_messages=_as_bool(compress_raw.get("protectUserMessages"), False),
- ),
- deduplication=DCPStrategyConfig(
- enabled=_as_bool(dedup_raw.get("enabled"), True),
- protected_tools=_tool_set(dedup_raw.get("protectedTools")),
- ),
- purge_errors=DCPPurgeErrorsConfig(
- enabled=_as_bool(purge_raw.get("enabled"), True),
- protected_tools=_tool_set(purge_raw.get("protectedTools")),
- turns=_as_int(purge_raw.get("turns"), 4, minimum=0),
- ),
- )
-
-
-def resolve_limit(limit: int | str, context_length: int) -> int:
- """Resolve an absolute token limit or percentage string."""
- if isinstance(limit, int):
- return limit
- raw = limit.strip()
- if raw.endswith("%"):
- try:
- pct = float(raw[:-1]) / 100.0
- except ValueError:
- return 0
- return int(context_length * pct)
- try:
- return int(raw)
- except ValueError:
- return 0
-
-
-def resolve_model_limit(
- limits: dict[str, int | str],
- *,
- provider: str | None,
- model: str | None,
- context_length: int,
- fallback: int | str,
-) -> int:
- """Resolve DCP per-model limits with a simple provider/model key."""
- keys = []
- if provider and model:
- keys.append(f"{provider}/{model}")
- if model:
- keys.append(model)
- for key in keys:
- if key in limits:
- return resolve_limit(limits[key], context_length)
- return resolve_limit(fallback, context_length)
diff --git a/agent/dcp_context_engine.py b/agent/dcp_context_engine.py
deleted file mode 100644
index 8909e957e559b..0000000000000
--- a/agent/dcp_context_engine.py
+++ /dev/null
@@ -1,656 +0,0 @@
-"""DCP-style model-guided context engine for Hermes Agent."""
-
-from __future__ import annotations
-
-import hashlib
-import json
-import re
-import time
-from collections import defaultdict, deque
-from typing import Any
-
-from agent.context_engine import ContextEngine
-from agent.dcp_config import (
- DCP_DEFAULT_PROTECTED_TOOLS,
- DCPConfig,
- parse_dcp_config,
- resolve_model_limit,
-)
-from agent.dcp_state import CompressionBlock, DCPSessionState
-
-_ERROR_RE = re.compile(r"\b(error|exception|traceback|failed|failure|timed out|timeout)\b", re.I)
-
-_DCP_SYSTEM_EXTENSION = (
- "DCP context management is active. Message refs look like m0001; "
- "compressed blocks look like b1. Use the compress tool when older work "
- "is complete or stale. Preserve concrete file paths, commands, errors, "
- "test results, decisions, constraints, and open questions. Do not compress "
- "the active task or very recent user turns."
-)
-
-# Maximum deactivated blocks to retain; older ones are evicted to bound memory.
-_MAX_INACTIVE_BLOCKS = 50
-
-
-class DCPContextEngine(ContextEngine):
- """Model-guided context engine inspired by Dynamic Context Pruning.
-
- The engine keeps canonical history intact. ``compress`` creates DCP state;
- ``transform_api_messages`` applies that state to the provider-bound copy.
- """
-
- def __init__(
- self,
- *,
- config: dict[str, Any] | DCPConfig | None = None,
- context_length: int = 0,
- model: str = "",
- provider: str = "",
- quiet_mode: bool = False,
- ) -> None:
- self.config = config if isinstance(config, DCPConfig) else parse_dcp_config(config)
- self.context_length = context_length or 0
- self.model = model or ""
- self.provider = provider or ""
- self.quiet_mode = quiet_mode
- self.last_prompt_tokens = 0
- self.last_completion_tokens = 0
- self.last_total_tokens = 0
- self.compression_count = 0
- self.threshold_tokens = self._min_limit()
- self.state = DCPSessionState()
- # Cache for message signatures keyed by id(msg) — invalidated when
- # the canonical message list changes. Avoids re-hashing the same
- # dicts on every API call.
- self._sig_cache: dict[int, str] = {}
-
- @property
- def name(self) -> str:
- return "dcp"
-
- def update_from_response(self, usage: dict[str, Any]) -> None:
- self.last_prompt_tokens = int(usage.get("prompt_tokens") or usage.get("input_tokens") or 0)
- self.last_completion_tokens = int(usage.get("completion_tokens") or usage.get("output_tokens") or 0)
- self.last_total_tokens = int(usage.get("total_tokens") or (self.last_prompt_tokens + self.last_completion_tokens))
- self.state.last_prompt_tokens = self.last_prompt_tokens
-
- def should_compress(self, prompt_tokens: int = None) -> bool:
- return False
-
- def should_compress_preflight(self, messages: list[dict[str, Any]]) -> bool:
- return False
-
- def has_content_to_compress(self, messages: list[dict[str, Any]]) -> bool:
- self._ensure_refs(messages)
- return len(self.state.index_by_ref) > 4
-
- def compress(
- self,
- messages: list[dict[str, Any]],
- current_tokens: int = None,
- focus_topic: str = None,
- ) -> list[dict[str, Any]]:
- if focus_topic:
- self.state.manual_mode = "compress-pending"
- self.state.pending_manual_focus = focus_topic
- return messages
-
- def on_session_start(self, session_id: str, **kwargs: Any) -> None:
- self.state.session_id = session_id
- model = kwargs.get("model")
- context_length = kwargs.get("context_length")
- if isinstance(model, str) and model:
- self.model = model
- if isinstance(context_length, int) and context_length > 0:
- self.context_length = context_length
- self.threshold_tokens = self._min_limit()
-
- def on_session_reset(self) -> None:
- super().on_session_reset()
- self.state = DCPSessionState(session_id=self.state.session_id)
- self._sig_cache.clear()
-
- def update_model(
- self,
- model: str,
- context_length: int,
- base_url: str = "",
- api_key: str = "",
- provider: str = "",
- api_mode: str = "",
- ) -> None:
- self.model = model or self.model
- self.provider = provider or self.provider
- self.context_length = context_length
- self.threshold_tokens = self._min_limit()
-
- def get_tool_schemas(self) -> list[dict[str, Any]]:
- if not self.config.enabled or self.config.compress.permission == "deny":
- return []
- return [self._compress_tool_schema()]
-
- def handle_tool_call(self, name: str, args: dict[str, Any], **kwargs: Any) -> str:
- if name != "compress":
- return json.dumps({"ok": False, "error": f"Unknown context engine tool: {name}"})
- messages = kwargs.get("messages")
- if not isinstance(messages, list):
- return json.dumps({"ok": False, "error": "compress requires current messages"})
- try:
- result = self._handle_compress(args, messages)
- except Exception as exc:
- return json.dumps({"ok": False, "error": str(exc)})
- return json.dumps(result)
-
- def transform_api_messages(
- self,
- api_messages: list[dict[str, Any]],
- *,
- canonical_messages: list[dict[str, Any]],
- system_prompt: str,
- tools: list[dict[str, Any]] | None,
- api_call_count: int,
- model: str,
- provider: str | None,
- session_id: str | None,
- ) -> list[dict[str, Any]]:
- if not self.config.enabled:
- return api_messages
-
- self.model = model or self.model
- self.provider = provider or self.provider
- self.state.session_id = session_id or self.state.session_id
-
- # Build / refresh refs from canonical messages. This also updates
- # turn counters and message-since-last-user tracking.
- self._ensure_refs(canonical_messages)
-
- # Match API messages to refs by content signature. We use a
- # role+content-based signature that excludes tool_calls (whose JSON
- # may have been re-serialised by _canonicalize_api_tool_calls between
- # the canonical list and the API copy) so that assistant tool-calling
- # messages still match.
- ref_by_api_index = self._match_api_messages_to_refs(api_messages, canonical_messages)
-
- # Shallow-copy the list and structurally clone messages we will
- # mutate. This avoids the O(n) cost of copy.deepcopy on every call
- # while still protecting the caller's message dicts.
- transformed = list(api_messages)
- mutated: set[int] = set()
-
- self._annotate_refs(transformed, ref_by_api_index, mutated)
- self._apply_blocks(transformed, ref_by_api_index, mutated)
-
- if self._automatic_strategies_enabled():
- if self.config.deduplication.enabled:
- self._apply_deduplication(transformed, mutated)
- if self.config.purge_errors.enabled:
- self._apply_purge_errors(transformed, mutated)
-
- self._inject_system_extension(transformed, mutated)
- self._inject_nudge(transformed, api_call_count=api_call_count, mutated=mutated)
- return transformed
-
- def get_status(self) -> dict[str, Any]:
- active_blocks = self.state.active_blocks()
- return {
- **super().get_status(),
- "engine": "dcp",
- "active_blocks": len(active_blocks),
- "message_refs": len(self.state.ref_by_message_key),
- "min_context_limit": self._min_limit(),
- "max_context_limit": self._max_limit(),
- "compress_mode": self.config.compress.mode,
- "compress_permission": self.config.compress.permission,
- }
-
- # -- Tool schema ------------------------------------------------------
-
- def _compress_tool_schema(self) -> dict[str, Any]:
- is_message_mode = self.config.compress.mode == "message"
- if is_message_mode:
- item_properties = {
- "messageId": {"type": "string", "description": "Message ref, e.g. m0042."},
- "topic": {"type": "string", "description": "Short label for this message."},
- "summary": {"type": "string", "description": "Complete technical summary replacing this message."},
- }
- item_required = ["messageId", "topic", "summary"]
- description = (
- "Compress individual high-volume messages by ref. Preserve concrete "
- "technical facts, file paths, commands, decisions, errors, and open questions."
- )
- else:
- item_properties = {
- "startId": {"type": "string", "description": "Starting message or block ref, e.g. m0004 or b2."},
- "endId": {"type": "string", "description": "Ending message or block ref, e.g. m0018 or b3."},
- "summary": {"type": "string", "description": "Complete technical summary replacing the range."},
- }
- item_required = ["startId", "endId", "summary"]
- description = (
- "Compress completed, stale context ranges by message/block ref. "
- "Use this when prior work is closed and a concise technical summary "
- "will preserve the useful state. Do not compress the active task."
- )
- return {
- "name": "compress",
- "description": description,
- "parameters": {
- "type": "object",
- "properties": {
- "topic": {"type": "string", "description": "Short 3-5 word label for this compression batch."},
- "content": {
- "type": "array",
- "items": {
- "type": "object",
- "properties": item_properties,
- "required": item_required,
- },
- },
- },
- "required": ["topic", "content"],
- },
- }
-
- # -- Tool handling ----------------------------------------------------
-
- def _handle_compress(self, args: dict[str, Any], messages: list[dict[str, Any]]) -> dict[str, Any]:
- self._ensure_refs(messages)
- topic = self._require_str(args, "topic")
- content = args.get("content")
- if not isinstance(content, list) or not content:
- raise ValueError("compress.content must be a non-empty array")
-
- is_message_mode = self.config.compress.mode == "message"
- created: list[int] = []
- deactivated: list[int] = []
- run_id = self.state.new_run_id()
-
- for item in content:
- if not isinstance(item, dict):
- raise ValueError("Each compression entry must be an object")
- summary = self._require_str(item, "summary")
-
- if is_message_mode:
- ref = self._require_str(item, "messageId")
- if ref not in self.state.index_by_ref:
- raise ValueError(f"Unknown message ref: {ref}")
- item_topic = item.get("topic") if isinstance(item.get("topic"), str) else topic
- message_refs = [ref]
- included_blocks: list[int] = []
- block = CompressionBlock(
- block_id=self.state.new_block_id(),
- run_id=run_id,
- mode="message",
- topic=item_topic,
- summary=self._augment_summary(summary, message_refs),
- message_refs=message_refs,
- included_block_ids=[],
- consumed_block_ids=[],
- created_at=time.time(),
- )
- else:
- start_ref = self._require_str(item, "startId")
- end_ref = self._require_str(item, "endId")
- message_refs, included_blocks = self._resolve_range(start_ref, end_ref)
- if not message_refs:
- raise ValueError(f"Range {start_ref}-{end_ref} does not cover any messages")
- item_topic = topic
- consumed_blocks: list[int] = []
- block_id = self.state.new_block_id()
- for included in included_blocks:
- old = self.state.blocks_by_id.get(included)
- if old and old.active:
- old.active = False
- old.deactivated_at = time.time()
- old.deactivated_by_block_id = block_id
- self.state.active_block_ids.discard(included)
- consumed_blocks.append(included)
- deactivated.append(included)
- block = CompressionBlock(
- block_id=block_id,
- run_id=run_id,
- mode="range",
- topic=item_topic,
- summary=self._augment_summary(summary, message_refs),
- start_ref=start_ref,
- end_ref=end_ref,
- message_refs=message_refs,
- included_block_ids=included_blocks,
- consumed_block_ids=consumed_blocks,
- created_at=time.time(),
- )
-
- self.state.blocks_by_id[block.block_id] = block
- self.state.active_block_ids.add(block.block_id)
- created.append(block.block_id)
-
- self.compression_count += len(created)
- self.state.turns_since_last_compress = 0
- self._evict_inactive_blocks()
- mode = "message" if is_message_mode else "range"
- return {
- "ok": True,
- "mode": mode,
- "created_blocks": created,
- "deactivated_blocks": deactivated,
- "active_blocks": sorted(self.state.active_block_ids),
- "message": f"Compressed {len(created)} {mode}(s) into {', '.join(f'b{i}' for i in created)}.",
- }
-
- # -- Transforms -------------------------------------------------------
-
- def _ensure_refs(self, messages: list[dict[str, Any]]) -> None:
- self._sig_cache.clear()
- self.state.index_by_ref.clear()
- for idx, msg in enumerate(messages):
- key = self._message_key(msg, idx)
- ref = self.state.ref_by_message_key.get(key)
- if ref is None:
- ref = self.state.new_message_ref()
- self.state.ref_by_message_key[key] = ref
- self.state.message_key_by_ref[ref] = key
- self.state.index_by_ref[ref] = idx
- user_indices = [idx for idx, msg in enumerate(messages) if msg.get("role") == "user"]
- if user_indices:
- last_user = user_indices[-1]
- if last_user != self.state.last_user_turn_index:
- self.state.turns_since_last_compress += 1
- self.state.last_user_turn_index = last_user
- self.state.messages_since_last_user = len(messages) - last_user - 1
-
- def _match_api_messages_to_refs(
- self,
- api_messages: list[dict[str, Any]],
- canonical_messages: list[dict[str, Any]],
- ) -> dict[int, str]:
- # Build a mapping from content signature (role + content only,
- # NOT tool_calls) to a deque of refs. We exclude tool_calls from
- # the signature because _canonicalize_api_tool_calls may have
- # re-serialised tool-call argument JSON with sort_keys=True on the
- # API copy, producing a different hash than the canonical message.
- refs_by_sig: dict[str, deque[str]] = defaultdict(deque)
- for idx, msg in enumerate(canonical_messages):
- key = self._message_key(msg, idx)
- ref = self.state.ref_by_message_key.get(key)
- if ref:
- refs_by_sig[self._content_signature(msg)].append(ref)
-
- out: dict[int, str] = {}
- for api_idx, msg in enumerate(api_messages):
- if msg.get("role") == "system":
- continue
- sig = self._content_signature(msg)
- queue = refs_by_sig.get(sig)
- if queue:
- out[api_idx] = queue.popleft()
- return out
-
- def _clone_if_needed(self, messages: list[dict[str, Any]], idx: int, mutated: set[int]) -> dict[str, Any]:
- """Clone a message dict before mutating it (copy-on-write)."""
- if idx not in mutated:
- messages[idx] = dict(messages[idx])
- mutated.add(idx)
- return messages[idx]
-
- def _annotate_refs(self, messages: list[dict[str, Any]], ref_by_api_index: dict[int, str], mutated: set[int]) -> None:
- for idx, ref in ref_by_api_index.items():
- msg = self._clone_if_needed(messages, idx, mutated)
- content = msg.get("content")
- marker = f''
- if isinstance(content, str):
- if marker not in content:
- msg["content"] = f"{content}\n\n{marker}" if content else marker
- elif isinstance(content, list):
- msg["content"] = content + [{"type": "text", "text": marker}]
-
- def _apply_blocks(self, messages: list[dict[str, Any]], ref_by_api_index: dict[int, str], mutated: set[int]) -> None:
- ref_to_api_index = {ref: idx for idx, ref in ref_by_api_index.items()}
- for block in self.state.active_blocks():
- covered = [ref for ref in block.message_refs if ref in ref_to_api_index]
- if not covered:
- continue
- anchor_ref = covered[0]
- anchor_idx = ref_to_api_index[anchor_ref]
- anchor = self._clone_if_needed(messages, anchor_idx, mutated)
- anchor["content"] = self._block_summary_text(block)
- for ref in covered[1:]:
- idx = ref_to_api_index[ref]
- msg = self._clone_if_needed(messages, idx, mutated)
- msg["content"] = f"[DCP: content moved into compressed block {block.ref}.]"
-
- def _apply_deduplication(self, messages: list[dict[str, Any]], mutated: set[int]) -> None:
- protected = DCP_DEFAULT_PROTECTED_TOOLS | self.config.deduplication.protected_tools
- latest_by_sig: dict[str, int] = {}
- result_by_call_id: dict[str, int] = {}
- calls: list[tuple[int, str, str]] = []
- for idx, msg in enumerate(messages):
- if msg.get("role") == "assistant":
- for tc in msg.get("tool_calls") or []:
- name = self._tool_name(tc)
- if not name or name in protected:
- continue
- call_id = tc.get("id") if isinstance(tc, dict) else None
- if not isinstance(call_id, str):
- continue
- sig = self._tool_signature(tc)
- calls.append((idx, call_id, sig))
- latest_by_sig[sig] = idx
- elif msg.get("role") == "tool":
- call_id = msg.get("tool_call_id")
- if isinstance(call_id, str):
- result_by_call_id[call_id] = idx
- protected_indices = self._turn_protected_indices(messages)
- for call_idx, call_id, sig in calls:
- if latest_by_sig.get(sig) == call_idx:
- continue
- result_idx = result_by_call_id.get(call_id)
- if result_idx is not None and result_idx not in protected_indices:
- msg = self._clone_if_needed(messages, result_idx, mutated)
- msg["content"] = "[DCP: duplicate tool output removed. Same tool and arguments were called again later.]"
-
- def _apply_purge_errors(self, messages: list[dict[str, Any]], mutated: set[int]) -> None:
- protected = DCP_DEFAULT_PROTECTED_TOOLS | self.config.purge_errors.protected_tools
- keep_tail = max(0, self.config.purge_errors.turns * 2)
- cutoff = max(0, len(messages) - keep_tail)
- call_name_by_id: dict[str, str] = {}
- for msg in messages:
- if msg.get("role") != "assistant":
- continue
- for tc in msg.get("tool_calls") or []:
- call_id = tc.get("id") if isinstance(tc, dict) else None
- name = self._tool_name(tc)
- if isinstance(call_id, str) and name:
- call_name_by_id[call_id] = name
- protected_indices = self._turn_protected_indices(messages)
- for idx, msg in enumerate(messages[:cutoff]):
- if idx in protected_indices or msg.get("role") != "tool":
- continue
- call_id = msg.get("tool_call_id")
- name = call_name_by_id.get(call_id) if isinstance(call_id, str) else None
- if name in protected:
- continue
- content = msg.get("content")
- if isinstance(content, str) and len(content) > 240 and _ERROR_RE.search(content):
- first_line = content.strip().splitlines()[0][:240]
- cloned = self._clone_if_needed(messages, idx, mutated)
- cloned["content"] = f"[DCP: old failed tool output pruned after {self.config.purge_errors.turns} turns. Error preserved: {first_line}]"
-
- def _inject_system_extension(self, messages: list[dict[str, Any]], mutated: set[int]) -> None:
- if self.config.compress.permission == "deny":
- return
- if messages and messages[0].get("role") == "system" and isinstance(messages[0].get("content"), str):
- if _DCP_SYSTEM_EXTENSION not in messages[0]["content"]:
- msg = self._clone_if_needed(messages, 0, mutated)
- msg["content"] = f"{msg['content']}\n\n{_DCP_SYSTEM_EXTENSION}"
-
- def _inject_nudge(self, messages: list[dict[str, Any]], *, api_call_count: int, mutated: set[int]) -> None:
- # Use the provider-reported token count from the last response if
- # available — avoids re-estimating tokens on every API call.
- prompt_tokens = self.last_prompt_tokens
- max_limit = self._max_limit()
- min_limit = self._min_limit()
- nudge: str | None = None
- if max_limit and prompt_tokens >= max_limit:
- force = (
- "Before continuing, call compress on any completed range if safe."
- if self.config.compress.nudge_force == "strong"
- else "Consider calling compress on completed older ranges before continuing."
- )
- nudge = f"DCP context pressure is high (~{prompt_tokens:,} tokens). {force}"
- elif min_limit and prompt_tokens >= min_limit and self.state.turns_since_last_compress >= self.config.compress.nudge_frequency:
- nudge = "DCP: context is growing. If an older topic is complete, use compress with the visible refs."
- elif self.state.messages_since_last_user >= self.config.compress.iteration_nudge_threshold:
- nudge = "DCP: many assistant/tool messages have accumulated since the last user turn. Compress closed context if safe."
- elif self.state.manual_mode == "compress-pending":
- focus = f" Focus: {self.state.pending_manual_focus}." if self.state.pending_manual_focus else ""
- nudge = f"DCP manual compression requested.{focus} Call compress before continuing if there is safe completed context."
- self.state.manual_mode = False
- self.state.pending_manual_focus = None
-
- if not nudge:
- return
- # Only inject into user messages — never into tool results or
- # assistant messages, which could violate provider message semantics.
- for idx in range(len(messages) - 1, -1, -1):
- msg = messages[idx]
- if msg.get("role") == "user" and isinstance(msg.get("content"), str):
- cloned = self._clone_if_needed(messages, idx, mutated)
- cloned["content"] = f"{cloned['content']}\n\n{nudge}"
- return
-
- # -- Helpers ----------------------------------------------------------
-
- def _message_key(self, msg: dict[str, Any], idx: int) -> str:
- return f"{idx}:{self._content_signature(msg)}"
-
- def _content_signature(self, msg: dict[str, Any]) -> str:
- """Signature based on role + content only.
-
- Excludes tool_calls and tool_call_id because the API copy may have
- been re-serialised (sorted JSON keys) by _canonicalize_api_tool_calls,
- which would produce a different hash than the canonical message.
- """
- cache_key = id(msg)
- cached = self._sig_cache.get(cache_key)
- if cached is not None:
- return cached
- clean = {
- "role": msg.get("role"),
- "content": msg.get("content"),
- }
- raw = json.dumps(clean, sort_keys=True, default=str, separators=(",", ":"))
- sig = hashlib.sha1(raw.encode("utf-8", "ignore")).hexdigest()
- self._sig_cache[cache_key] = sig
- return sig
-
- def _require_str(self, args: dict[str, Any], key: str) -> str:
- value = args.get(key)
- if not isinstance(value, str) or not value.strip():
- raise ValueError(f"compress.{key} must be a non-empty string")
- return value.strip()
-
- def _resolve_range(self, start_ref: str, end_ref: str) -> tuple[list[str], list[int]]:
- start_idx = self._resolve_ref_to_index(start_ref)
- end_idx = self._resolve_ref_to_index(end_ref)
- if start_idx is None:
- raise ValueError(f"Unknown startId: {start_ref}")
- if end_idx is None:
- raise ValueError(f"Unknown endId: {end_ref}")
- if end_idx < start_idx:
- start_idx, end_idx = end_idx, start_idx
- refs = [ref for ref, idx in self.state.index_by_ref.items() if start_idx <= idx <= end_idx]
- refs.sort(key=lambda ref: self.state.index_by_ref[ref])
- included_blocks = [
- block.block_id
- for block in self.state.active_blocks()
- if any(ref in refs for ref in block.message_refs)
- ]
- return refs, included_blocks
-
- def _resolve_ref_to_index(self, ref: str) -> int | None:
- if ref.startswith("m"):
- return self.state.index_by_ref.get(ref)
- if ref.startswith("b"):
- try:
- block_id = int(ref[1:])
- except ValueError:
- return None
- block = self.state.blocks_by_id.get(block_id)
- if not block or not block.message_refs:
- return None
- return self.state.index_by_ref.get(block.message_refs[0])
- return None
-
- def _augment_summary(self, summary: str, message_refs: list[str]) -> str:
- parts = [summary.strip()]
- if self.config.compress.protect_user_messages:
- parts.append(f"Covered refs: {', '.join(message_refs)}")
- return "\n\n".join(part for part in parts if part)
-
- def _block_summary_text(self, block: CompressionBlock) -> str:
- covers = f"{block.start_ref}-{block.end_ref}" if block.start_ref and block.end_ref else ", ".join(block.message_refs)
- return (
- f'\n'
- f"Summary: {block.summary}\n"
- f"Covers: {covers}\n"
- ""
- )
-
- def _tool_name(self, tool_call: Any) -> str | None:
- if not isinstance(tool_call, dict):
- return None
- function = tool_call.get("function")
- if isinstance(function, dict) and isinstance(function.get("name"), str):
- return function["name"]
- return None
-
- def _tool_signature(self, tool_call: dict[str, Any]) -> str:
- function = tool_call.get("function") if isinstance(tool_call.get("function"), dict) else {}
- name = function.get("name", "")
- args = function.get("arguments", "")
- try:
- args_obj = json.loads(args) if isinstance(args, str) else args
- args_norm = json.dumps(args_obj, sort_keys=True, separators=(",", ":"), default=str)
- except Exception:
- args_norm = str(args)
- return f"{name}::{args_norm}"
-
- def _automatic_strategies_enabled(self) -> bool:
- if self.config.manual_mode.enabled and not self.config.manual_mode.automatic_strategies:
- return False
- return True
-
- def _turn_protected_indices(self, messages: list[dict[str, Any]]) -> set[int]:
- if not self.config.turn_protection.enabled or self.config.turn_protection.turns <= 0:
- return set()
- user_indices = [idx for idx, msg in enumerate(messages) if msg.get("role") == "user"]
- if not user_indices:
- return set()
- start = user_indices[-self.config.turn_protection.turns] if len(user_indices) >= self.config.turn_protection.turns else user_indices[0]
- return set(range(start, len(messages)))
-
- def _evict_inactive_blocks(self) -> None:
- """Bound memory by evicting old deactivated blocks."""
- inactive = sorted(
- (bid for bid, b in self.state.blocks_by_id.items() if not b.active),
- key=lambda bid: self.state.blocks_by_id[bid].deactivated_at or 0,
- )
- for bid in inactive[_MAX_INACTIVE_BLOCKS:]:
- del self.state.blocks_by_id[bid]
-
- def _min_limit(self) -> int:
- return resolve_model_limit(
- self.config.compress.model_min_limits,
- provider=self.provider,
- model=self.model,
- context_length=self.context_length,
- fallback=self.config.compress.min_context_limit,
- )
-
- def _max_limit(self) -> int:
- return resolve_model_limit(
- self.config.compress.model_max_limits,
- provider=self.provider,
- model=self.model,
- context_length=self.context_length,
- fallback=self.config.compress.max_context_limit,
- )
diff --git a/agent/dcp_state.py b/agent/dcp_state.py
deleted file mode 100644
index 1f982ffdf50ba..0000000000000
--- a/agent/dcp_state.py
+++ /dev/null
@@ -1,70 +0,0 @@
-"""State model for the DCP context engine."""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from typing import Any, Literal
-
-
-@dataclass(slots=True)
-class CompressionBlock:
- block_id: int
- run_id: int
- mode: Literal["range", "message"]
- topic: str
- summary: str
- active: bool = True
- start_ref: str | None = None
- end_ref: str | None = None
- message_refs: list[str] = field(default_factory=list)
- included_block_ids: list[int] = field(default_factory=list)
- consumed_block_ids: list[int] = field(default_factory=list)
- created_at: float = 0.0
- deactivated_at: float | None = None
- deactivated_by_block_id: int | None = None
-
- @property
- def ref(self) -> str:
- return f"b{self.block_id}"
-
-
-@dataclass(slots=True)
-class DCPSessionState:
- session_id: str | None = None
- next_message_ref: int = 1
- next_block_id: int = 1
- next_run_id: int = 1
- ref_by_message_key: dict[str, str] = field(default_factory=dict)
- message_key_by_ref: dict[str, str] = field(default_factory=dict)
- index_by_ref: dict[str, int] = field(default_factory=dict)
- blocks_by_id: dict[int, CompressionBlock] = field(default_factory=dict)
- active_block_ids: set[int] = field(default_factory=set)
- last_prompt_tokens: int = 0
- last_user_turn_index: int = 0
- turns_since_last_compress: int = 0
- messages_since_last_user: int = 0
- manual_mode: bool | Literal["compress-pending"] = False
- pending_manual_focus: str | None = None
- stats: dict[str, Any] = field(default_factory=dict)
-
- def new_message_ref(self) -> str:
- ref = f"m{self.next_message_ref:04d}"
- self.next_message_ref += 1
- return ref
-
- def new_block_id(self) -> int:
- block_id = self.next_block_id
- self.next_block_id += 1
- return block_id
-
- def new_run_id(self) -> int:
- run_id = self.next_run_id
- self.next_run_id += 1
- return run_id
-
- def active_blocks(self) -> list[CompressionBlock]:
- return [
- self.blocks_by_id[block_id]
- for block_id in sorted(self.active_block_ids)
- if block_id in self.blocks_by_id and self.blocks_by_id[block_id].active
- ]
diff --git a/tests/agent/test_dcp_config.py b/tests/agent/test_dcp_config.py
deleted file mode 100644
index 5157fe1b3a6e6..0000000000000
--- a/tests/agent/test_dcp_config.py
+++ /dev/null
@@ -1,63 +0,0 @@
-from __future__ import annotations
-
-from agent.dcp_config import parse_dcp_config, resolve_limit, resolve_model_limit
-
-
-def test_dcp_config_defaults_match_supported_surface():
- cfg = parse_dcp_config({})
-
- assert cfg.enabled is True
- assert cfg.prune_notification == "detailed"
- assert cfg.compress.mode == "range"
- assert cfg.compress.permission == "allow"
- assert cfg.compress.max_context_limit == 100000
- assert cfg.compress.min_context_limit == 50000
- assert cfg.deduplication.enabled is True
- assert cfg.purge_errors.enabled is True
- assert cfg.purge_errors.turns == 4
-
-
-def test_dcp_config_parses_percent_limits_and_model_overrides():
- cfg = parse_dcp_config(
- {
- "compress": {
- "maxContextLimit": "80%",
- "minContextLimit": "40%",
- "modelMaxLimits": {"openai/test-model": "90%"},
- "modelMinLimits": {"test-model": 12345},
- }
- }
- )
-
- assert resolve_limit(cfg.compress.max_context_limit, 200000) == 160000
- assert resolve_limit(cfg.compress.min_context_limit, 200000) == 80000
- assert resolve_model_limit(
- cfg.compress.model_max_limits,
- provider="openai",
- model="test-model",
- context_length=200000,
- fallback=cfg.compress.max_context_limit,
- ) == 180000
- assert resolve_model_limit(
- cfg.compress.model_min_limits,
- provider="openai",
- model="test-model",
- context_length=200000,
- fallback=cfg.compress.min_context_limit,
- ) == 12345
-
-
-def test_dcp_config_rejects_invalid_choices_to_defaults():
- cfg = parse_dcp_config(
- {
- "compress": {
- "mode": "bad",
- "permission": "root",
- "nudgeForce": "loud",
- }
- }
- )
-
- assert cfg.compress.mode == "range"
- assert cfg.compress.permission == "allow"
- assert cfg.compress.nudge_force == "soft"
diff --git a/tests/agent/test_dcp_context_engine.py b/tests/agent/test_dcp_context_engine.py
deleted file mode 100644
index c937731960487..0000000000000
--- a/tests/agent/test_dcp_context_engine.py
+++ /dev/null
@@ -1,331 +0,0 @@
-from __future__ import annotations
-
-import json
-
-from agent.dcp_context_engine import DCPContextEngine
-
-
-def _tool_call(call_id: str, name: str, args: dict) -> dict:
- return {
- "id": call_id,
- "type": "function",
- "function": {"name": name, "arguments": json.dumps(args)},
- }
-
-
-def test_range_tool_schema_is_exposed_by_default():
- engine = DCPContextEngine(config={}, context_length=200000)
-
- schemas = engine.get_tool_schemas()
-
- assert len(schemas) == 1
- schema = schemas[0]
- assert schema["name"] == "compress"
- assert schema["parameters"]["required"] == ["topic", "content"]
- item = schema["parameters"]["properties"]["content"]["items"]
- assert item["required"] == ["startId", "endId", "summary"]
-
-
-def test_message_tool_schema_when_configured():
- engine = DCPContextEngine(config={"compress": {"mode": "message"}}, context_length=200000)
-
- schema = engine.get_tool_schemas()[0]
-
- item = schema["parameters"]["properties"]["content"]["items"]
- assert item["required"] == ["messageId", "topic", "summary"]
-
-
-def test_deny_permission_hides_compress_tool():
- engine = DCPContextEngine(config={"compress": {"permission": "deny"}}, context_length=200000)
-
- assert engine.get_tool_schemas() == []
-
-
-def test_disabled_engine_exposes_no_tool_and_returns_original_api_messages():
- engine = DCPContextEngine(config={"enabled": False}, context_length=200000)
- api_messages = [{"role": "user", "content": "hello"}]
-
- transformed = engine.transform_api_messages(
- api_messages,
- canonical_messages=[{"role": "user", "content": "hello"}],
- system_prompt="",
- tools=[],
- api_call_count=1,
- model="test-model",
- provider="openai",
- session_id="s1",
- )
-
- assert engine.get_tool_schemas() == []
- assert transformed is api_messages
- assert api_messages == [{"role": "user", "content": "hello"}]
-
-
-def test_transform_does_not_mutate_canonical_messages_and_adds_refs():
- engine = DCPContextEngine(config={}, context_length=200000)
- canonical = [{"role": "user", "content": "hello"}, {"role": "assistant", "content": "world"}]
- original = [msg.copy() for msg in canonical]
- api_messages = [{"role": "system", "content": "sys"}] + [msg.copy() for msg in canonical]
-
- transformed = engine.transform_api_messages(
- api_messages,
- canonical_messages=canonical,
- system_prompt="sys",
- tools=[],
- api_call_count=1,
- model="test-model",
- provider="openai",
- session_id="s1",
- )
-
- assert canonical == original
- assert '' in transformed[1]["content"]
- assert '' in transformed[2]["content"]
- assert "DCP context management is active" in transformed[0]["content"]
-
-
-def test_range_compress_creates_block_and_transform_applies_placeholder():
- engine = DCPContextEngine(config={}, context_length=200000)
- canonical = [
- {"role": "user", "content": "start"},
- {"role": "assistant", "content": "old work"},
- {"role": "user", "content": "new task"},
- ]
- engine._ensure_refs(canonical)
-
- result = json.loads(
- engine.handle_tool_call(
- "compress",
- {
- "topic": "old work",
- "content": [{"startId": "m0001", "endId": "m0002", "summary": "Old work summary."}],
- },
- messages=canonical,
- )
- )
-
- assert result["ok"] is True
- assert result["created_blocks"] == [1]
- transformed = engine.transform_api_messages(
- [msg.copy() for msg in canonical],
- canonical_messages=canonical,
- system_prompt="",
- tools=[],
- api_call_count=1,
- model="test-model",
- provider="openai",
- session_id="s1",
- )
- assert '' in transformed[0]["content"]
- assert "content moved into compressed block b1" in transformed[1]["content"]
- assert "new task" in transformed[2]["content"]
-
-
-def test_range_compress_consumes_overlapping_active_blocks():
- engine = DCPContextEngine(config={}, context_length=200000)
- canonical = [
- {"role": "user", "content": "phase one"},
- {"role": "assistant", "content": "phase one result"},
- {"role": "user", "content": "phase two"},
- {"role": "assistant", "content": "phase two result"},
- ]
- engine._ensure_refs(canonical)
-
- first = json.loads(
- engine.handle_tool_call(
- "compress",
- {"topic": "phase one", "content": [{"startId": "m0001", "endId": "m0002", "summary": "Phase one summary."}]},
- messages=canonical,
- )
- )
- second = json.loads(
- engine.handle_tool_call(
- "compress",
- {"topic": "both phases", "content": [{"startId": "b1", "endId": "m0004", "summary": "Both phases summary."}]},
- messages=canonical,
- )
- )
-
- assert first["created_blocks"] == [1]
- assert second["created_blocks"] == [2]
- assert second["deactivated_blocks"] == [1]
- assert engine.state.blocks_by_id[1].active is False
- assert engine.state.blocks_by_id[1].deactivated_by_block_id == 2
- assert engine.state.active_block_ids == {2}
-
-
-def test_multimodal_messages_get_text_ref_without_mutating_canonical_content():
- engine = DCPContextEngine(config={}, context_length=200000)
- canonical = [
- {
- "role": "user",
- "content": [
- {"type": "text", "text": "look at this"},
- {"type": "image_url", "image_url": {"url": "https://example.invalid/image.png"}},
- ],
- }
- ]
- api_messages = [{"role": "system", "content": "sys"}] + [msg.copy() for msg in canonical]
-
- transformed = engine.transform_api_messages(
- api_messages,
- canonical_messages=canonical,
- system_prompt="sys",
- tools=[],
- api_call_count=1,
- model="test-model",
- provider="openai",
- session_id="s1",
- )
-
- assert canonical[0]["content"] == [
- {"type": "text", "text": "look at this"},
- {"type": "image_url", "image_url": {"url": "https://example.invalid/image.png"}},
- ]
- assert transformed[1]["content"][-1] == {"type": "text", "text": ''}
-
-
-def test_message_compress_creates_message_block():
- engine = DCPContextEngine(config={"compress": {"mode": "message"}}, context_length=200000)
- canonical = [{"role": "user", "content": "huge pasted log"}]
- engine._ensure_refs(canonical)
-
- result = json.loads(
- engine.handle_tool_call(
- "compress",
- {"topic": "logs", "content": [{"messageId": "m0001", "topic": "log", "summary": "Useful log facts."}]},
- messages=canonical,
- )
- )
-
- assert result["ok"] is True
- assert result["mode"] == "message"
- assert result["created_blocks"] == [1]
-
-
-def test_deduplication_prunes_older_duplicate_tool_output():
- engine = DCPContextEngine(config={}, context_length=200000)
- messages = [
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("a", "read_file", {"path": "x"})]},
- {"role": "tool", "tool_call_id": "a", "content": "old output"},
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("b", "read_file", {"path": "x"})]},
- {"role": "tool", "tool_call_id": "b", "content": "new output"},
- ]
-
- engine._apply_deduplication(messages, set())
-
- assert "duplicate tool output removed" in messages[1]["content"]
- assert messages[3]["content"] == "new output"
-
-
-def test_deduplication_respects_protected_tools():
- engine = DCPContextEngine(config={}, context_length=200000)
- messages = [
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("a", "patch", {"path": "x"})]},
- {"role": "tool", "tool_call_id": "a", "content": "old output"},
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("b", "patch", {"path": "x"})]},
- {"role": "tool", "tool_call_id": "b", "content": "new output"},
- ]
-
- engine._apply_deduplication(messages, set())
-
- assert messages[1]["content"] == "old output"
-
-
-def test_purge_errors_preserves_error_summary():
- engine = DCPContextEngine(config={"strategies": {"purgeErrors": {"turns": 0}}}, context_length=200000)
- messages = [
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("a", "terminal", {"command": "bad"})]},
- {"role": "tool", "tool_call_id": "a", "content": "ERROR: failed\n" + "x" * 500},
- {"role": "user", "content": "next"},
- ]
-
- engine._apply_purge_errors(messages, set())
-
- assert "old failed tool output pruned" in messages[1]["content"]
- assert "ERROR: failed" in messages[1]["content"]
-
-
-def test_turn_protection_prevents_dedup_pruning_recent_messages():
- engine = DCPContextEngine(
- config={"turnProtection": {"enabled": True, "turns": 1}},
- context_length=200000,
- )
- messages = [
- {"role": "user", "content": "latest turn"},
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("a", "read_file", {"path": "x"})]},
- {"role": "tool", "tool_call_id": "a", "content": "old output"},
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("b", "read_file", {"path": "x"})]},
- {"role": "tool", "tool_call_id": "b", "content": "new output"},
- ]
-
- engine._apply_deduplication(messages, set())
-
- assert messages[1]["content"] == ""
- assert messages[2]["content"] == "old output"
-
-
-def test_manual_mode_can_disable_automatic_strategies_in_transform():
- engine = DCPContextEngine(
- config={"manualMode": {"enabled": True, "automaticStrategies": False}},
- context_length=200000,
- )
- canonical = [
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("a", "read_file", {"path": "x"})]},
- {"role": "tool", "tool_call_id": "a", "content": "old output"},
- {"role": "assistant", "content": "", "tool_calls": [_tool_call("b", "read_file", {"path": "x"})]},
- {"role": "tool", "tool_call_id": "b", "content": "new output"},
- ]
-
- transformed = engine.transform_api_messages(
- [msg.copy() for msg in canonical],
- canonical_messages=canonical,
- system_prompt="",
- tools=[],
- api_call_count=1,
- model="test-model",
- provider="openai",
- session_id="s1",
- )
-
- assert transformed[1]["content"].startswith("old output")
- assert transformed[3]["content"].startswith("new output")
-
-
-def test_manual_compress_request_injects_one_shot_nudge_without_mutating_history():
- engine = DCPContextEngine(config={}, context_length=200000)
- canonical = [
- {"role": "user", "content": "please compact old work"},
- {"role": "assistant", "content": "working"},
- ]
-
- returned = engine.compress(canonical, current_tokens=1234, focus_topic="old investigation")
- first = engine.transform_api_messages(
- [msg.copy() for msg in canonical],
- canonical_messages=canonical,
- system_prompt="",
- tools=[],
- api_call_count=1,
- model="test-model",
- provider="openai",
- session_id="s1",
- )
- second = engine.transform_api_messages(
- [msg.copy() for msg in canonical],
- canonical_messages=canonical,
- system_prompt="",
- tools=[],
- api_call_count=2,
- model="test-model",
- provider="openai",
- session_id="s1",
- )
-
- assert returned is canonical
- assert "DCP manual compression requested" in first[0]["content"]
- assert "old investigation" in first[0]["content"]
- assert "DCP manual compression requested" not in second[0]["content"]
- assert canonical == [
- {"role": "user", "content": "please compact old work"},
- {"role": "assistant", "content": "working"},
- ]
diff --git a/website/docs/developer-guide/agent-loop.md b/website/docs/developer-guide/agent-loop.md
index 676807bd146ba..24cee51082df7 100644
--- a/website/docs/developer-guide/agent-loop.md
+++ b/website/docs/developer-guide/agent-loop.md
@@ -70,13 +70,10 @@ run_conversation()
- chat_completions: OpenAI format as-is
- codex_responses: convert to Responses API input items
- anthropic_messages: convert via anthropic_adapter.py
- 6. Let the active context engine transform the API-call copy, if supported
- - DCP-style engines can add refs, compression placeholders, and nudges
- - canonical conversation history must remain unchanged
- 7. Inject ephemeral prompt layers (budget warnings, context pressure)
- 8. Apply prompt caching markers if on Anthropic
- 9. Make interruptible API call (_interruptible_api_call)
- 10. Parse response:
+ 6. Inject ephemeral prompt layers (budget warnings, context pressure)
+ 7. Apply prompt caching markers if on Anthropic
+ 8. Make interruptible API call (_interruptible_api_call)
+ 9. Parse response:
- If tool_calls: execute them, append results, loop back to step 5
- If text response: persist session, flush memory if needed, return
```
@@ -163,19 +160,6 @@ Some tools are intercepted by `run_agent.py` *before* reaching `handle_function_
These tools modify agent state directly and return synthetic tool results without going through the registry.
-
-### Context-engine tools
-
-The active context engine can expose tools via `get_tool_schemas()`. These tools
-are injected into the model-visible tool list and routed back to
-`handle_tool_call()` before normal registry dispatch.
-
-This is how DCP-style context management exposes a model-callable `compress`
-tool. The tool updates context-engine state, then the next API-call transform
-applies compression blocks to the outbound message copy. It should not mutate
-the canonical transcript unless the engine explicitly documents a
-transcript-mutating mode.
-
## Callback Surfaces
`AIAgent` supports platform-specific callbacks that enable real-time progress in the CLI, gateway, and ACP integrations:
diff --git a/website/docs/developer-guide/architecture.md b/website/docs/developer-guide/architecture.md
index 1807a14333483..6c1f6cafa412c 100644
--- a/website/docs/developer-guide/architecture.md
+++ b/website/docs/developer-guide/architecture.md
@@ -30,7 +30,7 @@ This page is the top-level map of Hermes Agent internals. Use it to orient yours
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ ┌──────┴───────┐ ┌──────┴───────┐ ┌──────┴───────┐ │
-│ │ Context Mgmt │ │ 3 API Modes │ │ Tool Registry│ │
+│ │ Compression │ │ 3 API Modes │ │ Tool Registry│ │
│ │ & Caching │ │ chat_compl. │ │ (registry.py)│ │
│ │ │ │ codex_resp. │ │ 70+ tools │ │
│ │ │ │ anthropic │ │ 28 toolsets │ │
@@ -64,7 +64,6 @@ hermes-agent/
│ ├── prompt_builder.py # System prompt assembly
│ ├── context_engine.py # ContextEngine ABC (pluggable)
│ ├── context_compressor.py # Default engine — lossy summarization
-│ ├── dcp_context_engine.py # Optional DCP-style model-guided context engine
│ ├── prompt_caching.py # Anthropic prompt caching
│ ├── auxiliary_client.py # Auxiliary LLM for side tasks (vision, summarization)
│ ├── model_metadata.py # Model context lengths, token estimation
@@ -136,15 +135,6 @@ hermes-agent/
└── tests/ # Pytest suite (~25,000 tests across ~1,250 files)
```
-
-### Context management
-
-Context management is handled through the `ContextEngine` interface. The default
-engine is `ContextCompressor`, which performs host-triggered summarization. Other
-engines can expose tools and transform the provider-bound API message copy. A
-DCP-style engine uses that path to keep the stored transcript complete while
-sending compressed blocks, refs, and nudges to the model.
-
## Data Flow
### CLI Session
diff --git a/website/docs/developer-guide/context-compression-and-caching.md b/website/docs/developer-guide/context-compression-and-caching.md
index 9ed19b4a554fe..0b71f4eb6d1f6 100644
--- a/website/docs/developer-guide/context-compression-and-caching.md
+++ b/website/docs/developer-guide/context-compression-and-caching.md
@@ -34,88 +34,6 @@ Configure via `hermes plugins` → Provider Plugins → Context Engine, or edit
For building a context engine plugin, see [Context Engine Plugins](/developer-guide/context-engine-plugin).
-
-## DCP Context Engine
-
-Hermes can also run a DCP-style context engine with:
-
-```yaml
-context:
- engine: "dcp"
-```
-
-DCP mode is different from the built-in `ContextCompressor`. The built-in
-compressor is host-driven: Hermes decides that the session is too large, calls
-an auxiliary summarization model, and replaces the stored message list with a
-compressed transcript. DCP mode is model-guided: Hermes exposes a `compress`
-tool, adds stable message and block references to the outbound request, and
-lets the active model compress completed ranges when it has enough semantic
-context to know what is safe to replace.
-
-The DCP invariant is:
-
-> The canonical session transcript remains complete. DCP transforms only the
-> API-call copy of the messages sent to the provider.
-
-A DCP engine owns separate compression state:
-
-- message refs such as `m0001`, `m0002`
-- compression block refs such as `b1`, `b2`
-- active block summaries
-- duplicate-tool and old-error pruning state
-- compression stats and nudge cadence
-
-### DCP `compress` tool
-
-When DCP mode is active, the context engine may expose a `compress` tool. The
-tool does not rewrite the stored transcript. Instead, it records compression
-blocks. The next API-call transform applies those blocks to the outbound copy.
-
-DCP supports two modes:
-
-- `range`: compress one or more contiguous spans using `{startId, endId, summary}`.
-- `message`: compress individual high-volume messages using `{messageId, topic, summary}`.
-
-Range mode is the default because it preserves chronology and usually gives
-the model enough context to summarize closed work accurately. Message mode is
-more surgical and should be treated as experimental until provider-format and
-cache behavior are well tested.
-
-### DCP nudges and automatic strategies
-
-DCP mode can inject ephemeral nudges when context pressure rises. Nudges tell
-the model to call `compress` before continuing if a completed topic is safe to
-compact. The engine may also run cheap automatic strategies over the outbound
-copy:
-
-- deduplicate repeated tool calls with the same tool name and arguments, keeping
- the latest output
-- purge bulky old failed-tool inputs while preserving the error text
-- protect recent user turns and configured protected tools
-
-These strategies are DCP-state transformations, not transcript edits.
-
-### Interaction with existing compression
-
-When `context.engine: "dcp"`, the built-in `compression:` settings do not drive
-normal compaction. DCP should return `False` from `should_compress()` during
-normal operation and rely on nudges plus the `compress` tool. The built-in
-`ContextCompressor` may still be used as an emergency fallback for hard context
-limit failures, but it should not run as a parallel primary compressor.
-
-Gateway session hygiene remains a safety net. Because hygiene compression
-operates before the agent starts and may mutate gateway history, DCP-aware
-hygiene behavior must be handled deliberately rather than implicitly reusing
-the built-in compressor path.
-
-### Prompt caching
-
-DCP must run before provider cache-control markers are applied so prompt caching
-sees the actual outgoing request. The transform should be deterministic and
-should avoid modifying old stable content on every turn. Compression blocks and
-automatic pruning should change the cached prefix only when compression state
-changes, not as a side effect of moving counters or timestamps.
-
## Dual Compression System
Hermes has two separate compression layers that operate independently:
From 0647bf98895426f0759599a30c8b0a0cf5640e37 Mon Sep 17 00:00:00 2001
From: kshitij <82637225+kshitijk4poor@users.noreply.github.com>
Date: Sat, 8 Aug 2026 23:09:45 +0530
Subject: [PATCH 2/2] Revert "fix: rewire DCP context engine to current main
architecture"
This reverts commit 9841a6c65161b9253c342f04b20f3a8bdb63884c.
---
agent/agent_init.py | 10 +---------
agent/context_engine.py | 33 ---------------------------------
agent/conversation_loop.py | 28 ----------------------------
hermes_cli/config_defaults.py | 26 --------------------------
4 files changed, 1 insertion(+), 96 deletions(-)
diff --git a/agent/agent_init.py b/agent/agent_init.py
index d71cfd6471fb0..0cb21e9446539 100644
--- a/agent/agent_init.py
+++ b/agent/agent_init.py
@@ -31,7 +31,6 @@ from typing import Any, Callable, Dict, List, Optional
from urllib.parse import parse_qs, urlparse, urlunparse
from agent.context_compressor import ContextCompressor
-from agent.dcp_context_engine import DCPContextEngine
from agent.iteration_budget import IterationBudget
from agent.memory_manager import StreamingContextScrubber
from agent.session_activity import ActivityProvenance
@@ -2412,14 +2411,7 @@ def init_agent(
except Exception:
pass
- if _engine_name == "dcp":
- _selected_engine = DCPContextEngine(
- config=_ctx_cfg.get("dcp", {}) if isinstance(_ctx_cfg, dict) else {},
- model=agent.model,
- provider=agent.provider,
- quiet_mode=agent.quiet_mode,
- )
- elif _engine_name != "compressor":
+ if _engine_name != "compressor":
# Try loading from plugins/context_engine//
try:
from plugins.context_engine import load_context_engine
diff --git a/agent/context_engine.py b/agent/context_engine.py
index 12393b42116c5..b772125c0cbe3 100644
--- a/agent/context_engine.py
+++ b/agent/context_engine.py
@@ -382,39 +382,6 @@ class ContextEngine(ABC):
"""
return True
- # -- Optional: API-call-time transform ---------------------------------
-
- def transform_api_messages(
- self,
- api_messages: List[Dict[str, Any]],
- *,
- canonical_messages: List[Dict[str, Any]],
- system_prompt: str,
- tools: List[Dict[str, Any]] | None,
- api_call_count: int,
- model: str,
- provider: str | None,
- session_id: str | None,
- ) -> List[Dict[str, Any]]:
- """Transform the provider-bound API-call copy of the transcript.
-
- Default returns ``api_messages`` unchanged. Engines may override this
- to add ephemeral refs, compression placeholders, or other context
- layers without mutating ``canonical_messages``.
-
- Contract:
- - Receives the API-call copy, not authoritative history.
- - Must not mutate ``canonical_messages``.
- - Must preserve valid OpenAI message ordering.
- - Must not separate an assistant ``tool_calls`` message from its
- required tool results.
- - Should run before prompt-cache marker placement so caching logic
- sees the actual outgoing request.
- - Should avoid churning the stable prefix on every call to preserve
- prompt-cache hit rates.
- """
- return api_messages
-
# -- Optional: session lifecycle ---------------------------------------
def on_session_start(self, session_id: str, **kwargs) -> None:
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index 596e9b6e07a95..2ab35ec945e11 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -1981,34 +1981,6 @@ def run_conversation(
# the OpenAI SDK. Sanitizing here prevents the 3-retry cycle.
_sanitize_messages_surrogates(api_messages)
- # Let the active context engine transform the API-call copy.
- # DCP uses this to add refs, compression block placeholders, and
- # nudges without mutating the canonical messages transcript.
- # Runs after sanitization/normalization so the transform sees the
- # final wire shape, and before prompt-cache marker placement so
- # caching logic sees the actual outgoing request.
- _ctx_engine = getattr(agent, "context_compressor", None)
- if _ctx_engine is not None:
- _transform_hook = getattr(_ctx_engine, "transform_api_messages", None)
- if callable(_transform_hook):
- try:
- api_messages = _transform_hook(
- api_messages,
- canonical_messages=messages,
- system_prompt=effective_system,
- tools=agent.tools,
- api_call_count=api_call_count,
- model=agent.model,
- provider=agent.provider,
- session_id=agent.session_id,
- )
- except Exception as _ctx_err:
- request_logger.warning(
- "Context engine transform_api_messages failed (session=%s): %s",
- getattr(agent, "session_id", None) or "-",
- _ctx_err,
- )
-
# NOTE (empty-content class fix): no send-time pad loop here. The
# single owner for "never send a turn strict wire validation rejects
# as empty" is ``repair_empty_non_final_messages``, which runs inside
diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py
index 276570461c07c..b3073d9517a2d 100644
--- a/hermes_cli/config_defaults.py
+++ b/hermes_cli/config_defaults.py
@@ -1665,32 +1665,6 @@ DEFAULT_CONFIG = {
# a plugin in plugins/context_engine// or ~/.hermes/plugins/.
"context": {
"engine": "compressor",
- # DCP context engine config — only active when engine == "dcp".
- "dcp": {
- "enabled": True,
- "compress": {
- "mode": "range",
- "permission": "allow",
- "maxContextLimit": 100000,
- "minContextLimit": 50000,
- "nudgeFrequency": 5,
- "iterationNudgeThreshold": 15,
- "nudgeForce": "soft",
- },
- "strategies": {
- "deduplication": {
- "enabled": True,
- },
- "purgeErrors": {
- "enabled": True,
- "turns": 4,
- },
- },
- "turnProtection": {
- "enabled": False,
- "turns": 4,
- },
- },
# Return freed glibc allocator pages after long-running agent/TUI
# cleanup boundaries. Unsupported platforms are safe no-ops.
"memory_trim": {