feat(agent): align execution guardrails with Hermes

This commit is contained in:
jxxghp
2026-09-30 18:49:52 +08:00
parent 810df1c0d9
commit fcd4cd1725
23 changed files with 2072 additions and 21 deletions

View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Nous Research
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

View File

@@ -0,0 +1 @@
"""Agent 每轮工具循环检测与受控停止契约。"""

View File

@@ -0,0 +1,56 @@
"""Hermes 回复结束纠偏;派生自 agent_runtime_helpers.py,MIT 许可见同目录。
保留短回复、尾部动作与语言上下文边界;中文尾部动作是同一规则的本地化。
"""
import re
_TRAILING_ACTION = re.compile(
r"(?:\blet me now\b|\bi(?:['\u2019])?ll now\b|\bi will now\b"
r"|\bnow i(?:['\u2019]ll| will)\b|\bnext[,:] i\b"
r"|(?:接下来|现在)[,,]?我(?:会|将|要)|我(?:现在|接下来)(?:会|将|要)|让我(?:现在|接下来))"
r"[^.!?\n。!?]{0,100}[.:。\u2026]?\s*$", re.IGNORECASE,
)
_ACK_FUTURE = re.compile(r"\b(i['’]ll|i will|let me|i can do that|i can help with that)\b")
_ACK_ACTIONS = ('look into', 'look at', 'inspect', 'scan', 'check', 'analyz', 'review', 'explore', 'read', 'open',
'run', 'test', 'fix', 'debug', 'search', 'find', 'walkthrough', 'report back', 'summarize')
_WORKSPACE = ('directory', 'current dir', 'cwd', 'repo', 'repository', 'codebase', 'project', 'folder',
'filesystem', 'file tree', 'files', 'path')
CONTINUATION_NUDGE = (
'[System: Continue now. Execute the required tool calls and only send your final answer '
'after completing the task.]'
)
FRAGMENT_NUDGE = (
'[System: Your previous message ended the turn with a fragment that is not a usable answer. '
'If the task is unfinished, continue it and then give the complete answer. If that fragment '
'WAS your complete answer, send it again exactly as before.]'
)
def trailing_continue_intent(text: str) -> bool:
"""只检测四百字符内回复的最后一百六十字符,不把中途描述当结束意图。"""
text = text.strip()
return bool(text and len(text) <= 400 and _TRAILING_ACTION.search(text[-160:]))
def intermediate_ack(text: str, user_text: str, *, tool_results: int) -> bool:
"""Codex 专属默认:无工具回执的短未来动作回复还须有工作区上下文。"""
text = text.strip().lower()
if tool_results or not text or len(text) > 1200 or not _ACK_FUTURE.search(text):
return False
if not any(marker in text for marker in _ACK_ACTIONS):
return False
user_text = user_text.lower()
return '/' in user_text or any(marker in user_text or marker in text for marker in _WORKSPACE)
def degenerate_final(text: str, user_text: str) -> bool:
"""只纠正异常标点开头或语言不符的极短碎片,正常数字、路径和中文短答不触发。"""
text = text.strip()
if not text or len(text) > 24 or text.endswith(('.', '!', '?', '。', '!', '?')):
return False
if text[0] in '?!,;:)]}' and len(text) > 1 and text[1].isalpha():
return True
if not any(ch.isalpha() for ch in text) or any(ch.isascii() and ch.isalnum() for ch in text):
return False
return not any(ch.isalpha() and not ch.isascii() for ch in user_text)

View File

