From fcd4cd172553ee55d39bed52d3abc59e56ee93c5 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Wed, 30 Sep 2026 18:49:52 +0800 Subject: [PATCH] feat(agent): align execution guardrails with Hermes --- app/agent/guardrails/LICENSE.hermes | 21 + app/agent/guardrails/__init__.py | 1 + app/agent/guardrails/completion.py | 56 ++ app/agent/guardrails/controller.py | 690 ++++++++++++++++++ app/agent/guardrails/results.py | 43 ++ app/agent/learning/session.py | 4 +- app/agent/middleware/guardrails.py | 314 ++++++++ app/agent/middleware/policy.py | 18 +- app/agent/middleware/subagents.py | 10 + app/agent/orchestrator.py | 24 +- app/agent/tasks.py | 3 +- docs/agent.md | 10 + docs/architecture-overview.md | 8 +- docs/development-setup.md | 4 + docs/refactor/optimization-checklist.md | 2 +- docs/rules/05-architecture.md | 1 + .../architecture/dependency-baseline.json | 30 +- tests/test_agent_background_output.py | 25 + tests/test_agent_guardrail_controller.py | 438 +++++++++++ tests/test_agent_guardrails.py | 368 ++++++++++ tests/test_agent_subagent_reuse.py | 8 +- tests/test_agent_summarization_streaming.py | 1 + tests/test_agent_task_runs.py | 14 +- 23 files changed, 2072 insertions(+), 21 deletions(-) create mode 100644 app/agent/guardrails/LICENSE.hermes create mode 100644 app/agent/guardrails/__init__.py create mode 100644 app/agent/guardrails/completion.py create mode 100644 app/agent/guardrails/controller.py create mode 100644 app/agent/guardrails/results.py create mode 100644 app/agent/middleware/guardrails.py create mode 100644 tests/test_agent_guardrail_controller.py create mode 100644 tests/test_agent_guardrails.py diff --git a/app/agent/guardrails/LICENSE.hermes b/app/agent/guardrails/LICENSE.hermes new file mode 100644 index 000000000..75410e733 --- /dev/null +++ b/app/agent/guardrails/LICENSE.hermes @@ -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. diff --git a/app/agent/guardrails/__init__.py b/app/agent/guardrails/__init__.py new file mode 100644 index 000000000..537ce5f71 --- /dev/null +++ b/app/agent/guardrails/__init__.py @@ -0,0 +1 @@ +"""Agent 每轮工具循环检测与受控停止契约。""" diff --git a/app/agent/guardrails/completion.py b/app/agent/guardrails/completion.py new file mode 100644 index 000000000..135708e50 --- /dev/null +++ b/app/agent/guardrails/completion.py @@ -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) diff --git a/app/agent/guardrails/controller.py b/app/agent/guardrails/controller.py new file mode 100644 index 000000000..62e5ccb4c --- /dev/null +++ b/app/agent/guardrails/controller.py @@ -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() diff --git a/app/agent/guardrails/results.py b/app/agent/guardrails/results.py new file mode 100644 index 000000000..a71447cfb --- /dev/null +++ b/app/agent/guardrails/results.py @@ -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 diff --git a/app/agent/learning/session.py b/app/agent/learning/session.py index fc143ec7d..ca39020b9 100644 --- a/app/agent/learning/session.py +++ b/app/agent/learning/session.py @@ -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 diff --git a/app/agent/middleware/guardrails.py b/app/agent/middleware/guardrails.py new file mode 100644 index 000000000..e0f8718d1 --- /dev/null +++ b/app/agent/middleware/guardrails.py @@ -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'.*?(?:|$)', '', 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 diff --git a/app/agent/middleware/policy.py b/app/agent/middleware/policy.py index 9347a7e28..298d939b7 100644 --- a/app/agent/middleware/policy.py +++ b/app/agent/middleware/policy.py @@ -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 diff --git a/app/agent/middleware/subagents.py b/app/agent/middleware/subagents.py index b595ffd2c..7749765a6 100644 --- a/app/agent/middleware/subagents.py +++ b/app/agent/middleware/subagents.py @@ -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}, " diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index 0c47d02de..483b35390 100644 --- a/app/agent/orchestrator.py +++ b/app/agent/orchestrator.py @@ -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}") diff --git a/app/agent/tasks.py b/app/agent/tasks.py index 1c8814cfd..13b86d97f 100644 --- a/app/agent/tasks.py +++ b/app/agent/tasks.py @@ -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 diff --git a/docs/agent.md b/docs/agent.md index 3c412ae17..d093c881d 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -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)。固定响应单测不代表真实模型完成率。 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 61383db51..45cab1a77 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -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、越层依赖或边界债务。循环控制器不执行工具，实际调用、权限、原始回执与取消仍由宿主掌握。 diff --git a/docs/development-setup.md b/docs/development-setup.md index 60859a95d..6b0aede88 100644 --- a/docs/development-setup.md +++ b/docs/development-setup.md @@ -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`。 diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 08f87021e..6825ff5f5 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -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 等基线均为零 | diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index efd62a639..30118ebee 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -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 | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index ee0973b47..d8c4b0c02 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -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", diff --git a/tests/test_agent_background_output.py b/tests/test_agent_background_output.py index 5aa72a32e..4990e87ef 100644 --- a/tests/test_agent_background_output.py +++ b/tests/test_agent_background_output.py @@ -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 diff --git a/tests/test_agent_guardrail_controller.py b/tests/test_agent_guardrail_controller.py new file mode 100644 index 000000000..fd29d3bb4 --- /dev/null +++ b/tests/test_agent_guardrail_controller.py @@ -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" diff --git a/tests/test_agent_guardrails.py b/tests/test_agent_guardrails.py new file mode 100644 index 000000000..356054832 --- /dev/null +++ b/tests/test_agent_guardrails.py @@ -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()) diff --git a/tests/test_agent_subagent_reuse.py b/tests/test_agent_subagent_reuse.py index a8ab88f2c..21441be72 100644 --- a/tests/test_agent_subagent_reuse.py +++ b/tests/test_agent_subagent_reuse.py @@ -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( diff --git a/tests/test_agent_summarization_streaming.py b/tests/test_agent_summarization_streaming.py index 31a3fa73a..4c51e3c87 100644 --- a/tests/test_agent_summarization_streaming.py +++ b/tests/test_agent_summarization_streaming.py @@ -328,6 +328,7 @@ def test_streaming_agent_uses_non_streaming_llm_for_model_middlewares(with_invoc "RuntimeConfigMiddleware", "PlanMiddleware", "MemoryMiddleware", + "ToolGuardrailsMiddleware", "PatchToolCallsMiddleware", "_FakeToolSelectorMiddleware", "FinalRequestCompactionMiddleware", diff --git a/tests/test_agent_task_runs.py b/tests/test_agent_task_runs.py index 9b3b5124b..e93a90dc1 100644 --- a/tests/test_agent_task_runs.py +++ b/tests/test_agent_task_runs.py @@ -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"]