diff --git a/core/agent_runtime/runner.py b/core/agent_runtime/runner.py
index 3c13850d..86d68cca 100644
--- a/core/agent_runtime/runner.py
+++ b/core/agent_runtime/runner.py
@@ -39,6 +39,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,
@@ -167,6 +169,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
@@ -348,6 +364,9 @@ async def _drain_injections(
         return injected_messages
 
     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
@@ -536,11 +555,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)
@@ -604,6 +645,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
 
@@ -1069,6 +1135,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):
@@ -1151,6 +1245,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
         )
diff --git a/core/events/session.py b/core/events/session.py
index 1991635b..c5acd107 100644
--- a/core/events/session.py
+++ b/core/events/session.py
@@ -61,6 +61,7 @@
     summarize_call,
     summarize_result,
 )
+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
@@ -394,6 +395,7 @@ def __init__(
         execution_profile: Any | None = None,
         tool_filter: Any | None = None,
         closure_callback: Any | None = None,
+        guards: LoopGuards | None = None,
     ) -> None:
         self._runner = AgentRunner(provider)
         self._provider = provider
@@ -449,6 +451,11 @@ def __init__(
         # this to ask the model for one final complete/blocked/continue decision;
         # ordinary Turns leave it unset.
         self._closure_callback = closure_callback
+        # 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
 
@@ -867,6 +874,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
diff --git a/core/harness/tools/spawn_agent.py b/core/harness/tools/spawn_agent.py
index d3b3af6f..2c4a4750 100644
--- a/core/harness/tools/spawn_agent.py
+++ b/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:
@@ -94,6 +95,22 @@ async def execute(self, **kwargs: Any) -> Any:
         name = str(kwargs.get("name") or "").strip() or None
         isolate = bool(kwargs.get("isolate", True))
         fork_turns = _parse_fork_turns(kwargs.get("fork_turns"))
+        # 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, name=name, isolate=isolate, fork_turns=fork_turns
diff --git a/core/loop/__init__.py b/core/loop/__init__.py
index 463ec55a..7aa9a362 100644
--- a/core/loop/__init__.py
+++ b/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}")
diff --git a/core/loop/guard_telemetry.py b/core/loop/guard_telemetry.py
new file mode 100644
index 00000000..049bc8c8
--- /dev/null
+++ b/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",
+]
diff --git a/core/loop/guards.py b/core/loop/guards.py
new file mode 100644
index 00000000..a98035ea
--- /dev/null
+++ b/core/loop/guards.py
@@ -0,0 +1,697 @@
+"""Loop guards (P3.5) — REASONIX 反漫游守卫的 DEEPCODE 落地。
+
+来源：``scripts/REASONIX_第五阶段深度分析报告.md`` / ``REASONIX_第六阶段深度分析报告.md``
+（逆向 reasonix.exe v1.21.5 的 applyBatchGuards 组件族）。
+
+落地的四个机制：
+
+1. :class:`EvidenceLedger` + :func:`EvidenceLedger.score_round`
+   — 证据账本（``evidence.(*Ledger).Record`` port，纯内存追加）+ 整轮 0-3 评分。
+2. :class:`ProgressGuard` — 连续零进展轮次 N，2/4/6 升级干预
+   （``progressGuard.observe`` port：N==2 温和提醒 / N==4 强制换策略 / N==6 强制收尾+阻塞）。
+3. :class:`StormBreaker` — 工具序列指纹（``batchStormSignature`` port：
+   concat 工具名），相同失败签名超过上限熔断。
+4. :func:`delegation_admission` — 委派准入（``agent.delegationAdmission`` port，
+   已按 DEEPCODE 本地并发委派场景适配）。
+
+评分方向裁决
+------------
+
+第五阶段 ``progressGuard.observe`` 反编译伪代码：
+
+    iVar4 = evidence.(*ProgressTracker).ScoreRound(guard[0], toolCalls, count, ctx);
+    if (iVar4 < 1) guard[1] += 1;   // 零新证据 → 连续计数++
+    else           guard[1] = 0;    // 有新证据 → 计数清零
+
+证明 ScoreRound 的**聚合返回值**语义为：``0`` = 本轮零证据（最差，streak++），
+``>= 1`` = 本轮有进展（streak 清零）。第六阶段 ``scoreReceipt`` 的单条打分标签
+（0=新证据 / 3=零证据）是 receipt 层的惩罚分，与聚合层方向相反（聚合时被反转）。
+
+落地以第五阶段伪代码为准：``SCORE_ZERO_EVIDENCE == 0``（最差），
+``SCORE_NEW_EVIDENCE == 3``（最好）。同时把 ``SCORE_REPEATED_CALLS``（完全重复）
+也视为零进展——REASONIX 干预消息原文明确把"repeated earlier reads or commands
+without new results"归入零进展，故 ``score < SCORE_PARTIAL_EVIDENCE`` 即触发 streak++。
+"""
+
+from __future__ import annotations
+
+import fnmatch
+import hashlib
+import json
+import threading
+from dataclasses import dataclass, field
+from typing import Any, Callable, Iterable
+
+# ---------------------------------------------------------------------------
+# 评分常量（0 = 最差 / 零证据，3 = 最好 / 新证据）——方向裁决见模块 docstring
+# ---------------------------------------------------------------------------
+
+SCORE_ZERO_EVIDENCE = 0  # 完全零证据（双零：无新文件无新结果）
+SCORE_REPEATED_CALLS = 1  # 重复工具调用（无新产出）
+SCORE_PARTIAL_EVIDENCE = 2  # 部分证据（相同调用但结果变化）
+SCORE_NEW_EVIDENCE = 3  # 本轮有新证据（正常）
+
+# ProgressGuard 升级阈值（REASONIX 第五阶段 §2.2）
+GUARD_NUDGE = 2  # 连续 N 轮零进展 → 温和提醒
+GUARD_STRATEGY = 4  # → 强制换策略
+GUARD_FORCE_ANSWER = 6  # → 强制出最终答案 + 阻塞
+
+# 委派准入白名单（REASONIX 第六阶段 §2.2，6 条全部保留）
+DELEGATION_RESEARCH_HINTS = (
+    "research",
+    "调研",
+    "查资料",
+    "查文档",
+    "search the web",
+    "look up online",
+)
+# DEEPCODE 适配：引用父上下文但不继承上下文 → 子 agent 缺上下文无法完成
+DELEGATION_CONTEXT_REFERENCE_HINTS = (
+    "上述",
+    "上面",
+    "之前说的",
+    "前面提到",
+    "我们刚才",
+    "我们之前",
+    "之前的讨论",
+    "前面",
+    "above",
+    "mentioned",
+    "earlier",
+)
+
+# 工具结果哈希截断长度（对齐 backpressure._TAIL_CHARS 的务实上限）
+_RESULT_HASH_CHARS = 2000
+
+
+def tool_call_signature(name: str, arguments: dict[str, Any] | None) -> str:
+    """规范化工具调用指纹：工具名 + 排序键的参数 JSON。"""
+    args = arguments or {}
+    canon = json.dumps(args, sort_keys=True, ensure_ascii=False, default=str)
+    return f"{name}:{canon}"
+
+
+def result_hash(result: Any) -> str | None:
+    """工具结果摘要哈希；空结果返回 None（无法作为证据）。"""
+    text = str(result)
+    if not text.strip():
+        return None
+    return hashlib.sha256(
+        text[:_RESULT_HASH_CHARS].encode("utf-8", "ignore")
+    ).hexdigest()
+
+
+@dataclass(frozen=True, slots=True)
+class ToolEvidence:
+    """单条工具调用证据（对齐 REASONIX Ledger 条目语义）。"""
+
+    tool: str
+    arg_signature: str
+    result_hash: str | None
+
+
+class EvidenceLedger:
+    """内存追加证据账本（纯内存，不落盘——同 REASONIX Ledger）。"""
+
+    def __init__(self) -> None:
+        self._seen_calls: set[tuple[str, str]] = set()
+        self._seen_results: set[str] = set()
+        self._rounds: list[list[ToolEvidence]] = []
+
+    @staticmethod
+    def _build(
+        calls: Iterable[tuple[str, dict[str, Any] | None, Any]]
+    ) -> list[ToolEvidence]:
+        """把 (tool_name, arguments, result) 三元组构建为证据列表（不落账）。"""
+        evidence: list[ToolEvidence] = []
+        for name, arguments, result in calls:
+            evidence.append(
+                ToolEvidence(
+                    tool=name,
+                    arg_signature=tool_call_signature(name, arguments),
+                    result_hash=result_hash(result),
+                )
+            )
+        return evidence
+
+    def _append(self, evidence: list[ToolEvidence]) -> None:
+        """把一轮证据写入账本（seen 集合 + 轮次历史）。"""
+        for item in evidence:
+            self._seen_calls.add((item.tool, item.arg_signature))
+            if item.result_hash is not None:
+                self._seen_results.add(item.result_hash)
+        if evidence:
+            self._rounds.append(evidence)
+
+    def record(self, calls: Iterable[tuple[str, dict[str, Any] | None, Any]]) -> list[ToolEvidence]:
+        """记录一轮 (tool_name, arguments, result) 三元组，返回本轮证据列表。"""
+        evidence = self._build(calls)
+        self._append(evidence)
+        return evidence
+
+    def score_round(
+        self, calls: Iterable[tuple[str, dict[str, Any] | None, Any]]
+    ) -> int:
+        """整轮评分 0-3（0=零证据最差，3=新证据最好）。
+
+        先与历史证据比对、后落账（compare-then-record）：若先写入
+        ``_seen_calls``/``_seen_results`` 再查重，本轮证据必然命中自身，
+        导致永远返回 ``SCORE_REPEATED_CALLS``——进度守卫会把每一轮都
+        当作零进展轮。
+        """
+        evidence = self._build(calls)
+        if not evidence:
+            return SCORE_ZERO_EVIDENCE
+        for item in evidence:
+            if (item.tool, item.arg_signature) not in self._seen_calls:
+                # 存在全新调用（工具+参数都未出现过）→ 本轮有新证据
+                self._append(evidence)
+                return SCORE_NEW_EVIDENCE
+        for item in evidence:
+            if item.result_hash is not None and item.result_hash not in self._seen_results:
+                # 相同调用但产生了新的结果内容 → 部分证据
+                self._append(evidence)
+                return SCORE_PARTIAL_EVIDENCE
+        self._append(evidence)
+        return SCORE_REPEATED_CALLS
+
+    @property
+    def rounds(self) -> int:
+        return len(self._rounds)
+
+
+@dataclass(frozen=True, slots=True)
+class GuardIntervention:
+    """一次进度守卫干预。"""
+
+    level: int  # 1=温和提醒 2=强制换策略 3=强制收尾+阻塞
+    streak: int  # 触发时的连续零进展轮数
+    message: str
+
+
+class ProgressGuard:
+    """连续零进展轮次守卫（``progressGuard.observe`` port）。
+
+    每轮评分后调用 :meth:`observe`：``score < SCORE_PARTIAL_EVIDENCE``
+    （零证据或完全重复）→ streak++；否则 streak 清零。
+    阈值 2/4/6 各触发一次（``>=`` 判定 + 级别递进，避免跳级漏触发/重复触发）。
+    """
+
+    def __init__(
+        self,
+        nudge: int = GUARD_NUDGE,
+        strategy: int = GUARD_STRATEGY,
+        force_answer: int = GUARD_FORCE_ANSWER,
+    ) -> None:
+        self._nudge = nudge
+        self._strategy = strategy
+        self._force = force_answer
+        self._streak = 0
+        self._delivered = 0
+        self._blocked = False
+
+    @property
+    def streak(self) -> int:
+        return self._streak
+
+    @property
+    def blocked(self) -> bool:
+        """N==6 触发后为 True（强制收尾熔断）。"""
+        return self._blocked
+
+    def observe(self, score: int) -> GuardIntervention | None:
+        if score < SCORE_PARTIAL_EVIDENCE:
+            self._streak += 1
+        else:
+            self._streak = 0
+            return None
+
+        if self._streak >= self._force and self._delivered < 3:
+            self._delivered = 3
+            self._blocked = True
+            return GuardIntervention(
+                3,
+                self._streak,
+                f"[progress guard] 连续 {self._streak} 轮工具调用未产生任何新证据"
+                "（没有新文件、新结果或变更）。请停止探索，现在直接给出最终答案，"
+                "说明已确认的内容与仍然未知的内容。",
+            )
+        if self._streak >= self._strategy and self._delivered < 2:
+            self._delivered = 2
+            return GuardIntervention(
+                2,
+                self._streak,
+                f"[progress guard] 连续 {self._streak} 轮仍无新证据。请立即更换策略："
+                "换一个角度或工具、委派一个聚焦的子任务、或缩小验证范围后再继续。",
+            )
+        if self._streak >= self._nudge and self._delivered < 1:
+            self._delivered = 1
+            return GuardIntervention(
+                1,
+                self._streak,
+                f"[progress guard] 最近 {self._streak} 轮工具调用反复执行了相同的"
+                "读取/命令，但没有产生新结果。请收窄调查范围或调整计划后再继续。",
+            )
+        return None
+
+
+class StormBreaker:
+    """工具序列风暴熔断（``batchStormSignature`` port）。
+
+    同批工具名 concat 成签名；同一失败签名超过 ``max_failures`` 次后熔断
+    （首次熔断注入一次提醒，之后保持 blocked 不再重复注入）。
+    """
+
+    def __init__(self, max_failures: int = 3) -> None:
+        self._max_failures = max_failures
+        self._fail_counts: dict[str, int] = {}
+        self._blocked = False
+        self._announced = False
+
+    @property
+    def blocked(self) -> bool:
+        return self._blocked
+
+    def observe_batch(self, tool_names: list[str], *, failed: bool) -> str | None:
+        """返回熔断提醒消息（仅首次熔断时）；否则 None。"""
+        if not failed or not tool_names:
+            return None
+        signature = "|".join(tool_names)
+        count = self._fail_counts.get(signature, 0) + 1
+        self._fail_counts[signature] = count
+        if count > self._max_failures:
+            self._blocked = True
+            if not self._announced:
+                self._announced = True
+                return (
+                    f"[loop guard] 工具序列 {signature} 已连续失败 {count} 次"
+                    f"（超过 {self._max_failures} 次上限），已熔断。请停止重复该序列，"
+                    "改用不同的方法。"
+                )
+        return None
+
+
+def delegation_admission(
+    task: str, *, fork_turns: str = "none"
+) -> tuple[str, str]:
+    """委派准入（``agent.delegationAdmission`` 的 DEEPCODE 适配版）。
+
+    REASONIX 原版：委派场景下仅研究意图 / 外部 URL 引用放行，否则
+    ``deny(local_fix_no_external_need)``。DEEPCODE 的 spawn_agent 是本地并发委派
+    （C2，非外部研究委派），照搬原版会把普通编程子任务全部拒绝，故适配为：
+
+    - 研究意图 / 外部 URL / 继承父上下文 → ``allow``
+    - 引用父上下文但不继承（fork_turns=none）→ ``deny``（子 agent 缺上下文无法完成）
+    - 其余自包含任务 → ``allow``
+
+    返回 ``(decision, reason)``，decision ∈ {"allow", "deny"}。
+    """
+    text = (task or "").strip()
+    if not text:
+        return "deny", "empty_task"
+    low = text.lower()
+    if any(hint in low for hint in DELEGATION_RESEARCH_HINTS):
+        return "allow", "user_requested_research"
+    if "http://" in low or "https://" in low:
+        return "allow", "external_source_cited"
+    if fork_turns != "none":
+        return "allow", "context_inherited"
+    if any(hint in text for hint in DELEGATION_CONTEXT_REFERENCE_HINTS):
+        return "deny", "task_references_parent_context"
+    return "allow", "self_contained"
+
+
+# ---------------------------------------------------------------------------
+# P3.6 单工具级守卫（REASONIX 第五阶段 §1.2–1.8 port）——runner 在工具执行前
+# 调用 ``LoopGuards.check_tool``，执行后调用 ``observe_tool_*`` 观察。
+# ---------------------------------------------------------------------------
+
+
+def _extract_paths(arguments: dict[str, Any] | None) -> list[str]:
+    """从工具参数中提取显式路径（path / file_path / filepath / file）。"""
+    args = arguments or {}
+    paths: list[str] = []
+    for key in ("path", "file_path", "filepath", "file"):
+        value = args.get(key)
+        if isinstance(value, str) and value.strip():
+            paths.append(value.strip())
+    return paths
+
+
+class ContextualToolGate:
+    """上下文工具门（``applyContextualToolGate`` / ``contextualToolGateOutcome`` port）。
+
+    REASONIX 侧按工具类型 hash 分派到专用匹配器，匹配器判定当前工作流上下文
+    是否允许该工具执行，不匹配 → ``blocked: tool %q is unavailable in the current
+    workflow context``（第六阶段 §3）。DEEPCODE 侧以 ``{工具 glob: 匹配器回调}``
+    注册表等价实现，回调返回非 None 即阻断（str 为原因）。默认空注册表 → 零成本。
+    """
+
+    def __init__(
+        self,
+        matchers: dict[str, Callable[[dict[str, Any]], str | None]] | None = None,
+    ) -> None:
+        self._matchers = dict(matchers or {})
+        self._lock = threading.Lock()
+        self._blocks: dict[str, int] = {}
+
+    def check(self, tool_name: str, arguments: dict[str, Any] | None) -> str | None:
+        """工具执行前调用；返回 None=放行，str=阻断消息（errors-as-data）。"""
+        args = arguments or {}
+        for pattern, matcher in self._matchers.items():
+            if fnmatch.fnmatch(tool_name, pattern):
+                reason = matcher(args)
+                if reason is not None:
+                    with self._lock:
+                        self._blocks[tool_name] = self._blocks.get(tool_name, 0) + 1
+                    return (
+                        f"blocked: tool {tool_name!r} is unavailable in the current "
+                        f"workflow context ({reason})"
+                    )
+        return None
+
+    @property
+    def blocks(self) -> dict[str, int]:
+        """各工具被阻断次数（只读快照）。"""
+        with self._lock:
+            return dict(self._blocks)
+
+
+class MutationDependencyBarrier:
+    """bash 变更依赖屏障（``applyMutationDependencyBarrier`` / ``shellPreflightExecution`` port）。
+
+    REASONIX 语义：bash 工具执行前预检"挂起的变更依赖"，存在依赖则阻断并记日志
+    ``[shell_preflight] blocked tool execution pending dependent mutations (bash %s)``。
+    DEEPCODE 适配：写/编辑类工具先 ``observe_mutation`` 登记待验证路径，read/grep/
+    test 等验证工具 ``observe_verify`` 清除；bash/exec 引用未验证路径时——默认**软提醒**
+    （开发循环 write→test 是合法路径，硬阻断会误伤），``hard_block=True`` 才做 REASONIX
+    等价硬阻断。
+    """
+
+    # 会登记 pending 路径的变更工具
+    _MUTATION_TOOLS = (
+        "write_file", "write", "edit_file", "edit", "apply_patch", "patch", "replace",
+    )
+    # 会清除 pending 的验证工具
+    _VERIFY_TOOLS = ("read_file", "read", "grep", "test", "pytest", "run_tests")
+    # 触发依赖预检的 shell 工具
+    _SHELL_TOOLS = ("bash", "exec", "execute_bash", "execute_commands")
+
+    def __init__(self, hard_block: bool = False) -> None:
+        self._hard_block = hard_block
+        self._pending: dict[str, str] = {}  # path -> 登记它的变更工具名
+        self._lock = threading.Lock()
+
+    def observe_mutation(self, tool_name: str, arguments: dict[str, Any] | None) -> None:
+        """变更工具执行成功后登记待验证路径。"""
+        if tool_name not in self._MUTATION_TOOLS:
+            return
+        with self._lock:
+            for path in _extract_paths(arguments):
+                self._pending[path] = tool_name
+
+    def observe_verify(self, tool_name: str, arguments: dict[str, Any] | None) -> None:
+        """验证工具（read/grep/test）命中 pending 路径时清除依赖。"""
+        if tool_name not in self._VERIFY_TOOLS:
+            return
+        with self._lock:
+            for path in _extract_paths(arguments):
+                self._pending.pop(path, None)
+
+    def check(self, tool_name: str, arguments: dict[str, Any] | None) -> str | None:
+        """bash/exec 执行前预检；返回 None=放行，str=提醒/阻断消息。"""
+        if tool_name not in self._SHELL_TOOLS:
+            return None
+        args = arguments or {}
+        command = str(args.get("command") or args.get("cmd") or "")
+        with self._lock:
+            hits = [path for path in self._pending if path and path in command]
+        if not hits:
+            return None
+        pending = ", ".join(hits[:3])
+        registrar = self._pending.get(hits[0], "?")
+        if self._hard_block:
+            return (
+                f"[shell_preflight] blocked tool execution pending dependent "
+                f"mutations (bash {tool_name}): {pending}"
+            )
+        return (
+            f"[shell_preflight] {tool_name} references unverified mutations "
+            f"({pending}) written by {registrar}. "
+            "Verify them first (read_file/grep/test) or proceed if intentional."
+        )
+
+    @property
+    def pending(self) -> dict[str, str]:
+        """挂起路径快照（只读）。"""
+        with self._lock:
+            return dict(self._pending)
+
+
+class DeliveryPolicyGates:
+    """配送策略门（``applyDeliveryPolicyGates`` port）。
+
+    REASONIX 语义：按配送策略检查工具调用，``remember`` 等记忆工具特判，消息
+    ``[delivery] tool %q rejected by delivery policy %q: %s (id=%d)``（第五阶段 §1.7）。
+    DEEPCODE 侧以 ``{工具 glob: 策略}`` 实现：策略为 "allow"/"deny" 或回调
+    ``Callable[[dict], bool]``（True=deny）。last-match-wins（与 PermissionEngine 一致）。
+    ``deny_memory_tools=True`` 时对记忆类工具特判。
+    """
+
+    _MEMORY_TOOLS = ("remember", "memorize", "memory_save", "memory_store")
+
+    def __init__(
+        self,
+        policies: dict[str, str | Callable[[dict[str, Any]], bool]] | None = None,
+        *,
+        deny_memory_tools: bool = False,
+    ) -> None:
+        self._policies = dict(policies or {})
+        self._deny_memory_tools = deny_memory_tools
+        self._lock = threading.Lock()
+        self._rejections: dict[str, int] = {}
+
+    def check(self, tool_name: str, arguments: dict[str, Any] | None) -> str | None:
+        args = arguments or {}
+        decision: str | None = None
+        matched: str | None = None
+        for pattern, policy in self._policies.items():
+            if fnmatch.fnmatch(tool_name, pattern):
+                matched = pattern
+                if callable(policy):
+                    decision = "deny" if policy(args) else "allow"
+                else:
+                    decision = policy
+        if self._deny_memory_tools and tool_name in self._MEMORY_TOOLS:
+            matched = "<memory-tools>"
+            decision = "deny"
+        if decision != "deny":
+            return None
+        with self._lock:
+            count = self._rejections.get(tool_name, 0) + 1
+            self._rejections[tool_name] = count
+        return (
+            f"[delivery] tool {tool_name!r} rejected by delivery policy "
+            f"{matched!r}: not allowed in the current delivery policy (id={count})"
+        )
+
+    @property
+    def rejections(self) -> dict[str, int]:
+        """各工具被拒次数（只读快照）。"""
+        with self._lock:
+            return dict(self._rejections)
+
+
+class RecoveryGate:
+    """恢复门（``applyRecoveryAndPermission`` 的 RecoveryGate 部分 port）。
+
+    REASONIX 语义：``recovery required before continuing: %s``（第五阶段 §1.7），
+    恢复未完成时阻断工具执行。DEEPCODE 侧通过 ``recovery_check`` 回调注入
+    "是否需要恢复"的判定（如子任务未收尾/审批未决），无回调 → 零成本。
+    """
+
+    def __init__(self, recovery_check: Callable[[], str | None] | None = None) -> None:
+        self._recovery_check = recovery_check
+
+    def check(self, tool_name: str, arguments: dict[str, Any] | None) -> str | None:
+        if self._recovery_check is None:
+            return None
+        need = self._recovery_check()
+        if need:
+            return f"recovery required before continuing: {need}"
+        return None
+
+
+class ToolResultMaintenanceView:
+    """工具结果维护视图（``applyToolResultMaintenanceView`` port）。
+
+    REASONIX 语义：基于 ``<evidence fingerprint>``（第五阶段 §1.7）对工具结果做
+    缓存指纹比对，维护"同一调用是否产生新结果"的视图。DEEPCODE 侧以
+    调用签名 → 结果指纹 的简单 dict 缓存实现（锁保护），``mark`` 返回是否为新结果。
+    """
+
+    def __init__(self) -> None:
+        self._cache: dict[str, str] = {}
+        self._lock = threading.Lock()
+
+    def fingerprint(self, tool_name: str, arguments: dict[str, Any] | None, result: Any) -> str:
+        """结果指纹 = 调用签名 + 结果哈希。"""
+        return (
+            f"{tool_call_signature(tool_name, arguments)}|"
+            f"{result_hash(result) or '<empty>'}"
+        )
+
+    def has_changed(self, tool_name: str, arguments: dict[str, Any] | None, result: Any) -> bool:
+        """是否与最近一次标记的结果不同。"""
+        key = tool_call_signature(tool_name, arguments)
+        fp = self.fingerprint(tool_name, arguments, result)
+        with self._lock:
+            return self._cache.get(key) != fp
+
+    def mark(self, tool_name: str, arguments: dict[str, Any] | None, result: Any) -> bool:
+        """标记结果指纹；返回 True=相对上次有新结果，False=完全重复。"""
+        key = tool_call_signature(tool_name, arguments)
+        fp = self.fingerprint(tool_name, arguments, result)
+        with self._lock:
+            changed = self._cache.get(key) != fp
+            self._cache[key] = fp
+        return changed
+
+
+class LoopGuards:
+    """REASONIX ``applyBatchGuards`` 的 Python 化组合：进度守卫 + 风暴熔断。
+
+    runner 主循环工具执行后调用 :meth:`observe_batch`，返回注入消息列表
+    （``{"role": "user", "content": ...}``），由调用方 append 进 messages。
+    """
+
+    def __init__(
+        self,
+        progress: ProgressGuard | None = None,
+        storm: StormBreaker | None = None,
+        contextual: ContextualToolGate | None = None,
+        mutation: MutationDependencyBarrier | None = None,
+        delivery: DeliveryPolicyGates | None = None,
+        recovery: RecoveryGate | None = None,
+        result_view: ToolResultMaintenanceView | None = None,
+    ) -> None:
+        self._progress = progress or ProgressGuard()
+        self._storm = storm or StormBreaker()
+        self._ledger = EvidenceLedger()
+        self._contextual = contextual or ContextualToolGate()
+        self._mutation = mutation or MutationDependencyBarrier()
+        self._delivery = delivery or DeliveryPolicyGates()
+        self._recovery = recovery or RecoveryGate()
+        self._result_view = result_view or ToolResultMaintenanceView()
+
+    @property
+    def progress(self) -> ProgressGuard:
+        return self._progress
+
+    @property
+    def storm(self) -> StormBreaker:
+        return self._storm
+
+    @property
+    def ledger(self) -> EvidenceLedger:
+        return self._ledger
+
+    @property
+    def blocked(self) -> bool:
+        """进度守卫熔断（N==6 强制收尾）——runner 以此短路后续工具执行。"""
+        return self._progress.blocked
+
+    @property
+    def streak(self) -> int:
+        return self._progress.streak
+
+    def observe_batch(
+        self,
+        tool_calls: Iterable[Any],
+        results: Iterable[Any],
+        events: Iterable[dict[str, str]],
+    ) -> list[dict[str, str]]:
+        """工具执行完成后调用，返回注入消息列表（可为空）。
+
+        ``tool_calls`` 元素需带 ``.name`` 与 ``.arguments``（或 dict 的 "name"/"arguments"）。
+        """
+        calls = [
+            (
+                getattr(tc, "name", None) or (tc.get("name") if isinstance(tc, dict) else None),
+                getattr(tc, "arguments", None) or (tc.get("arguments") if isinstance(tc, dict) else None),
+                result,
+            )
+            for tc, result in zip(tool_calls, results)
+        ]
+        injections: list[dict[str, str]] = []
+
+        score = self._ledger.score_round(calls)
+        intervention = self._progress.observe(score)
+        if intervention is not None:
+            injections.append({"role": "user", "content": intervention.message})
+
+        names = [c[0] for c in calls if c[0]]
+        failed = any((e.get("status") == "error") for e in events)
+        storm_message = self._storm.observe_batch(names, failed=failed)
+        if storm_message is not None:
+            injections.append({"role": "user", "content": storm_message})
+
+        return injections
+
+    def check_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> str | None:
+        """工具执行前的单工具守卫链（runner ``_run_tool`` 调用）。
+
+        链序：RecoveryGate（恢复未完成）→ ContextualToolGate（上下文可用性）→
+        MutationDependencyBarrier（bash 变更依赖预检）→ DeliveryPolicyGates
+        （配送策略）。返回 None=放行，str=阻断消息（errors-as-data）。
+        """
+        for gate in (self._recovery, self._contextual, self._mutation, self._delivery):
+            message = gate.check(tool_name, arguments)
+            if message is not None:
+                return message
+        return None
+
+    def observe_tool_mutation(
+        self, tool_name: str, arguments: dict[str, Any] | None
+    ) -> None:
+        """变更工具执行成功后登记待验证路径（MutationDependencyBarrier）。"""
+        self._mutation.observe_mutation(tool_name, arguments)
+
+    def observe_tool_verify(
+        self, tool_name: str, arguments: dict[str, Any] | None
+    ) -> None:
+        """验证工具执行成功后清除已验证路径（MutationDependencyBarrier）。"""
+        self._mutation.observe_verify(tool_name, arguments)
+
+    def observe_tool_result(
+        self, tool_name: str, arguments: dict[str, Any] | None, result: Any
+    ) -> bool:
+        """标记工具结果指纹（ToolResultMaintenanceView）；返回是否为新结果。"""
+        return self._result_view.mark(tool_name, arguments, result)
+
+
+__all__ = [
+    "SCORE_ZERO_EVIDENCE",
+    "SCORE_REPEATED_CALLS",
+    "SCORE_PARTIAL_EVIDENCE",
+    "SCORE_NEW_EVIDENCE",
+    "GUARD_NUDGE",
+    "GUARD_STRATEGY",
+    "GUARD_FORCE_ANSWER",
+    "DELEGATION_RESEARCH_HINTS",
+    "DELEGATION_CONTEXT_REFERENCE_HINTS",
+    "ToolEvidence",
+    "EvidenceLedger",
+    "GuardIntervention",
+    "ProgressGuard",
+    "StormBreaker",
+    "delegation_admission",
+    "LoopGuards",
+    "tool_call_signature",
+    "result_hash",
+    "ContextualToolGate",
+    "MutationDependencyBarrier",
+    "DeliveryPolicyGates",
+    "RecoveryGate",
+    "ToolResultMaintenanceView",
+]
diff --git a/tests/test_guard_telemetry.py b/tests/test_guard_telemetry.py
new file mode 100644
index 00000000..e1bcf4c4
--- /dev/null
+++ b/tests/test_guard_telemetry.py
@@ -0,0 +1,221 @@
+"""Tests for the P5 guard telemetry adapter (guard_event_callback → deepcode-telemetry).
+
+Covers the optional telemetry wiring implemented in
+:mod:`core.loop.guard_telemetry`:
+
+- Event mapping: ``blocked`` / ``injection`` / ``tool_block`` events drive the
+  correct counters/metrics on the telemetry engine.
+- Silent degradation: when the telemetry skill is unavailable or import fails,
+  ``make_telemetry_guard_callback`` returns ``None`` and nothing breaks.
+- Environment gate: telemetry is only assembled when
+  ``DEEPCODE_GUARD_TELEMETRY=1`` and no explicit callback is set.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.loop.guard_telemetry import (  # noqa: E402
+    ENV_GUARD_TELEMETRY,
+    install_guard_telemetry,
+    make_telemetry_guard_callback,
+)
+
+
+class _FakeEngine:
+    """Minimal fake telemetry engine recording counter/metric calls."""
+
+    def __init__(self) -> None:
+        self.counters: list[tuple[str, float]] = []
+        self.metrics: list[tuple[str, float]] = []
+
+    def increment_counter(self, name: str, delta: float = 1.0) -> None:
+        self.counters.append((name, delta))
+
+    def record_metric(self, name: str, value: float, unit: str = "") -> None:
+        self.metrics.append((name, value))
+
+
+def _fake_module(engine: _FakeEngine):
+    """Build a fake telemetry module exposing get_telemetry()."""
+    import types
+
+    mod = types.ModuleType("_fake_telemetry")
+    mod.get_telemetry = lambda: engine
+    return mod
+
+
+# -- 事件映射 ------------------------------------------------------------------
+
+
+def test_blocked_event_maps_to_counter_and_streak_metric(monkeypatch):
+    engine = _FakeEngine()
+    module = _fake_module(engine)
+    monkeypatch.setattr(
+        "core.loop.guard_telemetry._load_telemetry_module", lambda: module
+    )
+    monkeypatch.setattr("core.loop.guard_telemetry._attempted", False)
+    monkeypatch.setattr("core.loop.guard_telemetry._callbacks_cache", None)
+
+    callback = make_telemetry_guard_callback()
+    assert callback is not None
+    callback({"kind": "blocked", "tool": None, "level": 3, "streak": 6,
+              "message": "no new evidence"})
+
+    assert engine.counters == [("guard.blocked", 1.0)]
+    assert engine.metrics == [("guard.blocked.streak", 6.0)]
+
+
+def test_injection_event_maps_to_counter(monkeypatch):
+    engine = _FakeEngine()
+    module = _fake_module(engine)
+    monkeypatch.setattr(
+        "core.loop.guard_telemetry._load_telemetry_module", lambda: module
+    )
+    monkeypatch.setattr("core.loop.guard_telemetry._attempted", False)
+    monkeypatch.setattr("core.loop.guard_telemetry._callbacks_cache", None)
+
+    callback = make_telemetry_guard_callback()
+    assert callback is not None
+    callback({"kind": "injection", "message": "nudge"})
+
+    assert engine.counters == [("guard.injections", 1.0)]
+    assert engine.metrics == []
+
+
+def test_tool_block_event_maps_to_general_and_per_tool_counters(monkeypatch):
+    engine = _FakeEngine()
+    module = _fake_module(engine)
+    monkeypatch.setattr(
+        "core.loop.guard_telemetry._load_telemetry_module", lambda: module
+    )
+    monkeypatch.setattr("core.loop.guard_telemetry._attempted", False)
+    monkeypatch.setattr("core.loop.guard_telemetry._callbacks_cache", None)
+
+    callback = make_telemetry_guard_callback()
+    assert callback is not None
+    callback({"kind": "tool_block", "tool": "remember", "message": "denied"})
+
+    assert engine.counters == [
+        ("guard.tool_block", 1.0),
+        ("guard.tool_block.remember", 1.0),
+    ]
+    assert engine.metrics == []
+
+
+# -- 静默降级 ------------------------------------------------------------------
+
+
+def test_missing_skill_returns_none(monkeypatch):
+    monkeypatch.setattr(
+        "core.loop.guard_telemetry._load_telemetry_module", lambda: None
+    )
+    monkeypatch.setattr("core.loop.guard_telemetry._attempted", False)
+    monkeypatch.setattr("core.loop.guard_telemetry._callbacks_cache", None)
+
+    assert make_telemetry_guard_callback() is None
+
+
+def test_module_without_get_telemetry_returns_none(monkeypatch):
+    import types
+
+    mod = types.ModuleType("_fake_telemetry_no_getter")
+    monkeypatch.setattr(
+        "core.loop.guard_telemetry._load_telemetry_module", lambda: mod
+    )
+    monkeypatch.setattr("core.loop.guard_telemetry._attempted", False)
+    monkeypatch.setattr("core.loop.guard_telemetry._callbacks_cache", None)
+
+    assert make_telemetry_guard_callback() is None
+
+
+def test_telemetry_failure_silently_degrades(monkeypatch):
+    """遥测调用抛异常时不冒泡，绝不影响 agent 主循环。"""
+    engine = _FakeEngine()
+
+    def boom(*args, **kwargs):
+        raise RuntimeError("telemetry down")
+
+    engine.increment_counter = boom
+    module = _fake_module(engine)
+    monkeypatch.setattr(
+        "core.loop.guard_telemetry._load_telemetry_module", lambda: module
+    )
+    monkeypatch.setattr("core.loop.guard_telemetry._attempted", False)
+    monkeypatch.setattr("core.loop.guard_telemetry._callbacks_cache", None)
+
+    callback = make_telemetry_guard_callback()
+    assert callback is not None
+    callback({"kind": "blocked", "streak": 6, "message": "x"})  # 不应抛异常
+
+
+# -- 环境变量开关 ---------------------------------------------------------------
+
+
+def test_install_skips_without_env(monkeypatch):
+    monkeypatch.delenv(ENV_GUARD_TELEMETRY, raising=False)
+
+    class _Spec:
+        guard_event_callback = None
+
+    spec = _Spec()
+    install_guard_telemetry(spec)
+    assert spec.guard_event_callback is None
+
+
+def test_install_skips_when_env_not_one(monkeypatch):
+    monkeypatch.setenv(ENV_GUARD_TELEMETRY, "0")
+    monkeypatch.setattr(
+        "core.loop.guard_telemetry.make_telemetry_guard_callback",
+        lambda: lambda e: None,
+    )
+
+    class _Spec:
+        guard_event_callback = None
+
+    spec = _Spec()
+    install_guard_telemetry(spec)
+    assert spec.guard_event_callback is None
+
+
+def test_install_assembles_when_env_enabled(monkeypatch):
+    monkeypatch.setenv(ENV_GUARD_TELEMETRY, "1")
+
+    def fake_callback(event):
+        return None
+
+    monkeypatch.setattr(
+        "core.loop.guard_telemetry.make_telemetry_guard_callback",
+        lambda: fake_callback,
+    )
+
+    class _Spec:
+        guard_event_callback = None
+
+    spec = _Spec()
+    install_guard_telemetry(spec)
+    assert spec.guard_event_callback is fake_callback
+
+
+def test_install_never_overrides_existing_callback(monkeypatch):
+    monkeypatch.setenv(ENV_GUARD_TELEMETRY, "1")
+
+    def existing(event):
+        return None
+
+    class _Spec:
+        def __init__(self) -> None:
+            # 实例属性（而非类属性）：避免 Python 描述符协议把函数包装成
+            # bound method，导致 `is` 比较失败（每次访问都新建绑定对象）。
+            self.guard_event_callback = existing
+
+    spec = _Spec()
+    install_guard_telemetry(spec)
+    assert spec.guard_event_callback is existing
diff --git a/tests/test_guards.py b/tests/test_guards.py
new file mode 100644
index 00000000..4e6cfcb4
--- /dev/null
+++ b/tests/test_guards.py
@@ -0,0 +1,366 @@
+"""Tests for the P3.5 loop guards — REASONIX 反漫游守卫的 DEEPCODE 落地.
+
+Covers the four guard components ported from REASONIX §5/§6
+(``scripts/REASONIX_第五阶段深度分析报告.md`` / ``REASONIX_第六阶段深度分析报告.md``):
+
+- :class:`~core.loop.guards.EvidenceLedger` — per-round evidence scoring (0-3)
+- :class:`~core.loop.guards.ProgressGuard` — consecutive zero-progress rounds,
+  escalating at thresholds 2/4/6 (nudge → strategy change → force answer + block)
+- :class:`~core.loop.guards.StormBreaker` — tool-sequence signature circuit breaker
+- :func:`~core.loop.guards.delegation_admission` — spawn_agent admission gate
+- :class:`~core.loop.guards.LoopGuards` — the runner-facing integration entry
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.loop.guards import (  # noqa: E402
+    SCORE_NEW_EVIDENCE,
+    SCORE_PARTIAL_EVIDENCE,
+    SCORE_REPEATED_CALLS,
+    SCORE_ZERO_EVIDENCE,
+    ContextualToolGate,
+    DeliveryPolicyGates,
+    EvidenceLedger,
+    LoopGuards,
+    MutationDependencyBarrier,
+    ProgressGuard,
+    RecoveryGate,
+    StormBreaker,
+    ToolResultMaintenanceView,
+    delegation_admission,
+)
+
+
+def _tool(name: str, arguments: dict | None):
+    """Build a minimal object with .name / .arguments (like ToolCallRequest)."""
+    return type("TC", (), {"name": name, "arguments": arguments})()
+
+
+def _batch():
+    """Two tool calls with stable results — used for repeated-round tests."""
+    return (
+        [_tool("read_file", {"path": "x.py"}), _tool("grep", {"q": "foo"})],
+        ["content-1", "hits"],
+        [{"name": "read_file", "status": "ok"}, {"name": "grep", "status": "ok"}],
+    )
+
+
+# -- EvidenceLedger ------------------------------------------------------------
+
+
+def test_score_round_four_buckets():
+    """New call → new evidence; identical → repeated; same call new result → partial."""
+    ledger = EvidenceLedger()
+    scores = [
+        ledger.score_round([("read_file", {"path": "x.py"}, "content-1")]),  # 全新调用
+        ledger.score_round([("read_file", {"path": "x.py"}, "content-1")]),  # 完全相同
+        ledger.score_round([("read_file", {"path": "x.py"}, "content-CHANGED")]),  # 新结果
+        ledger.score_round([("grep", {"q": "foo"}, "hits")]),  # 全新调用
+        ledger.score_round([]),  # 空轮
+    ]
+    assert scores == [
+        SCORE_NEW_EVIDENCE,
+        SCORE_REPEATED_CALLS,
+        SCORE_PARTIAL_EVIDENCE,
+        SCORE_NEW_EVIDENCE,
+        SCORE_ZERO_EVIDENCE,
+    ]
+
+
+def test_score_round_compare_before_record():
+    """Regression: scoring must not pollute the seen-sets it is judged against."""
+    ledger = EvidenceLedger()
+    assert (
+        ledger.score_round([("read_file", {"path": "x.py"}, "c")])
+        == SCORE_NEW_EVIDENCE
+    )
+    assert (
+        ledger.score_round([("read_file", {"path": "x.py"}, "c")])
+        == SCORE_REPEATED_CALLS
+    )
+
+
+def test_record_tracks_rounds():
+    ledger = EvidenceLedger()
+    ledger.record([("ls", None, "out")])
+    ledger.record([("ls", None, "out")])
+    assert ledger.rounds == 2
+
+
+# -- ProgressGuard -------------------------------------------------------------
+
+
+def test_progress_guard_escalation_2_4_6():
+    """Consecutive zero-evidence rounds escalate: streak 2→nudge, 4→strategy, 6→force."""
+    pg = ProgressGuard()
+    seq = []
+    for _ in range(6):
+        iv = pg.observe(SCORE_REPEATED_CALLS)
+        seq.append(iv.level if iv else 0)
+    assert seq == [0, 1, 0, 2, 0, 3], seq  # 只在 2/4/6 触发，级别 1/2/3
+    assert pg.blocked is True and pg.streak == 6
+
+
+def test_progress_guard_streak_resets_on_new_evidence():
+    pg = ProgressGuard()
+    pg.observe(SCORE_REPEATED_CALLS)
+    pg.observe(SCORE_REPEATED_CALLS)
+    iv = pg.observe(SCORE_NEW_EVIDENCE)
+    assert iv is None and pg.streak == 0 and pg.blocked is False
+
+
+# -- StormBreaker --------------------------------------------------------------
+
+
+def test_storm_breaker_trips_after_three_failures_once():
+    sb = StormBreaker()
+    seen = []
+    for _ in range(5):
+        msg = sb.observe_batch(["bash", "grep"], failed=True)
+        seen.append(msg is not None)
+    assert seen == [False, False, False, True, False], seen  # 只在第 4 次宣布一次
+    assert sb.blocked is True
+
+
+def test_storm_breaker_success_resets_failure_count():
+    sb = StormBreaker()
+    sb.observe_batch(["bash"], failed=True)
+    sb.observe_batch(["bash"], failed=False)  # 成功批次重置
+    sb.observe_batch(["bash"], failed=True)
+    assert sb.blocked is False
+
+
+# -- delegation_admission ------------------------------------------------------
+
+
+def test_delegation_admission_branches():
+    assert delegation_admission("请调研一下这个库", fork_turns="none") == (
+        "allow",
+        "user_requested_research",
+    )
+    assert delegation_admission("看下 https://example.com 的文档", fork_turns="none") == (
+        "allow",
+        "external_source_cited",
+    )
+    assert delegation_admission("按上面的讨论实现", fork_turns="none") == (
+        "deny",
+        "task_references_parent_context",
+    )
+    assert delegation_admission("按上面的讨论实现", fork_turns="all") == (
+        "allow",
+        "context_inherited",
+    )
+    assert delegation_admission("实现一个排序算法", fork_turns="none") == (
+        "allow",
+        "self_contained",
+    )
+    assert delegation_admission("", fork_turns="none") == ("deny", "empty_task")
+
+
+# -- LoopGuards.observe_batch 集成 ---------------------------------------------
+
+
+def test_observe_batch_nudge_fires_on_third_repeated_round():
+    """REASONIX 语义：nudge 在 streak==2（连续 2 轮零进展）触发，即第 3 轮。"""
+    lg = LoopGuards()
+    injs1 = lg.observe_batch(*_batch())  # 新证据
+    assert injs1 == [], injs1
+    injs2 = lg.observe_batch(*_batch())  # 首次重复 → streak=1，无注入
+    assert injs2 == [] and lg.streak == 1, (injs2, lg.streak)
+    injs3 = lg.observe_batch(*_batch())  # 二次重复 → streak=2，nudge 注入
+    assert len(injs3) == 1 and injs3[0]["role"] == "user" and lg.streak == 2, (
+        injs3,
+        lg.streak,
+    )
+    assert "[progress guard]" in injs3[0]["content"]
+
+
+def test_observe_batch_storm_injection():
+    lg = LoopGuards()
+    calls, results, _ = _batch()
+    events = [{"name": "bash", "status": "error"}] * 4
+    messages: list[dict[str, str]] = []
+    for _ in range(4):
+        messages.extend(lg.observe_batch(calls, results, events))
+    storm_msgs = [m for m in messages if "storm" in m["content"] or "loop guard" in m["content"]]
+    assert len(storm_msgs) == 1  # 熔断只宣布一次
+
+
+def test_guards_blocked_short_circuits():
+    lg = LoopGuards()
+    # 第 1 轮是新证据（streak 清零）；从第 2 轮起连续 6 轮重复 → streak=6 → 熔断
+    for _ in range(7):
+        lg.observe_batch(*_batch())
+    assert lg.blocked is True
+    assert lg.streak == 6
+
+
+# -- ContextualToolGate --------------------------------------------------------
+
+
+def test_contextual_tool_gate_default_allows_everything():
+    gate = ContextualToolGate()
+    assert gate.check("bash", {"command": "ls"}) is None
+    assert gate.blocks == {}
+
+
+def test_contextual_tool_gate_blocks_on_matcher():
+    # 匹配器对 "bash" 返回非 None → 阻断；其他工具放行
+    def deny_bash(args: dict) -> str | None:
+        command = str(args.get("command") or "")
+        if "rm -rf" in command:
+            return "destructive command not allowed in this context"
+        return None
+
+    gate = ContextualToolGate(matchers={"bash*": deny_bash})
+    assert gate.check("bash", {"command": "ls"}) is None
+    message = gate.check("bash", {"command": "rm -rf /tmp/x"})
+    assert message is not None
+    assert "blocked: tool 'bash' is unavailable" in message
+    assert "destructive command" in message
+    assert gate.blocks == {"bash": 1}
+
+
+# -- MutationDependencyBarrier -------------------------------------------------
+
+
+def test_mutation_barrier_soft_reminder_default():
+    barrier = MutationDependencyBarrier()  # hard_block=False（默认软提醒）
+    barrier.observe_mutation("write_file", {"path": "src/main.py"})
+    assert barrier.pending == {"src/main.py": "write_file"}
+    message = barrier.check("bash", {"command": "python src/main.py"})
+    assert message is not None
+    assert "[shell_preflight]" in message
+    assert "unverified mutations" in message
+    assert "src/main.py" in message
+
+
+def test_mutation_barrier_hard_block():
+    barrier = MutationDependencyBarrier(hard_block=True)
+    barrier.observe_mutation("edit_file", {"path": "config.yaml"})
+    message = barrier.check("bash", {"command": "cat config.yaml"})
+    assert message is not None
+    assert "[shell_preflight] blocked tool execution pending dependent mutations" in message
+
+
+def test_mutation_barrier_verify_clears_pending():
+    barrier = MutationDependencyBarrier()
+    barrier.observe_mutation("write_file", {"path": "src/main.py"})
+    barrier.observe_verify("read_file", {"path": "src/main.py"})
+    assert barrier.pending == {}
+    assert barrier.check("bash", {"command": "python src/main.py"}) is None
+
+
+def test_mutation_barrier_non_shell_untouched():
+    barrier = MutationDependencyBarrier()
+    barrier.observe_mutation("write_file", {"path": "a.py"})
+    assert barrier.check("read_file", {"path": "a.py"}) is None  # 非 shell 工具不预检
+
+
+# -- DeliveryPolicyGates -------------------------------------------------------
+
+
+def test_delivery_policy_last_match_wins():
+    gates = DeliveryPolicyGates(
+        policies={"*": "allow", "remember": "deny", "web_*": "deny"}
+    )
+    assert gates.check("read_file", {}) is None
+    message = gates.check("remember", {"text": "x"})
+    assert message is not None
+    assert "[delivery] tool 'remember' rejected" in message
+    assert "id=1" in message
+    # last-match-wins：web_search 命中 "*"（allow）与 "web_*"（deny），后者生效
+    message2 = gates.check("web_search", {"q": "x"})
+    assert message2 is not None
+    assert gates.rejections == {"remember": 1, "web_search": 1}
+
+
+def test_delivery_policy_deny_memory_tools():
+    gates = DeliveryPolicyGates(deny_memory_tools=True)
+    message = gates.check("remember", {"text": "x"})
+    assert message is not None
+    assert "<memory-tools>" in message
+    assert gates.check("read_file", {}) is None
+
+
+def test_delivery_policy_callable():
+    def deny_big(args: dict) -> bool:
+        return len(str(args.get("text") or "")) > 10
+
+    gates = DeliveryPolicyGates(policies={"memorize": deny_big})
+    assert gates.check("memorize", {"text": "short"}) is None
+    assert gates.check("memorize", {"text": "this is a long text"}) is not None
+
+
+# -- RecoveryGate --------------------------------------------------------------
+
+
+def test_recovery_gate_noop_without_callback():
+    gate = RecoveryGate()
+    assert gate.check("bash", {}) is None
+
+
+def test_recovery_gate_blocks_when_recovery_needed():
+    gate = RecoveryGate(recovery_check=lambda: "subagent not finalized")
+    message = gate.check("bash", {"command": "ls"})
+    assert message == "recovery required before continuing: subagent not finalized"
+
+
+# -- ToolResultMaintenanceView -------------------------------------------------
+
+
+def test_result_view_mark_detects_new_results():
+    view = ToolResultMaintenanceView()
+    assert view.mark("read_file", {"path": "x.py"}, "content-1") is True
+    assert view.mark("read_file", {"path": "x.py"}, "content-1") is False  # 完全重复
+    assert view.mark("read_file", {"path": "x.py"}, "content-CHANGED") is True  # 新结果
+    assert view.has_changed("read_file", {"path": "x.py"}, "content-2") is True
+
+
+def test_result_view_fingerprint_includes_arguments():
+    view = ToolResultMaintenanceView()
+    view.mark("grep", {"q": "foo"}, "hits-a")
+    # 相同结果但不同参数 → 视为新调用
+    assert view.mark("grep", {"q": "bar"}, "hits-a") is True
+
+
+# -- LoopGuards.check_tool 链 --------------------------------------------------
+
+
+def test_check_tool_chains_all_gates():
+    lg = LoopGuards(
+        recovery=RecoveryGate(recovery_check=lambda: "pending approval"),
+        contextual=ContextualToolGate(),
+        mutation=MutationDependencyBarrier(),
+        delivery=DeliveryPolicyGates(policies={"remember": "deny"}),
+    )
+    # RecoveryGate 最先触发
+    assert lg.check_tool("bash", {}) == "recovery required before continuing: pending approval"
+
+
+def test_check_tool_delivery_gate_after_recovery_ok():
+    lg = LoopGuards(delivery=DeliveryPolicyGates(policies={"remember": "deny"}))
+    assert lg.check_tool("read_file", {}) is None
+    message = lg.check_tool("remember", {"text": "x"})
+    assert message is not None
+    assert "[delivery]" in message
+
+
+def test_observe_tool_mutation_and_result_through_loop_guards():
+    lg = LoopGuards()
+    lg.observe_tool_result("read_file", {"path": "x.py"}, "content-1")
+    assert lg.observe_tool_result("read_file", {"path": "x.py"}, "content-1") is False
+    lg.observe_tool_mutation("write_file", {"path": "y.py"})
+    assert lg._mutation.pending == {"y.py": "write_file"}  # 内部状态可达（测试专用）
+    message = lg.check_tool("bash", {"command": "run y.py"})
+    assert message is not None
+    assert "[shell_preflight]" in message