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