@@ -0,0 +1,690 @@
"""Pure tool-call loop guardrail primitives.
Adapted from NousResearch/hermes-agent f42f579cf8bac4918ac9599bece71618afadd846.
Copyright (c) 2025 Nous Research; MIT license in LICENSE.hermes.
The controller is side-effect free: it tracks per-turn tool-call observations
and returns decisions. Runtime code decides whether a decision becomes warning
guidance, a synthetic tool result, or a controlled turn halt.
"""
from __future__ import annotations
import hashlib
import json
from collections import deque
from dataclasses import asdict, dataclass, field, fields
from typing import Any, Mapping
from app.agent.guardrails.results import file_mutation_result_landed, is_guardrail_refusal, safe_json_loads
IDEMPOTENT_TOOL_NAMES = frozenset({
"read_file", "search_files", "web_search", "web_extract", "session_search", "skill_view", "skills_list",
"browser_snapshot", "browser_console", "browser_get_images", "mcp_filesystem_read_file",
"mcp_filesystem_read_text_file", "mcp_filesystem_read_multiple_files", "mcp_filesystem_list_directory",
"mcp_filesystem_list_directory_with_sizes", "mcp_filesystem_directory_tree", "mcp_filesystem_get_file_info",
"mcp_filesystem_search_files",
})
MUTATING_TOOL_NAMES = frozenset({
"terminal", "execute_code", "write_file", "patch", "todo_list", "memory", "skill_manage",
"browser_click", "browser_type", "browser_press", "browser_scroll", "browser_navigate",
"send_message", "cronjob_manage", "delegate_task", "process_manage",
})
# Pollers: legitimately re-invoked with identical args; the identical-call NOTICE never fires.
STALL_GUARD_REPEATABLE_TOOLS = frozenset({"process_manage"})
_STALL_GUARD_REPEATABLE_SUFFIXES = ("_get_result", "_poll") # generated / MCP poller conventions
# Nth consecutive identical (tool, args, result) call that fires the notice; 3 tolerates one double-check.
STALL_GUARD_IDENTICAL_CALL_THRESHOLD = 3
# Repeating multi-call cycles (A,B,A,B,... with identical args AND results) defeat the
# consecutive streak above — every alternation resets it, so a model replaying the same
# 2–4 call batch each iteration ran to the budget unflagged (port of can1357/oh-my-pi#10521,
# which widened their loop guard from single-call turns to whole tool-call batches).
# Longest cycle period detected; laps reuse the streak thresholds (notice at
# STALL_GUARD_IDENTICAL_CALL_THRESHOLD laps, halt at no_progress_block_after laps).
_STALL_GUARD_MAX_CYCLE_PERIOD = 4
# History window: enough for block_after laps of the longest cycle plus slack.
_STALL_GUARD_CYCLE_HISTORY = 64
# From the 2nd byte-identical repeat the duplicate payload becomes a reference stub; smaller results
# aren't worth it, errors never are. The args preview keeps WHAT was called if compression evicts the original.
IDENTICAL_RESULT_STUB_MIN_CHARS = 512
_RESULT_STUB_ARGS_PREVIEW_CHARS = 120
# Tools whose "failure" is normal work output (red test run, empty grep, page timeout).
# same_tool_failure (DIFFERENT commands) never halts these; only an exact-args replay with
# no intervening change, or an identical-result streak, can.
FAILURE_TOLERANT_TOOL_NAMES = frozenset({
"terminal", "execute_code", "process_manage", "process", "browser_navigate", "web_extract",
})
# A successful call to one of these marks progress for every failing signature still counted
# this turn: the next retry is a new experiment (edit -> re-run), not a replay.
PROGRESS_RESET_TOOL_NAMES = frozenset({
"write_file", "patch", "terminal", "execute_code", "browser_click", "browser_type", "browser_press",
"browser_navigate", "process_manage", "process", "delegate_task", "send_message", "cronjob",
"cronjob_manage", "todo", "todo_list", "memory", "skill_manage",
})
_BOOL_FIELDS = ("warnings_enabled", "hard_stop_enabled", "non_interactive_hard_stop_enabled")
# Threshold field -> (nested section, nested key). The flat legacy key is the field name itself.
_THRESHOLD_SOURCES: dict[str, tuple[str, str]] = {
"exact_failure_warn_after": ("warn_after", "exact_failure"),
"same_tool_failure_warn_after": ("warn_after", "same_tool_failure"),
"no_progress_warn_after": ("warn_after", "idempotent_no_progress"),
"exact_failure_block_after": ("hard_stop_after", "exact_failure"),
"same_tool_failure_halt_after": ("hard_stop_after", "same_tool_failure"),
"no_progress_block_after": ("hard_stop_after", "idempotent_no_progress"),
}
# Per-turn caps on runaway-prone tools (counters reset in reset_for_turn).
_DEFAULT_MAX_WEB_SEARCHES_PER_TURN = 50
_DEFAULT_MAX_SUBAGENTS_PER_TURN = 50
# Interactive surfaces plus bounded supervised task loops (subagent stopped by its parent;
# api_server has a live client) doing real edit -> re-run work keep the warn-only default.
_ATTENDED_PLATFORMS = frozenset({"cli", "tui", "desktop", "acp", "subagent", "api_server"})
def is_stall_guard_repeatable(tool_name: str) -> bool:
"""Whether a tool is exempt from the identical-call loop notice."""
return tool_name in STALL_GUARD_REPEATABLE_TOOLS or tool_name.endswith(_STALL_GUARD_REPEATABLE_SUFFIXES)
def _is_non_interactive_platform(platform: str | None) -> bool:
"""True for gateway/cron sessions where tool loops are unattended."""
if not isinstance(platform, str) or not platform.strip():
return False
return platform.strip().lower() not in _ATTENDED_PLATFORMS
@dataclass(frozen=True)
class LoopCapConfig:
"""Per-turn hard ceilings on web_search calls / subagent spawns; count total calls (not
repeats), fire regardless of ``hard_stop_enabled``; ``0`` disables a cap."""
max_web_searches: int = _DEFAULT_MAX_WEB_SEARCHES_PER_TURN
max_subagents: int = _DEFAULT_MAX_SUBAGENTS_PER_TURN
@classmethod
def from_mapping(cls, data: Mapping[str, Any] | None) -> "LoopCapConfig":
"""Build config from the ``tool_loop_guardrails.loop_caps`` section."""
if not isinstance(data, Mapping):
return cls()
defaults = cls()
return cls(**{f.name: _int_at_least(data.get(f.name), getattr(defaults, f.name), 0) for f in fields(cls)})
@dataclass(frozen=True)
class ToolCallGuardrailConfig:
"""Thresholds for per-turn tool-call loop detection. Warnings never prevent execution; hard
stops are opt-in on interactive platforms, default on for unattended gateway/cron platforms."""
warnings_enabled: bool = True
hard_stop_enabled: bool = False
non_interactive_hard_stop_enabled: bool = True
exact_failure_warn_after: int = 2
exact_failure_block_after: int = 5
same_tool_failure_warn_after: int = 3
same_tool_failure_halt_after: int = 8
no_progress_warn_after: int = 2
no_progress_block_after: int = 5
idempotent_tools: frozenset[str] = field(default_factory=lambda: IDEMPOTENT_TOOL_NAMES)
mutating_tools: frozenset[str] = field(default_factory=lambda: MUTATING_TOOL_NAMES)
progress_tools: frozenset[str] = field(default_factory=lambda: PROGRESS_RESET_TOOL_NAMES)
loop_caps: LoopCapConfig = field(default_factory=LoopCapConfig)
@classmethod
def from_mapping(
cls, data: Mapping[str, Any] | None, *, platform: str | None = None,
) -> "ToolCallGuardrailConfig":
"""Build config from `tool_loop_guardrails`; nested ``warn_after`` / ``hard_stop_after`` win over flat legacy keys."""
if not isinstance(data, Mapping):
data = {}
d = cls()
flags: dict[str, Any] = {name: _as_bool(data.get(name), getattr(d, name)) for name in _BOOL_FIELDS}
if flags["non_interactive_hard_stop_enabled"] and _is_non_interactive_platform(platform):
flags["hard_stop_enabled"] = True
def threshold(name: str, section_name: str, key: str) -> int:
"""Resolve nested configuration before legacy flat threshold keys."""
section = data.get(section_name)
nested = section.get(key, data.get(name)) if isinstance(section, Mapping) else data.get(name)
return _int_at_least(nested, getattr(d, name), 1)
thresholds: dict[str, Any] = {name: threshold(name, *src) for name, src in _THRESHOLD_SOURCES.items()}
return cls(loop_caps=LoopCapConfig.from_mapping(data.get("loop_caps")), **flags, **thresholds)
@dataclass(frozen=True)
class IdenticalCallObservation:
"""``notice`` is appended after the result, ``stub`` replaces a byte-identical duplicate result;
``kind`` names the detector behind the notice (``identical_call_streak`` / ``identical_cycle``)."""
notice: str | None = None
stub: str | None = None
kind: str | None = None
@dataclass(frozen=True)
class ToolCallSignature:
"""Stable, non-reversible identity for a tool name plus canonical args."""
tool_name: str
args_hash: str
@classmethod
def from_call(cls, tool_name: str, args: Mapping[str, Any] | None) -> "ToolCallSignature":
"""Hash canonical arguments so metadata never exposes credentials."""
return cls(tool_name=tool_name, args_hash=_sha256(canonical_tool_args(args or {})))
def to_metadata(self) -> dict[str, str]:
"""Return public metadata without raw argument values."""
return asdict(self)
@dataclass(frozen=True)
class ToolGuardrailDecision:
"""Decision returned by the tool-call guardrail controller."""
action: str = "allow" # allow | warn | block | halt
code: str = "allow"
message: str = ""
tool_name: str = ""
count: int = 0
signature: ToolCallSignature | None = None
@property
def allows_execution(self) -> bool:
"""Warnings allow execution; block and halt do not."""
return self.action in {"allow", "warn"}
@property
def should_halt(self) -> bool:
"""Both block and halt end an unattended turn."""
return self.action in {"block", "halt"}
def to_metadata(self) -> dict[str, Any]:
"""Return decision metadata without raw tool arguments."""
data = asdict(self)
if data["signature"] is None:
del data["signature"]
return data
def _canonical_json(value: Any) -> str:
"""Serialize nested tool arguments deterministically."""
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
def canonical_tool_args(args: Mapping[str, Any]) -> str:
"""Return sorted compact JSON for parsed tool arguments."""
if not isinstance(args, Mapping):
raise TypeError(f"tool args must be a mapping, got {type(args).__name__}")
return _canonical_json(args)
def classify_tool_failure(tool_name: str, result: str | None) -> tuple[bool, str]:
"""Fallback classifier used only when callers don't pass ``failed``; mirrors
``agent.display._detect_tool_failure`` so the guardrail never disagrees with the CLI's ``[error]`` tag."""
if result is None or file_mutation_result_landed(tool_name, result):
return False, ""
# A harness REFUSAL of a redundant call (repeated identical read/search) carries
# ``"error"`` for the model's benefit -- exactly what the substring test below keys
# on -- but nothing failed; counting it lets the cheap refusal feed the streak that
# fires the next, harder one. Mirrored in ``agent.display._detect_tool_failure``.
if is_guardrail_refusal(result):
return False, ""
if tool_name == "terminal":
data = safe_json_loads(result)
exit_code = data.get("exit_code") if isinstance(data, dict) else None
return (True, f" [exit {exit_code}]") if exit_code is not None and exit_code != 0 else (False, "")
if tool_name == "memory":
data = safe_json_loads(result)
if isinstance(data, dict) and data.get("success") is False and "exceed the limit" in data.get("error", ""):
return True, " [full]"
lower = result[:500].lower()
return (True, " [error]") if '"error"' in lower or '"failed"' in lower or result.startswith("Error") else (False, "")
# Guardrail verdict text injected into the conversation, keyed by decision code.
# ``same_tool_failure_warning`` is built by _tool_failure_recovery_hint (tool-specific).
_DECISION_MESSAGES: dict[str, str] = {
"repeated_exact_failure_block": (
"Blocked {tool_name}: the same tool call failed {count} times with identical arguments. "
"Stop retrying it unchanged; change strategy or explain the blocker."
),
"idempotent_no_progress_block": (
"Blocked {tool_name}: this read-only call returned the same result {count} times. "
"Stop repeating it unchanged; use the result already provided or try a different query."
),
"same_tool_failure_halt": (
"Stopped {tool_name}: it failed {count} times this turn. "
"Stop retrying the same failing tool path and choose a different approach."
),
"repeated_exact_failure_warning": (
"{tool_name} has failed {count} times with identical arguments. This looks like a loop; "
"inspect the error and change strategy instead of retrying it unchanged."
),
"idempotent_no_progress_warning": (
"{tool_name} returned the same result {count} times. Use the result already provided "
"or change the query instead of repeating it unchanged."
),
"identical_call_streak_halt": (
"Stopped {tool_name}: the same call with identical arguments returned the same result "
"{count} times in a row. Stop repeating it unchanged; use the result already provided or change strategy."
),
"identical_cycle_halt": (
"Stopped {tool_name}: the same repeating cycle of tool calls (period {period}) with identical "
"arguments and identical results has run {count} times. Repeating the batch unchanged is not "
"progress; use the results already provided or change strategy."
),
"loop_web_search_cap": (
"Blocked web_search: this turn has already made {cap} web searches, the per-turn limit. "
"This looks like a runaway search loop. Work with the results you already have and give the user your answer."
),
"loop_subagent_cap": (
"Blocked delegate_task: this turn has already spawned {count} subagents (limit {cap}). "
"This looks like a runaway delegation loop. Finish the work with the results you have and answer the user."
),
}
_IDENTICAL_CALL_NOTICE = (
"[hermes note: this is the {ordinal} consecutive identical call to "
"{tool_name} with identical arguments returning the same result. "
"Do not repeat it — change arguments, use a different tool, or "
"proceed with what you have.]"
)
_IDENTICAL_CYCLE_NOTICE = (
"[hermes note: the last {count} rounds repeated the same cycle of {period} tool calls "
"(ending with {tool_name}) with identical arguments and identical results. "
"Do not repeat the batch — change arguments, use a different tool, or "
"proceed with what you have.]"
)
# tool -> (LoopCapConfig field, controller counter attribute, decision code)
_LOOP_CAPS: dict[str, tuple[str, str, str]] = {
"web_search": ("max_web_searches", "_turn_web_search_count", "loop_web_search_cap"),
"delegate_task": ("max_subagents", "_turn_subagent_count", "loop_subagent_cap"),
}
class ToolCallGuardrailController:
"""Per-turn controller for repeated failed/non-progressing tool calls."""
def __init__(self, config: ToolCallGuardrailConfig | None = None):
"""Bind thresholds and initialize per-turn observations."""
self.config = config or ToolCallGuardrailConfig()
self.reset_for_turn()
def reset_for_turn(self) -> None:
"""Discard prior-turn signatures, counters, references and halt decisions."""
self._exact_failure_counts: dict[ToolCallSignature, int] = {}
self._same_tool_failure_counts: dict[str, int] = {}
# signature -> a mutating call succeeded since its last failure
self._progress_since_failure: dict[ToolCallSignature, bool] = {}
self._no_progress: dict[ToolCallSignature, tuple[str, int]] = {}
self._halt_decision: ToolGuardrailDecision | None = None
# Identical-call streak: CONSECUTIVE identical (tool, args, result) calls; any different call or
# result resets it, so re-reads after edits and varied polling are never flagged.
# Identical-call loop-breaker state (agent.stall_guards): tracks the CONSECUTIVE streak of identical
# (tool, canonical args) calls whose results were also identical. Per-turn, like everything else
# here. NOTE: open PR #85352 (patrykkopycinski) tracks no-progress loops ACROSS turns via a
# detection window — a different mechanism from this per-turn consecutive streak. Coordinate future
# work there.
self._identical_streak_sig: ToolCallSignature | None = None
self._identical_streak_result_hash: str = ""
self._identical_streak_count: int = 0
self._identical_streak_first_call_id: str = ""
# Batch-cycle loop breaker (port of can1357/oh-my-pi#10521): sequence of
# (signature, result_hash, repeatable) for every observed call this turn, so a repeating
# multi-call cycle (A,B,A,B,...) is caught even though it resets the consecutive streak above.
self._call_history: deque[tuple[ToolCallSignature, str, bool]] = deque(maxlen=_STALL_GUARD_CYCLE_HISTORY)
# tool_call_id -> spillover path, so a stub referencing a persisted-output preview can't dangle.
self._persisted_result_paths: dict[str, str] = {}
self._turn_web_search_count = 0
self._turn_subagent_count = 0
@property
def halt_decision(self) -> ToolGuardrailDecision | None:
"""Expose the detector decision that requires a controlled stop."""
return self._halt_decision
@property
def first_result_call_id(self) -> str:
"""Expose the result reference so the host can detect context compaction eviction."""
return self._identical_streak_first_call_id
def replace_result_reference(self, tool_call_id: str) -> None:
"""After the original leaves active context, retain the next full fresh result as the reference."""
self._identical_streak_first_call_id = tool_call_id
def _decide(
self, action: str, code: str, tool_name: str, count: int, signature: ToolCallSignature,
*, message: str | None = None, **fmt: Any,
) -> ToolGuardrailDecision:
"""Build a warn/block/halt decision; block/halt is also recorded as the turn's halt decision."""
if message is None:
message = _DECISION_MESSAGES[code].format(tool_name=tool_name, count=count, **fmt)
decision = ToolGuardrailDecision(action, code, message, tool_name, count, signature)
if decision.should_halt:
self._halt_decision = decision
return decision
def before_call(self, tool_name: str, args: Mapping[str, Any] | None) -> ToolGuardrailDecision:
"""Enforce call caps and already-observed failure or read replay thresholds."""
args = _coerce_args(args)
signature = ToolCallSignature.from_call(tool_name, args)
allow = ToolGuardrailDecision(tool_name=tool_name, signature=signature)
# Loop caps apply regardless of hard_stop_enabled (which only governs the detector).
cap_block = self._check_loop_cap(tool_name, args, signature)
if cap_block is not None or not self.config.hard_stop_enabled:
return cap_block or allow
# A mutation since this call last failed makes the retry a new experiment.
exact_count = 0 if self._progress_since_failure.get(signature) else self._exact_failure_counts.get(signature, 0)
if exact_count >= self.config.exact_failure_block_after:
return self._decide("block", "repeated_exact_failure_block", tool_name, exact_count, signature)
record = self._no_progress.get(signature) if self._is_idempotent(tool_name) else None
if record is not None and record[1] >= self.config.no_progress_block_after:
return self._decide("block", "idempotent_no_progress_block", tool_name, record[1], signature)
return allow
def after_call(
self, tool_name: str, args: Mapping[str, Any] | None, result: str | None,
*, failed: bool | None = None,
) -> ToolGuardrailDecision:
"""Count observed failures and unchanged reads; successful mutations allow a new experiment."""
args = _coerce_args(args)
signature = ToolCallSignature.from_call(tool_name, args)
if failed is None:
failed, _ = classify_tool_failure(tool_name, result)
warnings = self.config.warnings_enabled
if failed:
# An identical failing call is only a REPLAY if nothing landed in between;
# a mutation since the last identical failure restarts the exact-args streak.
if self._progress_since_failure.pop(signature, False):
self._exact_failure_counts.pop(signature, None)
exact_count = self._exact_failure_counts[signature] = self._exact_failure_counts.get(signature, 0) + 1
same_count = self._same_tool_failure_counts[tool_name] = self._same_tool_failure_counts.get(tool_name, 0) + 1
self._no_progress.pop(signature, None)
# same_tool_failure counts DIFFERENT args on one tool; for failure-tolerant
# tools a run of distinct red commands is diagnosis, not a loop — warn, never halt.
if (
# Hard-stop widening (#89069 / #100849 bundle): the per-turn no-progress BLOCK above only
# covers tools in idempotent_tools, so a model replaying the same successful
# `terminal`/`skill_view` call with a byte-identical result ran until the iteration budget.
# The consecutive-identical streak is tool-agnostic; when hard stops are enabled, halt at
# the same idempotent_no_progress threshold. Pollers stay exempt (an unchanged poll is
# progress).
self.config.hard_stop_enabled
and tool_name not in FAILURE_TOLERANT_TOOL_NAMES
and same_count >= self.config.same_tool_failure_halt_after
):
return self._decide("halt", "same_tool_failure_halt", tool_name, same_count, signature)
if warnings and exact_count >= self.config.exact_failure_warn_after:
return self._decide("warn", "repeated_exact_failure_warning", tool_name, exact_count, signature)
if warnings and same_count >= self.config.same_tool_failure_warn_after:
return self._decide(
"warn", "same_tool_failure_warning", tool_name, same_count, signature,
message=_tool_failure_recovery_hint(tool_name, same_count),
)
return ToolGuardrailDecision(tool_name=tool_name, count=exact_count, signature=signature)
self._exact_failure_counts.pop(signature, None)
self._same_tool_failure_counts.pop(tool_name, None)
# A successful mutation is progress for every failing signature still counted
# this turn. Pure loops never mutate between attempts, so the replay detector keeps its teeth.
if tool_name in self.config.progress_tools or file_mutation_result_landed(tool_name, result):
self._progress_since_failure.update(dict.fromkeys(self._exact_failure_counts, True))
self._same_tool_failure_counts.clear()
if not self._is_idempotent(tool_name):
self._no_progress.pop(signature, None)
return ToolGuardrailDecision(tool_name=tool_name, signature=signature)
result_hash = _result_hash(result, tool_name)
previous = self._no_progress.get(signature)
repeat_count = previous[1] + 1 if previous is not None and previous[0] == result_hash else 1
self._no_progress[signature] = (result_hash, repeat_count)
if warnings and repeat_count >= self.config.no_progress_warn_after:
return self._decide("warn", "idempotent_no_progress_warning", tool_name, repeat_count, signature)
return ToolGuardrailDecision(tool_name=tool_name, count=repeat_count, signature=signature)
def _is_idempotent(self, tool_name: str) -> bool:
"""Only explicitly known non-mutating tools receive the read replay check."""
return tool_name not in self.config.mutating_tools and tool_name in self.config.idempotent_tools
def observe_call(
self, tool_name: str, args: Mapping[str, Any] | None, result: str | None,
*, tool_call_id: str = "", failed: bool = False,
) -> IdenticalCallObservation:
"""Track consecutive identical calls; return notice + dedupe stub info.
``notice`` fires from the threshold-th consecutive identical (tool, args, result) call
(observational, pollers exempt). ``stub`` replaces the CURRENT result from the 2nd byte-identical
repeat — the tool still executed, only the context representation is deduplicated, so polling
semantics survive; pollers are NOT exempt here since an unchanged poll is where it saves most.
"""
is_plain_str = isinstance(result, str)
signature = ToolCallSignature.from_call(tool_name, _coerce_args(args))
result_hash = _result_hash(result, tool_name) if is_plain_str else ""
if is_plain_str and (signature, result_hash) == (self._identical_streak_sig, self._identical_streak_result_hash):
self._identical_streak_count += 1
else:
# New streak; non-string (multimodal) results never form one.
self._identical_streak_sig = signature if is_plain_str else None
self._identical_streak_result_hash = result_hash
self._identical_streak_count = 1 if is_plain_str else 0
self._identical_streak_first_call_id = tool_call_id or ""
count = self._identical_streak_count
notice = kind = None
if not is_stall_guard_repeatable(tool_name) and count >= STALL_GUARD_IDENTICAL_CALL_THRESHOLD:
notice, kind = _IDENTICAL_CALL_NOTICE.format(ordinal=_ordinal(count), tool_name=tool_name), "identical_call_streak"
# The no-progress BLOCK in before_call only covers idempotent_tools; this streak
# is tool-agnostic, so with hard stops on, halt at the same threshold (a model
# replaying a successful `terminal` call otherwise runs to the budget).
if self.config.hard_stop_enabled and count >= self.config.no_progress_block_after and self._halt_decision is None:
self._decide("halt", "identical_call_streak_halt", tool_name, count, signature)
# Batch-cycle detection (oh-my-pi#10521): a repeating multi-call cycle resets the
# consecutive streak on every alternation, so check the call history for a period-p lap.
if is_plain_str:
self._call_history.append((signature, result_hash, is_stall_guard_repeatable(tool_name)))
else:
self._call_history.clear()
if notice is None and is_plain_str:
cycle = self._detect_identical_cycle()
if cycle is not None:
period, laps = cycle
notice, kind = _IDENTICAL_CYCLE_NOTICE.format(count=laps, period=period, tool_name=tool_name), "identical_cycle"
if self.config.hard_stop_enabled and laps >= self.config.no_progress_block_after and self._halt_decision is None:
self._decide("halt", "identical_cycle_halt", tool_name, laps, signature, period=period)
stub = None
if isinstance(result, str) and count >= 2 and not failed and len(result) >= IDENTICAL_RESULT_STUB_MIN_CHARS:
stub = self._build_result_reference_stub(tool_name, args)
return IdenticalCallObservation(notice=notice, stub=stub, kind=kind)
def _detect_identical_cycle(self) -> tuple[int, int] | None:
"""Detect a repeating identical-call cycle ending at the latest observed call.
Returns ``(period, laps)`` for the smallest period 2..max whose trailing laps
(identical signature AND result per position) reach the notice threshold, else None.
Period 1 is the consecutive streak's job. A cycle made ONLY of poller-exempt tools
is exempt (an unchanged poll loop is legitimate waiting); one non-exempt call in
the cycle keeps the guard armed, matching the single-call exemption semantics.
"""
history = self._call_history
for period in range(2, _STALL_GUARD_MAX_CYCLE_PERIOD + 1):
if len(history) < period * STALL_GUARD_IDENTICAL_CALL_THRESHOLD:
continue
laps = 1
# Count how many consecutive trailing laps equal the final lap.
while True:
base = len(history) - period * (laps + 1)
if base < 0:
break
lap_equal = all(
history[base + i][:2] == history[len(history) - period + i][:2]
for i in range(period)
)
if not lap_equal:
break
laps += 1
if laps >= STALL_GUARD_IDENTICAL_CALL_THRESHOLD:
tail = [history[len(history) - period + i] for i in range(period)]
if all(repeatable for _, _, repeatable in tail):
continue
# A constant sub-cycle would already have fired at a smaller period.
return period, laps
return None
def record_persisted_result(self, tool_call_id: str, file_path: str) -> None:
"""Remember the spillover path a persisted result was saved to."""
if tool_call_id and file_path:
self._persisted_result_paths[tool_call_id] = file_path
def _build_result_reference_stub(self, tool_name: str, args: Mapping[str, Any] | None) -> str:
"""Reference stub for a byte-identical duplicate result (tool + args preview)."""
args_preview = canonical_tool_args(_coerce_args(args))
if len(args_preview) > _RESULT_STUB_ARGS_PREVIEW_CHARS:
args_preview = args_preview[:_RESULT_STUB_ARGS_PREVIEW_CHARS] + "…"
first_id = self._identical_streak_first_call_id
ref = f" (tool_call_id {first_id})" if first_id else ""
stub = (
f"[hermes note: this result is byte-identical to the {tool_name} "
f"result earlier this turn{ref}. Refer to that result; it has not "
f"changed. Args: {args_preview}]"
)
spill_path = self._persisted_result_paths.get(first_id) if first_id else None
if spill_path:
stub += f"\n[The referenced result was persisted to: {spill_path} — page through it with read_file if you need the full content.]"
return stub
def _check_loop_cap(
self, tool_name: str, args: Mapping[str, Any], signature: ToolCallSignature,
) -> ToolGuardrailDecision | None:
"""Block once a per-turn cap is reached (BEFORE the call, so the (cap+1)-th is refused), else advance
the counter and return None. delegate_task control actions spawn nothing and keep working after the cap."""
spec = _LOOP_CAPS.get(tool_name)
if spec is None:
return None
cap_field, count_attr, code = spec
cap, count = getattr(self.config.loop_caps, cap_field), getattr(self, count_attr)
increment = 1 if tool_name == "web_search" else (_subagent_spawn_count(args) if cap else 0)
if increment and cap and count >= cap:
return self._decide("block", code, tool_name, count, signature, cap=cap)
setattr(self, count_attr, count + increment)
return None
def toolguard_synthetic_result(decision: ToolGuardrailDecision) -> str:
"""Build a synthetic role=tool content string for a blocked tool call."""
return json.dumps({"error": decision.message, "guardrail": decision.to_metadata()}, ensure_ascii=False)
def append_toolguard_guidance(result: str, decision: ToolGuardrailDecision) -> str:
"""Append runtime guidance to the current tool result content."""
if decision.action not in {"warn", "halt"} or not decision.message:
return result
label = "Tool loop hard stop" if decision.action == "halt" else "Tool loop warning"
return (result or "") + f"\n\n[{label}: {decision.code}; count={decision.count}; {decision.message}]"
def _tool_failure_recovery_hint(tool_name: str, count: int) -> str:
"""Action-oriented guidance for recovering from repeated tool failures."""
common = (
f"{tool_name} has failed {count} times this turn. This looks like a loop. "
"Do not switch to text-only replies; keep using tools, but diagnose before retrying. "
"First inspect the latest error/output and verify your assumptions. "
)
if tool_name == "terminal":
return common + (
"For terminal failures, run a small diagnostic such as `pwd && ls -la` "
"in the same tool, then try an absolute path, a simpler command, a different "
"working directory, or a different tool such as read_file/write_file/patch."
)
return common + (
"Try different arguments, a narrower query/path, an absolute path when relevant, "
"or a different tool that can make progress. If the blocker is external, report "
"the blocker after one diagnostic attempt instead of repeating the same failing path."
)
def _ordinal(count: int) -> str:
"""Format the count used in model-facing loop notices."""
return f"{count}{'th' if 11 <= count % 100 <= 13 else {1: 'st', 2: 'nd', 3: 'rd'}.get(count % 10, 'th')}"
def _coerce_args(args: Mapping[str, Any] | None) -> Mapping[str, Any]:
"""Treat non-object arguments as absent for detector bookkeeping."""
return args if isinstance(args, Mapping) else {}
# execute_code reports per-call kernel bookkeeping — a running `kernel.execution_count` and wall-clock
# `duration_seconds` — that changes on every invocation even when the code did the same thing. Hashed
# as-is, every replay of an identical empty probe looks new and the identical-call streak never forms
# (a model re-ran one empty execute_code call 147 times unflagged). Only these known locations in
# execute_code's own result shape are dropped: for any other tool the same key names can be real output.
def _without_execute_code_metadata(parsed: Any) -> Any:
"""Ignore only kernel counters and wall time, preserving meaningful code output."""
if not isinstance(parsed, dict):
return parsed
cleaned = {k: v for k, v in parsed.items() if k != "duration_seconds"}
kernel = cleaned.get("kernel")
if isinstance(kernel, dict):
cleaned["kernel"] = {k: v for k, v in kernel.items() if k != "execution_count"}
return cleaned
def _result_hash(result: str | None, tool_name: str = "") -> str:
"""Compare structured results canonically while preserving other tools data fields."""
parsed = safe_json_loads(result or "")
if parsed is None:
return _sha256(result or "")
if tool_name == "execute_code":
parsed = _without_execute_code_metadata(parsed)
return _sha256(_canonical_json(parsed))
_BOOL_WORDS = {w: True for w in ("1", "true", "yes", "on", "enabled")} | {w: False for w in ("0", "false", "no", "off", "disabled")}
def _as_bool(value: Any, default: bool) -> bool:
"""Parse explicit boolean spellings without treating false text as true."""
if isinstance(value, (bool, int, float)):
return bool(value)
if isinstance(value, str):
return _BOOL_WORDS.get(value.strip().lower(), default)
return default
def _int_at_least(value: Any, default: int, minimum: int) -> int:
"""junk/None/below-minimum fall back to default (caps use minimum 0 so 0 = disabled)."""
try:
parsed = int(value)
except (TypeError, ValueError):
return default
return parsed if parsed >= minimum else default
def _subagent_spawn_count(args: Mapping[str, Any]) -> int:
"""Subagents one delegate_task call spawns: ``len(tasks)`` for a non-empty batch, else 1; control actions 0."""
if str(args.get("action") or "").strip().lower() in ("list", "steer", "stop"):
return 0
tasks = args.get("tasks")
return len(tasks) if isinstance(tasks, list) and tasks else 1
def _sha256(value: str) -> str:
# surrogatepass: web-scraped results can carry unpaired UTF-16 surrogates; a
# strict encode would raise and take down the conversation loop.
"""Hash output even when external data contains unmatched Unicode surrogates."""
return hashlib.sha256(value.encode("utf-8", "surrogatepass")).hexdigest()

