diff --git a/agent/context_engine.py b/agent/context_engine.py
index bbafcd29c017a..1de01ca0eb4a3 100644
--- a/agent/context_engine.py
+++ b/agent/context_engine.py
@@ -105,6 +105,26 @@ class ContextEngine(ABC):
"""
return False
+ 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``.
+ """
+ return api_messages
+
# -- Optional: manual /compress preflight ------------------------------
def has_content_to_compress(self, messages: List[Dict[str, Any]]) -> bool:
diff --git a/agent/dcp_config.py b/agent/dcp_config.py
new file mode 100644
index 0000000000000..aac674907863b
--- /dev/null
+++ b/agent/dcp_config.py
@@ -0,0 +1,256 @@
+"""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
new file mode 100644
index 0000000000000..1837cbe4e81cb
--- /dev/null
+++ b/agent/dcp_context_engine.py
@@ -0,0 +1,624 @@
+"""DCP-style model-guided context engine for Hermes Agent."""
+
+from __future__ import annotations
+
+import copy
+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
+from agent.model_metadata import estimate_messages_tokens_rough
+
+_ERROR_RE = re.compile(r"\b(error|exception|traceback|failed|failure|timed out|timeout)\b", re.I)
+
+
+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()
+
+ @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:
+ # DCP is normally model-guided through the compress tool, not host-driven.
+ 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]]:
+ # Manual /compress fallback: record a pending nudge and leave transcript intact.
+ 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)
+
+ def update_model(
+ self,
+ model: str,
+ context_length: int,
+ base_url: str = "",
+ api_key: str = "",
+ provider: 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 []
+ if self.config.compress.mode == "message":
+ return [self._message_tool_schema()]
+ return [self._range_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:
+ if self.config.compress.mode == "message":
+ result = self._handle_message_compress(args, messages)
+ else:
+ result = self._handle_range_compress(args, messages)
+ except Exception as exc: # fail loudly to the model, not to the agent loop
+ 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
+ self._ensure_refs(canonical_messages)
+
+ transformed = copy.deepcopy(api_messages)
+ ref_by_api_index = self._match_api_messages_to_refs(transformed, canonical_messages)
+ self._annotate_refs(transformed, ref_by_api_index)
+ self._apply_blocks(transformed, ref_by_api_index)
+
+ if self._automatic_strategies_enabled():
+ if self.config.deduplication.enabled:
+ self._apply_deduplication(transformed)
+ if self.config.purge_errors.enabled:
+ self._apply_purge_errors(transformed)
+
+ self._inject_system_extension(transformed)
+ self._inject_nudge(transformed, api_call_count=api_call_count)
+ 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 schemas -----------------------------------------------------
+
+ def _range_tool_schema(self) -> dict[str, Any]:
+ return {
+ "name": "compress",
+ "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."
+ ),
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "topic": {"type": "string", "description": "Short 3-5 word label for this compression batch."},
+ "content": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "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."},
+ },
+ "required": ["startId", "endId", "summary"],
+ },
+ },
+ },
+ "required": ["topic", "content"],
+ },
+ }
+
+ def _message_tool_schema(self) -> dict[str, Any]:
+ return {
+ "name": "compress",
+ "description": (
+ "Compress individual high-volume messages by ref. Preserve concrete "
+ "technical facts, file paths, commands, decisions, errors, and open questions."
+ ),
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "topic": {"type": "string", "description": "Short label for this compression batch."},
+ "content": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "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."},
+ },
+ "required": ["messageId", "topic", "summary"],
+ },
+ },
+ },
+ "required": ["topic", "content"],
+ },
+ }
+
+ # -- Tool handling ----------------------------------------------------
+
+ def _handle_range_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")
+
+ 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 range must be an object")
+ start_ref = self._require_str(item, "startId")
+ end_ref = self._require_str(item, "endId")
+ summary = self._require_str(item, "summary")
+ 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")
+ block_id = self.state.new_block_id()
+ consumed_blocks: list[int] = []
+ for included in included_blocks:
+ block = self.state.blocks_by_id.get(included)
+ if block and block.active:
+ block.active = False
+ block.deactivated_at = time.time()
+ block.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=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_id] = block
+ self.state.active_block_ids.add(block_id)
+ created.append(block_id)
+
+ self.compression_count += len(created)
+ self.state.turns_since_last_compress = 0
+ return {
+ "ok": True,
+ "mode": "range",
+ "created_blocks": created,
+ "deactivated_blocks": deactivated,
+ "active_blocks": sorted(self.state.active_block_ids),
+ "message": f"Compressed {len(created)} range(s) into {', '.join(f'b{i}' for i in created)}.",
+ }
+
+ def _handle_message_compress(self, args: dict[str, Any], messages: list[dict[str, Any]]) -> dict[str, Any]:
+ self._ensure_refs(messages)
+ batch_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")
+ created: list[int] = []
+ run_id = self.state.new_run_id()
+ for item in content:
+ if not isinstance(item, dict):
+ raise ValueError("Each compressed message must be an object")
+ ref = self._require_str(item, "messageId")
+ if ref not in self.state.index_by_ref:
+ raise ValueError(f"Unknown message ref: {ref}")
+ topic = item.get("topic") if isinstance(item.get("topic"), str) else batch_topic
+ summary = self._require_str(item, "summary")
+ block_id = self.state.new_block_id()
+ block = CompressionBlock(
+ block_id=block_id,
+ run_id=run_id,
+ mode="message",
+ topic=topic,
+ summary=self._augment_summary(summary, [ref]),
+ message_refs=[ref],
+ created_at=time.time(),
+ )
+ self.state.blocks_by_id[block_id] = block
+ self.state.active_block_ids.add(block_id)
+ created.append(block_id)
+ self.compression_count += len(created)
+ self.state.turns_since_last_compress = 0
+ return {
+ "ok": True,
+ "mode": "message",
+ "created_blocks": created,
+ "active_blocks": sorted(self.state.active_block_ids),
+ "message": f"Compressed {len(created)} message(s) into {', '.join(f'b{i}' for i in created)}.",
+ }
+
+ # -- Transforms -------------------------------------------------------
+
+ def _ensure_refs(self, messages: list[dict[str, Any]]) -> None:
+ 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]:
+ self._ensure_refs(canonical_messages)
+ refs_by_key: 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_key[self._message_signature(msg)].append(ref)
+
+ out: dict[int, str] = {}
+ for api_idx, msg in enumerate(api_messages):
+ if msg.get("role") == "system":
+ continue
+ sig = self._message_signature(msg)
+ queue = refs_by_key.get(sig)
+ if queue:
+ out[api_idx] = queue.popleft()
+ return out
+
+ def _annotate_refs(self, messages: list[dict[str, Any]], ref_by_api_index: dict[int, str]) -> None:
+ for idx, ref in ref_by_api_index.items():
+ msg = messages[idx]
+ 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]) -> 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 = messages[anchor_idx]
+ anchor["content"] = self._block_summary_text(block)
+ for ref in covered[1:]:
+ idx = ref_to_api_index[ref]
+ msg = messages[idx]
+ msg["content"] = f"[DCP: content moved into compressed block {block.ref}.]"
+
+ def _apply_deduplication(self, messages: list[dict[str, Any]]) -> 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:
+ messages[result_idx]["content"] = "[DCP: duplicate tool output removed. Same tool and arguments were called again later.]"
+
+ def _apply_purge_errors(self, messages: list[dict[str, Any]]) -> 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]
+ msg["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]]) -> None:
+ if self.config.compress.permission == "deny":
+ return
+ 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."
+ )
+ if messages and messages[0].get("role") == "system" and isinstance(messages[0].get("content"), str):
+ if "DCP context management is active" not in messages[0]["content"]:
+ messages[0]["content"] = f"{messages[0]['content']}\n\n{extension}"
+
+ def _inject_nudge(self, messages: list[dict[str, Any]], *, api_call_count: int) -> None:
+ prompt_tokens = estimate_messages_tokens_rough(messages)
+ 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
+ for msg in reversed(messages):
+ if msg.get("role") == "user" and isinstance(msg.get("content"), str):
+ msg["content"] = f"{msg['content']}\n\n{nudge}"
+ return
+ if messages and isinstance(messages[-1].get("content"), str):
+ messages[-1]["content"] = f"{messages[-1]['content']}\n\n{nudge}"
+
+ # -- Helpers ----------------------------------------------------------
+
+ def _message_key(self, msg: dict[str, Any], idx: int) -> str:
+ return f"{idx}:{self._message_signature(msg)}"
+
+ def _message_signature(self, msg: dict[str, Any]) -> str:
+ clean = {
+ "role": msg.get("role"),
+ "content": msg.get("content"),
+ "tool_call_id": msg.get("tool_call_id"),
+ "tool_calls": msg.get("tool_calls"),
+ }
+ raw = json.dumps(clean, sort_keys=True, default=str, separators=(",", ":"))
+ return hashlib.sha1(raw.encode("utf-8", "ignore")).hexdigest()
+
+ 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 _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
new file mode 100644
index 0000000000000..1f982ffdf50ba
--- /dev/null
+++ b/agent/dcp_state.py
@@ -0,0 +1,70 @@
+"""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/hermes_cli/config.py b/hermes_cli/config.py
index 2d11a868fc6fd..b606082736314 100644
--- a/hermes_cli/config.py
+++ b/hermes_cli/config.py
@@ -946,6 +946,55 @@ DEFAULT_CONFIG = {
# a plugin in plugins/context_engine// or ~/.hermes/plugins/.
"context": {
"engine": "compressor",
+ "dcp": {
+ "enabled": True,
+ "debug": False,
+ "pruneNotification": "detailed",
+ "pruneNotificationType": "chat",
+ "commands": {
+ "enabled": True,
+ "protectedTools": [],
+ },
+ "manualMode": {
+ "enabled": False,
+ "automaticStrategies": True,
+ },
+ "turnProtection": {
+ "enabled": False,
+ "turns": 4,
+ },
+ "experimental": {
+ "allowSubAgents": False,
+ "customPrompts": False,
+ },
+ "protectedFilePatterns": [],
+ "compress": {
+ "mode": "range",
+ "permission": "allow",
+ "showCompression": False,
+ "summaryBuffer": True,
+ "maxContextLimit": 100000,
+ "minContextLimit": 50000,
+ "modelMaxLimits": {},
+ "modelMinLimits": {},
+ "nudgeFrequency": 5,
+ "iterationNudgeThreshold": 15,
+ "nudgeForce": "soft",
+ "protectedTools": [],
+ "protectUserMessages": False,
+ },
+ "strategies": {
+ "deduplication": {
+ "enabled": True,
+ "protectedTools": [],
+ },
+ "purgeErrors": {
+ "enabled": True,
+ "turns": 4,
+ "protectedTools": [],
+ },
+ },
+ },
},
# Persistent memory -- bounded curated memory injected into system prompt
diff --git a/run_agent.py b/run_agent.py
index 919a5875b65ad..616c0a28ed5da 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -147,6 +147,7 @@ from agent.model_metadata import (
query_ollama_num_ctx,
)
from agent.context_compressor import ContextCompressor
+from agent.dcp_context_engine import DCPContextEngine
from agent.subdirectory_hints import SubdirectoryHintTracker
from agent.prompt_caching import apply_anthropic_cache_control
from agent.prompt_builder import build_skills_system_prompt, build_context_files_prompt, build_environment_hints, load_soul_md, TOOL_USE_ENFORCEMENT_GUIDANCE, TOOL_USE_ENFORCEMENT_MODELS, GOOGLE_MODEL_OPERATIONAL_GUIDANCE, OPENAI_MODEL_EXECUTION_GUIDANCE
@@ -2005,7 +2006,15 @@ class AIAgent:
except Exception:
pass
- if _engine_name != "compressor":
+ if _engine_name == "dcp":
+ _selected_engine = DCPContextEngine(
+ config=_ctx_cfg.get("dcp", {}) if isinstance(_ctx_cfg, dict) else {},
+ model=self.model,
+ provider=self.provider,
+ quiet_mode=self.quiet_mode,
+ )
+
+ if _engine_name != "compressor" and _selected_engine is None:
# Try loading from plugins/context_engine//
try:
from plugins.context_engine import load_context_engine
@@ -11149,6 +11158,24 @@ class AIAgent:
for idx, pfm in enumerate(self.prefill_messages):
api_messages.insert(sys_offset + idx, pfm.copy())
+ # Let the active context engine transform the provider-bound copy.
+ # DCP uses this to add refs, compression block placeholders, and
+ # nudges without mutating the canonical `messages` transcript.
+ if self.context_compressor is not None:
+ try:
+ api_messages = self.context_compressor.transform_api_messages(
+ api_messages,
+ canonical_messages=messages,
+ system_prompt=effective_system,
+ tools=self.tools,
+ api_call_count=api_call_count,
+ model=self.model,
+ provider=self.provider,
+ session_id=self.session_id,
+ )
+ except Exception as _ctx_transform_err:
+ logger.warning("Context engine API-message transform failed: %s", _ctx_transform_err)
+
# Apply Anthropic prompt caching for Claude models on native
# Anthropic, OpenRouter, and third-party Anthropic-compatible
# gateways. Auto-detected: if ``_use_prompt_caching`` is set,
diff --git a/tests/agent/test_dcp_config.py b/tests/agent/test_dcp_config.py
new file mode 100644
index 0000000000000..5157fe1b3a6e6
--- /dev/null
+++ b/tests/agent/test_dcp_config.py
@@ -0,0 +1,63 @@
+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
new file mode 100644
index 0000000000000..f8f1d72deb99d
--- /dev/null
+++ b/tests/agent/test_dcp_context_engine.py
@@ -0,0 +1,181 @@
+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_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_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)
+
+ 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)
+
+ 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)
+
+ 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)
+
+ assert messages[1]["content"] == ""
+ assert messages[2]["content"] == "old output"
diff --git a/tests/run_agent/test_plugin_context_engine_init.py b/tests/run_agent/test_plugin_context_engine_init.py
index 60e89889088ef..84407d868fa4e 100644
--- a/tests/run_agent/test_plugin_context_engine_init.py
+++ b/tests/run_agent/test_plugin_context_engine_init.py
@@ -7,6 +7,7 @@ context_length, causing the CLI status bar to show 'ctx --'.
from unittest.mock import MagicMock, patch
from agent.context_engine import ContextEngine
+from agent.dcp_context_engine import DCPContextEngine
class _StubEngine(ContextEngine):
@@ -88,4 +89,37 @@ def test_plugin_engine_update_model_args():
assert "model" in kw
assert "provider" in kw
# Should NOT pass api_mode — the ABC doesn't accept it
- assert "api_mode" not in kw
+ assert agent.context_compressor is engine
+
+
+def test_builtin_dcp_engine_selected_from_config():
+ cfg = {
+ "context": {
+ "engine": "dcp",
+ "dcp": {"compress": {"mode": "range", "permission": "allow"}},
+ },
+ "agent": {},
+ }
+
+ with (
+ patch("hermes_cli.config.load_config", return_value=cfg),
+ patch("agent.model_metadata.get_model_context_length", return_value=204_800),
+ patch("run_agent.get_tool_definitions", return_value=[]),
+ patch("run_agent.check_toolset_requirements", return_value={}),
+ patch("run_agent.OpenAI"),
+ ):
+ from run_agent import AIAgent
+
+ agent = AIAgent(
+ model="test/model",
+ api_key="test-key-1234567890",
+ base_url="https://openrouter.ai/api/v1",
+ quiet_mode=True,
+ skip_context_files=True,
+ skip_memory=True,
+ )
+
+ assert isinstance(agent.context_compressor, DCPContextEngine)
+ assert agent.context_compressor.context_length == 204_800
+ assert "compress" in agent._context_engine_tool_names
+ assert any(tool.get("function", {}).get("name") == "compress" for tool in agent.tools)