compat(plugins): temporary import-path shims for external plugins — ONE commit, revert on schedule

The Sep 2026 decomposition (PR #102117) makes internal import paths a non-API: names now live in
the focused modules that define them. This commit is the ONLY thing keeping the old paths alive,
so external plugins have time to update. It is deliberately a single, unsquashed commit:

    git revert <this sha>

removes every shim, stub and manifest at once on the announced date. Nothing in-tree may depend on
these pointers: scripts/check_compat_pointers.py (wired into lint.yml) fails CI if it does.

What it adds (see COMPAT_MANIFEST.md, compat_manifest.json):
- 332 facade modules get one delimited `PLUGIN-COMPAT` block appended at the end of the file
- 1,172 moved names resolved lazily via a module `__getattr__` (PEP 562) — never a top-level import,
  so no import cycles; facades that already had `__getattr__` get a chained one
- 592 third-party/stdlib names the old modules used to expose, with their original import statements
- 266 public definitions that had been deleted as unused, restored byte-for-byte from the pre-decomposition
  tree (+40 private helpers and 16 imports pulled in only because a restored definition needs them)
- 3 deleted modules recreated as re-export stubs (gateway/startup_watchdog, hermes_cli/observability/
  relay_runtime, tools/environments/modal_utils)
- private names (`_x`) get no pointer: they were never API (3,792 skipped)

Verified: all 335 touched modules import under a fresh HERMES_HOME and every manifest name resolves;
the lint reports zero in-tree uses; ruff clean; targeted suites unchanged.

.github/workflows/lint.yml

@@ -178,3 +178,9 @@ jobs:
 
       - name: Run footgun checker
         run: python scripts/check-windows-footguns.py --all
+
+      # The Sep 2026 decomposition kept old import paths alive for external plugins
+      # (PLUGIN-COMPAT blocks, see COMPAT_MANIFEST.md). They are removed on schedule by
+      # reverting one commit, so in-tree code must never depend on them.
+      - name: Forbid in-tree use of plugin-compat pointers
+        run: python scripts/check_compat_pointers.py

acp_adapter/auth.py

@@ -49,3 +49,14 @@ def build_auth_methods() -> list[Any]:
                      "Use this when Hermes has not been configured on this machine yet."),
     ))
     return methods
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def has_provider() -> bool:
+    """Return True if Hermes can resolve any runtime provider credentials."""
+    return detect_provider() is not None
+# ---- END PLUGIN-COMPAT ----

acp_adapter/edit_approval.py

