diff --git a/supervisor/app.py b/supervisor/app.py index 08c7abf..3a4b33d 100644 --- a/supervisor/app.py +++ b/supervisor/app.py @@ -380,6 +380,7 @@ def cmd_run_foreground(args): mode=config.pause_handling_mode, max_auto_interventions=config.max_auto_interventions, ), + runtime_recovery_policy=config.runtime_recovery_policy(), ) print(f"[DEBUG MODE] Foreground controller — for debugging only") diff --git a/supervisor/config.py b/supervisor/config.py index 786d089..e2b1b62 100644 --- a/supervisor/config.py +++ b/supervisor/config.py @@ -18,6 +18,15 @@ "judge_temperature", "judge_max_tokens", "worker_trust_level", "notification_channels", "pause_handling_mode", "max_auto_interventions", "poll_interval_sec", "read_lines", + "runtime_recovery_enabled", "runtime_recovery_profile", + "runtime_recovery_reset_timezone", "runtime_recovery_reset_grace_seconds", + "runtime_recovery_transient_delay_seconds", + "runtime_recovery_rate_limit_fallback_delay_seconds", + "runtime_recovery_quiet_windows", "runtime_recovery_max_attempts_per_run", + "provider_retry_enabled", "provider_retry_reset_timezone", + "provider_retry_reset_grace_seconds", "provider_retry_transient_delay_seconds", + "provider_retry_rate_limit_fallback_delay_seconds", + "provider_retry_skip_windows", "provider_retry_max_attempts_per_run", "explainer_model", "explainer_temperature", "explainer_max_tokens", "deep_explainer_model", "deep_explainer_temperature", "deep_explainer_max_tokens", "clarification_escalation_confidence", @@ -78,13 +87,20 @@ def coerce_config_value(key: str, value: str): ftype = known[key].type if value.lower() in ("null", "none", "~"): return None - if ftype in ("float", float): + if _field_accepts(ftype, "float", float): return float(value) - if ftype in ("int", int): + if _field_accepts(ftype, "int", int): return int(value) + if _field_accepts(ftype, "bool", bool): + return value.lower() in ("1", "true", "yes", "on") return value +def _field_accepts(ftype, type_name: str, pytype) -> bool: + text = str(ftype) + return ftype in (type_name, pytype) or type_name in text + + @dataclass class RuntimeConfig: # -- Execution Surface -- @@ -139,6 +155,25 @@ class RuntimeConfig: branch_confidence_threshold: float = 0.75 default_agent_timeout_sec: int = 300 + # -- Runtime recovery -- + runtime_recovery_enabled: bool | None = None + runtime_recovery_profile: str | None = None + runtime_recovery_reset_timezone: str | None = None + runtime_recovery_reset_grace_seconds: int | None = None + runtime_recovery_transient_delay_seconds: int | None = None + runtime_recovery_rate_limit_fallback_delay_seconds: int | None = None + runtime_recovery_quiet_windows: list[str] | None = None + runtime_recovery_max_attempts_per_run: int | None = None + + # Back-compat aliases for configs written during the initial provider-retry rollout. + provider_retry_enabled: bool = True + provider_retry_reset_timezone: str = "Asia/Shanghai" + provider_retry_reset_grace_seconds: int = 60 + provider_retry_transient_delay_seconds: int = 60 + provider_retry_rate_limit_fallback_delay_seconds: int = 300 + provider_retry_skip_windows: list[str] = field(default_factory=list) + provider_retry_max_attempts_per_run: int = 3 + # -- Notifications -- notification_channels: list[dict] = field(default_factory=lambda: [ {"kind": "tmux_display"}, @@ -171,10 +206,12 @@ def from_env(cls, prefix: str = "SUPERVISOR_") -> "RuntimeConfig": if field_name not in known: continue ftype = known[field_name].type - if ftype in ("float", float): + if _field_accepts(ftype, "float", float): data[field_name] = float(val) - elif ftype in ("int", int): + elif _field_accepts(ftype, "int", int): data[field_name] = int(val) + elif _field_accepts(ftype, "bool", bool): + data[field_name] = val.lower() in ("1", "true", "yes", "on") else: data[field_name] = val return cls(**data) @@ -216,10 +253,12 @@ def load(cls, config_path: str | Path | None = None) -> "RuntimeConfig": if field_name not in known: continue ftype = known[field_name].type - if ftype in ("float", float): + if _field_accepts(ftype, "float", float): setattr(base, field_name, float(val)) - elif ftype in ("int", int): + elif _field_accepts(ftype, "int", int): setattr(base, field_name, int(val)) + elif _field_accepts(ftype, "bool", bool): + setattr(base, field_name, val.lower() in ("1", "true", "yes", "on")) else: setattr(base, field_name, val) return base @@ -233,6 +272,52 @@ def effective_target(self) -> str: """Resolve the effective surface target (surface_target > pane_target).""" return self.surface_target or self.pane_target + def runtime_recovery_policy(self): + from supervisor.runtime_recovery import RuntimeRecoveryPolicy, policy_with_profile + + defaults = RuntimeConfig() + windows = ( + self.runtime_recovery_quiet_windows + if self.runtime_recovery_quiet_windows is not None + else self.provider_retry_skip_windows + ) + if isinstance(windows, str): + windows = [item.strip() for item in windows.split(",") if item.strip()] + return policy_with_profile(RuntimeRecoveryPolicy( + enabled=( + self.runtime_recovery_enabled + if self.runtime_recovery_enabled is not None + else self.provider_retry_enabled + ), + profile=self.runtime_recovery_profile or "", + reset_timezone=( + self.runtime_recovery_reset_timezone + if self.runtime_recovery_reset_timezone is not None + else self.provider_retry_reset_timezone + ), + reset_grace_seconds=( + self.provider_retry_reset_grace_seconds + if self.runtime_recovery_reset_grace_seconds is None + else self.runtime_recovery_reset_grace_seconds + ), + transient_delay_seconds=( + self.provider_retry_transient_delay_seconds + if self.runtime_recovery_transient_delay_seconds is None + else self.runtime_recovery_transient_delay_seconds + ), + rate_limit_fallback_delay_seconds=( + self.provider_retry_rate_limit_fallback_delay_seconds + if self.runtime_recovery_rate_limit_fallback_delay_seconds is None + else self.runtime_recovery_rate_limit_fallback_delay_seconds + ), + quiet_windows=tuple(windows or ()), + max_attempts_per_run=( + self.provider_retry_max_attempts_per_run + if self.runtime_recovery_max_attempts_per_run is None + else self.runtime_recovery_max_attempts_per_run + ), + )) + def default_config_yaml(self) -> str: """Render a commented YAML template suitable for ``init``.""" return ( @@ -253,6 +338,19 @@ def default_config_yaml(self) -> str: "# trust: low | standard | high (high = minimal supervision)\n" f"worker_trust_level: \"{self.worker_trust_level}\"\n" "\n" + "# Runtime recovery: orthogonal to workflow steps; retries provider/connectivity failures.\n" + f"runtime_recovery_enabled: {str(self.runtime_recovery_enabled if self.runtime_recovery_enabled is not None else True).lower()}\n" + "# Optional preset profile. Example: \"glm\" reserves UTC+8 12:00-18:00 and uses 5h fallback.\n" + f"runtime_recovery_profile: \"{self.runtime_recovery_profile or ''}\"\n" + "# Naive reset timestamps in Chinese Claude/GLM output are usually UTC+8.\n" + f"runtime_recovery_reset_timezone: \"{self.runtime_recovery_reset_timezone or self.provider_retry_reset_timezone}\"\n" + f"runtime_recovery_reset_grace_seconds: {self.runtime_recovery_reset_grace_seconds if self.runtime_recovery_reset_grace_seconds is not None else self.provider_retry_reset_grace_seconds}\n" + f"runtime_recovery_transient_delay_seconds: {self.runtime_recovery_transient_delay_seconds if self.runtime_recovery_transient_delay_seconds is not None else self.provider_retry_transient_delay_seconds}\n" + f"runtime_recovery_rate_limit_fallback_delay_seconds: {self.runtime_recovery_rate_limit_fallback_delay_seconds if self.runtime_recovery_rate_limit_fallback_delay_seconds is not None else self.provider_retry_rate_limit_fallback_delay_seconds}\n" + "# Optional quiet windows, e.g. [\"UTC+8 12:00-18:00\"]\n" + "runtime_recovery_quiet_windows: []\n" + f"runtime_recovery_max_attempts_per_run: {self.runtime_recovery_max_attempts_per_run if self.runtime_recovery_max_attempts_per_run is not None else self.provider_retry_max_attempts_per_run}\n" + "\n" "# LLM judge (set to null for stub/offline mode)\n" "# Examples: anthropic/claude-haiku-4-5-20251001, openai/gpt-4o-mini\n" f"judge_model: null\n" diff --git a/supervisor/daemon/server.py b/supervisor/daemon/server.py index a223067..ffb614c 100644 --- a/supervisor/daemon/server.py +++ b/supervisor/daemon/server.py @@ -479,6 +479,7 @@ def _run_worker(self, entry: RunEntry, spec, state) -> None: mode=self.config.pause_handling_mode, max_auto_interventions=self.config.max_auto_interventions, ), + runtime_recovery_policy=self.config.runtime_recovery_policy(), ) loop.run_sidecar( spec, state, terminal, diff --git a/supervisor/loop.py b/supervisor/loop.py index adcb55f..e89e4bd 100644 --- a/supervisor/loop.py +++ b/supervisor/loop.py @@ -3,6 +3,7 @@ import logging import signal import time +from datetime import datetime, timezone from supervisor.domain.enums import DeliveryState, TopState, DecisionType from supervisor.domain.models import ( @@ -26,6 +27,12 @@ from supervisor.interventions import AutoInterventionManager from supervisor.notifications import NotificationEvent, NotificationManager from supervisor.pause_summary import PAUSE_CLASSES, latest_human_escalation, summarize_state +from supervisor.runtime_recovery import ( + RuntimeRecoveryObservation, + RuntimeRecoveryPolicy, + detect_runtime_recovery, + seconds_until_recovery, +) from supervisor.progress import write_progress from supervisor.protocol.reason_code import ( ESC_AUTHORIZATION_REQUIRED, @@ -82,7 +89,8 @@ def __init__(self, store, judge_model: str | None = None, judge_temperature: float = 0.1, judge_max_tokens: int = 512, worker_profile: WorkerProfile | None = None, notification_manager: NotificationManager | None = None, - auto_intervention_manager: AutoInterventionManager | None = None): + auto_intervention_manager: AutoInterventionManager | None = None, + runtime_recovery_policy: RuntimeRecoveryPolicy | None = None): self.store = store self.judge_client = JudgeClient( model=judge_model, @@ -98,6 +106,7 @@ def __init__(self, store, judge_model: str | None = None, self.worker_profile = worker_profile or WorkerProfile() self.notification_manager = notification_manager or NotificationManager() self.auto_intervention_manager = auto_intervention_manager or AutoInterventionManager(mode="notify_only") + self.runtime_recovery_policy = runtime_recovery_policy or RuntimeRecoveryPolicy() # Set while a sidecar loop is active; consulted by helpers that need # to cooperate with daemon stop_event / SIGTERM. self._interrupted_ref = None @@ -904,6 +913,11 @@ def _run_sidecar_inner( effective_idle_timeout_sec = ZERO_POLL_IDLE_TIMEOUT_SEC pending_text = None last_activity_at = time.monotonic() + recovery_attempts: dict[str, int] = {} + recovery_total_attempts = 0 + scheduled_recovery: RuntimeRecoveryObservation | None = None + scheduled_recovery_signature = "" + exhausted_recovery_signatures: set[str] = set() # Delivery ack: transient monotonic time of last injection (not persisted) delivery_ack_deadline = 0.0 # 0 = not awaiting ack DELIVERY_ACK_TIMEOUT = 60 # seconds @@ -1041,6 +1055,79 @@ def _run_sidecar_inner( # 2. Parse checkpoint with identity checkpoints = adapter.parse_checkpoints(text, run_id=state.run_id, surface_id=surface_id) if not checkpoints: + wall_clock_now = datetime.now(timezone.utc) + recovery = detect_runtime_recovery( + text, + now=wall_clock_now, + policy=self.runtime_recovery_policy, + ) + if recovery is not None: + if recovery_total_attempts < self.runtime_recovery_policy.max_attempts_per_run: + if scheduled_recovery is None or scheduled_recovery_signature != recovery.signature: + scheduled_recovery = recovery + scheduled_recovery_signature = recovery.signature + self.store.append_session_event( + state.run_id, + "runtime_recovery_scheduled", + { + "kind": recovery.kind, + "reason": recovery.reason, + "retry_at": recovery.retry_at.isoformat(), + "attempt": recovery_total_attempts + 1, + "max_attempts": self.runtime_recovery_policy.max_attempts_per_run, + }, + ) + self.store.save(state) + elif recovery.signature not in exhausted_recovery_signatures: + exhausted_recovery_signatures.add(recovery.signature) + self.store.append_session_event( + state.run_id, + "runtime_recovery_exhausted", + { + "kind": recovery.kind, + "reason": recovery.reason, + "attempts": recovery_total_attempts, + "max_attempts": self.runtime_recovery_policy.max_attempts_per_run, + }, + ) + self.store.save(state) + + if ( + scheduled_recovery is not None + and seconds_until_recovery(scheduled_recovery.retry_at, now=wall_clock_now) <= 0 + ): + attempts = recovery_attempts.get(scheduled_recovery.signature, 0) + instruction = self._build_runtime_recovery_instruction( + state, scheduled_recovery + ) + state.last_injected_node_id = state.current_node_id + state.last_injected_attempt = state.current_attempt + state.last_injection_seq = state.checkpoint_seq + self.store.save(state) + if not self._inject_or_pause(state, terminal, instruction, spec=spec): + return + recovery_attempts[scheduled_recovery.signature] = attempts + 1 + recovery_total_attempts += 1 + self.store.append_session_event( + state.run_id, + "runtime_recovery_injected", + { + "kind": scheduled_recovery.kind, + "reason": scheduled_recovery.reason, + "attempt": recovery_total_attempts, + "max_attempts": self.runtime_recovery_policy.max_attempts_per_run, + }, + ) + scheduled_recovery = None + scheduled_recovery_signature = "" + delivery_ack_deadline = time.monotonic() + DELIVERY_ACK_TIMEOUT + time.sleep(effective_poll_interval) + continue + + if scheduled_recovery is not None: + time.sleep(effective_poll_interval) + continue + if effective_idle_timeout_sec and effective_idle_timeout_sec > 0: idle_for = now - last_activity_at if idle_for >= effective_idle_timeout_sec: @@ -1165,6 +1252,8 @@ def _run_sidecar_inner( if (state.checkpoint_seq > state.last_injection_seq and state.delivery_state not in (DeliveryState.IDLE, DeliveryState.STARTED_PROCESSING)): self._set_delivery_state(state, DeliveryState.STARTED_PROCESSING, reason="checkpoint received") + scheduled_recovery = None + scheduled_recovery_signature = "" logger.info("checkpoint: %s (id=%s)", checkpoint.summary, checkpoint.checkpoint_id) if checkpoint.status in {"working", "step_done", "workflow_done"}: self._reset_recovery_tracking(state, clear_escalations=False) @@ -1347,6 +1436,21 @@ def _get_cwd(self, terminal, state=None) -> str | None: return state.workspace_root return None + def _build_runtime_recovery_instruction(self, state, recovery: RuntimeRecoveryObservation) -> HandoffInstruction: + content = ( + "retry\n\n" + "Runtime recovery detected a provider/connectivity failure outside the task logic. " + f"Retry the last interrupted action and continue current_node={state.current_node_id}. " + f"Observed {recovery.kind}: {recovery.reason}" + ) + return HandoffInstruction.make( + content=content, + node_id=state.current_node_id, + current_attempt=state.current_attempt, + triggered_by_decision_id="", + trigger_type="runtime_recovery", + ) + def _wait_for_injection_window(self, state, terminal, *, instruction_id: str) -> tuple[bool, str]: readiness_fn = getattr(terminal, "injection_readiness", None) if not callable(readiness_fn): diff --git a/supervisor/runtime_recovery.py b/supervisor/runtime_recovery.py new file mode 100644 index 0000000..25f193d --- /dev/null +++ b/supervisor/runtime_recovery.py @@ -0,0 +1,250 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass, replace +from datetime import datetime, time, timedelta, timezone +from zoneinfo import ZoneInfo + + +@dataclass(frozen=True) +class RuntimeRecoveryPolicy: + enabled: bool | None = True + profile: str = "" + reset_timezone: str | None = "Asia/Shanghai" + reset_grace_seconds: int | None = 60 + transient_delay_seconds: int | None = 60 + rate_limit_fallback_delay_seconds: int | None = 300 + quiet_windows: tuple[str, ...] = () + max_attempts_per_run: int | None = 3 + + +@dataclass(frozen=True) +class RuntimeRecoveryObservation: + kind: str + reason: str + retry_at: datetime + signature: str + + +PROFILE_DEFAULTS: dict[str, RuntimeRecoveryPolicy] = { + "glm": RuntimeRecoveryPolicy( + profile="glm", + reset_timezone="Asia/Shanghai", + reset_grace_seconds=60, + transient_delay_seconds=300, + rate_limit_fallback_delay_seconds=5 * 60 * 60, + quiet_windows=("UTC+8 12:00-18:00",), + max_attempts_per_run=3, + ), +} + + +_RESET_PATTERNS = ( + re.compile(r"限额将在\s*(?P\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})\s*重置"), + re.compile(r"reset(?:s|ting)?(?:\s+at|\s+on|:)?\s*(?P\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})", re.I), + re.compile(r"expires?(?:\s+at|\s+on|:)?\s*(?P\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})", re.I), + re.compile(r"try again(?:\s+at|\s+after|:)?\s*(?P\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})", re.I), + re.compile(r"until\s*(?P\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})", re.I), +) + +_TRANSIENT_PATTERNS = ( + re.compile(r"empty or malformed response\s*\(HTTP 200\)", re.I), + re.compile(r"proxy or gateway intercepting the request", re.I), + re.compile(r"\bnetwork (?:is )?(?:down|unavailable|disconnected|connection lost)\b", re.I), + re.compile(r"\b(?:connection|network) (?:reset|refused|timed out|timeout)\b", re.I), +) + + +def policy_with_profile(policy: RuntimeRecoveryPolicy) -> RuntimeRecoveryPolicy: + profile = (policy.profile or "").strip().lower() + base = PROFILE_DEFAULTS.get(profile) + if base is None: + return policy + defaults = RuntimeRecoveryPolicy() + return replace( + base, + enabled=policy.enabled if policy.enabled is not None else base.enabled, + profile=profile, + reset_timezone=( + policy.reset_timezone + if policy.reset_timezone is not None and policy.reset_timezone != defaults.reset_timezone + else base.reset_timezone + ), + reset_grace_seconds=( + policy.reset_grace_seconds + if policy.reset_grace_seconds is not None and policy.reset_grace_seconds != defaults.reset_grace_seconds + else base.reset_grace_seconds + ), + transient_delay_seconds=( + policy.transient_delay_seconds + if policy.transient_delay_seconds is not None and policy.transient_delay_seconds != defaults.transient_delay_seconds + else base.transient_delay_seconds + ), + rate_limit_fallback_delay_seconds=( + policy.rate_limit_fallback_delay_seconds + if ( + policy.rate_limit_fallback_delay_seconds is not None + and policy.rate_limit_fallback_delay_seconds != defaults.rate_limit_fallback_delay_seconds + ) + else base.rate_limit_fallback_delay_seconds + ), + quiet_windows=policy.quiet_windows or base.quiet_windows, + max_attempts_per_run=( + policy.max_attempts_per_run + if policy.max_attempts_per_run is not None and policy.max_attempts_per_run != defaults.max_attempts_per_run + else base.max_attempts_per_run + ), + ) + + +def detect_runtime_recovery( + text: str, + *, + now: datetime | None = None, + policy: RuntimeRecoveryPolicy | None = None, +) -> RuntimeRecoveryObservation | None: + policy = policy_with_profile(policy or RuntimeRecoveryPolicy()) + if not policy.enabled or not text: + return None + + now = _aware(now or datetime.now(timezone.utc)) + lowered = text.lower() + reset_retry_at = _extract_reset_time(text, policy=policy) + + if reset_retry_at is not None: + retry_at = next_allowed_recovery_at(reset_retry_at, policy) + return RuntimeRecoveryObservation( + kind="rate_limit", + reason=_compact_reason(text), + retry_at=retry_at, + signature=_signature("rate_limit", text), + ) + + if "429" in lowered or "rate limit" in lowered or "使用上限" in text: + retry_at = now + timedelta(seconds=policy.rate_limit_fallback_delay_seconds or 0) + retry_at = next_allowed_recovery_at(retry_at, policy) + return RuntimeRecoveryObservation( + kind="rate_limit", + reason=_compact_reason(text), + retry_at=retry_at, + signature=_signature("rate_limit", text), + ) + + if any(pattern.search(text) for pattern in _TRANSIENT_PATTERNS): + retry_at = now + timedelta(seconds=policy.transient_delay_seconds or 0) + retry_at = next_allowed_recovery_at(retry_at, policy) + return RuntimeRecoveryObservation( + kind="transient_connectivity", + reason=_compact_reason(text), + retry_at=retry_at, + signature=_signature("transient_connectivity", text), + ) + + return None + + +def next_allowed_recovery_at(candidate: datetime, policy: RuntimeRecoveryPolicy) -> datetime: + policy = policy_with_profile(policy) + result = _aware(candidate) + for _ in range(max(1, len(policy.quiet_windows) + 1)): + moved = False + for raw in policy.quiet_windows: + window = _parse_quiet_window(raw, default_tz=policy.reset_timezone) + if window is None: + continue + tz, start, end = window + local = result.astimezone(tz) + if not _time_in_window(local.timetz().replace(tzinfo=None), start, end): + continue + local_end = _window_end(local, start, end) + result = local_end.astimezone(result.tzinfo or timezone.utc) + moved = True + if not moved: + return result + return result + + +def seconds_until_recovery(retry_at: datetime, *, now: datetime | None = None) -> float: + now = _aware(now or datetime.now(timezone.utc)) + return max(0.0, (_aware(retry_at).astimezone(timezone.utc) - now.astimezone(timezone.utc)).total_seconds()) + + +def _extract_reset_time(text: str, *, policy: RuntimeRecoveryPolicy) -> datetime | None: + for pattern in _RESET_PATTERNS: + match = pattern.search(text) + if not match: + continue + parsed = datetime.fromisoformat(match.group("ts").replace(" ", "T")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=_timezone(policy.reset_timezone)) + return parsed + timedelta(seconds=policy.reset_grace_seconds) + return None + + +def _parse_quiet_window(raw: str, *, default_tz: str) -> tuple[timezone | ZoneInfo, time, time] | None: + parts = raw.strip().split() + if not parts: + return None + if len(parts) == 1: + tz_name = default_tz + span = parts[0] + else: + tz_name = parts[0] + span = parts[1] + match = re.fullmatch(r"(?P\d{1,2}:\d{2})-(?P\d{1,2}:\d{2})", span) + if not match: + return None + return ( + _timezone(tz_name), + time.fromisoformat(match.group("start")), + time.fromisoformat(match.group("end")), + ) + + +def _timezone(name: str) -> timezone | ZoneInfo: + normalized = (name or "UTC").strip() + match = re.fullmatch(r"UTC([+-])(\d{1,2})(?::?(\d{2}))?", normalized, re.I) + if match: + sign = 1 if match.group(1) == "+" else -1 + hours = int(match.group(2)) + minutes = int(match.group(3) or "0") + return timezone(sign * timedelta(hours=hours, minutes=minutes)) + if normalized.upper() == "UTC": + return timezone.utc + try: + return ZoneInfo(normalized) + except Exception: + return timezone.utc + + +def _time_in_window(value: time, start: time, end: time) -> bool: + if start < end: + return start <= value < end + return value >= start or value < end + + +def _window_end(local_dt: datetime, start: time, end: time) -> datetime: + end_dt = local_dt.replace(hour=end.hour, minute=end.minute, second=0, microsecond=0) + if start >= end and local_dt.timetz().replace(tzinfo=None) >= start: + end_dt += timedelta(days=1) + if end_dt <= local_dt: + end_dt += timedelta(days=1) + return end_dt + + +def _aware(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + + +def _compact_reason(text: str) -> str: + lines = [line.strip() for line in text.splitlines() if line.strip()] + joined = " ".join(lines) + return joined[:300] + + +def _signature(kind: str, text: str) -> str: + normalized = re.sub(r"\s+", " ", text.strip().lower()) + normalized = re.sub(r"\d{4}-\d{2}-\d{2}[ t]\d{2}:\d{2}:\d{2}", "", normalized) + return f"{kind}:{normalized[:180]}" diff --git a/tests/test_runtime_recovery.py b/tests/test_runtime_recovery.py new file mode 100644 index 0000000..7e07953 --- /dev/null +++ b/tests/test_runtime_recovery.py @@ -0,0 +1,296 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from zoneinfo import ZoneInfo + +from supervisor.loop import SupervisorLoop +from supervisor.plan.loader import load_spec +from supervisor.config import RuntimeConfig +from supervisor.runtime_recovery import ( + RuntimeRecoveryPolicy, + detect_runtime_recovery, + next_allowed_recovery_at, + policy_with_profile, +) +from supervisor.storage.state_store import StateStore + + +class _Terminal: + def __init__(self, outputs: list[str]): + self.outputs = list(outputs) + self.injected: list[str] = [] + self._read_done = False + self.read_count = 0 + + def read(self, lines: int = 100) -> str: + self._read_done = True + self.read_count += 1 + if self.outputs: + return self.outputs.pop(0) + return "" + + def inject(self, text: str) -> None: + assert self._read_done + self.injected.append(text) + self._read_done = False + + +class _StopAfterRetry: + def __init__(self, terminal: _Terminal): + self.terminal = terminal + + def is_set(self) -> bool: + return any(text.startswith("retry") for text in self.terminal.injected) + + +class _StopAfterReads: + def __init__(self, terminal: _Terminal, count: int): + self.terminal = terminal + self.count = count + + def is_set(self) -> bool: + return self.terminal.read_count >= self.count + + +def _checkpoint(status: str = "working") -> str: + return ( + "\n" + f"status: {status}\n" + "current_node: write_test\n" + "summary: recovered with real work\n" + "evidence:\n" + " - command: pytest\n" + "candidate_next_actions:\n" + " - continue\n" + "needs:\n" + " - none\n" + "question_for_supervisor:\n" + " - none\n" + "\n" + ) + + +def test_detects_claude_429_reset_as_utc8_and_adds_retry_grace() -> None: + text = ( + "API Error: Request rejected (429) · 已达到 5 小时的使用上限。" + "您的限额将在 2026-05-15 04:07:12 重置。" + ) + now = datetime(2026, 5, 14, 20, 0, tzinfo=ZoneInfo("Asia/Shanghai")) + policy = RuntimeRecoveryPolicy(reset_timezone="Asia/Shanghai", reset_grace_seconds=60) + + retry = detect_runtime_recovery(text, now=now, policy=policy) + + assert retry is not None + assert retry.kind == "rate_limit" + assert retry.retry_at == datetime(2026, 5, 15, 4, 8, 12, tzinfo=ZoneInfo("Asia/Shanghai")) + + +def test_retry_policy_skips_configured_utc8_high_watermark() -> None: + policy = RuntimeRecoveryPolicy(quiet_windows=("UTC+8 12:00-18:00",)) + candidate = datetime(2026, 5, 15, 4, 30, tzinfo=timezone.utc) + + retry_at = next_allowed_recovery_at(candidate, policy) + + assert retry_at == datetime(2026, 5, 15, 10, 0, tzinfo=timezone.utc) + + +def test_detects_malformed_200_gateway_error_for_short_retry() -> None: + text = ( + "API Error: API returned an empty or malformed response (HTTP 200) " + "— check for a proxy or gateway intercepting the request" + ) + now = datetime(2026, 5, 14, 12, 0, tzinfo=timezone.utc) + policy = RuntimeRecoveryPolicy(transient_delay_seconds=90) + + retry = detect_runtime_recovery(text, now=now, policy=policy) + + assert retry is not None + assert retry.kind == "transient_connectivity" + assert retry.retry_at == datetime(2026, 5, 14, 12, 1, 30, tzinfo=timezone.utc) + + +def test_detects_english_expires_at_reset_text() -> None: + text = "API Error: Request rejected (429). Limit expires at 2026-05-15 04:07:12." + now = datetime(2026, 5, 14, 20, 0, tzinfo=ZoneInfo("Asia/Shanghai")) + policy = RuntimeRecoveryPolicy(reset_timezone="Asia/Shanghai", reset_grace_seconds=60) + + retry = detect_runtime_recovery(text, now=now, policy=policy) + + assert retry is not None + assert retry.retry_at == datetime(2026, 5, 15, 4, 8, 12, tzinfo=ZoneInfo("Asia/Shanghai")) + + +def test_detects_timestamp_only_reset_text() -> None: + text = "Limit expires at 2026-05-15 04:07:12." + now = datetime(2026, 5, 14, 20, 0, tzinfo=ZoneInfo("Asia/Shanghai")) + policy = RuntimeRecoveryPolicy(reset_timezone="Asia/Shanghai", reset_grace_seconds=60) + + retry = detect_runtime_recovery(text, now=now, policy=policy) + + assert retry is not None + assert retry.kind == "rate_limit" + assert retry.retry_at == datetime(2026, 5, 15, 4, 8, 12, tzinfo=ZoneInfo("Asia/Shanghai")) + + +def test_glm_profile_reserves_midday_and_uses_five_hour_fallback() -> None: + policy = policy_with_profile(RuntimeRecoveryPolicy(profile="glm")) + now = datetime(2026, 5, 15, 11, 30, tzinfo=ZoneInfo("Asia/Shanghai")) + + retry = detect_runtime_recovery( + "API Error: Request rejected (429)", + now=now, + policy=policy, + ) + + assert retry is not None + assert retry.retry_at == datetime(2026, 5, 15, 18, 0, tzinfo=ZoneInfo("Asia/Shanghai")) + + +def test_invalid_timezone_config_falls_back_to_utc() -> None: + policy = RuntimeRecoveryPolicy(reset_timezone="Not/AZone") + + retry = detect_runtime_recovery( + "API Error: Request rejected (429). Limit expires at 2026-05-15 04:07:12.", + now=datetime(2026, 5, 14, 20, 0, tzinfo=timezone.utc), + policy=policy, + ) + + assert retry is not None + assert retry.retry_at == datetime(2026, 5, 15, 4, 8, 12, tzinfo=timezone.utc) + + +def test_sidecar_injects_retry_for_malformed_http_200_error(tmp_path, monkeypatch) -> None: + spec = load_spec("specs/examples/linear_plan.example.yaml") + store = StateStore(str(tmp_path / "runtime")) + state = store.load_or_init(spec) + policy = RuntimeRecoveryPolicy(enabled=True, transient_delay_seconds=0) + loop = SupervisorLoop(store, runtime_recovery_policy=policy) + terminal = _Terminal([ + "", + ( + "⎿ API Error: API returned an empty or malformed response (HTTP 200) " + "— check for a proxy or gateway intercepting the request\n" + "✻ Cooked for 23m 57s\n" + ), + ]) + + monkeypatch.setattr("supervisor.loop.time.sleep", lambda seconds: None) + + final = loop.run_sidecar( + spec, + state, + terminal, + poll_interval=0, + read_lines=50, + stop_event=_StopAfterRetry(terminal), + ) + + retry_injections = [text for text in terminal.injected if text.startswith("retry")] + assert retry_injections + assert "empty or malformed response" in retry_injections[-1] + assert final.top_state.value in {"ATTACHED", "RUNNING"} + + +def test_sidecar_reuses_original_fallback_deadline_for_repeated_error(tmp_path, monkeypatch) -> None: + spec = load_spec("specs/examples/linear_plan.example.yaml") + store = StateStore(str(tmp_path / "runtime")) + state = store.load_or_init(spec) + policy = RuntimeRecoveryPolicy(enabled=True, rate_limit_fallback_delay_seconds=120) + loop = SupervisorLoop(store, runtime_recovery_policy=policy) + error = "API Error: Request rejected (429)" + terminal = _Terminal(["", error, error, error]) + times = iter([ + datetime(2026, 5, 15, 12, 0, tzinfo=timezone.utc), + datetime(2026, 5, 15, 12, 1, tzinfo=timezone.utc), + datetime(2026, 5, 15, 12, 2, tzinfo=timezone.utc), + datetime(2026, 5, 15, 12, 3, tzinfo=timezone.utc), + ]) + + class _FakeDateTime: + @staticmethod + def now(tz=None): + value = next(times) + return value.astimezone(tz) if tz else value.replace(tzinfo=None) + + monkeypatch.setattr("supervisor.loop.datetime", _FakeDateTime) + monkeypatch.setattr("supervisor.loop.time.sleep", lambda seconds: None) + + final = loop.run_sidecar( + spec, + state, + terminal, + poll_interval=0, + read_lines=50, + stop_event=_StopAfterRetry(terminal), + ) + + retry_injections = [text for text in terminal.injected if text.startswith("retry")] + assert retry_injections + assert final.top_state.value in {"ATTACHED", "RUNNING"} + + +def test_sidecar_clears_scheduled_recovery_after_real_checkpoint(tmp_path, monkeypatch) -> None: + spec = load_spec("specs/examples/linear_plan.example.yaml") + store = StateStore(str(tmp_path / "runtime")) + state = store.load_or_init(spec) + policy = RuntimeRecoveryPolicy(enabled=True, rate_limit_fallback_delay_seconds=120) + loop = SupervisorLoop(store, runtime_recovery_policy=policy) + terminal = _Terminal(["", "API Error: Request rejected (429)", _checkpoint(), ""]) + times = iter([ + datetime(2026, 5, 15, 12, 0, tzinfo=timezone.utc), + datetime(2026, 5, 15, 12, 3, tzinfo=timezone.utc), + ]) + + class _FakeDateTime: + @staticmethod + def now(tz=None): + value = next(times) + return value.astimezone(tz) if tz else value.replace(tzinfo=None) + + monkeypatch.setattr("supervisor.loop.datetime", _FakeDateTime) + monkeypatch.setattr("supervisor.loop.time.sleep", lambda seconds: None) + + loop.run_sidecar( + spec, + state, + terminal, + poll_interval=0, + read_lines=50, + stop_event=_StopAfterReads(terminal, 4), + ) + + assert not any(text.startswith("retry") for text in terminal.injected) + + +def test_runtime_recovery_config_fields_override_provider_aliases() -> None: + cfg = RuntimeConfig( + runtime_recovery_enabled=True, + provider_retry_enabled=False, + runtime_recovery_reset_timezone="UTC", + provider_retry_reset_timezone="Asia/Shanghai", + ) + + policy = cfg.runtime_recovery_policy() + + assert policy.enabled is True + assert policy.reset_timezone == "UTC" + + +def test_provider_aliases_are_used_when_runtime_recovery_fields_are_unset() -> None: + cfg = RuntimeConfig( + runtime_recovery_enabled=None, + runtime_recovery_reset_timezone=None, + runtime_recovery_transient_delay_seconds=None, + runtime_recovery_max_attempts_per_run=None, + provider_retry_enabled=False, + provider_retry_reset_timezone="UTC", + provider_retry_transient_delay_seconds=42, + provider_retry_max_attempts_per_run=7, + ) + + policy = cfg.runtime_recovery_policy() + + assert policy.enabled is False + assert policy.reset_timezone == "UTC" + assert policy.transient_delay_seconds == 42 + assert policy.max_attempts_per_run == 7