feat(loop): add REASONIX anti-wandering loop guards

Port the REASONIX anti-wandering guard family (evidence ledger, progress
guard, storm breaker, delegation admission) into the agent runtime loop:

- core/loop/guards.py: guard engine (stdlib-only, no internal deps)
- core/loop/guard_telemetry.py: optional telemetry seam
- runner.py: blocked short-circuit, observe_batch injection, per-tool
  check_tool gate, success-path observation wiring
- session.py: thread LoopGuards instance through AgentSession/AgentRunSpec
- spawn_agent.py: delegation admission gate (REASONIX delegationAdmission)
- loop/__init__.py: public exports

Origin: Python port of the applyBatchGuards component family from
DeepSeek-Reasonix (https://github.com/esengine/DeepSeek-Reasonix, MIT
licensed), adapted to DeepCode's run-loop conventions. Implementation is
a fresh Python rewrite, not a code copy.

Author: DeepCodeClone

core/agent_runtime/runner.py

@@ -49,6 +49,8 @@
 )
 from core.agent_runtime.tools.base import ToolResult
 from core.agent_runtime.tools.registry import ToolRegistry
+from core.loop.guard_telemetry import install_guard_telemetry
+from core.loop.guards import LoopGuards
 from core.providers.base import LLMProvider, LLMResponse, ToolCallRequest
 from core.providers.timeouts import (
     resolve_request_timeout_s,
@@ -190,6 +192,20 @@ class AgentRunSpec:
     # so a well-behaved hook stops blocking after its first continuation.
     stop_hook: Any | None = None
 
+    # Loop guards (REASONIX §5/§6 port, P3.5): anti-wandering circuit breakers.
+    # ``LoopGuards.observe_batch`` runs after each tool batch; triggered
+    # interventions are injected as user messages. Once the progress guard
+    # forces a final answer (N consecutive zero-evidence rounds), further tool
+    # execution is short-circuited. Absent (None) means zero cost.
+    guards: LoopGuards | None = None
+
+    # Guard-event telemetry seam (REASONIX P1): optional callable invoked when a
+    # loop guard triggers (blocked short-circuit / observe_batch injection /
+    # check_tool block). Receives a structured dict
+    # ``{"kind", "tool", "level", "streak", "message"}``; default None keeps the
+    # existing logger.warning behavior.
+    guard_event_callback: Callable[[dict], None] | None = None
+
     def allowed_tool_names(self) -> frozenset[str] | None:
         if self.tool_filter is None:
             return None
@@ -400,6 +416,9 @@ def _injection_note_source(item: Any) -> str:
         return "injection"
 
     async def run(self, spec: AgentRunSpec) -> AgentRunResult:
+        # P5: optional guard telemetry wiring (DEEPCODE_GUARD_TELEMETRY=1
+        # auto-connects; zero overhead by default).
+        install_guard_telemetry(spec)
         hook = spec.hook or AgentHook()
         messages = list(spec.initial_messages)
         final_content: str | None = None
@@ -568,11 +587,33 @@ async def record_compaction_response(response: LLMResponse) -> None:
 
                 await hook.before_execute_tools(context)
 
-                results, new_events, fatal_error = await self._execute_tools(
-                    spec,
-                    response.tool_calls,
-                    external_lookup_counts,
-                )
+                # Progress guard forced a final answer (REASONIX §2.2 level 3):
+                # short-circuit tool execution and feed errors-as-data instead,
+                # so the model reads the block reason and wraps up cleanly.
+                if spec.guards is not None and spec.guards.blocked:
+                    block_reason = (
+                        "Error: blocked by loop guard — no new evidence for "
+                        f"{spec.guards.streak} consecutive tool rounds. "
+                        "Produce your final answer now."
+                    )
+                    blocked_results = [block_reason for _ in response.tool_calls]
+                    results, new_events, fatal_error = blocked_results, [], None
+                    if spec.guard_event_callback is not None:
+                        spec.guard_event_callback(
+                            {
+                                "kind": "blocked",
+                                "tool": None,
+                                "level": 3,
+                                "streak": spec.guards.streak,
+                                "message": block_reason,
+                            }
+                        )
+                else:
+                    results, new_events, fatal_error = await self._execute_tools(
+                        spec,
+                        response.tool_calls,
+                        external_lookup_counts,
+                    )
                 tool_events.extend(new_events)
                 context.tool_results = list(results)
                 context.tool_events = list(new_events)
@@ -657,6 +698,31 @@ async def record_compaction_response(response: LLMResponse) -> None:
                 if drained:
                     had_injections = True
                     sampling_limit.reset()
+
+                if spec.guards is not None:
+                    guard_injections = spec.guards.observe_batch(
+                        response.tool_calls, results, new_events
+                    )
+                    if guard_injections:
+                        self._append_injected_messages(messages, guard_injections)
+                        for injection in guard_injections:
+                            logger.warning(
+                                "Loop guard on turn {} for {}: {}",
+                                current_iteration,
+                                spec.session_key or "default",
+                                injection["content"][:120],
+                            )
+                            if spec.guard_event_callback is not None:
+                                spec.guard_event_callback(
+                                    {
+                                        "kind": "injection",
+                                        "tool": None,
+                                        "level": None,
+                                        "streak": spec.guards.streak,
+                                        "message": injection["content"],
+                                    }
+                                )
+
                 await hook.after_iteration(context)
                 continue
 
@@ -1167,6 +1233,34 @@ async def _run_tool(
             )
             return result, event, None
 
+        # Loop guards (REASONIX §1.2–1.8 port, P3.6) — per-tool governance gate.
+        # Permission is the security layer (checked first); guards are the
+        # loop-governance layer (checked second). Blocks are errors-as-data.
+        if spec.guards is not None:
+            guard_message = spec.guards.check_tool(
+                tool_call.name, tool_call.arguments
+            )
+            if guard_message is not None:
+                event = {
+                    "name": tool_call.name,
+                    "status": "denied",
+                    "detail": guard_message.replace("\n", " ").strip()[:120],
+                }
+                if spec.guard_event_callback is not None:
+                    spec.guard_event_callback(
+                        {
+                            "kind": "tool_block",
+                            "tool": tool_call.name,
+                            "level": None,
+                            "streak": spec.guards.streak,
+                            "message": guard_message,
+                        }
+                    )
+                result = self._compose_hook_context(
+                    f"Error: {guard_message}" + _HINT, pre_contexts
+                )
+                return result, event, None
+
         prepare_call = getattr(spec.tools, "prepare_call", None)
         tool, params, prep_error = None, tool_call.arguments, None
         if callable(prepare_call):
@@ -1299,6 +1393,15 @@ async def _run_tool(
         elif len(detail) > 120:
             detail = detail[:120] + "..."
         event = {"name": tool_call.name, "status": "ok", "detail": detail}
+        if spec.guards is not None:
+            # Post-execution guard observation (REASONIX P3.6 port): result
+            # fingerprint maintenance + mutation dependency tracking. Only the
+            # success path reaches here, so pending-state updates are accurate.
+            spec.guards.observe_tool_result(
+                tool_call.name, tool_call.arguments, result
+            )
+            spec.guards.observe_tool_mutation(tool_call.name, tool_call.arguments)
+            spec.guards.observe_tool_verify(tool_call.name, tool_call.arguments)
         return await self._finish_tool(
             spec, tool_call, result, event, None, pre_contexts
         )

core/events/session.py

@@ -63,6 +63,7 @@
 )
 from core.mcp.models import McpStartupError
 from core.mcp.runtime import McpSessionRuntime
+from core.loop.guards import LoopGuards
 from core.providers.base import LLMProvider
 from core.providers.catalog import context_window_for
 from core.reasoning import ReasoningAvailability, ReasoningChannel
@@ -398,6 +399,7 @@ def __init__(
         tool_filter: Any | None = None,
         closure_callback: Any | None = None,
         mcp_runtime: McpSessionRuntime | None = None,
+        guards: LoopGuards | None = None,
     ) -> None:
         self._runner = AgentRunner(provider)
         self._provider = provider
@@ -457,6 +459,11 @@ def __init__(
         # ordinary Turns leave it unset.
         self._closure_callback = closure_callback
         self._mcp_runtime = mcp_runtime
+        # Loop guards (REASONIX P3.5/P3.6 port): anti-wandering circuit
+        # breakers observed across tool batches; None keeps the feature dormant
+        # at zero cost. LoopTask passes the same instance every round so
+        # ProgressGuard/StormBreaker state survives across rounds.
+        self._guards = guards
         # Secret-free immutable selection used by persistence/frontends.
         self.execution_profile = execution_profile
 
@@ -931,6 +938,7 @@ def visible_tool_names() -> tuple[str, ...] | None:
             stop_hook=stop_hook,
             pre_compact_hook=pre_compact_hook,
             post_compact_hook=post_compact_hook,
+            guards=self._guards,
             tool_filter=(
                 visible_tool_names
                 if self._skill_runtime is not None or self._tool_filter is not None

core/harness/tools/spawn_agent.py

@@ -17,6 +17,7 @@
 
 from core.agent_runtime.tools.base import Tool, tool_parameters
 from core.harness.agents.control import AgentControl, AgentLimitError
+from core.loop.guards import delegation_admission
 
 
 def _parse_fork_turns(value: Any) -> str | int:
@@ -154,6 +155,22 @@ async def execute(self, **kwargs: Any) -> Any:
         output_schema = kwargs.get("output_schema")
         if output_schema is not None and not isinstance(output_schema, dict):
             return "Error: 'output_schema' must be a JSON object."
+        # Delegation admission (REASONIX §1.10 delegationAdmission adaptation):
+        # spawn_agent is a local concurrent delegation (C2); self-contained
+        # tasks are allowed by default. Only tasks that reference the parent
+        # conversation but would not inherit it are rejected (the sub-agent
+        # cannot see your messages).
+        decision, reason = delegation_admission(task, fork_turns=str(fork_turns))
+        if decision == "deny":
+            return (
+                "Error: delegation denied (local_fix_no_external_need). "
+                f"The subtask references the parent conversation but "
+                f"fork_turns is '{fork_turns}', so the sub-agent inherits no "
+                "context and cannot act on those references. Either do the "
+                "work yourself, or rewrite the task to be self-contained, or "
+                "pass fork_turns='all'/'<N>' to inherit the needed context. "
+                f"(reason: {reason})"
+            )
         try:
             agent_id = self._control.spawn(
                 task,

core/loop/__init__.py

@@ -1,5 +1,42 @@
 """Optional maintenance utilities outside the interactive Agent loop."""
 
-from core.loop.autodream import AutodreamResult, consolidate_memory
+from core.loop.guards import (
+    EvidenceLedger,
+    GuardIntervention,
+    LoopGuards,
+    ProgressGuard,
+    StormBreaker,
+    delegation_admission,
+)
 
-__all__ = ["AutodreamResult", "consolidate_memory"]
+__all__ = [
+    "AutodreamResult",
+    "consolidate_memory",
+    # REASONIX anti-wandering guards (P3.5)
+    "EvidenceLedger",
+    "GuardIntervention",
+    "ProgressGuard",
+    "StormBreaker",
+    "LoopGuards",
+    "delegation_admission",
+]
+
+
+def __getattr__(name: str):
+    """Lazily expose the autodream API.
+
+    ``core.loop.autodream`` imports ``core.agent_setup`` (and transitively
+    ``core.compat.agent`` -> ``core.agent_runtime.runner``), so eagerly
+    importing it here would create a cycle when ``runner`` itself imports this
+    package (e.g. for the REASONIX loop guards). Load it only on first access
+    -- by then package init has completed and the import chain is safe.
+    """
+    if name in ("AutodreamResult", "consolidate_memory"):
+        from core.loop.autodream import AutodreamResult, consolidate_memory
+
+        value = (
+            AutodreamResult if name == "AutodreamResult" else consolidate_memory
+        )
+        globals()[name] = value
+        return value
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

core/loop/guard_telemetry.py

@@ -0,0 +1,112 @@
+"""Guard-event telemetry adapter (P5) — guard_event_callback → deepcode-telemetry.
+
+REASONIX 守卫事件（``blocked`` / ``injection`` / ``tool_block``）的可选遥测接入。
+默认零开销：未设置 ``DEEPCODE_GUARD_TELEMETRY=1`` 时不做任何事；遥测 skill 缺失或
+调用失败时静默降级，绝不影响 agent 主循环。
+
+设计要点
+--------
+- 纯 stdlib，不依赖 loguru / core 内部（避免循环导入与回归风险）。
+- 通过 ``importlib.util.spec_from_file_location`` 动态加载
+  ``.deepcode/skills/deepcode-telemetry/telemetry.py``，不污染 ``sys.path``。
+- 模块级缓存：每个进程只尝试加载一次（``_attempted`` / ``_callbacks_cache``）。
+- 事件映射（kind → 遥测调用）：
+    - ``blocked``    → increment_counter("guard.blocked") + record_metric("guard.blocked.streak")
+    - ``injection``  → increment_counter("guard.injections")
+    - ``tool_block`` → increment_counter("guard.tool_block") + 按工具细分计数器
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+from pathlib import Path
+from typing import Any, Callable
+
+ENV_GUARD_TELEMETRY = "DEEPCODE_GUARD_TELEMETRY"
+
+# core/loop/guard_telemetry.py → parents[2] = F:/DEEPCODE
+_TELEMETRY_SKILL_REL = Path(".deepcode") / "skills" / "deepcode-telemetry" / "telemetry.py"
+
+_attempted = False
+_callbacks_cache: Callable[[dict], None] | None = None
+
+
+def _load_telemetry_module() -> Any | None:
+    """动态加载 deepcode-telemetry skill 模块；失败返回 None（静默降级）。"""
+    skill_py = Path(__file__).resolve().parents[2] / _TELEMETRY_SKILL_REL
+    if not skill_py.is_file():
+        return None
+    try:
+        spec = importlib.util.spec_from_file_location("_deepcode_guard_telemetry", skill_py)
+        if spec is None or spec.loader is None:
+            return None
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+    except Exception:
+        return None
+
+
+def _build_callback(engine: Any) -> Callable[[dict], None]:
+    """构造守卫事件 → 遥测调用回调。所有遥测调用异常静默吞掉。"""
+
+    def on_guard_event(event: dict) -> None:
+        try:
+            kind = event.get("kind")
+            if kind == "blocked":
+                engine.increment_counter("guard.blocked", 1)
+                streak = event.get("streak")
+                if streak is not None:
+                    engine.record_metric("guard.blocked.streak", float(streak))
+            elif kind == "injection":
+                engine.increment_counter("guard.injections", 1)
+            elif kind == "tool_block":
+                engine.increment_counter("guard.tool_block", 1)
+                tool = event.get("tool")
+                if tool:
+                    engine.increment_counter(f"guard.tool_block.{tool}", 1)
+        except Exception:
+            pass  # telemetry 失败静默降级，绝不影响 agent 主循环
+
+    return on_guard_event
+
+
+def make_telemetry_guard_callback() -> Callable[[dict], None] | None:
+    """惰性构造遥测守卫回调（进程级缓存，只尝试一次）。"""
+    global _attempted, _callbacks_cache
+    if _attempted:
+        return _callbacks_cache
+    _attempted = True
+    try:
+        module = _load_telemetry_module()
+        if module is None or not hasattr(module, "get_telemetry"):
+            _callbacks_cache = None
+            return None
+        _callbacks_cache = _build_callback(module.get_telemetry())
+    except Exception:
+        _callbacks_cache = None
+    return _callbacks_cache
+
+
+def install_guard_telemetry(spec: Any) -> None:
+    """runner ``run()`` 开头调用：可选装配遥测回调，默认零开销。
+
+    - 已有 ``guard_event_callback`` → 不覆盖。
+    - ``DEEPCODE_GUARD_TELEMETRY`` 未设为 "1" → 不装配。
+    - 遥测 skill 不可用 → 静默保持 None。
+    """
+    if spec.guard_event_callback is not None:
+        return
+    if os.environ.get(ENV_GUARD_TELEMETRY, "0") != "1":
+        return
+    callback = make_telemetry_guard_callback()
+    if callback is not None:
+        spec.guard_event_callback = callback
+
+
+__all__ = [
+    "ENV_GUARD_TELEMETRY",
+    "make_telemetry_guard_callback",
+    "install_guard_telemetry",
+]