Merge pull request #81891 from kshitijk4poor/revert/dcp-context-engine
revert: remove DCP context engine
This commit is contained in:
kshitij
GitHub
commit
e65204b953
|
|
@ -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/<name>/
|
||||
try:
|
||||
from plugins.context_engine import load_context_engine
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -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'<dcp-ref id="{ref}" />'
|
||||
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<dcp-nudge>{nudge}</dcp-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'<dcp-compressed-block id="{block.ref}" topic="{block.topic}">\n'
|
||||
f"Summary: {block.summary}\n"
|
||||
f"Covers: {covers}\n"
|
||||
"</dcp-compressed-block>"
|
||||
)
|
||||
|
||||
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,
|
||||
)
|
||||
|
|
@ -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
|
||||
]
|
||||
|
|
@ -1665,32 +1665,6 @@ DEFAULT_CONFIG = {
|
|||
# a plugin in plugins/context_engine/<name>/ 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": {
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
@ -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 '<dcp-ref id="m0001" />' in transformed[1]["content"]
|
||||
assert '<dcp-ref id="m0002" />' 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 '<dcp-compressed-block id="b1" topic="old work">' 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": '<dcp-ref id="m0001" />'}
|
||||
|
||||
|
||||
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"},
|
||||
]
|
||||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Reference in New Issue