Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 108 additions & 5 deletions core/agent_runtime/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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
)
Expand Down
8 changes: 8 additions & 0 deletions core/events/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down
17 changes: 17 additions & 0 deletions core/harness/tools/spawn_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down
41 changes: 39 additions & 2 deletions core/loop/__init__.py
Original file line number Diff line number Diff line change
@@ -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}")
112 changes: 112 additions & 0 deletions core/loop/guard_telemetry.py
Original file line number Diff line number Diff line change
@@ -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",
]
Loading
Loading