View File

@@ -0,0 +1,43 @@
"""Hermes 工具结果分类的纯 JSON 部分;MIT 来源见 controller.py 和 LICENSE.hermes。"""
import json
from typing import Any
FILE_MUTATING_TOOL_NAMES = frozenset({'write_file', 'patch'})
GUARDRAIL_REFUSAL_KEY = 'guardrail_refusal'
def safe_json_loads(value: str) -> Any:
"""无法解析的自然语言结果返回 None,不能让检测异常中断真实工具。"""
try:
return json.loads(value)
except (TypeError, ValueError, RecursionError):
return None
def is_guardrail_refusal(result: Any) -> bool:
"""Return True when ``result`` (JSON string or parsed dict) is a harness refusal."""
data = result
if isinstance(result, str):
try:
data = json.loads(result.strip())
except Exception:
return False
return isinstance(data, dict) and data.get(GUARDRAIL_REFUSAL_KEY) is True
def file_mutation_result_landed(tool_name: str, result: Any) -> bool:
"""Return True when a file mutation result proves the write landed."""
if tool_name not in FILE_MUTATING_TOOL_NAMES or not isinstance(result, str):
return False
try:
data = json.loads(result.strip())
except Exception:
return False
if not isinstance(data, dict) or data.get("error"):
return False
if tool_name == "write_file":
return "bytes_written" in data
if tool_name == "patch":
return data.get("success") is True
return False

View File

@@ -15,6 +15,7 @@ from app.agent.learning.review import ReviewLoop, ReviewSnapshot
from app.agent.learning.skills import SkillLibrary
from app.agent.learning.tools import LearningTools
from app.agent.tools.base import run_agent_blocking
from app.agent.tools.result import TOOL_OBSERVATION_MARKER
from app.application.messaging.interaction.agent import is_agent_learning_command
from app.runtime.log import logger
from app.runtime.tasks import get_task_registry
@@ -69,7 +70,8 @@ class LearningSession:
"""先让旧复盘退出,再推进真实用户轮次;恢复会话用历史 human 条数初始化。"""
await self.wait_cancelled()
if not self.initialized:
self.turns = max(0, sum(isinstance(message, HumanMessage) for message in messages) - 1) % 10
self.turns = max(0, sum(isinstance(message, HumanMessage) and not message.additional_kwargs.get(TOOL_OBSERVATION_MARKER)
for message in messages) - 1) % 10
self.initialized = True
self.turns = (self.turns + 1) % 10
self.memory_due = self.turns == 0

View File

