"""Turn-local idle / checklist guard to cut narration-only and no-progress loops.""" from __future__ import annotations import os from dataclasses import dataclass, field from typing import Any, Literal IdleAction = Literal["continue", "nudge", "early_finalize"] def _env_int(name: str, default: int, *, min_v: int = 1, max_v: int = 20) -> int: raw = str(os.getenv(name) or "").strip() if not raw: return default try: n = int(raw) except Exception: return default return max(min_v, min(int(n), max_v)) def idle_guard_enabled() -> bool: raw = str(os.getenv("AIA_TURN_IDLE_GUARD") or "1").strip().lower() return raw not in ("0", "false", "no", "off") @dataclass class RoundStats: round_idx: int had_tool_calls: bool ok_count: int = 0 fail_count: int = 0 schema_fail_count: int = 0 retry_guard_count: int = 0 tool_names: list[str] = field(default_factory=list) @dataclass class IdleGuardDecision: action: IdleAction reason: str nudge_text: str = "" @dataclass class TurnIdleTracker: """Tracks per-turn progress and decides nudge / early finalize.""" lang: str = "en" short_intent: str | None = None max_idle_rounds: int = field(default_factory=lambda: _env_int("AIA_TURN_IDLE_MAX_ROUNDS", 2)) max_schema_fails: int = field(default_factory=lambda: _env_int("AIA_TURN_IDLE_MAX_SCHEMA_FAILS", 3)) rounds: list[RoundStats] = field(default_factory=list) nudged: bool = False total_ok: int = 0 total_schema_fails: int = 0 def record_from_traces( self, *, round_idx: int, had_tool_calls: bool, round_traces: list[dict[str, Any]], results_by_id: dict[str, tuple[dict[str, Any], int]] | None = None, ) -> RoundStats: ok = 0 fail = 0 schema = 0 guard = 0 names: list[str] = [] # Prefer live results when available (richer failure_class). payloads: list[dict[str, Any]] = [] if results_by_id: for _cid, (payload, _dur) in results_by_id.items(): if isinstance(payload, dict): payloads.append(payload) for tr in round_traces or []: name = str((tr or {}).get("name") or "").strip() if name: names.append(name) if not payloads: if (tr or {}).get("ok") is True: ok += 1 elif (tr or {}).get("ok") is False: fail += 1 for payload in payloads: if payload.get("ok") is True: ok += 1 continue fail += 1 klass = str(payload.get("failure_class") or payload.get("error_class") or "").lower() code = str(payload.get("error_code") or "").lower() if klass == "schema_validation" or code in {"tool_invalid_arguments", "invalid_arguments"}: schema += 1 if klass == "retry_guard" or code in { "identical_retry_blocked", "retry_forbidden_blocked", "tool_loop_guard", "cli_call_budget_exceeded", "cli_fail_budget_exceeded", "cli_batch_budget_exceeded", "report_first_required", }: guard += 1 stats = RoundStats( round_idx=int(round_idx), had_tool_calls=bool(had_tool_calls), ok_count=int(ok), fail_count=int(fail), schema_fail_count=int(schema), retry_guard_count=int(guard), tool_names=names, ) self.rounds.append(stats) self.total_ok += int(ok) self.total_schema_fails += int(schema) return stats def decide_after_assistant_no_tools(self) -> IdleGuardDecision: """Model returned text without tool_calls.""" if not idle_guard_enabled(): return IdleGuardDecision(action="continue", reason="disabled") # Short-intent recipes require an evidence tool before answering. intent = str(self.short_intent or "").strip() if intent and intent != "continue" and self.total_ok == 0 and not self.nudged: self.nudged = True return IdleGuardDecision( action="nudge", reason="short_intent_narration_only", nudge_text=self._nudge_text(reason="narration_only"), ) return IdleGuardDecision(action="continue", reason="allow_text_reply") def decide_after_tools(self, stats: RoundStats) -> IdleGuardDecision: if not idle_guard_enabled(): return IdleGuardDecision(action="continue", reason="disabled") if self.total_schema_fails >= int(self.max_schema_fails) and self.total_ok == 0: return IdleGuardDecision( action="early_finalize", reason="schema_fail_budget", nudge_text=self._nudge_text(reason="schema_fail_budget"), ) # Count trailing idle rounds: tools ran but zero successes. idle_streak = 0 for r in reversed(self.rounds): if r.had_tool_calls and r.ok_count == 0: idle_streak += 1 continue break if idle_streak >= int(self.max_idle_rounds) and self.total_ok == 0: return IdleGuardDecision( action="early_finalize", reason="idle_no_progress", nudge_text=self._nudge_text(reason="idle_no_progress"), ) # One soft nudge when a round is all retry_guard / schema with no success yet. if ( stats.had_tool_calls and stats.ok_count == 0 and (stats.schema_fail_count + stats.retry_guard_count) > 0 and self.total_ok == 0 and not self.nudged ): self.nudged = True return IdleGuardDecision( action="nudge", reason="round_no_progress", nudge_text=self._nudge_text(reason="round_no_progress"), ) return IdleGuardDecision(action="continue", reason="progress_or_recoverable") def _nudge_text(self, *, reason: str) -> str: is_zh = str(self.lang or "").strip().lower().startswith("zh") intent = str(self.short_intent or "").strip() from runtime.tools.playbook_contracts import short_intent_first_step step = short_intent_first_step(intent) if is_zh: lines = ["[Idle guard] 本轮未取得有效工具证据，禁止空转。"] if step: tool, example = step lines.append(f"立即调用：{tool} 参数示例 {example}") elif reason == "schema_fail_budget": lines.append("参数多次不合法：按 tool 返回的 example/playbook 修正后只重试一次，然后直接作答。") else: lines.append("改参数或换 fallback 工具；若仍无证据则基于已知信息直接答复用户。") return "\n".join(lines) lines = ["[Idle guard] No usable tool evidence yet — stop spinning."] if step: tool, example = step lines.append(f"Call now: {tool} with example args {example}") elif reason == "schema_fail_budget": lines.append( "Repeated invalid arguments: fix once using the tool example/playbook, then answer." ) else: lines.append("Change args or switch fallback tools; if still blocked, answer from what you have.") return "\n".join(lines) __all__ = [ "IdleGuardDecision", "RoundStats", "TurnIdleTracker", "idle_guard_enabled", ]