@@ -226,3 +226,19 @@ def make_acp_edit_approval_requester(
         return getattr(outcome, "outcome", None) == "selected" and getattr(outcome, "option_id", None) == "allow_once"
 
     return _requester
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from concurrent.futures import TimeoutError as FutureTimeout  # noqa: F401,E402
+
+def clear_edit_approval_requester() -> None:
+    """Clear the current requester; primarily used by tests."""
+
+    _EDIT_APPROVAL_REQUESTER.set(None)
+
+def get_edit_approval_requester() -> EditApprovalRequester | None:
+    return _EDIT_APPROVAL_REQUESTER.get()
+# ---- END PLUGIN-COMPAT ----

acp_adapter/events.py

@@ -174,3 +174,11 @@ def make_step_cb(
                 tool_call_ids.pop(tool_name, None)
 
     return _step
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import json  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

acp_adapter/server.py

@@ -972,3 +972,38 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
         self.session_manager.save_session(session_id)
         logger.info("Session %s: config option %s updated", session_id, config_id)
         return SetSessionConfigOptionResponse(config_options=[])
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from acp.schema import AgentThoughtChunk  # noqa: F401,E402
+from acp.schema import AudioContentBlock  # noqa: F401,E402
+from acp.schema import AvailableCommand  # noqa: F401,E402
+from acp.schema import AvailableCommandsUpdate  # noqa: F401,E402
+from acp.schema import BlobResourceContents  # noqa: F401,E402
+from acp.schema import EmbeddedResourceContentBlock  # noqa: F401,E402
+from acp.schema import ImageContentBlock  # noqa: F401,E402
+from pathlib import Path  # noqa: F401,E402
+from acp.schema import ResourceContentBlock  # noqa: F401,E402
+from acp.schema import TextResourceContents  # noqa: F401,E402
+from acp.schema import UnstructuredCommandInput  # noqa: F401,E402
+import base64  # noqa: F401,E402
+import json  # noqa: F401,E402
+from urllib.parse import unquote  # noqa: F401,E402
+from urllib.parse import urlparse  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'ACP_MAX_MODELS_PER_PROVIDER': ('acp_adapter.model_catalog', 'ACP_MAX_MODELS_PER_PROVIDER'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

acp_adapter/session.py

@@ -422,3 +422,11 @@ class SessionManager:
         # ACP stdio: stdout is protocol-only JSON-RPC; agent chatter goes to stderr.
         agent._print_fn = _acp_stderr_print
         return agent
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from threading import Lock  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/agent_init.py

@@ -2304,3 +2304,23 @@ def init_agent(
 
 
 __all__ = ["init_agent"]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'ToolGuardrailDecision': ('agent.tool_guardrails', 'ToolGuardrailDecision'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/agent_runtime_helpers.py

@@ -3210,3 +3210,30 @@ __all__ = [
     "extract_api_error_context", "apply_pending_steer_to_tool_results", "_iter_pool_sockets",
     "force_close_tcp_sockets",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def agent_runtime_owns_post_tool_hook(agent: Any, function_name: str) -> bool:
+    """Return True when an agent-level tool path emits its own post hook."""
+    if function_name in AGENT_RUNTIME_POST_HOOK_TOOL_NAMES:
+        return True
+    if getattr(agent, "_context_engine_tool_names", None) and function_name in agent._context_engine_tool_names:
+        return True
+    memory_manager = getattr(agent, "_memory_manager", None)
+    return bool(memory_manager and memory_manager.has_tool(function_name))
+
+def intent_ack_continuation_enabled(agent) -> bool:
+    """Whether intent-ack continuation should fire at all for this turn.
+
+    The ``codex_ack_continuations < 2`` per-turn cap and the
+    ``looks_like_codex_intermediate_ack`` detector are applied by the caller;
+    this only decides the on/off gate. Callers that also need to know whether
+    the workspace requirement applies should use ``intent_ack_continuation_mode``
+    directly (``"codex_only"`` ⇒ require_workspace=True, ``"all"`` ⇒ False).
+    """
+    return intent_ack_continuation_mode(agent) != "off"
+# ---- END PLUGIN-COMPAT ----

agent/anthropic_adapter.py

@@ -673,3 +673,45 @@ def create_anthropic_message(
                 "%sAnthropic Messages stream unavailable; falling back to messages.create(): %s", log_prefix, exc
             )
     return messages_api.create(**{k: v for k, v in api_kwargs.items() if k != "stream"})
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from pathlib import Path  # noqa: F401,E402
+from typing import Tuple  # noqa: F401,E402
+import copy  # noqa: F401,E402
+import json  # noqa: F401,E402
+import os  # noqa: F401,E402
+import platform  # noqa: F401,E402
+import secrets  # noqa: F401,E402
+import stat  # noqa: F401,E402
+from urllib.parse import urlparse  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'CredentialPersistError': ('agent.anthropic_credentials', 'CredentialPersistError'),
+    'base_url_host_matches': ('utils', 'base_url_host_matches'),
+    'base_url_hostname': ('utils', 'base_url_hostname'),
+    'claude_code_credentials_path': ('agent.anthropic_credentials', 'claude_code_credentials_path'),
+    'get_hermes_home': ('hermes_constants', 'get_hermes_home'),
+    'is_claude_code_token_valid': ('agent.anthropic_credentials', 'is_claude_code_token_valid'),
+    'is_rotation_consumed_uncommitted': ('agent.anthropic_credentials', 'is_rotation_consumed_uncommitted'),
+    'mark_rotation_consumed_uncommitted': ('agent.anthropic_credentials', 'mark_rotation_consumed_uncommitted'),
+    'read_claude_code_credentials': ('agent.anthropic_credentials', 'read_claude_code_credentials'),
+    'read_hermes_oauth_credentials': ('agent.anthropic_credentials', 'read_hermes_oauth_credentials'),
+    'refresh_anthropic_oauth_pure': ('agent.anthropic_credentials', 'refresh_anthropic_oauth_pure'),
+    'resolve_anthropic_token': ('agent.anthropic_credentials', 'resolve_anthropic_token'),
+    'run_hermes_oauth_login_pure': ('agent.anthropic_credentials', 'run_hermes_oauth_login_pure'),
+    'run_oauth_setup_token': ('agent.anthropic_credentials', 'run_oauth_setup_token'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/aux_accounting.py

@@ -89,3 +89,14 @@ def record_aux_usage(
         )
     except Exception:
         logger.debug("Aux usage recording failed (non-fatal)", exc_info=True)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def get_accounting_context() -> Optional[tuple]:
+    """Return ``(session_db, session_id)`` for the active turn, or ``None``."""
+    return _accounting.get()
+# ---- END PLUGIN-COMPAT ----

agent/auxiliary_client.py

@@ -7319,3 +7319,32 @@ async def _async_call_llm_impl(
         if result is _RERAISE_ORIGINAL:
             raise
         return result
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from pathlib import Path  # noqa: F401,E402
+import copy  # noqa: F401,E402
+
+NOUS_EXTRA_BODY = _nous_extra_body()
+
+def get_async_text_auxiliary_client(task: str = "", *, main_runtime: Optional[Dict[str, Any]] = None):
+    """Return (async_client, model_slug) for async consumers.
+
+    For standard providers returns (AsyncOpenAI, model). For Codex returns
+    (AsyncCodexAuxiliaryClient, model) which wraps the Responses API.
+    Returns (None, None) when no provider is available.
+    """
+    provider, model, base_url, api_key, api_mode = _resolve_task_provider_model(task or None)
+    return resolve_provider_client(
+        provider,
+        model=model,
+        async_mode=True,
+        explicit_base_url=base_url,
+        explicit_api_key=api_key,
+        api_mode=api_mode,
+        main_runtime=main_runtime,
+    )
+# ---- END PLUGIN-COMPAT ----

agent/backend_identity.py

@@ -128,3 +128,24 @@ def should_skip_candidate(
     True when it is the same backend as ``failed`` along the axis ``scope`` invalidated.
     Every fallback/dedup/skip site must call this."""
     return _SCOPE_PREDICATES.get(scope, same_deployment)(candidate, failed)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+_REASON_SCOPES = {
+    "auth error": FailureScope.CREDENTIAL,
+    "payment error": FailureScope.CREDENTIAL,
+    "rate limit": FailureScope.MODEL,
+    "model incompatible with route": FailureScope.MODEL,
+    "invalid provider response": FailureScope.MODEL,
+    "connection error": FailureScope.MODEL,
+    "timeout": FailureScope.MODEL,
+}
+
+def classify_failure_scope(reason: Optional[str]) -> FailureScope:
+    """Map a human-readable failure reason to the identity axis it kills."""
+    return _REASON_SCOPES.get((reason or "").strip().lower(), FailureScope.MODEL)
+# ---- END PLUGIN-COMPAT ----

agent/background_review.py

@@ -1120,3 +1120,39 @@ __all__ = [
     "_MEMORY_REVIEW_PROMPT", "_SKILL_REVIEW_PROMPT", "_COMBINED_REVIEW_PROMPT", "load_background_review_settings",
     "spawn_background_review_thread", "summarize_background_review_actions", "build_memory_write_metadata",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from pathlib import Path  # noqa: F401,E402
+
+def is_background_review_enabled(
+    task_cfg: Optional[Dict[str, Any]] = None,
+) -> bool:
+    """Return whether automatic post-turn background review may spawn.
+
+    Controlled by ``auxiliary.background_review.enabled`` (default ``true``).
+    Explicit ``/refine`` (``focus`` set) bypasses this gate — same contract as
+    zeroing the nudge intervals, which stops automatic forks but leaves manual
+    refine working (issue #87250).
+
+    Prefer :func:`load_background_review_settings` at the spawn call site so
+    the task block is not re-read on the same turn.
+    """
+    if task_cfg is not None:
+        try:
+            from utils import is_truthy_value
+
+            return is_truthy_value(task_cfg.get("enabled"), default=True)
+        except Exception:
+            logger.warning(
+                "Failed to interpret background_review.enabled; leaving "
+                "automatic review enabled (fail-open)",
+                exc_info=True,
+            )
+            return True
+    enabled, _ = load_background_review_settings()
+    return enabled
+# ---- END PLUGIN-COMPAT ----

agent/bedrock_adapter.py

@@ -998,3 +998,109 @@ def get_bedrock_context_length(model_id: str, region: str = "", probe: bool = Tr
         return probed
     matches = [key for key in BEDROCK_CONTEXT_LENGTHS if key in model_id.lower()]
     return BEDROCK_CONTEXT_LENGTHS[max(matches, key=len)] if matches else BEDROCK_DEFAULT_CONTEXT_LENGTH
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+CONTEXT_OVERFLOW_PATTERNS = [
+    re.compile(r"ValidationException.*(?:input is too long|max input token|input token.*exceed)", re.IGNORECASE),
+    re.compile(r"ValidationException.*(?:exceeds? the (?:maximum|max) (?:number of )?(?:input )?tokens)", re.IGNORECASE),
+    re.compile(r"ModelStreamErrorException.*(?:Input is too long|too many input tokens)", re.IGNORECASE),
+]
+
+OVERLOAD_PATTERNS = [
+    re.compile(r"ModelNotReadyException", re.IGNORECASE),
+    re.compile(r"ModelTimeoutException", re.IGNORECASE),
+    re.compile(r"InternalServerException", re.IGNORECASE),
+]
+
+THROTTLE_PATTERNS = [
+    re.compile(r"ThrottlingException", re.IGNORECASE),
+    re.compile(r"Too many concurrent requests", re.IGNORECASE),
+    re.compile(r"ServiceQuotaExceededException", re.IGNORECASE),
+]
+
+def call_converse_stream(
+    region: str,
+    model: str,
+    messages: List[Dict],
+    tools: Optional[List[Dict]] = None,
+    max_tokens: Optional[int] = 4096,
+    temperature: Optional[float] = None,
+    top_p: Optional[float] = None,
+    stop_sequences: Optional[List[str]] = None,
+    guardrail_config: Optional[Dict] = None,
+) -> SimpleNamespace:
+    """Call Bedrock ConverseStream API and return an OpenAI-compatible response.
+
+    Consumes the full stream and returns the assembled response. For true
+    streaming with delta callbacks, use ``iter_converse_stream()`` instead.
+    """
+    client = _get_bedrock_runtime_client(region)
+    kwargs = build_converse_kwargs(
+        model=model,
+        messages=messages,
+        tools=tools,
+        max_tokens=max_tokens,
+        temperature=temperature,
+        top_p=top_p,
+        stop_sequences=stop_sequences,
+        guardrail_config=guardrail_config,
+    )
+
+    try:
+        response = client.converse_stream(**kwargs)
+    except Exception as exc:
+        retry_kwargs = recover_from_cache_point_rejection(exc, kwargs)
+        if retry_kwargs is not None:
+            return normalize_converse_stream_events(
+                client.converse_stream(**retry_kwargs)
+            )
+        if is_streaming_access_denied_error(exc):
+            # IAM allows bedrock:InvokeModel but not
+            # InvokeModelWithResponseStream — permanent for this session.
+            # Fall back to the non-streaming converse() path.
+            logger.info(
+                "bedrock: converse_stream denied by IAM on (region=%s, model=%s) — "
+                "falling back to non-streaming converse().",
+                region, model,
+            )
+            return normalize_converse_response(client.converse(**kwargs))
+        if is_stale_connection_error(exc):
+            logger.warning(
+                "bedrock: stale-connection error on converse_stream(region=%s, "
+                "model=%s): %s — evicting cached client so the next call reconnects.",
+                region, model, type(exc).__name__,
+            )
+            invalidate_runtime_client(region)
+        raise
+    return normalize_converse_stream_events(response)
+
+def is_context_overflow_error(error_message: str) -> bool:
+    """Return True if the error indicates the input context was too large.
+
+    When this returns True, the agent should compress context and retry
+    rather than treating it as a fatal error.
+    """
+    return any(p.search(error_message) for p in CONTEXT_OVERFLOW_PATTERNS)
+
+def classify_bedrock_error(error_message: str) -> str:
+    """Classify a Bedrock error for retry/failover decisions.
+
+    Returns:
+      - ``"context_overflow"`` — input too long, compress and retry
+      - ``"rate_limit"`` — throttled, backoff and retry
+      - ``"overloaded"`` — model temporarily unavailable, retry with delay
+      - ``"unknown"`` — unclassified error
+    """
+    if is_context_overflow_error(error_message):
+        return "context_overflow"
+    if any(p.search(error_message) for p in THROTTLE_PATTERNS):
+        return "rate_limit"
+    if any(p.search(error_message) for p in OVERLOAD_PATTERNS):
+        return "overloaded"
+    return "unknown"
+# ---- END PLUGIN-COMPAT ----

agent/bounded_response.py

@@ -77,3 +77,26 @@ def read_streaming_error_body(
             "bounded error-body read: capped at %d bytes (max=%d)", sum(len(c) for c in chunks), max_bytes,
         )
     return b"".join(chunks).decode("utf-8", errors="replace")
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Optional  # noqa: F401,E402
+
+def read_error_body_or_default(
+    response: httpx.Response,
+    *,
+    max_bytes: int = DEFAULT_ERROR_BODY_MAX_BYTES,
+    timeout_s: float = DEFAULT_ERROR_BODY_TIMEOUT_S,
+) -> Optional[str]:
+    """Like ``read_streaming_error_body`` but returns ``None`` on empty body.
+
+    Convenience for callers that distinguish "no body" from "empty string".
+    """
+    text = read_streaming_error_body(
+        response, max_bytes=max_bytes, timeout_s=timeout_s
+    )
+    return text or None
+# ---- END PLUGIN-COMPAT ----

agent/browser_provider.py

@@ -55,3 +55,12 @@ class BrowserProvider(ProviderBase):
     def emergency_cleanup(self, session_id: str) -> None:
         """Best-effort teardown from atexit / signal handlers. Must tolerate missing credentials and
         network errors; must not raise."""
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Any  # noqa: F401,E402
+from typing import Optional  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/browser_registry.py

@@ -59,3 +59,26 @@ def _resolve(configured: Optional[str]) -> Optional[BrowserProvider]:
         ):
             return provider
     return None
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+from typing import List  # noqa: F401,E402
+import threading  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'hermes_home_key': ('hermes_constants', 'hermes_home_key'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/codex_runtime.py

@@ -969,3 +969,21 @@ __all__ = [
     "run_codex_app_server_turn", "run_codex_stream",
     "_consume_codex_event_stream", "make_codex_app_server_event_bridge",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None):
+    """Backward-compatible alias for the unified event-driven path.
+
+    Historically this was the fallback when the SDK's high-level
+    ``responses.stream(...)`` helper raised on shape drift.  The primary
+    path now does exactly what the fallback did, so this just forwards.
+    Kept as a public symbol because tests and a small number of call sites
+    still reference it by name.
+    """
+    return run_codex_stream(agent, api_kwargs, client=client)
+# ---- END PLUGIN-COMPAT ----

agent/coding_context.py

@@ -554,3 +554,34 @@ def build_coding_workspace_block(cwd: Optional[str | Path] = None) -> str:
     if f.context_files:
         lines.append(f"- Context files: {', '.join(f.context_files)}")
     return "\n".join(lines)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def coding_system_blocks(
+    *,
+    platform: Optional[str] = None,
+    cwd: Optional[str | Path] = None,
+    config: Optional[dict[str, Any]] = None,
+    model: Optional[str] = None,
+) -> list[str]:
+    """Stable system-prompt blocks for the current posture (empty when general).
+
+    ``model`` steers the brief's edit-format nudge toward the model's family.
+    """
+    return resolve_runtime_mode(
+        platform=platform, cwd=cwd, config=config, model=model
+    ).system_blocks()
+
+_PROFILES: dict[str, ContextProfile] = {
+    GENERAL_PROFILE.name: GENERAL_PROFILE,
+    CODING_PROFILE.name: CODING_PROFILE,
+}
+
+def get_profile(name: str) -> ContextProfile:
+    """Return a registered profile, falling back to ``general``."""
+    return _PROFILES.get(name, GENERAL_PROFILE)
+# ---- END PLUGIN-COMPAT ----

agent/context_compressor.py

@@ -4894,3 +4894,23 @@ def reference_handoff_would_drive_next_model_call(messages: Optional[List[Dict[s
 def is_user_originated_turn(message: Any) -> bool:
     """True for human-authored user turns (not compaction scaffolding); dispatchers must use this, not a bare role check."""
     return user_originated_turn_view(message) is not None
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'tool_result_id_variants': ('agent.message_sanitization', 'tool_result_id_variants'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/conversation_compression.py

@@ -4013,3 +4013,13 @@ __all__ = [
     "check_compression_model_feasibility", "replay_compression_warning", "compress_context",
     "try_shrink_image_parts_in_messages",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+class CompressionExecutorSaturatedError(RuntimeError):
+    """All compression pool slots are occupied; submission was refused."""
+# ---- END PLUGIN-COMPAT ----

agent/conversation_loop.py

@@ -1536,3 +1536,63 @@ def run_conversation(
 
 
 __all__ = ["run_conversation"]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import os  # noqa: F401,E402
+import random  # noqa: F401,E402
+import ssl  # noqa: F401,E402
+import sys  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE': ('agent.conversation_compression', 'COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE'),
+    'COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE': ('agent.conversation_compression', 'COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE'),
+    'COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE': ('agent.conversation_compression', 'COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE'),
+    'COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE': ('agent.conversation_compression', 'COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE'),
+    'FailoverReason': ('agent.error_classifier', 'FailoverReason'),
+    'KawaiiSpinner': ('agent.display', 'KawaiiSpinner'),
+    'PARTIAL_STREAM_STUB_ID': ('hermes_constants', 'PARTIAL_STREAM_STUB_ID'),
+    'PRE_API_COMPRESSION_STATUS_TEMPLATE': ('agent.conversation_compression', 'PRE_API_COMPRESSION_STATUS_TEMPLATE'),
+    'adaptive_rate_limit_backoff': ('agent.retry_utils', 'adaptive_rate_limit_backoff'),
+    'anchored_context_tokens': ('agent.model_metadata', 'anchored_context_tokens'),
+    'automatic_compaction_status_message': ('agent.context_engine', 'automatic_compaction_status_message'),
+    'capture_usage_anchor': ('agent.model_metadata', 'capture_usage_anchor'),
+    'classify_api_error': ('agent.error_classifier', 'classify_api_error'),
+    'close_interrupted_tool_sequence': ('agent.message_sanitization', 'close_interrupted_tool_sequence'),
+    'coalesce_tool_call_id': ('agent.message_sanitization', 'coalesce_tool_call_id'),
+    'compose_user_api_content': ('agent.turn_context', 'compose_user_api_content'),
+    'compression_blocked_transiently': ('agent.conversation_compression', 'compression_blocked_transiently'),
+    'compression_skipped_due_to_lock': ('agent.conversation_compression', 'compression_skipped_due_to_lock'),
+    'context_compression_timed_out': ('agent.conversation_compression', 'context_compression_timed_out'),
+    'conversation_history_after_compression': ('agent.conversation_compression', 'conversation_history_after_compression'),
+    'env_var_enabled': ('utils', 'env_var_enabled'),
+    'estimate_messages_tokens_rough': ('agent.model_metadata', 'estimate_messages_tokens_rough'),
+    'estimate_request_tokens_rough': ('agent.model_metadata', 'estimate_request_tokens_rough'),
+    'estimate_usage_cost': ('agent.usage_pricing', 'estimate_usage_cost'),
+    'get_context_length_from_provider_error': ('agent.model_metadata', 'get_context_length_from_provider_error'),
+    'has_incomplete_scratchpad': ('agent.trajectory', 'has_incomplete_scratchpad'),
+    'is_output_cap_error': ('agent.model_metadata', 'is_output_cap_error'),
+    'is_repetition_dominated': ('agent.repetition_guard', 'is_repetition_dominated'),
+    'is_zai_coding_overload_error': ('agent.retry_utils', 'is_zai_coding_overload_error'),
+    'jittered_backoff': ('agent.retry_utils', 'jittered_backoff'),
+    'normalize_usage': ('agent.usage_pricing', 'normalize_usage'),
+    'parse_available_output_tokens_from_error': ('agent.model_metadata', 'parse_available_output_tokens_from_error'),
+    'reanchor_current_turn_user_idx': ('agent.turn_context', 'reanchor_current_turn_user_idx'),
+    'save_context_length': ('agent.model_metadata', 'save_context_length'),
+    'serialized_messages_bytes': ('agent.message_sanitization', 'serialized_messages_bytes'),
+    'splice_provider_projection': ('agent.provider_projection', 'splice_provider_projection'),
+    'zai_coding_overload_retry_ceiling': ('agent.retry_utils', 'zai_coding_overload_retry_ceiling'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/credential_sources.py

@@ -259,3 +259,23 @@ _REGISTRY: List[RemovalStep] = [
         description="Custom provider config.yaml api_key field",
     ),
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'register': ('tools.slash_confirm', 'register'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/display.py

@@ -1083,3 +1083,14 @@ def get_cute_tool_message(tool_name: str, args: dict, duration: float, result: s
         safe_name = tool_name[:9] if isinstance(tool_name, str) and tool_name else "tool"
         safe_duration = f"{duration:.1f}s" if isinstance(duration, (int, float)) else "done"
         return f"┊ ⚡ {safe_name:9} completed  {safe_duration}"
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def get_friendly_tool_labels() -> bool:
+    """Return whether friendly tool labels are enabled."""
+    return _friendly_tool_labels
+# ---- END PLUGIN-COMPAT ----

agent/error_surface.py

@@ -155,3 +155,12 @@ def build_error_surface_from_exception(exc: BaseException, provider: str = "", m
     except Exception:  # pragma: no cover — never break the error path
         logger.debug("error_surface: exception classification failed", exc_info=True)
         return None
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+LAYER_RUNTIME = "runtime"
+# ---- END PLUGIN-COMPAT ----

agent/estop.py

@@ -136,3 +136,11 @@ def check_paused(component: str, logger: logging.Logger) -> bool:
             component, suffix, sentinel_path(),
         )
     return True
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import os  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/file_safety.py

@@ -358,3 +358,86 @@ def get_container_mirror_warning(path: str, mirror_prefix: str | None = None) ->
         "the real HERMES_HOME.",
         "after explicit user direction, retry",
     )
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+PROFILE_SCOPED_AREAS = ("skills", "plugins", "cron", "memories")
+
+def classify_cross_profile_target(path: str) -> Optional[dict]:
+    """Classify a write target as cross-profile if it lands in another
+    profile's scoped area (skills/plugins/cron/memories).
+
+    Returns ``None`` when the target is outside Hermes scope, or is inside
+    the ACTIVE profile, or doesn't hit a profile-scoped area. Otherwise
+    returns a dict with:
+
+      * ``active_profile``: name of the profile the agent is running as
+      * ``target_profile``: name of the profile the path belongs to
+      * ``area``: which scoped area (``"skills"``, ``"plugins"``, etc.)
+      * ``target_path``: the resolved path string
+
+    The caller decides what to do with the result — surface a warning to
+    the model, prompt the user, or (with explicit consent /
+    ``cross_profile=True``) proceed anyway.
+    """
+    try:
+        target = Path(os.path.expanduser(str(path))).resolve()
+        root_real = _hermes_root_path().resolve()
+    except (OSError, RuntimeError):
+        return None
+
+    target_profile: Optional[str] = None
+    area: Optional[str] = None
+
+    try:
+        rel = target.relative_to(root_real)
+    except ValueError:
+        return None
+
+    parts = rel.parts
+    if not parts:
+        return None
+
+    if parts[0] in PROFILE_SCOPED_AREAS:
+        # ``<root>/<area>/...`` → default profile.
+        target_profile = "default"
+        area = parts[0]
+    elif (
+        parts[0] == "profiles"
+        and len(parts) >= 3
+        and parts[2] in PROFILE_SCOPED_AREAS
+    ):
+        # ``<root>/profiles/<name>/<area>/...`` → named profile.
+        target_profile = parts[1]
+        area = parts[2]
+    else:
+        return None
+
+    active_profile = _resolve_active_profile_name()
+    if target_profile == active_profile:
+        # In-profile write — not a cross-profile event.
+        return None
+
+    return {
+        "active_profile": active_profile,
+        "target_profile": target_profile,
+        "area": area,
+        "target_path": str(target),
+    }
+
+def get_cross_profile_warning(path: str) -> Optional[str]:
+    """RETIRED (maintainer decision): always returns ``None``.
+
+    The cross-profile write guard was removed — profiles were never
+    isolated (same OS user; the terminal tool writes anywhere), so the
+    block was ceremony that cost every schema real tokens and taught a
+    bypass arg. The system prompt's active-profile hint remains the only
+    steering; the classifier below survives for that hint and for
+    diagnostics. Kept as a stub so external callers/plugins fail soft.
+    """
+    return None
+# ---- END PLUGIN-COMPAT ----

agent/image_gen_provider.py

@@ -129,3 +129,13 @@ def error_response(
         "success": False, "image": None, "error": error, "error_type": error_type,
         "model": model, "prompt": prompt, "aspect_ratio": aspect_ratio, "provider": provider,
     }
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import base64  # noqa: F401,E402
+import datetime  # noqa: F401,E402
+import uuid  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/image_gen_registry.py

@@ -45,3 +45,26 @@ def get_active_provider() -> Optional[ImageGenProvider]:
         return available[0]
     fal = snapshot.get("fal")
     return fal if fal is not None and _available(fal) else None
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+from typing import List  # noqa: F401,E402
+import threading  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'hermes_home_key': ('hermes_constants', 'hermes_home_key'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/learning_graph_render.py

@@ -417,3 +417,25 @@ def render_frames(payload: dict[str, Any], *, cols: int = 80, rows: int = 16, fr
         "buckets": _bucket_rows(buckets, payload), "summary": build_summary(payload), "axis": axis_labels(payload),
         "count": len(payload.get("nodes", [])), "cols": cols, "rows": rows,
     }
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+Grid = list  # list[Row]
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'Run': ('hermes_cli.kanban_db', 'Run'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/lsp/eventlog.py

@@ -128,3 +128,29 @@ __all__ = [
     "log_server_unavailable", "log_timeout", "log_server_error", "log_spawn_failed", "log_reaped",
     "reset_announce_caches",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+_announced_no_server: set = set()     # keys: (server_id,)
+
+def _announce_once(bucket: set, key: Tuple) -> bool:
+    """Return True if *key* has not been announced for *bucket* yet.
+
+    Atomically marks the key as announced so concurrent callers
+    cannot both win the race and double-log.
+    """
+    with _announce_lock:
+        if key in bucket:
+            return False
+        bucket.add(key)
+        return True
+
+def log_no_server_configured(server_id: str) -> None:
+    """No spawn recipe for this language.  WARNING once."""
+    if _announce_once(_announced_no_server, (server_id,)):
+        _emit(server_id, logging.WARNING, "no server configured")
+# ---- END PLUGIN-COMPAT ----

agent/lsp/protocol.py

@@ -136,3 +136,12 @@ __all__ = [
     "encode_message", "read_message", "make_request", "make_notification", "make_response",
     "make_error_response", "classify_message",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+ERROR_REQUEST_CANCELLED = -32800
+# ---- END PLUGIN-COMPAT ----

agent/monitoring/cron_health.py

@@ -168,3 +168,24 @@ __all__ = [
     "CronHealthSnapshot", "build_cron_health_snapshot", "classify_cron_error", "emit_execution_state",
     "project_execution_event",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import hashlib  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'GatewayHealthSnapshot': ('agent.monitoring.gateway_health', 'GatewayHealthSnapshot'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/monitoring/emitter.py

@@ -165,3 +165,12 @@ def reset_emitter_for_tests(emitter: Optional[MonitoringEmitter] = None) -> None
 
 
 __all__ = ["MonitoringEmitter", "get_emitter", "emit", "reset_emitter_for_tests"]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+TelemetryEmitter = MonitoringEmitter
+# ---- END PLUGIN-COMPAT ----

agent/monitoring/gateway_health.py

@@ -329,3 +329,24 @@ __all__ = [
     "GatewayMetric", "GatewayHealthSnapshot", "GatewayDiagnosticLogHandler",
     "build_gateway_health_snapshot", "classify_gateway_error", "source_logger_for_export",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def redact_gateway_message(message: Any) -> str:
+    """Redact gateway diagnostic free text for operator-owned export.
+
+    Single scrub path: everything goes through
+    ``agent.monitoring.redaction.redact_for_export`` (unconditional
+    secrets + PII), then is length-bounded.
+    """
+    try:
+        from agent.monitoring.redaction import redact_for_export
+        redacted = redact_for_export(str(message or "")) or ""
+    except Exception:
+        redacted = "[redaction-unavailable]"
+    return redacted[:500]
+# ---- END PLUGIN-COMPAT ----

agent/monitoring/gateway_health_export.py

@@ -334,3 +334,11 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu
 
 
 __all__ = ["GatewayHealthExportRuntime", "start_gateway_health_export"]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import re  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/nous_rate_guard.py

@@ -162,3 +162,24 @@ def _has_exhausted_bucket_in_object(state: Any) -> bool:
         if _is_exhausted(remaining, reset):
             return True
     return False
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import tempfile  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'atomic_replace': ('utils', 'atomic_replace'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/outbound_webhooks.py

@@ -335,3 +335,13 @@ def _deliver(delivery: Dict[str, Any]) -> None:
         "outbound webhook delivery failed after %d attempt(s) (event=%s target=%s): %s",
         MAX_DELIVERY_ATTEMPTS, event, label, last_error,
     )
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from pathlib import Path  # noqa: F401,E402
+from datetime import datetime  # noqa: F401,E402
+from datetime import timezone  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/pet/generate/atlas.py

@@ -619,3 +619,24 @@ def _check_atlas_cells(atlas) -> tuple[list[str], list[str], list[str]]:
     if residue:
         errors.append(f"{residue} transparent pixels retain RGB residue")
     return errors, warnings, filled_states
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import io  # noqa: F401,E402
+import io  # noqa: F401,E402
+
+COLUMNS = max(count for _, _, count in ROW_SPECS)
+
+FRAME_COUNTS: dict[str, int] = {state: count for state, _, count in ROW_SPECS}
+
+ROWS = len(ROW_SPECS)
+
+def atlas_to_webp_bytes(atlas) -> bytes:
+    """Encode an atlas image to lossless WebP bytes (the on-disk pet format)."""
+    buf = io.BytesIO()
+    atlas.save(buf, format="WEBP", lossless=True, quality=100, method=6, exact=True)
+    return buf.getvalue()
+# ---- END PLUGIN-COMPAT ----

agent/prompt_builder.py

@@ -1586,3 +1586,24 @@ def build_context_files_prompt(
         return ""
     return ("# Project Context\n\nThe following project context files have been loaded and should be followed:\n\n"
             + "\n".join(sections))
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import List  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'org_id_of_path': ('agent.skill_utils', 'org_id_of_path'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/prompt_cache_boundary.py

@@ -56,3 +56,15 @@ def find_stable_prefix(content: str) -> Optional[str]:
         if best is not None:
             _prefixes.move_to_end(best)  # after the scan: never mutate mid-iteration
         return best
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def clear_stable_prefixes() -> None:
+    """Test isolation helper."""
+    with _lock:
+        _prefixes.clear()
+# ---- END PLUGIN-COMPAT ----

agent/proxy_sources/iron_proxy.py

@@ -1036,3 +1036,11 @@ __all__ = [
     "merge_mappings", "mint_proxy_token", "reload_proxy", "start_proxy", "stop_proxy",
     "write_mappings", "write_proxy_config",
 ]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import stat  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/reasoning_effort.py

@@ -133,3 +133,12 @@ def requested_effort(reasoning_config: Optional[dict]) -> Optional[str]:
     if not isinstance(reasoning_config, dict) or reasoning_config.get("enabled") is False:
         return None
     return str(reasoning_config.get("effort") or "").strip().lower() or None
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+CODEX_RESPONSES_EFFORTS: tuple[str, ...] = CODEX_GPT56_EFFORTS
+# ---- END PLUGIN-COMPAT ----

agent/relay_llm.py

@@ -880,3 +880,11 @@ def _run_awaitable(
     if _has_running_event_loop():
         raise RuntimeError(loop_error)
     return asyncio.run(value)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from dataclasses import dataclass  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/relay_runtime.py

@@ -1248,3 +1248,91 @@ def _reset_for_tests() -> None:
     HOST_REGISTRY.shutdown_all()
     _PLUGIN_CONFIGURATION.reset_for_tests()
     _PROFILE_KEY_CACHE.clear()
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from enum import auto  # noqa: F401,E402
+
+def emit_mark(
+    name: str,
+    *,
+    session_id: str,
+    data: Any = None,
+    metadata: Any = None,
+) -> bool:
+    """Emit a fail-open Relay mark under a Hermes session."""
+    runtime = get_runtime(create=False)
+    if runtime is None:
+        return False
+    try:
+        return runtime.emit_mark(
+            name,
+            {"session_id": session_id},
+            data=data,
+            metadata=metadata,
+        )
+    except Exception:
+        logger.warning("Hermes Relay mark failed: %s", name, exc_info=True)
+        return False
+
+def ensure_session(*, session_id: str, **context: Any) -> RelaySession | None:
+    """Create or return the shared Relay session used by Hermes core."""
+    runtime = get_runtime()
+    if runtime is None:
+        return None
+    try:
+        return runtime.ensure_session({"session_id": session_id, **context})
+    except Exception:
+        logger.warning("Hermes Relay session initialization failed", exc_info=True)
+        return None
+
+def get_host(
+    *,
+    create: bool = True,
+    profile_key: str | None = None,
+) -> RelayHost | None:
+    """Return the explicit real or reduced-capability host for a profile."""
+    return HOST_REGISTRY.for_profile(profile_key, create=create)
+
+def get_session_handle(session_id: str) -> Any:
+    """Return the shared Relay handle for direct core instrumentation."""
+    runtime = get_runtime(create=False)
+    return None if runtime is None else runtime.get_session_handle(session_id)
+
+def run_in_session(
+    session_id: str,
+    callback: Callable[..., Any],
+    *args: Any,
+    **kwargs: Any,
+) -> Any:
+    """Run a scope, LLM, or tool API against a shared Hermes session."""
+    runtime = get_runtime()
+    if runtime is None:
+        raise RuntimeError("Hermes Relay runtime is unavailable")
+    session = runtime.get_session(session_id)
+    if session is None:
+        session = runtime.ensure_session({"session_id": session_id})
+    if session is None:
+        raise RuntimeError("Hermes Relay session is unavailable")
+    return runtime.run_in_session(session, callback, *args, **kwargs)
+
+async def run_in_session_async(
+    session_id: str,
+    callback: Callable[..., Any],
+    *args: Any,
+    **kwargs: Any,
+) -> Any:
+    """Await a Relay operation inside a shared Hermes session context."""
+    runtime = get_runtime()
+    if runtime is None:
+        raise RuntimeError("Hermes Relay runtime is unavailable")
+    session = runtime.get_session(session_id)
+    if session is None:
+        session = runtime.ensure_session({"session_id": session_id})
+    if session is None:
+        raise RuntimeError("Hermes Relay session is unavailable")
+    return await runtime.run_in_session_async(session, callback, *args, **kwargs)
+# ---- END PLUGIN-COMPAT ----

agent/relay_tools.py

@@ -104,3 +104,12 @@ def _run_awaitable(value: Any) -> Any:
     return relay_llm._run_awaitable(
         value, loop_error="Synchronous Hermes Relay tool execution cannot run on an active event-loop thread",
     )
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import asyncio  # noqa: F401,E402
+import inspect  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/review_idle_queue.py

@@ -201,3 +201,11 @@ def _managed_server_idle() -> bool:
 
 # Module singleton — one queue per process, like the load-progress watcher.
 QUEUE = ReviewIdleQueue()
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import List  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/secret_sources/_cache.py

@@ -186,3 +186,24 @@ class SecretCache(Generic[K]):
     def clear(self, home_path: Optional[Path] = None) -> None:
         self.memory.clear()
         self.disk.clear(home_path)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'FetchResult': ('agent.secret_sources.base', 'FetchResult'),
+    'is_valid_env_name': ('agent.secret_sources.base', 'is_valid_env_name'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/secret_sources/bitwarden.py

@@ -526,3 +526,112 @@ def clear_caches(home_path: Optional[Path] = None) -> None:
 
 
 _reset_cache_for_tests = clear_caches
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import stat  # noqa: F401,E402
+
+def apply_bitwarden_secrets(
+    *,
+    enabled: bool,
+    access_token_env: str = "BWS_ACCESS_TOKEN",
+    project_id: str = "",
+    override_existing: bool = False,
+    cache_ttl_seconds: float = 300,
+    auto_install: bool = True,
+    server_url: str = "",
+    home_path: Optional[Path] = None,
+    encrypted_cache_enabled: bool = False,
+    encrypted_cache_max_stale_seconds: float = 0,
+) -> FetchResult:
+    """Pull secrets from BSM and set them on ``os.environ``.
+
+    This is the function ``load_hermes_dotenv()`` calls after the .env
+    files have loaded.  It is intentionally defensive — any failure
+    returns a :class:`FetchResult` with ``error`` set; it never raises.
+
+    ``server_url`` selects the Bitwarden region or self-hosted endpoint
+    (e.g. ``https://vault.bitwarden.eu`` for EU Cloud).  Empty string
+    means use ``bws``'s default (US Cloud).
+
+    Parameters mirror the ``secrets.bitwarden.*`` config keys so the
+    caller can just splat the dict in.
+    """
+    result = FetchResult()
+
+    if not enabled:
+        return result
+
+    access_token = os.environ.get(access_token_env, "").strip()
+    if not access_token:
+        result.error = (
+            f"secrets.bitwarden.enabled is true but {access_token_env} is "
+            "not set.  Run `hermes secrets bitwarden setup`."
+        )
+        return result
+
+    if not project_id:
+        result.error = (
+            "secrets.bitwarden.project_id is empty.  "
+            "Run `hermes secrets bitwarden setup`."
+        )
+        return result
+
+    binary = find_bws(install_if_missing=auto_install)
+    result.binary_path = binary
+    if binary is None:
+        result.error = (
+            "bws binary not available and auto-install is disabled.  "
+            "Run `hermes secrets bitwarden setup` to install."
+        )
+        return result
+
+    try:
+        secrets, warnings = fetch_bitwarden_secrets(
+            access_token=access_token,
+            project_id=project_id,
+            binary=binary,
+            cache_ttl_seconds=cache_ttl_seconds,
+            server_url=server_url,
+            home_path=home_path,
+            encrypted_cache_enabled=encrypted_cache_enabled,
+            encrypted_cache_max_stale_seconds=encrypted_cache_max_stale_seconds,
+        )
+    except RuntimeError as exc:
+        result.error = str(exc)
+        return result
+
+    result.secrets = secrets
+    result.warnings.extend(warnings)
+
+    for key, value in secrets.items():
+        if key == access_token_env:
+            # Don't let BSM clobber the very token we used to fetch
+            # itself — that would be a footgun if someone stored the
+            # token as a BSM secret too.
+            result.skipped.append(key)
+            continue
+        if not override_existing and os.environ.get(key):
+            result.skipped.append(key)
+            continue
+        os.environ[key] = value
+        result.applied.append(key)
+
+    return result
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'DiskCache': ('agent.secret_sources._cache', 'DiskCache'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/secret_sources/command.py

@@ -190,3 +190,192 @@ class CommandSource(SecretSource):
             return result
         result.secrets = secrets
         return result
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def apply_command_secrets(
+    *,
+    command: str,
+    override_existing: bool = False,
+    timeout_seconds: float = _COMMAND_TIMEOUT_SECONDS,
+    max_output_bytes: int = _MAX_OUTPUT_BYTES,
+    home_path: Optional[Path] = None,
+) -> FetchResult:
+    """Run the helper once at startup and set its KEY=VALUE output on
+    ``os.environ``.
+
+    LEGACY shim retained for API symmetry with ``apply_bitwarden_secrets``;
+    the startup path goes through :class:`CommandSource` + the registry
+    orchestrator instead (which owns precedence and the environ writes).
+    """
+    result = FetchResult()
+
+    command = (command or "").strip()
+    if not command:
+        result.error = (
+            "secrets.command.enabled is true but secrets.command.command is "
+            "empty.  Set the helper command in config.yaml."
+        )
+        return result
+
+    if _is_windows():
+        result.warnings.append(
+            "the 'command' secret source is POSIX-only (needs /bin/sh); "
+            "skipping on Windows"
+        )
+        return result
+
+    # The list/enumerate path: run the helper exactly ONCE with an empty
+    # HERMES_SECRET_KEY and parse its stdout as a dotenv blob.
+    stdout = _run_helper(command, "", timeout_seconds, max_output_bytes)
+    if stdout is None:
+        # _run_helper already logged structured fields to stderr.
+        result.warnings.append(
+            "helper command failed at startup; no secrets applied "
+            "(process env / .env values remain in effect)"
+        )
+        return result
+
+    secrets = _parse_dotenv_map(stdout)
+    result.secrets = secrets
+    if not secrets:
+        result.warnings.append(
+            "helper output was not a KEY=VALUE map; nothing applied at "
+            "startup (a bare-value helper still resolves single keys on demand)"
+        )
+        return result
+
+    for key, value in secrets.items():
+        if value.strip() == "":
+            # Whitespace-only placeholder entries are "no value" — applying
+            # them would flow into an Authorization header → guaranteed 401.
+            result.skipped.append(key)
+            continue
+        if not override_existing and os.environ.get(key):
+            # Process env / .env win — same precedence as bitwarden.
+            result.skipped.append(key)
+            continue
+        os.environ[key] = value
+        result.applied.append(key)
+
+    return result
+
+def parse_secret_output(stdout: str, wanted_key: str) -> Optional[str]:
+    """Parse a secret-fetch helper's stdout.  Supports BOTH shapes:
+
+    * a bare value (single secret): the whole trimmed stdout is the value.
+    * a dotenv blob (KEY=VALUE lines): parse them and return the entry for
+      ``wanted_key``.
+
+    Mirrors the TS ``parseSecretOutput`` exactly, including the cross-key
+    misroute guard and the base64-padding disambiguation.
+    """
+    text = stdout.replace("\r\n", "\n")
+    lines = text.split("\n")
+
+    # 1. Exact dotenv match wins: scan for a `wanted_key=...` line.  This
+    #    is deterministic and never returns another key's value.
+    dotenv_lines = [
+        line
+        for line in (raw.strip() for raw in lines)
+        if line and not line.startswith("#") and _ENV_LINE.match(line)
+    ]
+    for line in dotenv_lines:
+        m = _ENV_LINE.match(line)
+        assert m is not None  # filtered above
+        if m.group(1) == wanted_key:
+            value = unquote_dotenv_value(m.group(2))
+            # Whitespace-only (e.g. a quoted `K="  "` placeholder) is "no
+            # value": it would otherwise flow into an Authorization header
+            # → guaranteed 401.
+            return value if value.strip() != "" else None
+
+    # 2. The output is a multi-key dotenv dump that does NOT contain the
+    #    wanted key → None, rather than mis-returning an unrelated line as
+    #    a bare value.  Only >=2 env-shaped lines count as a dump: a SINGLE
+    #    non-matching env-shaped line falls through to the bare-value
+    #    branch, because a bare secret can itself match the KEY=VALUE shape
+    #    (e.g. base64 with '=' padding, "dGVzdA==") and must not be
+    #    misclassified as a dump.
+    if len(dotenv_lines) > 1:
+        return None
+
+    # 3. Otherwise treat the whole output as a single bare value (a per-key
+    #    helper that printed just the secret).  Trim first so whitespace-only
+    #    output (a ' '/'\t' placeholder entry) resolves to None, never a "key".
+    value = text.strip()
+    if value == "":
+        return None
+
+    # SECURITY (S2): a single env-shaped line for a DIFFERENT key must not
+    # be returned as the wanted secret.  A sloppy helper (e.g. `head -1
+    # env-file`, or a grep that matched the wrong line) emitting
+    # `OTHER_KEY=realvalue` would otherwise flow — key name, '=' and the
+    # OTHER key's value — into an Authorization header sent to the WANTED
+    # key's endpoint: cross-provider credential leakage, not just a 401.
+    # Disambiguation from a bare base64 secret: base64 padding only ever
+    # produces an env-shaped line whose "value" part is empty or all '='
+    # (`dGVzdA==` → key `dGVzdA`, value `=`), so a non-trivial value part
+    # after a non-matching key means a misrouted dotenv entry → None.
+    env_shaped = _ENV_LINE.match(value)
+    if (
+        env_shaped
+        and env_shaped.group(1) != wanted_key
+        and re.fullmatch(r"=*", env_shaped.group(2).strip()) is None
+    ):
+        return None
+    return value
+
+def get_command_secret(
+    *,
+    command: str,
+    key: str,
+    timeout_seconds: float = _COMMAND_TIMEOUT_SECONDS,
+    max_output_bytes: int = _MAX_OUTPUT_BYTES,
+) -> Optional[str]:
+    """Resolve a single secret by running the helper with the key in
+    ``HERMES_SECRET_KEY``.  Returns None on any failure — never raises."""
+    command = (command or "").strip()
+    if not command:
+        return None
+    stdout = _run_helper(command, key, timeout_seconds, max_output_bytes)
+    if stdout is None:
+        return None
+    return parse_secret_output(stdout, key)
+
+def list_command_secrets(
+    *,
+    command: str,
+    timeout_seconds: float = _COMMAND_TIMEOUT_SECONDS,
+    max_output_bytes: int = _MAX_OUTPUT_BYTES,
+) -> Dict[str, str]:
+    """Enumerate secrets by running the helper ONCE with an empty key.
+
+    Returns the dotenv map ONLY when the helper emits a KEY=VALUE blob;
+    a bare-value helper returns ``{}``.  Never raises.
+    """
+    command = (command or "").strip()
+    if not command:
+        return {}
+    stdout = _run_helper(command, "", timeout_seconds, max_output_bytes)
+    if stdout is None:
+        return {}
+    return _parse_dotenv_map(stdout)
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'get_source_environment': ('agent.secret_sources.base', 'get_source_environment'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/secret_sources/onepassword.py

@@ -335,3 +335,24 @@ def clear_caches(home_path: Optional[Path] = None) -> None:
 
 
 _reset_cache_for_tests = clear_caches
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import hashlib  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'DiskCache': ('agent.secret_sources._cache', 'DiskCache'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/shell_hooks.py

@@ -590,3 +590,11 @@ def run_once(spec: ShellHookSpec, kwargs: Dict[str, Any]) -> Dict[str, Any]:
     result = _spawn(spec, _serialize_payload(spec.event, kwargs))
     result["parsed"] = _evaluate_result(spec, result)
     return result
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import shlex  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/skill_bundles.py

@@ -204,3 +204,11 @@ def delete_bundle(name: str) -> Path:
 def get_bundle(name: str) -> Optional[Dict[str, Any]]:
     """Look up a bundle by name (slug-normalized)."""
     return get_skill_bundles().get(f"/{_slugify(name)}")
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import re  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/skill_utils.py

@@ -767,3 +767,20 @@ def parse_qualified_name(name: str) -> Tuple[Optional[str], str]:
 def is_valid_namespace(candidate: Optional[str]) -> bool:
     """Check whether *candidate* is a valid namespace (``[a-zA-Z0-9_-]+``)."""
     return bool(candidate) and bool(_NAMESPACE_RE.match(candidate))
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def get_scan_ordered_skills_dirs() -> List[Path]:
+    """All skill dirs in precedence order: project → local → external.
+
+    First-wins name deduplication over this order gives project skills
+    priority over profile-local and external ones.
+    """
+    dirs = list(get_project_skills_dirs())
+    dirs.extend(get_all_skills_dirs())
+    return dirs
+# ---- END PLUGIN-COMPAT ----

agent/ssl_guard.py

@@ -60,3 +60,19 @@ def verify_ca_bundle() -> None:
     except Exception as exc:
         raise _ssl_err(f"certifi is not importable: {exc}") from exc
     _validate_bundle_path("certifi", str(certifi.where()), require_substantial=True)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def verify_ca_bundle_with_fallback() -> None:
+    """Backward-compatible wrapper for older call sites.
+
+    The old PR name mentioned a platform fallback, but allowing startup with a
+    broken certifi bundle still leaves httpx/OpenAI and requests call sites
+    failing later. Keep the wrapper name but enforce the same check.
+    """
+    verify_ca_bundle()
+# ---- END PLUGIN-COMPAT ----

agent/system_prompt.py

@@ -709,3 +709,23 @@ def format_tools_for_system_message(agent: Any) -> str:
 
 __all__ = ["build_system_prompt_parts", "build_system_prompt", "invalidate_system_prompt",
            "restore_plugin_prompt_sections", "format_tools_for_system_message"]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'OPENAI_MODEL_EXECUTION_GUIDANCE': ('agent.prompt_builder', 'OPENAI_MODEL_EXECUTION_GUIDANCE'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/terminal_env_registry.py

@@ -81,3 +81,25 @@ def plugin_strip_env_keys() -> frozenset:
         except Exception:
             logger.debug("Terminal environment provider strip_env_keys raised", exc_info=True)
     return frozenset(keys)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+import threading  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'hermes_home_key': ('hermes_constants', 'hermes_home_key'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/transcript_repair.py

@@ -84,3 +84,11 @@ def sync_flushed_message_markers(batch_msgs: List[Dict[str, Any]], batch_rows: L
             written["_row_id"] = row["_row_id"]
         if "_canonical_content" in row:
             written["content"] = row["_canonical_content"]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Optional  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/transcription_provider.py

@@ -34,3 +34,25 @@ class TranscriptionProvider(CatalogProviderBase):
         ``prompt`` (from ``stt.prompt`` or a ``pre_transcription`` hook) as a vocabulary hint;
         unknown keys must be ignored.
         """
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import List  # noqa: F401,E402
+import logging  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'logger': ('agent.i18n', 'logger'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/transcription_registry.py

@@ -43,3 +43,27 @@ _registry: ProviderRegistry[TranscriptionProvider] = ProviderRegistry(
     on_builtin_collision=_warn_builtin_collision,
 )
 _registry.export(globals())
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+from typing import List  # noqa: F401,E402
+from typing import Optional  # noqa: F401,E402
+import threading  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'hermes_home_key': ('hermes_constants', 'hermes_home_key'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/transports/chat_completions.py

@@ -565,3 +565,11 @@ class ChatCompletionsTransport(ProviderTransport):
 from agent.transports import register_transport  # noqa: E402
 
 register_transport("chat_completions", ChatCompletionsTransport)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/transports/codex.py

@@ -674,3 +674,13 @@ class ResponsesApiTransport(ProviderTransport):
 from agent.transports import register_transport  # noqa: E402
 
 register_transport("codex_responses", ResponsesApiTransport)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+from typing import List  # noqa: F401,E402
+from typing import Tuple  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/transports/codex_app_server.py

@@ -276,3 +276,12 @@ def check_codex_binary(
     if version < min_version:
         return False, f"codex {have} is older than required {'.'.join(map(str, min_version))}. Run: npm i -g @openai/codex"
     return True, have
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from dataclasses import field  # noqa: F401,E402
+import time  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/tts_registry.py

@@ -43,3 +43,27 @@ _registry: ProviderRegistry[TTSProvider] = ProviderRegistry(
     builtin_names=_BUILTIN_NAMES, on_builtin_collision=_warn_builtin_collision,
 )
 _registry.export(globals())
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+from typing import List  # noqa: F401,E402
+from typing import Optional  # noqa: F401,E402
+import threading  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'hermes_home_key': ('hermes_constants', 'hermes_home_key'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/turn_context.py

@@ -1001,3 +1001,27 @@ def build_api_messages(
     if effective_system:
         api_messages = [{"role": "system", "content": effective_system}] + api_messages
     return api_messages, effective_system
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'IDLE_COMPACTION_STATUS_TEMPLATE': ('agent.conversation_compression', 'IDLE_COMPACTION_STATUS_TEMPLATE'),
+    'PREFLIGHT_COMPRESSION_STATUS_TEMPLATE': ('agent.conversation_compression', 'PREFLIGHT_COMPRESSION_STATUS_TEMPLATE'),
+    'automatic_compaction_status_message': ('agent.context_engine', 'automatic_compaction_status_message'),
+    'compression_skipped_due_to_lock': ('agent.conversation_compression', 'compression_skipped_due_to_lock'),
+    'conversation_history_after_compression': ('agent.conversation_compression', 'conversation_history_after_compression'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/turn_retry_state.py

@@ -50,3 +50,11 @@ class TurnRetryState:
     # A user correction cancelled the in-flight request: append a role-safe checkpoint +
     # user message, rebuild the payload, and retry the same logical iteration.
     restart_with_redirected_messages: bool = False
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from dataclasses import fields  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/usage_pricing.py

@@ -618,3 +618,12 @@ def format_token_count_compact(value: int) -> str:
     if "." in text:
         text = text.rstrip("0").rstrip(".")
     return f"{sign}{text}{suffix}"
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+DEFAULT_PRICING = {"input": 0.0, "output": 0.0}
+# ---- END PLUGIN-COMPAT ----

agent/video_gen_provider.py

@@ -262,3 +262,13 @@ class OpenAICompatibleVideoGenProvider(VideoGenProvider):
             close = getattr(client, "close", None)
             if callable(close):
                 close()
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import base64  # noqa: F401,E402
+import datetime  # noqa: F401,E402
+import uuid  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/video_gen_registry.py

@@ -44,3 +44,26 @@ def get_active_provider() -> Optional[VideoGenProvider]:
         if is_available_safe(p, logger, "video_gen provider %s.is_available() raised %s")
     ]
     return available[0] if len(available) == 1 else None
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+from typing import List  # noqa: F401,E402
+import threading  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'hermes_home_key': ('hermes_constants', 'hermes_home_key'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

agent/web_search_provider.py

@@ -78,3 +78,11 @@ class WebSearchProvider(ProviderBase):
         raise NotImplementedError(
             f"{self.name} does not support extract (override supports_extract)"
         )
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Optional  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

agent/web_search_registry.py

@@ -199,3 +199,26 @@ def get_active_search_provider() -> Optional[WebSearchProvider]:
 def get_active_extract_provider() -> Optional[WebSearchProvider]:
     """Resolve the currently-active web extract provider."""
     return _resolve(_configured_backend("extract"), capability="extract")
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+from typing import List  # noqa: F401,E402
+import threading  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'hermes_home_key': ('hermes_constants', 'hermes_home_key'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

cli.py

@@ -4574,3 +4574,69 @@ if __name__ == "__main__":
     import fire
 
     fire.Fire(main)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from prompt_toolkit.layout.menus import CompletionsMenu  # noqa: F401,E402
+from prompt_toolkit.filters import Condition  # noqa: F401,E402
+from prompt_toolkit.layout import ConditionalContainer  # noqa: F401,E402
+from prompt_toolkit.layout.processors import ConditionalProcessor  # noqa: F401,E402
+from prompt_toolkit.layout.dimension import Dimension  # noqa: F401,E402
+from prompt_toolkit.history import FileHistory  # noqa: F401,E402
+from prompt_toolkit.layout import FormattedTextControl  # noqa: F401,E402
+from prompt_toolkit.layout import HSplit  # noqa: F401,E402
+from prompt_toolkit.key_binding import KeyBindings  # noqa: F401,E402
+from prompt_toolkit.layout import Layout  # noqa: F401,E402
+from prompt_toolkit.styles import Style as PTStyle  # noqa: F401,E402
+from rich.panel import Panel  # noqa: F401,E402
+from prompt_toolkit.layout.processors import PasswordProcessor  # noqa: F401,E402
+from prompt_toolkit.layout.processors import Processor  # noqa: F401,E402
+from prompt_toolkit.widgets import TextArea  # noqa: F401,E402
+from prompt_toolkit.layout.processors import Transformation  # noqa: F401,E402
+from prompt_toolkit.layout import Window  # noqa: F401,E402
+from prompt_toolkit.layout import WindowAlign  # noqa: F401,E402
+import base64  # noqa: F401,E402
+import concurrent.futures  # noqa: F401,E402
+import copy  # noqa: F401,E402
+from rich import box as rich_box  # noqa: F401,E402
+import tempfile  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'AIAgent': ('run_agent', 'AIAgent'),
+    'CanonicalUsage': ('agent.usage_pricing', 'CanonicalUsage'),
+    'DEFAULT_BROWSER_CDP_URL': ('hermes_cli.browser_connect', 'DEFAULT_BROWSER_CDP_URL'),
+    'HERMES_AGENT_LOGO': ('hermes_cli.banner', 'HERMES_AGENT_LOGO'),
+    'HERMES_CADUCEUS': ('hermes_cli.banner', 'HERMES_CADUCEUS'),
+    'SlashCommandAutoSuggest': ('hermes_cli.commands_completion', 'SlashCommandAutoSuggest'),
+    'SlashCommandCompleter': ('hermes_cli.commands_completion', 'SlashCommandCompleter'),
+    'build_welcome_banner': ('hermes_cli.banner', 'build_welcome_banner'),
+    'display_hermes_home': ('hermes_constants', 'display_hermes_home'),
+    'estimate_usage_cost': ('agent.usage_pricing', 'estimate_usage_cost'),
+    'get_all_toolsets': ('toolsets', 'get_all_toolsets'),
+    'get_job': ('cron.jobs', 'get_job'),
+    'get_toolset_for_tool': ('model_tools', 'get_toolset_for_tool'),
+    'get_toolset_info': ('toolsets', 'get_toolset_info'),
+    'init_skin_from_config': ('hermes_cli.skin_engine', 'init_skin_from_config'),
+    'is_browser_debug_ready': ('hermes_cli.browser_connect', 'is_browser_debug_ready'),
+    'is_table_divider': ('agent.markdown_tables', 'is_table_divider'),
+    'looks_like_table_row': ('agent.markdown_tables', 'looks_like_table_row'),
+    'manual_chrome_debug_command': ('hermes_cli.browser_connect', 'manual_chrome_debug_command'),
+    'print_config_warnings': ('hermes_cli.config', 'print_config_warnings'),
+    'prompt_for_secret': ('hermes_cli.callbacks', 'prompt_for_secret'),
+    'set_friendly_tool_labels': ('agent.display', 'set_friendly_tool_labels'),
+    'set_tool_preview_max_len': ('agent.display', 'set_tool_preview_max_len'),
+    'setup_logging': ('hermes_logging', 'setup_logging'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

cron/jobs.py

@@ -3123,3 +3123,14 @@ def rewrite_skill_refs(
             save_jobs(jobs)
             logger.info("Curator rewrote skill references in %d cron job(s)", len(rewrites))
         return {"rewrites": rewrites, "jobs_updated": len(rewrites), "jobs_scanned": len(jobs)}
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def clear_drift_alerted(job_id: str) -> None:
+    """Clear the drift alert-dedup marker (resolution matches again)."""
+    _set_alert_flag(job_id, "drift_alerted", False)
+# ---- END PLUGIN-COMPAT ----

cron/scheduler.py

@@ -3865,3 +3865,29 @@ if __name__ == "__main__":
             0 if _run_external_worker_payload(args.external_worker_file, args.ack_file) else 1
         )
     tick(verbose=True)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import asyncio  # noqa: F401,E402
+import shutil  # noqa: F401,E402
+import signal  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'BOT_CHAT_PLATFORM': ('cron.scheduler_delivery', 'BOT_CHAT_PLATFORM'),
+    'SharedRouteAdapters': ('cron.scheduler_preflight', 'SharedRouteAdapters'),
+    'cron_delivery_targets': ('cron.scheduler_delivery', 'cron_delivery_targets'),
+    'parse_bot_chat_deliver_token': ('cron.scheduler_delivery', 'parse_bot_chat_deliver_token'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

cron/scheduler_provider.py

@@ -543,3 +543,26 @@ class InProcessCronScheduler(CronScheduler):
             if ok:
                 consecutive_failures = 0
             stop_event.wait(_backoff_wait_seconds(interval, consecutive_failures))
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def provider_supports_fire_cancel(provider: Any) -> bool:
+    """Return whether ``fire_claimed`` accepts a ``cancel_event`` kwarg."""
+    try:
+        parameters = inspect.signature(provider.fire_claimed).parameters.values()
+    except (TypeError, ValueError):
+        return False
+    return any(
+        parameter.kind is inspect.Parameter.VAR_KEYWORD
+        or (
+            parameter.name == "cancel_event"
+            and parameter.kind
+            in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
+        )
+        for parameter in parameters
+    )
+# ---- END PLUGIN-COMPAT ----

gateway/browser_control_artifacts.py

@@ -321,3 +321,13 @@ class ArtifactRateLimiter:
         """Drop the recorded hits for ``key`` (tests/diagnostics)."""
         with self._lock:
             self._hits.pop(key, None)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+class ArtifactOverwrite(ArtifactError):
+    """An artifact id already exists and the store refuses to overwrite it."""
+# ---- END PLUGIN-COMPAT ----

gateway/browser_control_broker.py

@@ -510,3 +510,16 @@ _GLOBAL_BROKER = BrowserControlBroker()
 def get_browser_control_broker() -> BrowserControlBroker:
     """Process-local broker shared by API and dashboard Gateway transports."""
     return _GLOBAL_BROKER
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+BROWSER_CONTROL_ALL_CAPABILITIES = frozenset(
+    BROWSER_CONTROL_CAPABILITIES
+    | BROWSER_CONTROL_ARTIFACT_CAPABILITIES
+    | BROWSER_CONTROL_DEVELOPER_CAPABILITIES
+)
+# ---- END PLUGIN-COMPAT ----

gateway/config.py

@@ -863,3 +863,11 @@ def _apply_env_overrides(config: GatewayConfig) -> None:
     """Apply environment variable overrides to config (see ``gateway.config_env``)."""
     from gateway.config_env import _apply_env_overrides as _impl
     _impl(config)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import json  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

gateway/delivery_ledger.py

@@ -352,3 +352,33 @@ def ledger_enabled(config: Optional[Dict[str, Any]] = None) -> bool:
         return value.strip().lower() not in {"false", "0", "no", "off"} if isinstance(value, str) else bool(value)
     except Exception:
         return True
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import json  # noqa: F401,E402
+import json  # noqa: F401,E402
+
+def debug_rows(limit: int = 20) -> str:
+    """Human-readable dump for ad-hoc inspection (sqlite3-free path)."""
+    with _DB_LOCK, _transaction() as conn:
+        rows = conn.execute(
+            """SELECT obligation_id, session_key, state, attempts,
+                      created_at, updated_at, last_error
+               FROM delivery_obligations
+               ORDER BY updated_at DESC LIMIT ?""",
+            (limit,),
+        ).fetchall()
+    return json.dumps(
+        [
+            {
+                "id": r[0], "session": r[1], "state": r[2], "attempts": r[3],
+                "created_at": r[4], "updated_at": r[5], "last_error": r[6],
+            }
+            for r in rows
+        ],
+        indent=2,
+    )
+# ---- END PLUGIN-COMPAT ----

gateway/disk_status.py

@@ -58,3 +58,24 @@ def collect_disk_status(home: Optional[Path] = None) -> Dict[str, Any]:
     status.update(total_mb=total_mb, free_mb=free_mb, used_percent=round((usage.used / usage.total) * 100, 1),
                   pressure=classify_disk_pressure(free_mb, total_mb))
     return status
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import logging  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'logger': ('gateway.run', 'logger'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

gateway/hosted_room_discussion.py

@@ -782,3 +782,11 @@ def plan_publication(
         payload={**_turn_coordinates(task), "seen_through_seq": task.seen_through_seq, **extra},
         authority_gateway_id=room.gateway_id, authority_epoch=room.authority_epoch))
     return PublicationPlan(task_id=task.identity.task_id, terminal_kind=terminal_kind, events=tuple(effects))
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import json  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

gateway/hosted_room_driver.py

@@ -857,3 +857,14 @@ def prune_published_terminal_tasks(
             f"DELETE FROM hosted_room_driver_tasks WHERE room_id=? AND task_id IN ({','.join('?' * len(candidates))})",
             (room_id, *candidates))
         return max(0, int(deleted.rowcount))
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Iterator  # noqa: F401,E402
+from pathlib import Path  # noqa: F401,E402
+from contextlib import contextmanager  # noqa: F401,E402
+import re  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

gateway/hosted_room_execution_policy.py

@@ -110,3 +110,12 @@ __all__ = [
     "MAX_POLICY_ITERATIONS", "POLICY_VERSION", "RoomExecutionPolicy", "RoomExecutionPolicyError",
     "bind_room_execution_policy", "current_room_execution_policy", "execution_policy_mapping",
     "reset_room_execution_policy"]
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import json  # noqa: F401,E402
+import re  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

gateway/hosted_room_peer.py

@@ -488,3 +488,57 @@ def room_grant_needs_dispatch_refresh(token: str, *, now: float | None = None, l
         return clock(now) + max(0.0, float(leeway_seconds)) >= expires_at
     except Exception:
         return True
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import time  # noqa: F401,E402
+
+@dataclass(frozen=True)
+class RoomLinkProbe:
+    """One gateway-verified route candidate."""
+
+    mode: LinkMode
+    verified: bool
+    encrypted: bool
+    latency_ms: float
+
+_LINK_PRIORITY = {
+    "direct": 0,
+    "overlay": 1,
+    "relay": 2,
+    "pull": 3,
+    "desktop": 4,
+}
+
+def select_room_link(
+    probes: Iterable[RoomLinkProbe],
+    *,
+    desktop_available: bool,
+) -> RoomLinkProbe | None:
+    """Choose the fastest safe route without weakening encryption."""
+    candidates = [
+        probe
+        for probe in probes
+        if probe.verified
+        and probe.encrypted
+        and probe.mode != "desktop"
+        and math.isfinite(probe.latency_ms)
+        and probe.latency_ms >= 0
+    ]
+    if candidates:
+        return min(
+            candidates,
+            key=lambda item: (_LINK_PRIORITY[item.mode], item.latency_ms),
+        )
+    if desktop_available:
+        return RoomLinkProbe(
+            mode="desktop",
+            verified=True,
+            encrypted=True,
+            latency_ms=0,
+        )
+    return None
+# ---- END PLUGIN-COMPAT ----

gateway/hosted_room_replicas.py

@@ -284,3 +284,26 @@ def demote_room(
     return {
         "room_id": room_id, "authority_gateway_id": observed_gateway_id, "authority_epoch": observed_epoch,
         "idempotent": False}
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from pathlib import Path  # noqa: F401,E402
+import time  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'MAX_EVENT_JSON_BYTES': ('gateway.hosted_rooms', 'MAX_EVENT_JSON_BYTES'),
+    'MAX_ROOM_ID_CHARS': ('gateway.hosted_rooms', 'MAX_ROOM_ID_CHARS'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

gateway/hosted_rooms.py

@@ -1158,3 +1158,14 @@ def read_events(
         if not fits(events):
             raise HostedRoomError("hosted room event exceeds replay page limit")
     return build_page(events)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Iterator  # noqa: F401,E402
+from typing import NoReturn  # noqa: F401,E402
+from contextlib import contextmanager  # noqa: F401,E402
+import time  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

gateway/kanban_watchers.py

@@ -304,3 +304,28 @@ class GatewayKanbanWatchersMixin:
             await self._sleep_between_ticks(interval)
 
         self._release_kanban_dispatcher_lock()
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Callable  # noqa: F401,E402
+from contextvars import Context  # noqa: F401,E402
+import logging  # noqa: F401,E402
+import re  # noqa: F401,E402
+import sqlite3  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    't': ('agent.i18n', 't'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

gateway/memory_status.py

@@ -110,3 +110,24 @@ def collect_memory_status(
         status["boot_id"] = started_at if isinstance(started_at, str) and started_at else None
 
     return status
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import logging  # noqa: F401,E402
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'logger': ('gateway.run', 'logger'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/helpers.py

@@ -511,3 +511,98 @@ def _chunk_newline_preferred(text, limit, len_fn):
     if remaining:
         chunks.append(remaining)
     return chunks
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Dict  # noqa: F401,E402
+from typing import TYPE_CHECKING  # noqa: F401,E402
+import asyncio  # noqa: F401,E402
+import asyncio  # noqa: F401,E402
+
+class TextBatchAggregator:
+    """Aggregates rapid-fire text events into single messages.
+
+    Replaces the ``_enqueue_text_event`` / ``_flush_text_batch`` pattern
+    previously duplicated in telegram, discord, matrix, wecom, and feishu.
+
+    Usage::
+
+        self._text_batcher = TextBatchAggregator(
+            handler=self._message_handler,
+            batch_delay=0.6,
+            split_threshold=1900,
+        )
+
+        # In message dispatch:
+        if msg_type == MessageType.TEXT and self._text_batcher.is_enabled():
+            self._text_batcher.enqueue(event, session_key)
+            return
+    """
+
+    def __init__(
+        self,
+        handler,
+        *,
+        batch_delay: float = 0.6,
+        split_delay: float = 2.0,
+        split_threshold: int = 4000,
+    ):
+        self._handler = handler
+        self._batch_delay = batch_delay
+        self._split_delay = split_delay
+        self._split_threshold = split_threshold
+        self._pending: Dict[str, "MessageEvent"] = {}
+        self._pending_tasks: Dict[str, asyncio.Task] = {}
+
+    def is_enabled(self) -> bool:
+        """Return True if batching is active (delay > 0)."""
+        return self._batch_delay > 0
+
+    def enqueue(self, event: "MessageEvent", key: str) -> None:
+        """Add *event* to the pending batch for *key*."""
+        chunk_len = len(event.text or "")
+        existing = self._pending.get(key)
+        if not existing:
+            event._last_chunk_len = chunk_len  # type: ignore[attr-defined]
+            self._pending[key] = event
+        else:
+            existing.text = f"{existing.text}\n{event.text}"
+            existing._last_chunk_len = chunk_len  # type: ignore[attr-defined]
+
+        # Cancel prior flush timer, start a new one
+        prior = self._pending_tasks.get(key)
+        if prior and not prior.done():
+            prior.cancel()
+        self._pending_tasks[key] = asyncio.create_task(self._flush(key))
+
+    async def _flush(self, key: str) -> None:
+        """Wait then dispatch the batched event for *key*."""
+        current_task = self._pending_tasks.get(key)
+        pending = self._pending.get(key)
+        last_len = getattr(pending, "_last_chunk_len", 0) if pending else 0
+
+        # Use longer delay when the last chunk looks like a split message
+        delay = self._split_delay if last_len >= self._split_threshold else self._batch_delay
+        await asyncio.sleep(delay)
+
+        event = self._pending.pop(key, None)
+        if event:
+            try:
+                await self._handler(event)
+            except Exception:
+                logger.exception("[TextBatchAggregator] Error dispatching batched event for %s", key)
+
+        if self._pending_tasks.get(key) is current_task:
+            self._pending_tasks.pop(key, None)
+
+    def cancel_all(self) -> None:
+        """Cancel all pending flush tasks."""
+        for task in self._pending_tasks.values():
+            if not task.done():
+                task.cancel()
+        self._pending_tasks.clear()
+        self._pending.clear()
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/qqbot/adapter.py

@@ -1695,3 +1695,12 @@ class QQAdapter(BasePlatformAdapter):
             return True
         self._seen_messages[msg_id] = now
         return False
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import base64  # noqa: F401,E402
+import mimetypes  # noqa: F401,E402
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/qqbot/chunked_upload.py

@@ -290,3 +290,14 @@ def _compute_file_hashes(file_path: str, file_size: int) -> Dict[str, str]:
             bytes_read += len(chunk)
     full_md5 = md5.hexdigest()
     return {"md5": full_md5, "sha1": sha1.hexdigest(), "md5_10m": md5_10m.hexdigest() if need_10m else full_md5}
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+from typing import Optional  # noqa: F401,E402
+import functools  # noqa: F401,E402
+
+ApiRequestFn = Callable[..., Awaitable[Dict[str, Any]]]
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/qqbot/keyboards.py

@@ -189,3 +189,97 @@ def parse_interaction_event(raw: Dict[str, Any]) -> InteractionEvent:
         channel_id=str(raw.get("channel_id", "")), guild_id=str(raw.get("guild_id", "")),
         button_data=str(resolved.get("button_data", "")), button_id=str(resolved.get("button_id", "")),
         resolver_user_id=str(resolved.get("user_id", "")))
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import logging  # noqa: F401,E402
+from typing import Awaitable  # noqa: F401,E402
+from typing import Callable  # noqa: F401,E402
+from typing import Awaitable  # noqa: F401,E402
+from typing import Callable  # noqa: F401,E402
+import logging  # noqa: F401,E402
+
+logger = logging.getLogger(__name__)
+
+PostMessageFn = Callable[..., Awaitable[Dict[str, Any]]]
+
+class ApprovalSender:
+    """Send an approval-request message with an inline keyboard.
+
+    Decoupled from the adapter via callables so it can be unit-tested in
+    isolation. Pass the adapter's ``_send_message_with_keyboard`` helper
+    (or any equivalent) as ``post_message``.
+    """
+
+    def __init__(
+        self,
+        post_c2c: PostMessageFn,
+        post_group: PostMessageFn,
+        log_tag: str = "QQBot",
+    ) -> None:
+        self._post_c2c = post_c2c
+        self._post_group = post_group
+        self._log_tag = log_tag
+
+    async def send(
+        self,
+        chat_type: str,
+        chat_id: str,
+        req: ApprovalRequest,
+        msg_id: Optional[str] = None,
+    ) -> bool:
+        """Send an approval message to *chat_id*.
+
+        :param chat_type: ``'c2c'`` or ``'group'``.
+        :param chat_id: User openid or group openid.
+        :param req: :class:`ApprovalRequest`.
+        :param msg_id: Reply-to message id (required for passive messages).
+        :returns: ``True`` on success, ``False`` on failure.
+        """
+        text = build_approval_text(req)
+        keyboard = build_approval_keyboard(req.session_key)
+
+        logger.info(
+            "[%s] Sending approval request to %s:%s (session=%.20s…)",
+            self._log_tag, chat_type, chat_id, req.session_key,
+        )
+
+        try:
+            if chat_type == "c2c":
+                await self._post_c2c(chat_id, text, msg_id, keyboard)
+            elif chat_type == "group":
+                await self._post_group(chat_id, text, msg_id, keyboard)
+            else:
+                logger.warning(
+                    "[%s] Approval: unsupported chat_type %r",
+                    self._log_tag, chat_type,
+                )
+                return False
+            logger.info(
+                "[%s] Approval message sent to %s:%s",
+                self._log_tag, chat_type, chat_id,
+            )
+            return True
+        except Exception as exc:
+            logger.error(
+                "[%s] Failed to send approval message to %s:%s: %s",
+                self._log_tag, chat_type, chat_id, exc,
+            )
+            return False
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'logger': ('gateway.platforms.qqbot.adapter', 'logger'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/signal.py

@@ -974,3 +974,25 @@ class SignalAdapter(BasePlatformAdapter):
         result = await self._rpc("getContact", {"account": self.account, "contactAddress": chat_id})
         name = (result.get("name") or result.get("profileName")) if isinstance(result, dict) else None
         return {"name": name or chat_id, "type": "dm", "chat_id": chat_id}
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+TYPING_INTERVAL = 8.0  # seconds between typing indicator refreshes
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'DEFAULT_EXT_TO_MIME': ('gateway.platforms.media_cache', 'DEFAULT_EXT_TO_MIME'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/weixin.py

@@ -1265,3 +1265,13 @@ async def send_weixin_direct(
         adapter._send_session = adapter._session = session
         adapter._token_store = token_store
         return await _deliver_direct(adapter, chat_id, message, media_files, context_token)
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import struct  # noqa: F401,E402
+
+MSG_TYPE_USER = 1
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/yuanbao.py

@@ -2802,3 +2802,205 @@ class YuanbaoAdapter(BasePlatformAdapter):
         """Current valid sign token (module-level cache)."""
         return await SignManager.get_token(self._app_key, self._app_secret, self._api_domain, route_env=self._route_env)
 
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+AUTH_FAILED_CODES = {4001, 4002, 4003}      # permanent auth failure, re-sign token
+
+AUTH_RETRYABLE_CODES = {4010, 4011, 4099}   # transient, can retry with same token
+
+class FileUrlHandler(MediaSendHandler):
+    """Strategy: send file from a URL (download → COS → TIMFileElem)."""
+
+    async def acquire_file(self, adapter, **kwargs):
+        file_url: str = kwargs["file_url"]
+        logger.info("[%s] FileUrlHandler: downloading %s", adapter.name, file_url)
+        file_bytes, content_type = await media_download_url(
+            file_url, max_size_mb=adapter.MEDIA_MAX_SIZE_MB,
+        )
+        filename = kwargs.get("filename")
+        if not filename:
+            path_part = file_url.split("?")[0]
+            filename = os.path.basename(path_part) or "file"
+        if not content_type or content_type == "application/octet-stream":
+            content_type = guess_mime_type(filename) or "application/octet-stream"
+        return file_bytes, filename, content_type
+
+    def build_msg_body(self, upload_result, **kwargs):
+        return build_file_msg_body(
+            url=upload_result["url"],
+            filename=kwargs["filename"],
+            uuid=kwargs["file_uuid"],
+            size=upload_result["size"],
+        )
+
+class GroupQueryService:
+    """Encapsulates all group query operations (both low-level WS calls and
+    higher-level AI-tool-facing wrappers).
+
+    Responsibilities:
+      - Low-level WS encode/decode for group info and member list queries
+      - Chat-id parsing, error wrapping and result filtering for AI tools
+      - Member cache population on the adapter
+    """
+
+    def __init__(self, adapter: "YuanbaoAdapter") -> None:
+        self._adapter = adapter
+
+    # ------------------------------------------------------------------
+    # Low-level WS query methods
+    # ------------------------------------------------------------------
+
+    async def query_group_info_raw(self, group_code: str) -> Optional[dict]:
+        """Query group info via WS (group name, owner, member count, etc.).
+
+        Returns:
+            Decoded dict or None on failure.
+        """
+        adapter = self._adapter
+        if adapter._connection.ws is None:
+            return None
+        encoded = encode_query_group_info(group_code)
+        from gateway.platforms.yuanbao_proto import decode_conn_msg as _decode
+        decoded = _decode(encoded)
+        req_id = decoded["head"]["msg_id"]
+        try:
+            response = await adapter._connection.send_biz_request(encoded, req_id=req_id)
+            head = response.get("head", {})
+            status = head.get("status", 0)
+            if status != 0:
+                logger.warning("[%s] query_group_info failed: status=%d", adapter.name, status)
+                return None
+            biz_data = response.get("data", b"") or response.get("body", b"")
+            if biz_data and isinstance(biz_data, bytes):
+                return decode_query_group_info_rsp(biz_data)
+            return {"group_code": group_code}
+        except asyncio.TimeoutError:
+            logger.warning("[%s] query_group_info timeout: group=%s", adapter.name, group_code)
+            return None
+        except Exception as exc:
+            logger.warning("[%s] query_group_info failed: %s", adapter.name, exc)
+            return None
+
+    async def get_group_member_list_raw(
+        self, group_code: str, offset: int = 0, limit: int = 200
+    ) -> Optional[dict]:
+        """Query group member list via WS.
+
+        Returns:
+            Decoded dict or None on failure.  Also populates adapter._member_cache.
+        """
+        adapter = self._adapter
+        if adapter._connection.ws is None:
+            return None
+        encoded = encode_get_group_member_list(group_code, offset=offset, limit=limit)
+        from gateway.platforms.yuanbao_proto import decode_conn_msg as _decode
+        decoded = _decode(encoded)
+        req_id = decoded["head"]["msg_id"]
+        try:
+            response = await adapter._connection.send_biz_request(encoded, req_id=req_id)
+            head = response.get("head", {})
+            status = head.get("status", 0)
+            if status != 0:
+                logger.warning("[%s] get_group_member_list failed: status=%d", adapter.name, status)
+                return None
+            biz_data = response.get("data", b"") or response.get("body", b"")
+            if biz_data and isinstance(biz_data, bytes):
+                result = decode_get_group_member_list_rsp(biz_data)
+            else:
+                result = {"members": [], "next_offset": 0, "is_complete": True}
+            if result and result.get("members"):
+                adapter._member_cache[group_code] = (time.time(), result["members"])
+            return result
+        except asyncio.TimeoutError:
+            logger.warning("[%s] get_group_member_list timeout: group=%s", adapter.name, group_code)
+            return None
+        except Exception as exc:
+            logger.warning("[%s] get_group_member_list failed: %s", adapter.name, exc)
+            return None
+
+    # ------------------------------------------------------------------
+    # AI-tool-facing wrappers (chat_id parsing + filtering)
+    # ------------------------------------------------------------------
+
+    async def query_group_info(self, chat_id: str) -> dict:
+        """AI tool: Query current group info.
+
+        No parameters needed (group_code extracted from session context).
+        Returns group name, owner, member count, etc.
+        """
+        if not chat_id.startswith("group:"):
+            return {"error": "This command is only available in group chats"}
+        group_code = chat_id[len("group:"):]
+        result = await self.query_group_info_raw(group_code)
+        if result is None:
+            return {"error": "Failed to query group info"}
+        return result
+
+    async def query_session_members(
+        self,
+        chat_id: str,
+        action: str = "list_all",
+        name: Optional[str] = None,
+    ) -> dict:
+        """AI tool: Query group member list.
+
+        Args:
+            chat_id: Chat ID (extracted from session context)
+            action: 'find' (search by name) | 'list_bots' (list bots) | 'list_all' (list all)
+            name: Search keyword when action='find'
+
+        Returns:
+            {"members": [...], "total": int, "mentionHint": str}
+        """
+        if not chat_id.startswith("group:"):
+            return {"error": "This command is only available in group chats"}
+        group_code = chat_id[len("group:"):]
+        result = await self.get_group_member_list_raw(group_code)
+        if result is None:
+            return {"error": "Failed to query group members"}
+
+        members = result.get("members", [])
+
+        if action == "find" and name:
+            query = name.lower()
+            members = [
+                m for m in members
+                if query in (m.get("nickname", "") or "").lower()
+                or query in (m.get("name_card", "") or "").lower()
+                or query in (m.get("user_id", "") or "").lower()
+            ]
+        elif action == "list_bots":
+            members = [m for m in members if "bot" in (m.get("nickname", "") or "").lower()]
+
+        # Construct mentionHint
+        mention_hint = ""
+        if members and len(members) <= 10:
+            names = [m.get("name_card") or m.get("nickname") or m.get("user_id", "") for m in members]
+            mention_hint = "Mention with @name: " + ", ".join(names)
+
+        return {
+            "members": members[:50],  # Limit return count
+            "total": len(members),
+            "mentionHint": mention_hint,
+        }
+
+REPLY_REF_TTL_S = 300.0            # Reference dedup TTL (5 minutes)
+
+def get_active_adapter() -> Optional["YuanbaoAdapter"]:
+    """Delegate to ``YuanbaoAdapter.get_active()``."""
+    return YuanbaoAdapter.get_active()
+
+async def send_yuanbao_direct(
+    adapter: "YuanbaoAdapter",
+    chat_id: str,
+    message: str,
+    media_files: Optional[List[Tuple[str, bool]]] = None,
+) -> Dict[str, Any]:
+    """Delegate to ``OutboundManager.send_direct``."""
+    return await adapter._outbound.send_direct(chat_id, message, media_files)
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/yuanbao_media.py

@@ -317,3 +317,12 @@ def _basename_from_url(url: str) -> str:
         return os.path.basename(urllib.parse.urlparse(url).path)
     except Exception:
         return ""
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+COS_USE_ACCELERATE = True
+# ---- END PLUGIN-COMPAT ----

gateway/platforms/yuanbao_proto.py

@@ -538,3 +538,79 @@ def decode_get_group_member_list_rsp(data: bytes) -> Optional[dict]:
         }
     except Exception:
         return None
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+import logging  # noqa: F401,E402
+
+DEBUG_MODE = False
+
+def _encode_forward_multimedia(media: dict) -> bytes:
+    buf = b""
+    for fn, key in [(1, "type"), (2, "url"), (4, "file_name"), (15, "media_id")]:
+        v = media.get(key, "")
+        if v:
+            buf += _encode_field(fn, WT_LEN, _encode_string(str(v)))
+    for fn, key in [(5, "file_size"), (6, "width"), (7, "height")]:
+        v = media.get(key, 0)
+        if v:
+            buf += _encode_field(fn, WT_VARINT, _encode_varint(int(v)))
+    return buf
+
+def _encode_forward_msg_content(content: dict) -> bytes:
+    buf = _encode_field(1, WT_VARINT, _encode_varint(int(content.get("type", 0))))
+    text = content.get("text", "")
+    if text:
+        buf += _encode_field(2, WT_LEN, _encode_string(str(text)))
+    for media in content.get("multimedia") or []:
+        buf += _encode_field(3, WT_LEN, _encode_message(_encode_forward_multimedia(media)))
+    return buf
+
+def _encode_forward_msg(msg: dict) -> bytes:
+    buf = b""
+    sender = msg.get("sender", "")
+    if sender:
+        buf += _encode_field(1, WT_LEN, _encode_string(str(sender)))
+    time_val = msg.get("time", 0)
+    if time_val:
+        buf += _encode_field(2, WT_VARINT, _encode_varint(int(time_val)))
+    plain = msg.get("plainText", "")
+    if plain:
+        buf += _encode_field(3, WT_LEN, _encode_string(str(plain)))
+    for mc in msg.get("msgContent") or []:
+        buf += _encode_field(4, WT_LEN, _encode_message(_encode_forward_msg_content(mc)))
+    return buf
+
+def encode_forward_msg_data(data: dict) -> bytes:
+    """Encode ForwardMsgData protobuf bytes (inverse of ``decode_forward_msg_data``).
+
+    Mainly used to build mock / test data; production code never needs to encode this.
+    """
+    buf = _encode_field(1, WT_VARINT, _encode_varint(int(data.get("sub_type", 0))))
+    for fn, key in [(2, "begin_time"), (3, "end_time")]:
+        v = data.get(key, 0)
+        if v:
+            buf += _encode_field(fn, WT_VARINT, _encode_varint(int(v)))
+    nick = data.get("nick_name", "")
+    if nick:
+        buf += _encode_field(4, WT_LEN, _encode_string(str(nick)))
+    for msg in data.get("msg") or []:
+        buf += _encode_field(5, WT_LEN, _encode_message(_encode_forward_msg(msg)))
+    return buf
+
+
+_PLUGIN_COMPAT_LAZY = {
+    'logger': ('gateway.platforms.base', 'logger'),
+}
+
+
+def __getattr__(name):  # PEP 562 — lazy so no import cycles
+    target = _PLUGIN_COMPAT_LAZY.get(name)
+    if target is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    import importlib
+    return getattr(importlib.import_module(target[0]), target[1])
+# ---- END PLUGIN-COMPAT ----

gateway/relay/__init__.py

@@ -650,3 +650,22 @@ def register_relay_adapter(force: bool = False, url: Optional[str] = None) -> bo
         )
     )
     return True
+
+
+# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
+# Names external plugins imported from this module before the Sep 2026 decomposition.
+# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
+# The whole block is removed by reverting the commit that added it.
+
+def relay_bot_username(platform: str) -> Optional[str]:
+    """The bot's deep-link username/handle for a platform (e.g. Telegram's
+    ``@handle`` for ``t.me/<handle>``), read from the per-platform entry in
+    ``GATEWAY_RELAY_BOT_IDS``. None when absent (most platforms don't need one).
+    """
+    entry = _relay_bot_ids_map().get(platform)
+    if isinstance(entry, dict):
+        username = entry.get("username")
+        if username:
+            return str(username).lstrip("@")
+    return None
+# ---- END PLUGIN-COMPAT ----