@@ -0,0 +1,314 @@
"""把 Hermes 每轮循环检测接入真实工具回执,不把重复读取变成缓存命中。"""
import asyncio
import json
import re
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field, replace
from typing import Any
from langchain.agents.middleware.types import AgentMiddleware, ModelResponse, ToolCallRequest
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langgraph.config import get_config
from langgraph.graph.message import add_messages
from langgraph.types import Command
from app.agent.guardrails.completion import (
CONTINUATION_NUDGE,
FRAGMENT_NUDGE,
degenerate_final,
intermediate_ack,
trailing_continue_intent,
)
from app.agent.guardrails.controller import (
ToolCallGuardrailConfig,
ToolCallGuardrailController,
append_toolguard_guidance,
toolguard_synthetic_result,
)
from app.agent.policy.api import resolve_api_operation
from app.agent.policy.contracts import ActionEffect, AuthSource, ExecutionOutcome, ToolOrigin, ToolPolicyContext
from app.agent.tools.base import TOOL_RESULT_RECORDER
from app.agent.tools.impl.api import MoviePilotApiTool
from app.agent.tools.result import EXECUTION_OUTCOME_KEY, TOOL_OBSERVATION_MARKER, inspect_tool_result
TURN_STATUS_KEY = 'moviepilot_turn_status'
GUARDRAIL_KEY = 'moviepilot_guardrail'
GUARDRAIL_REPLY_PREFIX = '本轮未完成:'
_TOOL_ALIASES = {'task': 'delegate_task', 'update_plan': 'todo_list', 'get_tool_execution': 'process_manage',
'edit_file': 'patch', 'apply_patch': 'patch', 'search_web': 'web_search',
'send_local_file': 'send_message', 'send_voice_message': 'send_message'}
_BROWSER_ALIASES = {'goto': 'browser_navigate', 'open_tab': 'browser_navigate', 'focus_tab': 'browser_navigate',
'click': 'browser_click', 'click_ref': 'browser_click', 'fill': 'browser_type', 'fill_ref': 'browser_type',
'select': 'browser_click', 'select_ref': 'browser_click', 'snapshot': 'browser_snapshot',
'get_content': 'browser_snapshot', 'screenshot': 'browser_get_images', 'wait': 'process_manage'}
@dataclass
class CallObservation:
"""只在一个工具批次内保留原始回执,按模型发出顺序消费,避免并行完成乱序。"""
name: str
arguments: dict[str, Any]
content: str | None
outcome: ExecutionOutcome
@dataclass
class GuardRun:
"""并行子任务共享编译图但不共享本轮计数,临时结果不写进 checkpoint。"""
controller: ToolCallGuardrailController
pending: dict[str, CallObservation] = field(default_factory=dict)
outcomes: set[ExecutionOutcome] = field(default_factory=set)
continuations: int = 0
has_tools: bool = False
codex_mode: bool = False
message_updates: list[ToolMessage] = field(default_factory=list)
identity: object | None = None
class ToolGuardrailsMiddleware(AgentMiddleware): # type: ignore[misc]
"""交互式 Web/API 默认警告,消息网关及后台默认有界停止;新用户轮次重置。"""
def __init__(self, context: ToolPolicyContext) -> None:
"""入口来自宿主认证信息,模型不能切换有人值守或后台策略。"""
self.context = context
unattended = (context.origin not in {ToolOrigin.SUBAGENT, ToolOrigin.AGENT_API}
and (context.origin is ToolOrigin.BACKGROUND or context.auth_source is AuthSource.CHANNEL))
self.config = ToolCallGuardrailConfig(hard_stop_enabled=unattended)
self._runs: dict[str, GuardRun] = {}
@staticmethod
def _scope() -> str:
"""LangGraph 宿主配置提供线程身份,不能由模型工具参数选择。"""
try:
return str(get_config().get('configurable', {}).get('thread_id') or '')
except RuntimeError:
return ''
def _run(self) -> GuardRun:
"""按真实图线程隔离运行状态,异常遗留也不能使缓存无界增长。"""
scope = self._scope()
if scope not in self._runs:
if len(self._runs) >= 64:
raise RuntimeError('循环检测运行状态未释放,拒绝继续创建任务')
self._runs[scope] = GuardRun(ToolCallGuardrailController(self.config))
return self._runs[scope]
@property
def controller(self) -> ToolCallGuardrailController:
"""返回当前图线程自己的检测器。"""
return self._run().controller
@property
def pending(self) -> dict[str, CallObservation]:
"""原始结果只归属当前运行,不能被并行子任务消费。"""
return self._run().pending
@property
def outcomes(self) -> set[ExecutionOutcome]:
"""保存本轮实际观察到的结果状态,未知不推断为成功。"""
return self._run().outcomes
def release(self, thread_id: str) -> None:
"""宿主在取消或异常路径也释放该子任务的临时结果。"""
self._runs.pop(thread_id, None)
def _begin(self, runtime: Any) -> GuardRun:
"""以 LangGraph 每次运行的 control 身份重置,复用现有模型节点不增加预算开销。"""
run = self._run()
identity = getattr(runtime, 'control', None)
if run.identity is not identity:
run = GuardRun(ToolCallGuardrailController(self.config), identity=identity)
self._runs[self._scope()] = run
return run
async def awrap_model_call(self, request: Any, handler: Callable[[Any], Awaitable[Any]]) -> Any:
"""记录当前模型实际提供的工具及 Codex 协议,结束检查不能凭模型名称猜入口。"""
run = self._begin(request.runtime)
run.has_tools = bool(request.tools)
run.codex_mode = bool(getattr(request.model, 'use_responses_api', False)
and 'chatgpt.com/backend-api/codex' in str(getattr(request.model, 'openai_api_base', '') or ''))
prepared = await self.prepare_model({'messages': request.messages}, None)
if prepared:
updates = prepared['messages']
run.message_updates = [message for message in updates if isinstance(message, ToolMessage)]
if prepared.get('jump_to') == 'end':
return ModelResponse(result=[updates[-1]])
request = request.override(messages=add_messages(request.messages, updates))
try:
return await handler(request)
except BaseException:
self.release(self._scope())
raise
async def after_model_update(self, state: dict[str, Any], runtime: Any) -> dict[str, Any] | None:
"""登记请求;只有短尾部行动或 Codex 特定碎片才最多追加两次继续提示。"""
del runtime
messages = state.get('messages', [])
if not messages or not isinstance(messages[-1], AIMessage):
return None
active_ids = {message.id for message in messages}
updates = [message for message in self._run().message_updates if message.id in active_ids]
self._run().message_updates.clear()
message = messages[-1]
for call in message.tool_calls:
self.pending[call['id']] = CallObservation(call['name'], dict(call.get('args') or {}), None, ExecutionOutcome.UNKNOWN)
continuation = None if message.tool_calls or not self._run().has_tools else self._continue(messages, message)
if continuation:
if continuation.get('jump_to') != 'model':
self.release(self._scope())
return {**continuation, 'messages': [*updates, *continuation['messages']]}
if not message.tool_calls:
self.release(self._scope())
return {'messages': updates} if updates else None
def _continue(self, messages: list[Any], message: AIMessage) -> dict[str, Any] | None:
"""合成提示不进入真实用户历史;预算耗尽仍报告部分完成,不伪造工具成果。"""
run = self._run()
text = re.sub(r'<think>.*?(?:</think>|$)', '', message.text, flags=re.DOTALL).strip()
user_text = next((item.text for item in reversed(messages) if isinstance(item, HumanMessage)
and not item.additional_kwargs.get(TOOL_OBSERVATION_MARKER)), '')
tool_results = 0
for previous in reversed(messages[:-1]):
if isinstance(previous, HumanMessage):
break
tool_results += isinstance(previous, ToolMessage)
stalled = trailing_continue_intent(text)
fragment = run.codex_mode and tool_results > 0 and degenerate_final(text, user_text)
ack = run.codex_mode and intermediate_ack(text, user_text, tool_results=sum(isinstance(item, ToolMessage) for item in messages))
if not (stalled or fragment or ack):
return None
if run.continuations >= 2:
return {'messages': [message.model_copy(update={
'content': GUARDRAIL_REPLY_PREFIX + '模型仍只描述下一步,未给出完成结果。\n' + text,
'additional_kwargs': {**message.additional_kwargs, TURN_STATUS_KEY: 'partial',
GUARDRAIL_KEY: {'code': 'continuation_exhausted', 'count': 2}},
})]}
run.continuations += 1
nudge = HumanMessage(content=FRAGMENT_NUDGE if fragment and not stalled else CONTINUATION_NUDGE,
additional_kwargs={TOOL_OBSERVATION_MARKER: True})
return {'messages': [nudge], 'jump_to': 'model'}
def _name(self, request: ToolCallRequest) -> str:
"""映射宿主多动作工具,轮询豁免不能扩大到启动、写入或终止动作。"""
name, arguments = request.tool_call['name'], request.tool_call.get('args') or {}
if name == 'execute_command':
return 'process_manage' if arguments.get('action') in {'read', 'wait', 'status', 'list'} else 'terminal'
if name == 'subagent_task':
return 'process_manage' if arguments.get('action') in {'list', 'status', 'wait'} else 'delegate_task'
if name in _TOOL_ALIASES:
return _TOOL_ALIASES[name]
if name == 'browse_webpage':
return _BROWSER_ALIASES.get(str(arguments.get('action') or ''), str(name))
if name == 'agent_task' and arguments.get('action') in {'create', 'update', 'run', 'delete'}:
return 'cronjob_manage'
if isinstance(request.tool, MoviePilotApiTool):
operation_id = str(arguments.get('operation_id') or '')
operation = resolve_api_operation(operation_id)
name = f'{name}:{operation_id}'
if operation is not None:
config = self.controller.config
if operation.effect is ActionEffect.SAFE_READ:
self.controller.config = replace(config, idempotent_tools=config.idempotent_tools | {name})
elif operation.effect not in {ActionEffect.SENSITIVE_READ, ActionEffect.UNKNOWN}:
self.controller.config = replace(config, progress_tools=config.progress_tools | {name})
return str(name)
async def awrap_tool_call(self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], Awaitable[Any]]) -> Any:
"""执行前检查上轮证据和总预算,执行后暂存原文而不改变真实工具结果。"""
name = self._name(request)
arguments = dict(request.tool_call.get('args') or {})
if name == 'delegate_task' and arguments.get('action') in {'cancel', 'update'}:
arguments['action'] = {'cancel': 'stop', 'update': 'steer'}[arguments['action']]
decision = self.controller.before_call(name, arguments)
if not decision.allows_execution:
return ToolMessage(content=toolguard_synthetic_result(decision), tool_call_id=request.tool_call['id'],
name=request.tool_call['name'], status='error',
additional_kwargs={EXECUTION_OUTCOME_KEY: 'failed', GUARDRAIL_KEY: decision.to_metadata()})
self.pending[request.tool_call['id']] = CallObservation(name, arguments, None, ExecutionOutcome.UNKNOWN)
raw: dict[str, str] = {}
previous = TOOL_RESULT_RECORDER.get()
def record(tool_name: str, text: str) -> dict[str, Any]:
"""在通用截断前观察原文,同时保留外层归档器的精确引用。"""
raw[tool_name] = text
return previous(tool_name, text) if previous else {}
token = TOOL_RESULT_RECORDER.set(record)
try:
result = await handler(request)
except asyncio.CancelledError:
self.release(self._scope())
raise
finally:
TOOL_RESULT_RECORDER.reset(token)
messages = result.update.get('messages', []) if isinstance(result, Command) and isinstance(result.update, dict) else [result]
for message in messages:
if isinstance(message, ToolMessage) and message.tool_call_id == request.tool_call['id']:
content = raw.get(request.tool_call['name'], message.content)
outcome = inspect_tool_result(message)
self.pending[message.tool_call_id] = CallObservation(name, arguments, content if isinstance(content, str) else None, outcome)
self.outcomes.add(outcome)
return result
def _observe(self, message: ToolMessage, active_ids: set[str]) -> ToolMessage:
"""在结果进入下一次模型请求前附加提示;被压缩掉的引用不能继续产生存根。"""
item = self.pending.pop(message.tool_call_id)
if message.additional_kwargs.get(GUARDRAIL_KEY):
return message
if item.content is None:
item.content = message.content if isinstance(message.content, str) else None
item.outcome = inspect_tool_result(message)
self.outcomes.add(item.outcome)
failed = item.outcome in {ExecutionOutcome.FAILED, ExecutionOutcome.UNKNOWN}
decision = self.controller.after_call(item.name, item.arguments, item.content, failed=failed)
observation = self.controller.observe_call(item.name, item.arguments, item.content,
tool_call_id=message.tool_call_id, failed=failed)
if not isinstance(message.content, str):
return message
stub = observation.stub
if self.controller.first_result_call_id not in active_ids:
self.controller.replace_result_reference(message.tool_call_id)
stub = None
# 保留可分页的原始回执元数据;存根不隐藏 pending/unknown 等宿主状态。
content = f'{stub}\nexecution_outcome={item.outcome.value}' if stub else message.content
if stub:
content += self._fresh_reference(message.content)
content = append_toolguard_guidance(content, decision)
if observation.notice:
content += '\n\n' + observation.notice
return message.model_copy(update={'content': content})
@staticmethod
def _fresh_reference(content: str) -> str:
"""大结果缓存会淘汰旧引用,重复存根指向这次实际执行刚保存的正文。"""
try:
payload = json.loads(content)
except (TypeError, ValueError):
return ''
if isinstance(payload, dict) and payload.get('tool_result_truncated') and payload.get('result_id'):
return f"\nFull fresh result: read_tool_result(result_id={payload['result_id']!r}, offset=0)."
return ''
async def prepare_model(self, state: dict[str, Any], runtime: Any) -> dict[str, Any] | None:
"""批次结果按请求顺序判定短循环;触发停止时保留计划和未知副作用证据。"""
del runtime
messages = state.get('messages', [])
active_ids = {message.tool_call_id for message in messages if isinstance(message, ToolMessage)}
by_id = {message.tool_call_id: message for message in messages if isinstance(message, ToolMessage)}
updates = []
for message in messages:
if isinstance(message, AIMessage):
for call in message.tool_calls:
if call['id'] in self.pending and call['id'] in by_id:
updates.append(self._observe(by_id[call['id']], active_ids))
decision = self.controller.halt_decision
if decision:
outcome = 'unknown' if ExecutionOutcome.UNKNOWN in self.outcomes else 'waiting' if ExecutionOutcome.PENDING in self.outcomes else 'blocked'
text = (f'{GUARDRAIL_REPLY_PREFIX}在 {decision.tool_name} 上重复尝试且没有取得进展,已停止继续调用。'
'任务尚未完成,请依据最近回执调整方法;已执行的操作不会因停止而撤销。')
updates.append(AIMessage(content=text, additional_kwargs={TURN_STATUS_KEY: outcome, GUARDRAIL_KEY: decision.to_metadata()}))
return {'messages': updates, 'jump_to': 'end'}
return {'messages': updates} if updates else None

View File

@@ -3,7 +3,7 @@
import asyncio
import json
from collections.abc import Awaitable, Callable
from typing import Any
from typing import TYPE_CHECKING, Any
from langchain.agents.middleware import AgentMiddleware, ToolCallRequest, hook_config
from langchain_core.messages import AIMessage, ToolMessage
@@ -29,6 +29,9 @@ from app.agent.tools.impl.mcp import McpExternalTool
from app.agent.tools.result import EXECUTION_OUTCOME_KEY, ToolExecutionError, annotate_tool_result
from app.agent.tools.tags import ToolTag
if TYPE_CHECKING:
from app.agent.middleware.guardrails import ToolGuardrailsMiddleware
POLICY_DENIED_MESSAGE = "当前宿主策略不允许执行该工具。"
POLICY_UNAVAILABLE_MESSAGE = "宿主策略暂时不可用,未执行该工具。"
TOOL_TIMEOUT_MESSAGE = "工具执行超时,已停止等待结果;若工具包含外部写操作,操作可能仍在继续,请先确认实际状态再重试。"
@@ -56,16 +59,25 @@ class AgentPolicyMiddleware(AgentMiddleware): # type: ignore[misc]
orchestrator: AgentToolPolicyOrchestrator = DEFAULT_TOOL_POLICY_ORCHESTRATOR,
catalog: ToolCatalogSnapshot | None = None,
tools: list[Any] | None = None,
guardrails: "ToolGuardrailsMiddleware | None" = None,
) -> None:
"""绑定宿主可信上下文和共享策略编排器。"""
self.guardrails = guardrails
self.context = context
self.orchestrator = orchestrator
self.catalog = catalog
self._tools = {tool.name: tool for tool in (tools or []) if getattr(tool, "name", None)}
@hook_config(can_jump_to=["end"]) # type: ignore[misc]
@hook_config(can_jump_to=["end", "model"]) # type: ignore[misc]
async def aafter_model(self, state: dict[str, Any], runtime: Any) -> Any:
"""在 ToolNode 前暂停需要用户确认的敏感设置读取。"""
"""复用原有模型后置节点合并检测更新,不消耗额外图递归步数。"""
paused = await self._sensitive_pause(state)
if paused is not None:
return paused
return await self.guardrails.after_model_update(state, runtime) if self.guardrails else None
async def _sensitive_pause(self, state: dict[str, Any]) -> Any:
"""敏感设置暂停优先于自动续行,绝不因纠偏放行未确认的调用。"""
messages = state.get("messages") or []
if not messages or not isinstance(messages[-1], AIMessage):
return None

View File

@@ -24,6 +24,7 @@ from langchain_core.tools import BaseTool, StructuredTool
from pydantic import BaseModel, Field
from app.agent.llm.helper import LLMHelper
from app.agent.middleware.guardrails import ToolGuardrailsMiddleware
from app.agent.middleware.policy import AgentPolicyMiddleware
from app.agent.middleware.summarization import (
ContextPreservingSummarizationMiddleware,
@@ -444,6 +445,7 @@ class _SubAgentAgentProvider:
self._policy_context = policy_context or _default_subagent_policy_context(tools)
self._catalog = catalog
self._agents = {}
self._guardrails: dict[str, ToolGuardrailsMiddleware] = {}
self._default_agent_name = "general-purpose"
def _resolve_profile(self, _agent_name: Optional[str] = None) -> _SubAgentProfile:
@@ -464,6 +466,7 @@ class _SubAgentAgentProvider:
logger.info(
f"创建子代理图: subagent_type={profile.name}, tools={len(subagent_tools)}"
)
guardrails = ToolGuardrailsMiddleware(self._policy_context)
agent = create_agent(
model=self._model,
tools=[*subagent_tools, *self._server_tools],
@@ -473,7 +476,9 @@ class _SubAgentAgentProvider:
AgentPolicyMiddleware(
context=self._policy_context,
catalog=subagent_catalog,
guardrails=guardrails,
),
guardrails,
FinalRequestCompactionMiddleware(
summarizer=ContextPreservingSummarizationMiddleware(
model=self._model,
@@ -484,6 +489,7 @@ class _SubAgentAgentProvider:
],
)
self._agents[profile.name] = agent
self._guardrails[profile.name] = guardrails
return profile.name, agent
async def run_task(
@@ -527,6 +533,10 @@ class _SubAgentAgentProvider:
f"task_id={log_task_id}, error={summarize_error(err)}"
)
raise
finally:
guardrails = self._guardrails.get(agent_name)
if guardrails:
guardrails.release(f"subagent-{agent_name}-{thread_suffix}")
final_text = _extract_final_text(result)
logger.info(
f"子代理调用完成: subagent_type={agent_name}, "

View File

@@ -30,6 +30,7 @@ from app.agent.llm.tools import ServerToolRegistry
from app.agent.mcp import agent_mcp_manager
from app.agent.memory import MemoryManager, memory_manager
from app.agent.middleware.config import RuntimeConfigMiddleware
from app.agent.middleware.guardrails import GUARDRAIL_KEY, ToolGuardrailsMiddleware
from app.agent.middleware.invocation import InvocationMiddleware
from app.agent.middleware.jobs import (
JobsMiddleware,
@@ -1976,12 +1977,14 @@ class MoviePilotAgent:
)
# 中间件
guardrails = ToolGuardrailsMiddleware(policy_context)
middlewares = [
# 宿主策略必须位于最外层,确保插件覆盖工具基类也不能绕过。
AgentPolicyMiddleware(
context=policy_context,
catalog=tool_catalog,
tools=tools,
guardrails=guardrails,
),
# 运行中补充消息只在模型回合边界进入同一张图,不启动并行 Agent。
*([SteeringMiddleware()] if self._steering_inbox.running else []),
@@ -2000,6 +2003,8 @@ class MoviePilotAgent:
# 稳定偏好与原始会话证据分别管理。
memory_middleware,
*recall_middlewares,
# 以工具原文及模型请求顺序检测循环,外层归档仍保留真实回执。
guardrails,
*([learning_middleware] if learning_middleware else []),
# 错误工具调用修复
PatchToolCallsMiddleware(),
@@ -2286,6 +2291,21 @@ class MoviePilotAgent:
stripper.process(content, on_token)
stripper.flush(on_token)
final_notice = MoviePilotAgent._guardrail_final_text(agent, config)
if final_notice:
on_token('\n\n' + final_notice)
@staticmethod
def _guardrail_final_text(agent: Any, config: dict[str, Any]) -> str:
"""受控停止由宿主生成,不产生模型 token;流式入口须显式交付最终说明。"""
get_state = getattr(agent, 'get_state', None)
if not callable(get_state):
return ''
messages = get_state(config).values.get('messages', [])
if isinstance(messages, list) and messages and isinstance(messages[-1], AIMessage):
if messages[-1].additional_kwargs.get(GUARDRAIL_KEY):
return str(LLMHelper.extract_text_content(messages[-1].content))
return ''
@staticmethod
def _sanitize_recovery_message(message: BaseMessage) -> BaseMessage:
@@ -2544,7 +2564,9 @@ class MoviePilotAgent:
display_text = await self._save_final_agent_state(agent, agent_config)
execution_success = True
if self._learning_enabled() and self._learning is not None:
self._learning.finish(delivered=bool(display_text))
final_state = agent.get_state(agent_config).values.get('messages', [])
halted = bool(final_state and final_state[-1].additional_kwargs.get(GUARDRAIL_KEY))
self._learning.finish(delivered=bool(display_text) and not halted)
except asyncio.CancelledError:
logger.info(f"Agent执行被取消: session_id={self.session_id}")

View File

@@ -6,6 +6,7 @@ from typing import Callable, Optional
from app.agent.contracts import ReplyMode
from app.agent.lifecycle import AgentLifecycleOwner
from app.agent.middleware.guardrails import GUARDRAIL_REPLY_PREFIX
from app.agent.middleware.jobs import filter_active_jobs, load_jobs_metadata
from app.agent.orchestrator import AGENT_EXECUTION_ERROR_PREFIX, HEARTBEAT_SESSION_PREFIX
from app.agent.prompt import prompt_manager
@@ -114,7 +115,7 @@ class AgentTaskOwner(AgentLifecycleOwner):
)
result_text = str(result or "").strip()
success = not result_text.startswith(
(AGENT_EXECUTION_ERROR_PREFIX, "处理消息时发生错误")
(AGENT_EXECUTION_ERROR_PREFIX, GUARDRAIL_REPLY_PREFIX, "处理消息时发生错误")
)
except asyncio.CancelledError:
success = False

View File

@@ -142,6 +142,16 @@ Agent 人格是全局运行时配置,不按用户隔离;所有用户共享
这些命令限定到当前用户;工具参数和历史消息不能代替用户输入。旧活动摘要文件保持原地,不参与学习检索。实现参考固定 Hermes 提交 `f42f579cf8bac4918ac9599bece71618afadd846`,派生匹配器、威胁模式和复盘提示词保留 MIT 许可。
## 执行纠偏与停止
工具循环按 Hermes 固定源码的阈值和恢复规则判断:同参数失败2次提示、5次后阻止继续重放;同一工具不同参数失败3次提示、8次可停止,但终端诊断等允许连续失败的工具不会因此被中断。相同只读结果2次提示、5次后阻止重放;连续相同调用3次提示,并检测最近64条中的2–4步重复循环。成功修改后重试属于新实验,改变参数或真实结果也会重新计数。
Web、兼容 API 和受父任务监督的子代理默认只警告,渠道网关与后台任务默认强制停止;网页搜索与子任务启动分别使用每轮50次的预算,按 Hermes 语义在批次派发前检查,已放行的同一并行批次可以越过该阈值。终端与子任务的明确 read/wait/status/list 动作允许正常轮询,不把所有同名工具动作一概豁免。每个真实图线程独立计数,按模型发出顺序处理并行回执,以宿主每次图运行的身份重置。检测复用已有模型包装与策略节点,不额外消耗递归步数,也不提高原有执行上限。
相同长结果从第二次起可在模型上下文中显示为前次结果引用,工具仍然实际执行,原始回执仍会归档。结果变化恢复完整正文,压缩移除原引用后也重新给出完整结果;超长结果存根附带本次 `read_tool_result` 引用,避免指向已淘汰缓存。未知副作用和等待状态保持显式,不能因循环停止变成成功。宿主生成的停止说明会进入流式回复,定时任务将其记录为未成功,本轮不会触发成功后的学习。
短回复以“接下来我会…”或 Hermes 的英文尾部动作结束却没有调用工具时,最多追加两次原版继续提示;这是相同规则的中文本地化。Codex Responses 入口额外采用原版工作区确认和极短异常碎片规则;普通数字、文件名和符合用户语言的短答案不触发。合成提示不作为用户消息归档或计入学习轮次,耗尽后明确标记部分完成。`moviepilot_turn_status` 记录 blocked/waiting/unknown/partial;这些是宿主停止状态,不是业务完成证明。
## 验证范围
任务评测与离线轨迹验收的区别、场景及命令见 [Agent 任务评测](agent-evaluation.md)。固定响应单测不代表真实模型完成率。

View File

@@ -762,8 +762,8 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 1086 |
| 内部导入边 | 9,199 |
| Python 模块 | 1091 |
| 内部导入边 | 9,218 |
| 非平凡 SCC | 1(精确 containment 的 TMDB 移植包环) |
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
| Direct egress | 48(债务已清零,48 条精确 containment) |
@@ -773,7 +773,7 @@ flowchart LR
| Model/Oper 自动事务与自建 Session | 0 |
| 组合根外 `SystemConfigOper()` | 0 |
整理失败反馈由 `app.application.transfer.feedback` 集中投影;已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列;网络连通性检测按 `app.application.nettest` 归档,并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理;音乐标签准入与批次共识由 `app.chain.transfer.music` 负责,预览证据由 `app.application.transfer.projection` 只读投影;宿主依赖边现为 9,199,音乐来源链与识别缓存复用领域身份核验,音乐交互包未新增 SCC,插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器,依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`,文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端,经 `RequestUtils` 出站,宿主不再导入 pyTelegramBotAPI,启用 Telegram 的实例也不会因该 SDK 加载 redis;该依赖暂时保留给插件使用。Web Push 协议(aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递)由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现,消息端点与 `WebPushModule` 共用,宿主不再导入 pywebpush 及其连带的 aiohttp;该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端(经 `RequestUtils` 出站)与网络适配器 `app.adapters.network.feishu` 事件长连接(websocket-client 同步收发 pbbp2 帧),宿主不再导入 lark-oapi 及其上万个生成模块,长连接也不再依赖 SDK 的模块级事件循环;插件经 `app.sdk.feishu` 复用同一长连接,lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放,默认关闭,未开放时不生成文档。当前 `app.startup.lifecycle` 为 578、`app.factory` 为 591、`app.main` 为 593。音乐调用观察 `app.application.music.observation` 新增一个轻量模块,记录来源结果及请求预算;AcoustID及路径识别复用此上下文传递有界指纹候选,保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`,各启动入口增加一个模块;性能基线只同步模块数量,原有耗时预算、历史采样和生命周期资源约束保持有效。
整理失败反馈由 `app.application.transfer.feedback` 集中投影;已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列;网络连通性检测按 `app.application.nettest` 归档,并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理;音乐标签准入与批次共识由 `app.chain.transfer.music` 负责,预览证据由 `app.application.transfer.projection` 只读投影;宿主依赖边现为 9,218,音乐来源链与识别缓存复用领域身份核验,音乐交互包未新增 SCC,插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器,依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`,文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端,经 `RequestUtils` 出站,宿主不再导入 pyTelegramBotAPI,启用 Telegram 的实例也不会因该 SDK 加载 redis;该依赖暂时保留给插件使用。Web Push 协议(aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递)由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现,消息端点与 `WebPushModule` 共用,宿主不再导入 pywebpush 及其连带的 aiohttp;该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端(经 `RequestUtils` 出站)与网络适配器 `app.adapters.network.feishu` 事件长连接(websocket-client 同步收发 pbbp2 帧),宿主不再导入 lark-oapi 及其上万个生成模块,长连接也不再依赖 SDK 的模块级事件循环;插件经 `app.sdk.feishu` 复用同一长连接,lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放,默认关闭,未开放时不生成文档。当前 `app.startup.lifecycle` 为 578、`app.factory` 为 591、`app.main` 为 593。音乐调用观察 `app.application.music.observation` 新增一个轻量模块,记录来源结果及请求预算;AcoustID及路径识别复用此上下文传递有界指纹候选,保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`,各启动入口增加一个模块;性能基线只同步模块数量,原有耗时预算、历史采样和生命周期资源约束保持有效。
架构专项验证分为两个 CI 投影:`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
语义门禁,`Check host architecture snapshot` 再执行快照测试及一次
@@ -841,3 +841,5 @@ Agent 历史回忆由 `application.messaging.recall` 的 DTO/Port 和有界 work
历史检索新增 11 个宿主模块,宿主模块总数为 1,072;依赖图新增 78 条、移除 3 条依赖边,没有新增循环。启动冷路径仅增加 2 个模块;SQLite 实现与 Agent runtime 仍在组合函数执行时加载。性能基线仅更新模块计数,不修改时间预算或历史采样。
Agent 个人技能与后台学习新增13个宿主模块、61条内部依赖边,宿主为1085模块/9198边;没有新增SCC、具体Adapter越层依赖或边界债务。冷启动不物化学习实现,三入口模块数量仍为578/590/592,耗时预算和历史采样未调整。
Agent Hermes 执行纠偏新增5个宿主模块、19条内部依赖边,当前1090模块/9217边;没有新增SCC、越层依赖或边界债务。循环控制器不执行工具,实际调用、权限、原始回执与取消仍由宿主掌握。

View File

@@ -277,3 +277,7 @@ CI 上传 `complexity-report` 工件,门禁失败时也保留已生成的指
`tests/test_agent_learning_review.py` 用临时目录与确定性模型核对 Hermes 学习契约,
不访问实际 LLM 或个人记忆。匹配器、威胁模式和提示词的上游来源及 MIT 许可在
`app/agent/learning/LICENSE.hermes`;不得以这些离线测试替代固定真实模型的效果评测。
`tests/test_agent_guardrail_controller.py` 保留 Hermes 原始循环规则的纯行为回归,
`tests/test_agent_guardrails.py` 使用真实 LangGraph 验证并行顺序、逐运行隔离、正常轮询、
未知结果、尾部行动继续提示和流式停止交付。派生代码许可在 `app/agent/guardrails/LICENSE.hermes`。

View File

@@ -96,7 +96,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| 指标 | 当前值 | 解释 |
|---|---:|---|
| 宿主 Python 模块 / 内部依赖边 | 1086 / 9,199 | `dependency-baseline.json` 当前快照;系统设置合同按 `app.application.settings` 包组织,系统设置 API 端点独立按 `app.api.endpoints.settings` 归档;消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理;已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档;网络连通性检测按 `app.application.nettest` 归档 |
| 宿主 Python 模块 / 内部依赖边 | 1091 / 9,218 | `dependency-baseline.json` 当前快照;系统设置合同按 `app.application.settings` 包组织,系统设置 API 端点独立按 `app.api.endpoints.settings` 归档;消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理;已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档;网络连通性检测按 `app.application.nettest` 归档 |
| 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

View File

@@ -96,6 +96,7 @@ to make the directory tree look symmetrical.
| `app/agent/middleware/output.py` | Bounded, expiring in-memory tool output and thread-scoped pagination; never persists raw tool results |
| `app/agent/middleware/vision.py` | Request-only tool-image observations after compaction; complete tool-reply batches, per-invocation visual fallback and unchanged user authorization |
| `app/agent/tools/result.py` | Pure interpretation of explicit tool outcomes and portable tool-image history; image observation copies never replace original user attachments |
| `app/agent/guardrails/` / `app/agent/middleware/guardrails.py` | Agent 专属的 Hermes 循环检测与结束纠偏;纯控制器只产出决策,中间件在实际工具派发和模型边界应用,按宿主图线程隔离临时状态。不可替代授权、持久写入回执或业务成功校验;检测复用已有模型包装及策略后置节点,不能因增加图节点缩减任务预算;子任务 owner 在取消路径释放检测状态 |
| `app/agent/api/arguments.py` | Canonical API request fingerprints from the generated operation schema and the executor's GET projection; no endpoint imports or live discovery |
| `app/agent/shell.py` | Shared command interpreter, login mode, launch directory and subprocess text-encoding policy for run, pipe and PTY; preserves Windows default priority and UTF-8 behavior |
| `app/agent/terminal/` | `ownership.py` owns host task identity, invocation context and lazy scope closure; `session.py` owns process state, input serialization and UTF-8 capture; `output.py` owns pure paging and bounded projections; `manager.py` owns launch, access grants and process lifecycle. Package root contains no implementation exports |

View File

@@ -930,8 +930,8 @@
"runtime_only": true
}
},
"edge_count": 9199,
"edge_sha256": "ad0a66de22bf8cbef0a94ebb5b3ebd1e857977f78139d1cf3c27f9290db14664",
"edge_count": 9218,
"edge_sha256": "50c110ffa461d8c05104774cbd16d7d28ba0ad33fbfe16c847e61dcc4a5ff6bf",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -1200,6 +1200,9 @@
"app.agent.capabilities.adapter -> app.runtime.settings",
"app.agent.contracts -> app.schemas",
"app.agent.contracts -> app.schemas.types",
"app.agent.guardrails.controller -> app.agent",
"app.agent.guardrails.controller -> app.agent.guardrails",
"app.agent.guardrails.controller -> app.agent.guardrails.results",
"app.agent.history.importer -> app.agent",
"app.agent.history.importer -> app.agent.history",
"app.agent.history.importer -> app.agent.history.message",
@@ -1257,6 +1260,7 @@
"app.agent.learning.session -> app.agent.learning.tools",
"app.agent.learning.session -> app.agent.tools",
"app.agent.learning.session -> app.agent.tools.base",
"app.agent.learning.session -> app.agent.tools.result",
"app.agent.learning.session -> app.application",
"app.agent.learning.session -> app.application.messaging",
"app.agent.learning.session -> app.application.messaging.interaction",
@@ -1402,6 +1406,18 @@
"app.agent.middleware.config -> app.agent.middleware",
"app.agent.middleware.config -> app.agent.middleware.utils",
"app.agent.middleware.config -> app.agent.runtime",
"app.agent.middleware.guardrails -> app.agent",
"app.agent.middleware.guardrails -> app.agent.guardrails",
"app.agent.middleware.guardrails -> app.agent.guardrails.completion",
"app.agent.middleware.guardrails -> app.agent.guardrails.controller",
"app.agent.middleware.guardrails -> app.agent.policy",
"app.agent.middleware.guardrails -> app.agent.policy.api",
"app.agent.middleware.guardrails -> app.agent.policy.contracts",
"app.agent.middleware.guardrails -> app.agent.tools",
"app.agent.middleware.guardrails -> app.agent.tools.base",
"app.agent.middleware.guardrails -> app.agent.tools.impl",
"app.agent.middleware.guardrails -> app.agent.tools.impl.api",
"app.agent.middleware.guardrails -> app.agent.tools.result",
"app.agent.middleware.invocation -> app.agent",
"app.agent.middleware.invocation -> app.agent.policy",
"app.agent.middleware.invocation -> app.agent.policy.api",
@@ -1509,6 +1525,7 @@
"app.agent.middleware.subagents -> app.agent.llm",
"app.agent.middleware.subagents -> app.agent.llm.helper",
"app.agent.middleware.subagents -> app.agent.middleware",
"app.agent.middleware.subagents -> app.agent.middleware.guardrails",
"app.agent.middleware.subagents -> app.agent.middleware.policy",
"app.agent.middleware.subagents -> app.agent.middleware.summarization",
"app.agent.middleware.subagents -> app.agent.middleware.terminal",
@@ -1554,6 +1571,7 @@
"app.agent.orchestrator -> app.agent.memory",
"app.agent.orchestrator -> app.agent.middleware",
"app.agent.orchestrator -> app.agent.middleware.config",
"app.agent.orchestrator -> app.agent.middleware.guardrails",
"app.agent.orchestrator -> app.agent.middleware.invocation",
"app.agent.orchestrator -> app.agent.middleware.jobs",
"app.agent.orchestrator -> app.agent.middleware.learning",
@@ -1684,6 +1702,7 @@
"app.agent.tasks -> app.agent.contracts",
"app.agent.tasks -> app.agent.lifecycle",
"app.agent.tasks -> app.agent.middleware",
"app.agent.tasks -> app.agent.middleware.guardrails",
"app.agent.tasks -> app.agent.middleware.jobs",
"app.agent.tasks -> app.agent.orchestrator",
"app.agent.tasks -> app.agent.prompt",
@@ -10133,7 +10152,7 @@
"app.workflow.actions.transfer_file -> app.workflow",
"app.workflow.actions.transfer_file -> app.workflow.actions"
],
"module_count": 1086,
"module_count": 1091,
"modules": [
"app",
"app.adapters",
@@ -10197,6 +10216,10 @@
"app.agent.capabilities",
"app.agent.capabilities.adapter",
"app.agent.contracts",
"app.agent.guardrails",
"app.agent.guardrails.completion",
"app.agent.guardrails.controller",
"app.agent.guardrails.results",
"app.agent.history",
"app.agent.history.importer",
"app.agent.history.message",
@@ -10235,6 +10258,7 @@
"app.agent.memory",
"app.agent.middleware",
"app.agent.middleware.config",
"app.agent.middleware.guardrails",
"app.agent.middleware.invocation",
"app.agent.middleware.jobs",
"app.agent.middleware.learning",

View File

@@ -31,41 +31,55 @@ pytestmark = pytest.mark.anyio
class _FakeGraphState:
"""保存测试所需的消息快照,避免访问真实持久化。"""
def __init__(self, messages):
"""保存该测试图的预设消息或故障。"""
self.values = {"messages": messages}
class _FakeAgent:
"""提供可控的非流式图结果,不调用真实模型。"""
def __init__(self, messages):
"""保存该测试图的预设消息或故障。"""
self._messages = messages
async def ainvoke(self, _payload, config=None):
"""返回预设结果或抛出故障,不执行外部调用。"""
return None
def get_state(self, _config):
"""提供宿主结束处理所需的可恢复快照。"""
return _FakeGraphState(self._messages)
class _FakeFailingAgent:
"""在调用边界注入故障,验证宿主错误回执。"""
def __init__(self, error):
"""保存该测试图的预设消息或故障。"""
self._error = error
async def ainvoke(self, _payload, config=None):
"""返回预设结果或抛出故障,不执行外部调用。"""
raise self._error
def get_state(self, _config):
"""提供宿主结束处理所需的可恢复快照。"""
return _FakeGraphState([])
class _FakeStreamingFailingAgent(_FakeFailingAgent):
"""在流式迭代边界注入同一故障。"""
async def astream(self, _messages, **_kwargs):
"""维持异步流接口,按测试预设结束或失败。"""
if _kwargs.get("__keep_async_generator__"):
yield None
raise self._error
class _FakeStreamingAgent(_FakeAgent):
"""不产生额外 token,供测试读取最终图状态。"""
async def astream(self, _messages, **_kwargs):
"""维持异步流接口,按测试预设结束或失败。"""
if _kwargs.get("__keep_async_generator__"):
yield None
@@ -125,6 +139,7 @@ class TestAgentBackgroundOutput:
"""验证后台任务的回复策略、流式结束行为和工具装配。"""
async def test_non_streaming_image_unsupported_error_sends_friendly_notice(self):
"""非流式图片能力错误只向用户交付友好说明。"""
agent = MoviePilotAgent(session_id="image-test", user_id="user-1")
agent.channel = "Telegram"
agent.source = "telegram-test"
@@ -159,6 +174,7 @@ class TestAgentBackgroundOutput:
assert agent._streamed_output == UNSUPPORTED_IMAGE_INPUT_MESSAGE
async def test_streaming_image_unsupported_error_sends_friendly_notice(self):
"""流式图片能力错误沿用同一用户可见说明。"""
agent = MoviePilotAgent(session_id="image-test", user_id="user-1")
agent.channel = "Telegram"
agent.source = "telegram-test"
@@ -291,6 +307,7 @@ class TestAgentBackgroundOutput:
captured = {}
async def _execute_agent(messages):
"""记录模拟图实际收到的当前用户输入。"""
captured["messages"] = messages
return "消息已发送", {}
@@ -313,6 +330,7 @@ class TestAgentBackgroundOutput:
assert len(captured["messages"]) == 2
async def test_background_non_streaming_sends_when_reply_mode_dispatch(self):
"""显式派发模式允许后台任务发送最终回复。"""
agent = MoviePilotAgent(session_id="bg-test", user_id="system")
agent.channel = None
agent.source = None
@@ -366,6 +384,7 @@ class TestAgentBackgroundOutput:
assert agent._streamed_output == "后台结果"
async def test_heartbeat_check_jobs_captures_final_reply_and_keeps_message_tools(self):
"""心跳捕获最终回复但保留任务实际需要的消息工具。"""
manager = AgentManager()
with (
@@ -389,6 +408,7 @@ class TestAgentBackgroundOutput:
assert kwargs["allow_message_tools"] is True
async def test_heartbeat_check_jobs_skips_when_no_active_jobs(self):
"""没有启用任务时不调用模型执行心跳。"""
manager = AgentManager()
with (
@@ -462,6 +482,7 @@ class TestAgentBackgroundOutput:
"runtime",
"PlanMiddleware",
"memory",
"ToolGuardrailsMiddleware",
"patch",
"FinalRequestCompactionMiddleware",
"VisionMiddleware",
@@ -570,6 +591,7 @@ class TestAgentBackgroundOutput:
"runtime",
"PlanMiddleware",
"memory",
"ToolGuardrailsMiddleware",
"patch",
"FinalRequestCompactionMiddleware",
"VisionMiddleware",
@@ -765,6 +787,7 @@ class TestAgentBackgroundOutput:
"runtime",
"PlanMiddleware",
"memory",
"ToolGuardrailsMiddleware",
"patch",
"FinalRequestCompactionMiddleware",
"VisionMiddleware",
@@ -772,10 +795,12 @@ class TestAgentBackgroundOutput:
]
async def test_run_background_prompt_forces_disable_message_tools_when_capture_only(self):
"""仅捕获的后台入口强制移除主动消息发送能力。"""
captured = {}
manager = AgentManager()
async def fake_process(task):
"""捕获后台参数而不触发真实 Agent 推理。"""
captured["message"] = task.message
captured["reply_mode"] = task.reply_mode
captured["allow_message_tools"] = task.allow_message_tools

View File

@@ -0,0 +1,438 @@
"""Hermes 循环原语的原始行为回归;MIT 来源见 app/agent/guardrails/LICENSE.hermes。"""
import json
from app.agent.guardrails.controller import (
LoopCapConfig,
ToolCallGuardrailConfig,
ToolCallGuardrailController,
ToolCallSignature,
canonical_tool_args,
classify_tool_failure,
)
_PYTEST = {'command': 'pytest tests/test_x.py -q'}
_RED = '{"output": "1 failed", "exit_code": 1}'
def _HARD():
"""构造 Hermes 无人值守的默认强制停止策略。"""
return ToolCallGuardrailController(ToolCallGuardrailConfig(hard_stop_enabled=True))
def test_tool_call_signature_hashes_canonical_nested_unicode_args_without_exposing_raw_args():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
args_a = {
"z": [{"β": "☤", "a": 1}],
"a": {"y": 2, "x": "secret-token-value"},
}
args_b = {
"a": {"x": "secret-token-value", "y": 2},
"z": [{"a": 1, "β": "☤"}],
}
assert canonical_tool_args(args_a) == canonical_tool_args(args_b)
sig_a = ToolCallSignature.from_call("web_search", args_a)
sig_b = ToolCallSignature.from_call("web_search", args_b)
assert sig_a == sig_b
assert len(sig_a.args_hash) == 64
metadata = sig_a.to_metadata()
assert metadata == {"tool_name": "web_search", "args_hash": sig_a.args_hash}
assert "secret-token-value" not in json.dumps(metadata)
assert "☤" not in json.dumps(metadata)
def test_config_parses_nested_warn_and_hard_stop_thresholds():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
cfg = ToolCallGuardrailConfig.from_mapping(
{
"warnings_enabled": False,
"hard_stop_enabled": True,
"warn_after": {
"exact_failure": 3,
"same_tool_failure": 4,
"idempotent_no_progress": 5,
},
"hard_stop_after": {
"exact_failure": 6,
"same_tool_failure": 7,
"idempotent_no_progress": 8,
},
}
)
assert cfg.warnings_enabled is False
assert cfg.hard_stop_enabled is True
assert cfg.exact_failure_warn_after == 3
assert cfg.same_tool_failure_warn_after == 4
assert cfg.no_progress_warn_after == 5
assert cfg.exact_failure_block_after == 6
assert cfg.same_tool_failure_halt_after == 7
assert cfg.no_progress_block_after == 8
def test_gateway_platform_defaults_to_hard_stop_without_changing_interactive_defaults():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
interactive_configs = [
ToolCallGuardrailConfig.from_mapping({}, platform=platform)
for platform in ("cli", "tui", "desktop", "acp")
]
telegram_cfg = ToolCallGuardrailConfig.from_mapping({}, platform="telegram")
cron_cfg = ToolCallGuardrailConfig.from_mapping({}, platform="cron")
assert all(cfg.hard_stop_enabled is False for cfg in interactive_configs)
assert telegram_cfg.hard_stop_enabled is True
assert cron_cfg.hard_stop_enabled is True
def test_non_interactive_hard_stop_can_be_disabled_explicitly():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
cfg = ToolCallGuardrailConfig.from_mapping(
{"non_interactive_hard_stop_enabled": False},
platform="telegram",
)
assert cfg.hard_stop_enabled is False
assert cfg.non_interactive_hard_stop_enabled is False
def test_default_repeated_identical_failed_call_warns_without_blocking():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController()
args = {"query": "same"}
decisions = []
for _ in range(5):
assert controller.before_call("web_search", args).action == "allow"
decisions.append(
controller.after_call("web_search", args, '{"error":"boom"}', failed=True)
)
assert decisions[0].action == "allow"
assert [d.action for d in decisions[1:]] == ["warn", "warn", "warn", "warn"]
assert {d.code for d in decisions[1:]} == {"repeated_exact_failure_warning"}
assert controller.before_call("web_search", args).action == "allow"
assert controller.halt_decision is None
def test_hard_stop_enabled_blocks_repeated_exact_failure_before_next_execution():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController(
ToolCallGuardrailConfig(
hard_stop_enabled=True,
exact_failure_warn_after=2,
exact_failure_block_after=2,
same_tool_failure_halt_after=99,
)
)
args = {"query": "same"}
assert controller.before_call("web_search", args).action == "allow"
first = controller.after_call("web_search", args, '{"error":"boom"}', failed=True)
assert first.action == "allow"
assert controller.before_call("web_search", args).action == "allow"
second = controller.after_call("web_search", args, '{"error":"boom"}', failed=True)
assert second.action == "warn"
assert second.code == "repeated_exact_failure_warning"
blocked = controller.before_call("web_search", args)
assert blocked.action == "block"
assert blocked.code == "repeated_exact_failure_block"
assert blocked.count == 2
def test_skill_read_tools_are_idempotent_and_block_repeated_identical_success_output():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
cases = [
(
"skill_view",
{"name": "gui-agent-ml-operations"},
'{"success":true,"name":"gui-agent-ml-operations","content":"same"}',
),
(
"skills_list",
{"category": "mlops"},
'{"success":true,"skills":[{"name":"gui-agent-ml-operations"}]}',
),
]
for tool_name, args, result in cases:
controller = ToolCallGuardrailController(
ToolCallGuardrailConfig(
hard_stop_enabled=True,
no_progress_warn_after=2,
no_progress_block_after=2,
)
)
assert controller.before_call(tool_name, args).action == "allow"
assert controller.after_call(tool_name, args, result, failed=False).action == "allow"
assert controller.before_call(tool_name, args).action == "allow"
warn = controller.after_call(tool_name, args, result, failed=False)
assert warn.action == "warn"
assert warn.code == "idempotent_no_progress_warning"
blocked = controller.before_call(tool_name, args)
assert blocked.action == "block"
assert blocked.code == "idempotent_no_progress_block"
def test_mutating_or_unknown_tools_are_not_blocked_for_repeated_identical_success_output_by_default():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController(
ToolCallGuardrailConfig(no_progress_warn_after=2, no_progress_block_after=2)
)
for _ in range(3):
assert controller.before_call("write_file", {"path": "/tmp/x", "content": "x"}).action == "allow"
assert controller.after_call("write_file", {"path": "/tmp/x", "content": "x"}, "ok", failed=False).action == "allow"
assert controller.before_call("custom_tool", {"x": 1}).action == "allow"
assert controller.after_call("custom_tool", {"x": 1}, "ok", failed=False).action == "allow"
def test_identical_call_streak_halts_any_tool_when_hard_stop_enabled():
# #89069 / #100849 bundle: a model replaying the same SUCCESSFUL
# terminal/skill_view call with a byte-identical result is not covered by
# the idempotent_tools no-progress block. The consecutive-identical
# streak (observe_call) is tool-agnostic; under hard_stop it must halt.
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController(
ToolCallGuardrailConfig(hard_stop_enabled=True, no_progress_block_after=5)
)
args = {"command": "hermes config get memory.provider"}
for i in range(1, 5):
controller.after_call("terminal", args, "local\n", failed=False)
controller.observe_call("terminal", args, "local\n", failed=False)
assert controller.halt_decision is None, f"halted early at {i}"
controller.after_call("terminal", args, "local\n", failed=False)
controller.observe_call("terminal", args, "local\n", failed=False)
halt = controller.halt_decision
assert halt is not None and halt.should_halt
assert halt.code == "identical_call_streak_halt"
assert halt.tool_name == "terminal" and halt.count == 5
def test_identical_call_streak_never_halts_when_hard_stop_disabled_or_for_pollers():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
soft = ToolCallGuardrailController(
ToolCallGuardrailConfig(hard_stop_enabled=False, no_progress_block_after=2)
)
for _ in range(6):
soft.observe_call("terminal", {"command": "ls"}, "a\nb\n", failed=False)
assert soft.halt_decision is None # notice-only in interactive sessions
hard = ToolCallGuardrailController(
ToolCallGuardrailConfig(hard_stop_enabled=True, no_progress_block_after=2)
)
for _ in range(6):
hard.observe_call("process_manage", {"action": "poll", "session_id": "p1"}, "running", failed=False)
assert hard.halt_decision is None # an unchanged poll is legitimate progress
# A changed result resets the streak.
for i in range(6):
hard.observe_call("terminal", {"command": "date"}, f"t{i}", failed=False)
assert hard.halt_decision is None
def test_loop_cap_zero_disables_and_junk_falls_back():
# 0 is a legitimate "unlimited" value; negatives / junk fall back to default.
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
assert LoopCapConfig.from_mapping({"max_web_searches": 0}).max_web_searches == 0
assert LoopCapConfig.from_mapping({"max_web_searches": -5}).max_web_searches == LoopCapConfig().max_web_searches
assert LoopCapConfig.from_mapping({"max_subagents": "nope"}).max_subagents == LoopCapConfig().max_subagents
def test_web_search_cap_blocks_after_limit_regardless_of_hard_stop():
# Loop caps fire even with hard_stop_enabled=False (the per-turn loop
# detector's flag). Each distinct query avoids the loop detector so we know
# the block came from the loop cap, not exact-failure repetition.
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController(
ToolCallGuardrailConfig(
hard_stop_enabled=False,
loop_caps=LoopCapConfig(max_web_searches=3),
)
)
for i in range(3):
assert controller.before_call("web_search", {"query": f"q{i}"}).action == "allow"
decision = controller.before_call("web_search", {"query": "q4"})
assert decision.action == "block"
assert decision.code == "loop_web_search_cap"
assert decision.should_halt is True
def _run_red(c, args=_PYTEST):
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
assert c.before_call("terminal", args).allows_execution
return c.after_call("terminal", args, _RED, failed=True)
def test_fix_retest_loop_is_never_hard_stopped():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
c = _HARD()
for i in range(12):
d = _run_red(c)
assert not d.should_halt, f"halted on red run {i + 1}"
# the model edits between runs — a landed mutation is progress
c.after_call("patch", {"path": "x.py", "old_string": "a", "new_string": f"b{i}"},
'{"success": true, "diff": "..."}', failed=False)
assert c.halt_decision is None
assert c.before_call("terminal", _PYTEST).allows_execution
def test_pure_replay_with_no_intervening_change_is_still_blocked():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
c = _HARD()
for _ in range(5):
_run_red(c)
d = c.before_call("terminal", _PYTEST)
assert d.action == "block" and d.code == "repeated_exact_failure_block"
def test_intervening_mutation_resets_the_replay_streak_only_once():
# 4 reds, one edit, then 4 reds with NO edit: the second run of 4 is a
# fresh streak, and the 5th unchanged retry after it is blocked.
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
c = _HARD()
for _ in range(4):
_run_red(c)
c.after_call("write_file", {"path": "x.py", "content": "y"}, '{"bytes_written": 1}', failed=False)
for _ in range(5):
assert c.before_call("terminal", _PYTEST).allows_execution
c.after_call("terminal", _PYTEST, _RED, failed=True)
assert c.before_call("terminal", _PYTEST).action == "block"
def test_distinct_failing_terminal_commands_warn_but_never_halt():
# A diagnostic sweep: grep with no matches, missing binaries, red builds.
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
c = _HARD()
for i in range(12):
args = {"command": f"grep -q needle{i} haystack.txt"}
d = c.after_call("terminal", args, _RED, failed=True)
assert not d.should_halt, f"same_tool halt on distinct command #{i + 1}"
assert c.halt_decision is None
# ...while a non-tolerant tool failing 8 distinct ways still halts.
c2 = _HARD()
last = None
for i in range(8):
last = c2.after_call("send_message", {"to": f"u{i}"}, '{"error": "no route"}', failed=True)
assert last.should_halt and last.code == "same_tool_failure_halt"
def test_browser_retry_after_action_is_not_a_replay():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
c = _HARD()
nav = {"url": "https://example.test/app"}
for _ in range(8):
assert c.before_call("browser_navigate", nav).allows_execution
c.after_call("browser_navigate", nav, '{"error": "timeout"}', failed=True)
c.after_call("browser_click", {"selector": "#retry"}, '{"ok": true}', failed=False)
assert c.halt_decision is None
def test_supervised_task_platforms_keep_warning_only_default():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
for platform in ("subagent", "api_server", "cli"):
cfg = ToolCallGuardrailConfig.from_mapping({}, platform=platform)
assert cfg.hard_stop_enabled is False, platform
for platform in ("telegram", "discord", "cron", "kanban"):
cfg = ToolCallGuardrailConfig.from_mapping({}, platform=platform)
assert cfg.hard_stop_enabled is True, platform
def test_a_real_tool_error_is_still_a_failure():
"""The exemption is keyed on the marker, not on the word: a body that
genuinely failed still counts, or the streak that stops a real loop is gone."""
real = '{"error": "ENOENT: no such file"}'
assert classify_tool_failure("read_file", real)[0] is True
# The marker is only honoured as the literal boolean, never as truthy prose.
assert classify_tool_failure("read_file", '{"error": "x", "guardrail_refusal": "yes"}')[0] is True
def test_identical_streak_ignores_volatile_execution_metadata():
# execute_code results carry per-call metadata (execution_count, duration_seconds) that
# changes on every run. Hashing it made each empty replay look "new", so a model re-ran
# the same empty probe 147 times without the identical-call halt ever firing.
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController(
ToolCallGuardrailConfig(hard_stop_enabled=True, no_progress_block_after=5)
)
args = {"code": "import subprocess\nprint(subprocess.run(['true']).returncode) if False else None"}
def result(n):
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
return json.dumps({
"status": "success", "output": "", "exit_code": 0, "tool_calls_made": 0,
"duration_seconds": 1.0 + n / 100,
"kernel": {"mode": "session", "reused": True, "execution_count": 100 + n, "state_reset": False},
"stdout_truncated": False, "stdout_bytes_captured": 0,
})
for i in range(4):
controller.observe_call("execute_code", args, result(i), failed=False)
assert controller.halt_decision is None, f"halted early at {i}"
controller.observe_call("execute_code", args, result(4), failed=False)
halt = controller.halt_decision
assert halt is not None, "volatile metadata defeated the identical-call streak"
assert halt.code == "identical_call_streak_halt"
def test_identical_streak_still_resets_when_real_output_changes():
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController(
ToolCallGuardrailConfig(hard_stop_enabled=True, no_progress_block_after=3)
)
args = {"code": "print(1)"}
for i in range(6):
out = json.dumps({"status": "success", "output": f"line {i}", "duration_seconds": 1.0,
"kernel": {"execution_count": i}})
controller.observe_call("execute_code", args, out, failed=False)
assert controller.halt_decision is None, "different real output must not count as a replay"
def test_other_tools_keep_duration_and_execution_count_as_real_output():
# Only execute_code's known metadata locations are volatile. For any other tool these keys
# can be the actual answer (e.g. a job-status tool reporting how long a job ran), so results
# that differ only there are different results and must not form an identical streak.
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController(
ToolCallGuardrailConfig(hard_stop_enabled=True, no_progress_block_after=3)
)
args = {"job": "build-42"}
for i in range(6):
out = json.dumps({"status": "done", "duration_seconds": 10 + i,
"stats": {"execution_count": 100 + i}})
controller.observe_call("mcp_ci_job_status", args, out, failed=False)
assert controller.halt_decision is None, "a real change in duration_seconds/execution_count was treated as a replay"
def test_execute_code_replay_streak_notice_fires_on_warn_only_desktop_config():
# #124072: on an interactive surface hard stops are off, so the appended notice is
# the only signal the model gets. 186 no-op print("...") cells whose results differed
# only in kernel.execution_count / duration_seconds produced zero notices.
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
controller = ToolCallGuardrailController(ToolCallGuardrailConfig.from_mapping({}, platform="desktop"))
args = {"code": 'print("...")'}
def result(n):
"""验证 Hermes 原始循环检测合同,保持阈值、恢复及回执语义。"""
return json.dumps({
"status": "success", "output": "...\n", "exit_code": 0, "tool_calls_made": 0,
"duration_seconds": 0.001 * n,
"kernel": {"mode": "session", "reused": True, "execution_count": n, "state_reset": False},
"stdout_truncated": False, "stdout_bytes_captured": 4, "stdout_bytes_total": 4,
"stdout_bytes_omitted": 0,
})
notices = [
controller.observe_call("execute_code", args, result(i), tool_call_id=f"c{i}").notice
for i in range(1, 7)
]
assert notices[:2] == [None, None]
assert all(n is not None and "consecutive identical call to execute_code" in n for n in notices[2:]), notices
assert controller.halt_decision is None, "warn-only surfaces must not halt"

View File

@@ -0,0 +1,368 @@
"""真实 LangChain 图中验证循环停止、并行顺序、轮询和原始工具结果。"""
import asyncio
import json
from types import SimpleNamespace
import pytest
from langchain.agents import create_agent as _create_agent
from langchain.agents.middleware.types import ToolCallRequest
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import StructuredTool
from langgraph.checkpoint.memory import MemorySaver
from app.agent.guardrails.completion import degenerate_final, trailing_continue_intent
from app.agent.guardrails.controller import ToolCallGuardrailConfig
from app.agent.middleware.guardrails import GUARDRAIL_KEY, TURN_STATUS_KEY, ToolGuardrailsMiddleware
from app.agent.middleware.output import ToolOutputMiddleware
from app.agent.middleware.policy import AgentPolicyMiddleware
from app.agent.orchestrator import MoviePilotAgent
from app.agent.policy.contracts import AuthSource, PrincipalType, ToolOrigin, ToolPolicyContext
from app.agent.tools.base import format_tool_result_for_agent
from app.agent.tools.result import TOOL_OBSERVATION_MARKER, messages_for_persistence
def create_agent(**kwargs):
"""使用生产的策略节点组合,保持循环检测不新增模型前后图节点。"""
middleware = kwargs['middleware']
guard = next(item for item in middleware if isinstance(item, ToolGuardrailsMiddleware))
policy = next((item for item in middleware if isinstance(item, AgentPolicyMiddleware)), None)
if policy is None:
middleware.insert(0, AgentPolicyMiddleware(context=guard.context, guardrails=guard))
else:
policy.guardrails = guard
return _create_agent(**kwargs)
class GuardModel(FakeMessagesListChatModel):
"""固定工具调用模型,保证测试不依赖真实服务或模型概率。"""
def bind_tools(self, _tools, **_kwargs):
"""使用真实 ToolNode,模型仅返回预定消息。"""
return self
def context(*, unattended=True):
"""只有宿主入口身份决定是否默认强制停止。"""
return ToolPolicyContext(session_id='guard', user_id='alice', origin=ToolOrigin.AGENT_INTERACTIVE,
principal_type=PrincipalType.HUMAN,
auth_source=AuthSource.CHANNEL if unattended else AuthSource.WEB_SESSION,
agent_context={'is_admin': False})
def call(index, *names, arguments=None):
"""构造唯一调用编号但参数相同的模型工具批次。"""
return AIMessage(content='', tool_calls=[dict(name=name, args=arguments or {}, id=f'{index}-{name}') for name in names])
@pytest.mark.parametrize('unattended, expected', [(True, 5), (False, 7)])
def test_successful_replay_stops_only_unattended_and_resets_next_turn(unattended, expected):
"""成功但无进展的普通工具也被检测,前台只提示,新轮次不继承停止。"""
async def scenario():
"""实际执行工具,检查第六次是否被省掉,而非只测试检测器状态。"""
executed = []
async def probe():
"""稳定输出模拟没有取得进展的探测。"""
executed.append(True)
return '相同结果'
middleware = ToolGuardrailsMiddleware(context(unattended=unattended))
model = GuardModel(responses=[*[call(i, 'probe') for i in range(7)], AIMessage(content='结束')])
graph = create_agent(model=model, tools=[StructuredTool.from_function(coroutine=probe, name='probe')], middleware=[middleware])
result = await graph.ainvoke({'messages': [HumanMessage(content='执行任务')]})
assert len(executed) == expected
assert bool(result['messages'][-1].additional_kwargs.get(GUARDRAIL_KEY)) is unattended
if unattended:
assert result['messages'][-1].additional_kwargs[TURN_STATUS_KEY] == 'blocked'
assert '尚未完成' in result['messages'][-1].content
assert not middleware._runs
assert not any(name.startswith('ToolGuardrailsMiddleware.') for name in graph.get_graph().nodes)
model.responses = [*[call(i + 20, 'probe') for i in range(7)], AIMessage(content='再次结束')]
model.i = 0
executed.clear()
again = await graph.ainvoke({'messages': [HumanMessage(content='重新核验')]})
assert len(executed) == expected
assert bool(again['messages'][-1].additional_kwargs.get(GUARDRAIL_KEY)) is unattended
assert not middleware._runs
asyncio.run(scenario())
def test_parallel_two_call_cycles_follow_emission_order():
"""工具完成次序交错时仍按模型声明顺序检测 A/B 循环。"""
async def scenario():
"""A 在偶数轮较慢,B 在奇数轮较慢,结果始终相同。"""
counts = {'a': 0, 'b': 0}
async def a():
"""返回稳定 A 结果,并交替让出事件循环。"""
counts['a'] += 1
if counts['a'] % 2:
await asyncio.sleep(.002)
return 'A'
async def b():
"""返回稳定 B 结果,与 A 的完成顺序交替。"""
counts['b'] += 1
if not counts['b'] % 2:
await asyncio.sleep(.002)
return 'B'
middleware = ToolGuardrailsMiddleware(context())
graph = create_agent(model=GuardModel(responses=[call(i, 'a', 'b') for i in range(8)]),
tools=[StructuredTool.from_function(coroutine=a), StructuredTool.from_function(coroutine=b)],
middleware=[middleware])
result = await graph.ainvoke({'messages': [HumanMessage(content='比较')]})
assert counts == {'a': 5, 'b': 5}
assert result['messages'][-1].additional_kwargs[GUARDRAIL_KEY]['code'] == 'identical_cycle_halt'
asyncio.run(scenario())
def test_polling_is_exempt_but_still_executes_and_deduplicates_fresh_results():
"""正常等待不会被当成循环,每次仍执行;较长重复回执仅压缩上下文表示。"""
async def scenario():
"""等待七次之后得到完成结果,状态变化必须使正文重新完整出现。"""
executed = []
async def execute_command(action: str):
"""模拟真实终端等待协议,不启动任何进程。"""
executed.append(action)
return json.dumps(dict(execution_outcome='pending' if len(executed) < 7 else 'succeeded', output='x' * 1000))
middleware = ToolGuardrailsMiddleware(context())
graph = create_agent(model=GuardModel(responses=[*[call(i, 'execute_command', arguments={'action': 'wait'}) for i in range(7)], AIMessage(content='结束')]),
tools=[StructuredTool.from_function(coroutine=execute_command)], middleware=[middleware])
result = await graph.ainvoke({'messages': [HumanMessage(content='等待')]})
outputs = [message for message in result['messages'] if isinstance(message, ToolMessage)]
assert len(executed) == 7 and middleware.controller.halt_decision is None
assert 'byte-identical' in outputs[1].content and 'pending' in outputs[1].content
assert json.loads(outputs[-1].content)['execution_outcome'] == 'succeeded'
asyncio.run(scenario())
def test_raw_result_changes_survive_identical_truncated_previews():
"""只在被截掉中部发生的真实结果变化也必须清除重复判断。"""
async def scenario():
"""截断原文记录器与循环观察器组合,确认没有把相同预览当完整结果。"""
count = 0
async def read_file():
"""生成中段变化的长文件。"""
nonlocal count
count += 1
return format_tool_result_for_agent(json.dumps({'content': 'a' * 80000 + str(count) + 'b' * 80000}), tool_name='read_file')
policy = context()
guard = ToolGuardrailsMiddleware(policy)
graph = create_agent(model=GuardModel(responses=[*[call(i, 'read_file') for i in range(7)], AIMessage(content='完成')]),
tools=[StructuredTool.from_function(coroutine=read_file)], middleware=[ToolOutputMiddleware(policy), guard])
result = await graph.ainvoke({'messages': [HumanMessage(content='连续读取')]}, {'configurable': {'thread_id': 'guard'}})
assert count == 7 and guard.controller.halt_decision is None
previews = [json.loads(message.content)['content_preview'] for message in result['messages'] if isinstance(message, ToolMessage)]
assert len(set(previews)) == 1
asyncio.run(scenario())
@pytest.mark.parametrize('tool_name', ['web_search', 'search_web'])
def test_evicted_reference_uses_fresh_full_result_and_cap_blocks_before_execution(tool_name):
"""压缩后不产生悬空引用;总调用上限在实际调用之前执行。"""
async def scenario():
"""直接使用真实 ToolCallRequest,隔离验证存根恢复和调用预算。"""
guard = ToolGuardrailsMiddleware(context(unattended=False))
guard.controller.config = ToolCallGuardrailConfig(loop_caps=SimpleNamespace(max_web_searches=2, max_subagents=50))
executed = []
async def handler(request):
"""计数实际执行,第三次必须无法到达这里。"""
executed.append(True)
return ToolMessage(content='x' * 700, tool_call_id=request.tool_call['id'], name='web_search', id=request.tool_call['id'])
outputs = []
for index in range(3):
request = ToolCallRequest(tool_call=call(index, tool_name).tool_calls[0], tool=None, state={}, runtime=None)
message = await guard.awrap_tool_call(request, handler)
result = await guard.prepare_model({'messages': [call(index, tool_name), message]}, None)
outputs.append(result['messages'][-1])
assert len(executed) == 2
assert outputs[1].content.startswith('x' * 700)
assert outputs[-1].additional_kwargs[GUARDRAIL_KEY]['code'] == 'loop_web_search_cap'
asyncio.run(scenario())
def test_moviepilot_file_edit_restarts_failed_terminal_experiment():
"""宿主 edit_file 名称映射到 Hermes patch,真实修改后重试不能累积成原样失败循环。"""
async def scenario():
"""七次失败之间每次有不同修复,最终正常返回,不在第五次失败处截断。"""
attempts = []
async def execute_command(command: str):
"""模拟未通过的验证命令。"""
attempts.append(command)
return json.dumps({'execution_outcome': 'failed', 'exit_code': 1})
async def edit_file(content: str):
"""模拟已经落盘的不同修复。"""
attempts.append(content)
return json.dumps({'success': True})
responses = []
for index in range(7):
responses.extend([call(index, 'execute_command', arguments={'command': 'pytest'}),
call(index, 'edit_file', arguments={'content': str(index)})])
graph = create_agent(model=GuardModel(responses=[*responses, AIMessage(content='已完成诊断')]),
tools=[StructuredTool.from_function(coroutine=execute_command), StructuredTool.from_function(coroutine=edit_file)],
middleware=[ToolGuardrailsMiddleware(context())])
result = await graph.ainvoke({'messages': [HumanMessage(content='修复并验证')]}, {'recursion_limit': 200})
assert len(attempts) == 14 and result['messages'][-1].content == '已完成诊断'
asyncio.run(scenario())
def test_parallel_graph_runs_do_not_share_counters_or_pending_results():
"""子代理可共享图中间件,但两个真实 thread_id 各执行五次后独立停止。"""
async def scenario():
"""同一中间件及相同工具调用 ID 用于两个并行运行,验证作用域来自宿主。"""
count = 0
async def probe():
"""交错执行,制造两个图同时等待的窗口。"""
nonlocal count
count += 1
await asyncio.sleep(.001)
return 'same'
guard = ToolGuardrailsMiddleware(context())
tool = StructuredTool.from_function(coroutine=probe)
graphs = [create_agent(model=GuardModel(responses=[call(i, 'probe') for i in range(8)]),
tools=[tool], middleware=[guard]) for _ in range(2)]
results = await asyncio.gather(*[graph.ainvoke({'messages': [HumanMessage(content='任务')]},
{'configurable': {'thread_id': str(index)}})
for index, graph in enumerate(graphs)])
assert count == 10
assert all(result['messages'][-1].additional_kwargs[TURN_STATUS_KEY] == 'blocked' for result in results)
assert not guard._runs
asyncio.run(scenario())
def test_outer_policy_exception_receipts_remain_unknown_when_halted():
"""检测也覆盖策略层转换的异常,超时可能已经发生的副作用不能写成成功。"""
async def scenario():
"""工具在循环中抛出超时,外层真实策略生成未知回执。"""
attempts = []
async def probe():
"""不访问外部服务,仅模拟结果未确认。"""
attempts.append(True)
raise TimeoutError('unconfirmed')
policy = context()
graph = create_agent(model=GuardModel(responses=[call(i, 'probe') for i in range(8)]),
tools=[StructuredTool.from_function(coroutine=probe)],
middleware=[AgentPolicyMiddleware(context=policy), ToolGuardrailsMiddleware(policy)])
result = await graph.ainvoke({'messages': [HumanMessage(content='执行')]})
assert len(attempts) == 5
assert result['messages'][-1].additional_kwargs[TURN_STATUS_KEY] == 'unknown'
asyncio.run(scenario())
@pytest.mark.parametrize('text', ['Let me now inspect the file.', '接下来我会检查文件。'])
def test_announced_action_gets_only_two_nudges_and_not_fake_user_history(text):
"""英文和中文尾部动作使用同一有界继续机制,模型始终不行动时明确部分完成。"""
async def scenario():
"""始终只返回下一步计划,确保不会无限追加模型调用。"""
async def probe():
"""不会实际被调用的可用工具。"""
raise AssertionError('unexpected execution')
graph = create_agent(model=GuardModel(responses=[AIMessage(content=text, id=f'ack-{index}') for index in range(3)]),
tools=[StructuredTool.from_function(coroutine=probe)], middleware=[ToolGuardrailsMiddleware(context())])
result = await graph.ainvoke({'messages': [HumanMessage(content='检查文件')]})
messages = result['messages']
assert sum(bool(message.additional_kwargs.get(TOOL_OBSERVATION_MARKER)) for message in messages) == 2
assert messages[-1].additional_kwargs[TURN_STATUS_KEY] == 'partial'
assert len([message for message in messages if isinstance(message, AIMessage)]) == 3
assert sum(isinstance(message, HumanMessage) for message in messages_for_persistence(messages)) == 1
asyncio.run(scenario())
def test_continuation_recovers_to_actual_work_and_normal_answer():
"""一次纠偏后实际调用工具并给出答案,不把正常完成回复继续追问。"""
async def scenario():
"""真实图走过承诺、继续提示、工具执行及最终答案四种节点。"""
attempts = []
async def probe():
"""记录已实际执行的核验。"""
attempts.append(True)
return 'observed'
graph = create_agent(model=GuardModel(responses=[AIMessage(content='Let me now check the file.'), call(1, 'probe'), AIMessage(content='已核验。')]),
tools=[StructuredTool.from_function(coroutine=probe)], middleware=[ToolGuardrailsMiddleware(context())])
result = await graph.ainvoke({'messages': [HumanMessage(content='检查文件')]})
assert attempts == [True] and result['messages'][-1].content == '已核验。'
assert GUARDRAIL_KEY not in result['messages'][-1].additional_kwargs
asyncio.run(scenario())
def test_short_answers_and_nonterminal_plans_do_not_trigger_continuation():
"""保留 Hermes 窄匹配边界,数字、正常中文短答、已交付内容和礼貌提议都不触发。"""
for text in ('42', 'SQLite', 'report.csv', '你好。', 'Done.', ':8080', 'Let me now check. The answer is 42.', "I'll help if needed."):
assert not trailing_continue_intent(text)
assert not degenerate_final(text, '正常中文问题')
assert not trailing_continue_intent('x' * 400 + 'Let me now check.')
assert not degenerate_final('是', '是否完成')
assert degenerate_final('?warming up', 'Check the service')
def test_stream_delivers_host_generated_guardrail_stop():
"""没有模型 token 的停止说明也会进入真实流式回调,不能只存到 checkpoint。"""
async def scenario():
"""执行五轮无进展后由宿主结束,通过 MoviePilot 流式入口读取说明。"""
async def probe():
"""返回稳定读取结果。"""
return 'same'
graph = create_agent(model=GuardModel(responses=[call(index, 'probe') for index in range(8)]),
tools=[StructuredTool.from_function(coroutine=probe)], middleware=[ToolGuardrailsMiddleware(context())],
checkpointer=MemorySaver())
tokens = []
await MoviePilotAgent._stream_agent_tokens(graph, {'messages': [HumanMessage(content='任务')]},
{'configurable': {'thread_id': 'stream-guard'}}, tokens.append)
assert ''.join(tokens).count('本轮未完成:') == 1
assert '任务尚未完成' in ''.join(tokens)
asyncio.run(scenario())
def test_duplicate_preview_references_latest_live_output_after_eviction():
"""重复工具仍执行,超过八项输出缓存容量后存根仍引用最新可读正文。"""
async def scenario():
"""执行十二次相同长结果,检查最后引用仍在宿主缓存中。"""
calls = []
async def read_file():
"""读取相同长文本以触发存根而非跳过真实读取。"""
calls.append(True)
return format_tool_result_for_agent('x' * 100000, tool_name='read_file')
policy = context(unattended=False)
output = ToolOutputMiddleware(policy)
graph = create_agent(model=GuardModel(responses=[*[call(index, 'read_file') for index in range(12)], AIMessage(content='已读')]),
tools=[StructuredTool.from_function(coroutine=read_file)], middleware=[output, ToolGuardrailsMiddleware(policy)])
result = await graph.ainvoke({'messages': [HumanMessage(content='读取')]}, {'configurable': {'thread_id': 'eviction'}})
last = next(message for message in reversed(result['messages']) if isinstance(message, ToolMessage))
assert len(calls) == 12
assert 'byte-identical' in last.content
assert any(identifier in last.content for identifier in output._results), (last.content, list(output._results))
asyncio.run(scenario())

View File

@@ -79,10 +79,14 @@ async def test_turn_cleanup_never_reopens_sealed_controller():
@pytest.mark.anyio
async def test_subagent_compacts_large_input_before_model_call(monkeypatch):
"""独立子图也执行最终请求压缩,大量工具输出不会绕过预算边界。"""
sequence = 0
@tool
def diagnostic() -> str:
"""读取诊断日志,不访问外部服务。"""
return "诊断日志中的连接信息。" * 450
"""每次读取不同日志;重复内容现在会变为引用,不能用于触发压缩预算。"""
nonlocal sequence
sequence += 1
return f"日志批次 {sequence}:" + "诊断日志中的连接信息。" * 450
diagnostic.tags = ["read", "system"]
model = _TaskModel(

View File

@@ -328,6 +328,7 @@ def test_streaming_agent_uses_non_streaming_llm_for_model_middlewares(with_invoc
"RuntimeConfigMiddleware",
"PlanMiddleware",
"MemoryMiddleware",
"ToolGuardrailsMiddleware",
"PatchToolCallsMiddleware",
"_FakeToolSelectorMiddleware",
"FinalRequestCompactionMiddleware",

View File

@@ -386,8 +386,9 @@ async def test_query_task_returns_owner_scoped_ten_recent_runs(monkeypatch) -> N
@pytest.mark.anyio
async def test_agent_manager_records_manual_trigger_source(monkeypatch) -> None:
"""真实执行入口应把手动触发来源写入对应 run。"""
@pytest.mark.parametrize('reply, success', [('完成', True), ('本轮未完成:工具循环已停止', False)])
async def test_agent_manager_records_manual_trigger_source(monkeypatch, reply, success) -> None:
"""真实入口记录手动触发来源,受控停止不能把定时任务标为成功。"""
from app.agent import orchestrator
from app.runtime.config import settings
@@ -402,16 +403,17 @@ async def test_agent_manager_records_manual_trigger_source(monkeypatch) -> None:
captured = {}
async def process_message(**kwargs):
"""返回正常结果或循环停止结果,验证任务运行持久状态。"""
captured.update(kwargs)
return "完成"
return reply
manager.process_message = process_message
assert await manager.execute_scheduled_task(task.id, trigger_source="manual") == (
True,
"完成",
success,
reply,
)
runs = AgentTaskOper().list_runs(task.id)
assert len(runs) == 1
assert runs[0].trigger_source == "manual"
assert runs[0].status == "success"
assert runs[0].status == ('success' if success else 'failed')
assert "定时任务已手动触发" in captured["message"]