From 707083da97de2f6ae104dba1b3b2f2f7eac7f4c0 Mon Sep 17 00:00:00 2001 From: Anu Kheru Date: Sun, 3 May 2026 18:42:07 -0300 Subject: [PATCH 1/6] =?UTF-8?q?ah:=20WS-AUTO-002=20AFK=20Work=20Loop=20?= =?UTF-8?q?=E2=80=94=20autonomous=20worker=20+=20dynamic=20heartbeats?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit While AH is in afk_manual or afk_auto, a new background thread (agent/afk_worker.py) ticks every 30 min, picks a tagged in_progress task from configured status files, routes it to a per-type provider/ model (Sonnet via OpenRouter for code, MiniMax for research/doc/ summary), and delegates execution to a leaf sub-agent. Each cycle is appended to ~/AFK_LOG.md (markdown) and ~/.hermes/afk_log.jsonl (machine-readable). MAX_DEPTH=50 finally has a chain to consume. Heartbeats (HB1/HB2/HB3) now embed a deterministic recap built from afk_log.jsonl — counts per status, top 3 completed tasks, current quota, cooldown level — replacing the aspirational "to be filled by AH later" wording. Safety: - §9.1 §9.3.2 §9.3.4: all parameters via ~/.hermes/config.yaml afk_worker.* (interval, status_files, threshold, model_routing, toolsets, log paths, picked-index path) - §2.4 SECCOMP: thread daemon in hermes-agent.service, no fork+exec - §2.5 budget: MiniMax quota gate at 80% (5h OR weekly) - Tags: [priority:P0|P1|P2], [afk:research|code_review|...], [afk:no], [blocked:*], [afk:decision_structurelle] (auto Discord block + skip) - Picked-index dedupe (~/.hermes/afk_worker_picked.json) avoids re-picking completed tasks; auto-purged when AH exits AFK - Cooldown: 3 errors -> cooldown_level++ (skip 2/4 ticks; halt at 3) - delegate_task gains 4 optional params (model_override, provider_override, base_url_override, api_key_override) - non- breaking, lets the worker route per-call without mutating config Tests: 33 new + 53 existing AFK tests all green (no regression). --- agent/afk_heartbeat.py | 131 ++++- agent/afk_scheduler.py | 10 + agent/afk_worker.py | 885 +++++++++++++++++++++++++++++++++ gateway/run.py | 40 +- tests/agent/test_afk_worker.py | 586 ++++++++++++++++++++++ tools/delegate_tool.py | 24 + 6 files changed, 1665 insertions(+), 11 deletions(-) create mode 100644 agent/afk_worker.py create mode 100644 tests/agent/test_afk_worker.py diff --git a/agent/afk_heartbeat.py b/agent/afk_heartbeat.py index fd4ffcfd757c..2da91f2cb02a 100644 --- a/agent/afk_heartbeat.py +++ b/agent/afk_heartbeat.py @@ -18,8 +18,11 @@ from __future__ import annotations +import json import logging +import os from datetime import datetime, timedelta, timezone +from pathlib import Path from typing import Optional from agent.afk_state import ( @@ -42,6 +45,10 @@ HB_DAYS = (3, 7, 14) STAND_BY_DELAY_HOURS_AFTER_HB3 = 24 +# WS-AUTO-002 — dynamic recap config defaults (mirror afk_worker.DEFAULT_*) +DEFAULT_LOG_JSONL = "~/.hermes/afk_log.jsonl" +DEFAULT_RECAP_N_CYCLES = 20 + def _parse_iso(ts: Optional[str]) -> Optional[datetime]: if not ts: @@ -52,6 +59,108 @@ def _parse_iso(ts: Optional[str]) -> Optional[datetime]: return None +def _build_dynamic_recap( + state: AFKState, + n_recent_cycles: int = DEFAULT_RECAP_N_CYCLES, + cfg: Optional[dict] = None, +) -> str: + """Build a deterministic recap of the AFK Work Loop activity since the + user entered AFK. + + Reads ~/.hermes/afk_log.jsonl (written by agent.afk_worker), filters to + cycles with started_at >= state.entered_at, and produces a markdown + summary suitable for embedding in a Discord heartbeat message. + + On any failure (file missing, corrupt JSONL, etc.) returns a fallback + string explicitly stating the data was unavailable — never raises. + """ + if cfg is None: + try: + from hermes_cli.config import load_config + cfg = (load_config() or {}).get("afk_worker") or {} + except Exception: + cfg = {} + + jsonl_path = Path( + os.path.expanduser(cfg.get("log_jsonl_path") or DEFAULT_LOG_JSONL) + ) + if not jsonl_path.is_file(): + return ( + "_(Aucun cycle AFK enregistré pour le moment — `~/.hermes/" + "afk_log.jsonl` absent.)_" + ) + + entered = _parse_iso(state.entered_at) if state.entered_at else None + cycles: list[dict] = [] + try: + with jsonl_path.open(encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + ts = _parse_iso(obj.get("started_at") or obj.get("ts")) + if entered is not None and ts is not None and ts < entered: + continue + cycles.append(obj) + except OSError as e: + return f"_(Erreur lecture `{jsonl_path}` : {e}.)_" + + if not cycles: + return "_(0 cycle enregistré depuis le début de cet AFK.)_" + + cycles = cycles[-n_recent_cycles:] + n_total = len(cycles) + n_completed = sum(1 for c in cycles if c.get("status") == "completed") + n_skip_q = sum(1 for c in cycles if c.get("status") == "skipped_quota") + n_skip_t = sum(1 for c in cycles if c.get("status") == "skipped_no_task") + n_blocked = sum(1 for c in cycles if c.get("status") == "skipped_decision_blocked") + n_errors = sum(1 for c in cycles if c.get("status") == "error") + + completed_titles = [ + (c.get("task") or {}).get("title", "?") + for c in reversed(cycles) + if c.get("status") == "completed" + ][:3] + + last = cycles[-1] + qa = last.get("quota_after") or {} + quota_line = "" + pct_5h = qa.get("minimax_5h_pct") + pct_w = qa.get("minimax_weekly_pct") + if pct_5h is not None or pct_w is not None: + quota_line = ( + f"\n• Quota MiniMax au dernier tick : " + f"{pct_5h if pct_5h is not None else '?'}% (5h) / " + f"{pct_w if pct_w is not None else '?'}% (semaine)" + ) + + cooldown_line = "" + if (state.cooldown_level or 0) > 0: + cooldown_line = ( + f"\n• ⚠️ Cooldown level **{state.cooldown_level}** actif " + f"(erreurs consécutives sur les délégations)" + ) + + titles_block = "" + if completed_titles: + titles_block = "\n• Dernières tâches complétées :\n - " + "\n - ".join( + completed_titles + ) + + return ( + f"📊 **Récap AFK** — {n_total} cycle(s) depuis le début " + f"(complétés: {n_completed}, skip quota: {n_skip_q}, " + f"skip no-task: {n_skip_t}, blocked: {n_blocked}, errors: {n_errors})" + f"{titles_block}" + f"{quota_line}" + f"{cooldown_line}" + ) + + def _format_hb_message(hb_cycle: int, days_in_afk: int, state: AFKState) -> str: """Build the Discord message for a heartbeat at the given cycle.""" common_tail = ( @@ -64,24 +173,27 @@ def _format_hb_message(hb_cycle: int, days_in_afk: int, state: AFKState) -> str: intervals_str = ", ".join(f"J{int(d)}" if d == int(d) else f"J{d}" for d in HB_DAYS) + # WS-AUTO-002 — replace the aspirational "to be filled by AH later" + # wording with a deterministic recap built from afk_log.jsonl. + recap = _build_dynamic_recap(state) + if hb_cycle == 1: return ( f"💓 **Heartbeat 1/{len(HB_DAYS)} — J+{days_in_afk}** " f"(intervals: {intervals_str}).\n\n" f"Ça fait {days_in_afk} jours que tu es en AFK ({state.mode}). " f"J'espère que tu vas bien.\n\n" - f"📊 État de mon travail (à compléter par AH lui-même au prochain " - f"démarrage de session — ce ping est juste l'alerte schedule).\n" - f"Voir ~/AFK_LOG.md pour le récap détaillé." + f"{recap}\n\n" + f"Détails complets : `~/AFK_LOG.md`." ) + common_tail if hb_cycle == 2: return ( f"💓 **Heartbeat 2/{len(HB_DAYS)} — J+{days_in_afk}**.\n\n" - f"Toujours pas de nouvelles depuis le HB1. Je continue mon travail " - f"selon les missions whitelist.\n\n" - f"📊 Récap à voir dans ~/AFK_LOG.md (mis à jour par AH à chaque " - f"changement d'état)." + f"Toujours pas de nouvelles depuis le HB1. Le worker AFK continue " + f"de tourner sur la backlog whitelist.\n\n" + f"{recap}\n\n" + f"Détails : `~/AFK_LOG.md`." ) + common_tail if hb_cycle == 3: @@ -92,8 +204,9 @@ def _format_hb_message(hb_cycle: int, days_in_afk: int, state: AFKState) -> str: f"Khéri, ça fait {days_in_afk} jours sans nouvelles. Si tu ne réponds " f"pas dans les **{sb_str}**, je passe en mode **stand_by** complet :\n" f"• Plus de nouveaux commits / missions\n" - f"• Lecture / monitoring uniquement\n" - f"• Récap final posté ici\n\n" + f"• Worker AFK arrêté\n" + f"• Lecture / monitoring uniquement\n\n" + f"{recap}\n\n" f"J'espère que tout va bien de ton côté." ) + common_tail diff --git a/agent/afk_scheduler.py b/agent/afk_scheduler.py index 474b5a636b02..f1d7d32901a0 100644 --- a/agent/afk_scheduler.py +++ b/agent/afk_scheduler.py @@ -144,6 +144,16 @@ def process_user_message(content: str, now_utc: Optional[datetime] = None) -> tu # Persist if anything changed (mode flip OR last_user_msg_at update). save_state(state) + # WS-AUTO-002 OQ4 — purge the AFK worker picked-index whenever AH flips + # back to normal. This guarantees the next AFK starts with a clean slate + # (no stale "already completed" entries from a previous session). + if state.mode != old_mode and state.mode == MODE_NORMAL: + try: + from agent.afk_worker import purge_picked_index + purge_picked_index() + except Exception: + logger.debug("afk-worker picked-index purge failed", exc_info=True) + if state.mode != old_mode: logger.info("AFK state: %s -> %s (user msg: %r)", old_mode, state.mode, content[:80]) diff --git a/agent/afk_worker.py b/agent/afk_worker.py new file mode 100644 index 000000000000..ec562c983dd2 --- /dev/null +++ b/agent/afk_worker.py @@ -0,0 +1,885 @@ +"""AFK Work Loop — WS-AUTO-002. + +Background thread that, while AH is in afk_manual or afk_auto, periodically: + 1. Checks MiniMax quota headroom + 2. Picks a tagged in_progress task from configured status files + 3. Routes to a model based on the task type (matrix from config.yaml) + 4. Delegates execution to a leaf sub-agent via delegate_task + 5. Appends the cycle to ~/AFK_LOG.md + ~/.hermes/afk_log.jsonl + +Driven by gateway/run.py at service startup; stops on stop_event. + +Spec reference: ~/Documents/Projects/ACOS-HERMES/.workspace/WS_AFK_WORK_LOOP_SPEC.md +""" + +from __future__ import annotations + +import fcntl +import json +import logging +import os +import re +import tempfile +import threading +import time +from dataclasses import dataclass, field, asdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from agent.afk_state import ( + AFKState, + MODE_STAND_BY, + MODE_NORMAL, + AFK_MODES, + load_state, + save_state, + now_iso, +) + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Defaults — all overridable via ~/.hermes/config.yaml afk_worker.* +# --------------------------------------------------------------------------- + +DEFAULT_INTERVAL_MIN = 30 +DEFAULT_QUOTA_THRESHOLD_PCT = 80 +DEFAULT_LOG_MD = "~/AFK_LOG.md" +DEFAULT_LOG_JSONL = "~/.hermes/afk_log.jsonl" +DEFAULT_PICKED_INDEX = "~/.hermes/afk_worker_picked.json" +DEFAULT_STAND_BY_BEHAVIOR = "stop" # Q8 + +# Cooldown (§3.5) +COOLDOWN_ERROR_THRESHOLD = 3 +COOLDOWN_LEVEL_TICK_SKIPS: dict[int, float] = {1: 2, 2: 4, 3: float("inf")} +ERROR_COOLDOWN_TICKS = 3 # ticks to skip a task that errored before retry + +# Tag regexes (§3.4) +RE_PRIORITY = re.compile(r"\[priority:(P0|P1|P2)\]", re.IGNORECASE) +RE_AFK_TYPE = re.compile(r"\[afk:([a-z_]+)\]", re.IGNORECASE) +RE_BLOCKED = re.compile(r"\[blocked:[^\]]+\]", re.IGNORECASE) +RE_AFK_NO = re.compile(r"\[afk:no\]", re.IGNORECASE) +RE_TAG_ANY = re.compile(r"\[[a-z_]+:[^\]]+\]", re.IGNORECASE) + +# Markdown task line — captures the checkbox marker and the rest. +# Eligible markers: ' ' (todo) and '/' (in_progress). +RE_TASK_LINE = re.compile(r"^\s*-\s*\[([ /])\]\s+(.+?)\s*$") +RE_HEADING = re.compile(r"^(#{1,6})\s+(.+?)\s*$") + +# Priority sort key (lower = higher priority) +_PRIORITY_RANK = {"P0": 0, "P1": 1, "P2": 2} + + +# --------------------------------------------------------------------------- +# Module-level parent-agent getter (set by gateway/run.py at startup) +# --------------------------------------------------------------------------- + +_RUNTIME_PARENT_AGENT_GETTER = None + + +def install_parent_agent_getter(fn) -> None: + """Register a callable that returns the active parent_agent. + + Called by gateway/run.py during startup. Lets the worker thread reach + into the gateway's session-keyed agent map without holding a reference + cycle. + """ + global _RUNTIME_PARENT_AGENT_GETTER + _RUNTIME_PARENT_AGENT_GETTER = fn + + +def _get_runtime_parent_agent(): + if _RUNTIME_PARENT_AGENT_GETTER is None: + return None + try: + return _RUNTIME_PARENT_AGENT_GETTER() + except Exception: + logger.exception("afk-worker: parent_agent getter raised") + return None + + +# --------------------------------------------------------------------------- +# Dataclasses +# --------------------------------------------------------------------------- + + +@dataclass +class TaskCandidate: + source_file: str + section: str + line_number: int + raw_line: str + title: str + priority: str # "P0" | "P1" | "P2" + afk_type: str + is_excluded: bool = False + exclusion_reason: Optional[str] = None + + def task_key(self) -> str: + return f"{self.source_file}::{self.section}::{self.title}" + + def to_jsonable(self) -> dict[str, Any]: + return { + "source_file": self.source_file, + "section": self.section, + "line_number": self.line_number, + "title": self.title, + "priority": self.priority, + "afk_type": self.afk_type, + } + + +@dataclass +class CycleResult: + tick_id: str + started_at: str + completed_at: Optional[str] = None + afk_mode: Optional[str] = None + status: str = "pending" + task: Optional[TaskCandidate] = None + model_used: Optional[dict[str, str]] = None + delegation_summary: Optional[str] = None + delegation_api_calls: Optional[int] = None + delegation_duration_s: Optional[float] = None + delegation_status: Optional[str] = None + delegation_error: Optional[str] = None + quota_before: Optional[dict] = None + quota_after: Optional[dict] = None + notes: list[str] = field(default_factory=list) + + +CYCLE_STATUSES = ( + "completed", + "skipped_quota", + "skipped_no_task", + "skipped_decision_blocked", + "skipped_cooldown", + "error", + "interrupted", +) + + +# --------------------------------------------------------------------------- +# AFKWorker +# --------------------------------------------------------------------------- + + +class AFKWorker: + """Background thread driving the AFK Work Loop.""" + + def __init__( + self, + stop_event: threading.Event, + adapters=None, + loop=None, + ): + self.stop_event = stop_event + self.adapters = adapters + self.loop = loop + self._delegation_lock = threading.Lock() # Q6: 1 delegation at a time + self._cooldown_skip_remaining = 0 + self._consecutive_errors = 0 + + # ───────────────────────── thread entry-point ───────────────────── + + def run(self) -> None: + """Loop until stop_event is set.""" + cfg = self._load_cfg() + if not cfg.get("enabled", True): + logger.info("afk-worker disabled via config; thread exits") + return + + interval_s = self._interval_seconds(cfg) + logger.info("afk-worker thread started (interval=%ds)", interval_s) + + # Initial sleep so the worker doesn't stampede right after a service + # restart (heartbeat ticker is doing its own thing). + if self.stop_event.wait(interval_s): + return + + while not self.stop_event.is_set(): + try: + self._tick(cfg) + except Exception: + logger.exception("afk-worker tick raised; will retry next interval") + self._consecutive_errors += 1 + if self._consecutive_errors >= COOLDOWN_ERROR_THRESHOLD: + self._raise_cooldown() + + # Re-load config each tick so live edits apply without restart + cfg = self._load_cfg() + if not cfg.get("enabled", True): + logger.info("afk-worker disabled via config; thread exits") + return + interval_s = self._interval_seconds(cfg) + + if self.stop_event.wait(interval_s): + return + logger.info("afk-worker thread stopped") + + # ───────────────────────── single tick ──────────────────────────── + + def _tick(self, cfg: dict) -> None: + cycle = CycleResult( + tick_id=self._build_tick_id(), + started_at=now_iso(), + ) + + state = load_state() + cycle.afk_mode = state.mode + + # Gate 1: not in AFK → silent skip (no log entry). + # This also catches MODE_STAND_BY (Q8 stop): is_afk() only returns + # True for afk_manual/afk_auto, so stand_by silent-skips here too. + # Silent rather than "skipped_stand_by" log entries because stand_by + # can last indefinitely and we don't want 48 log lines/day of noise. + if not state.is_afk(): + return + + # Gate 2: cooldown + if self._cooldown_skip_remaining > 0: + self._cooldown_skip_remaining -= 1 + cycle.status = "skipped_cooldown" + cycle.notes.append( + f"cooldown_level={state.cooldown_level} " + f"skips_remaining={self._cooldown_skip_remaining}" + ) + cycle.completed_at = now_iso() + self._log_cycle(cycle, cfg) + return + + # Gate 4: MiniMax quota + threshold = float( + (cfg.get("quota") or {}).get( + "minimax_skip_threshold_pct", DEFAULT_QUOTA_THRESHOLD_PCT + ) + ) + cycle.quota_before = self._read_quota() + if self._quota_exceeds(cycle.quota_before, threshold): + cycle.status = "skipped_quota" + cycle.notes.append(f"quota>={threshold}% skip") + cycle.completed_at = now_iso() + self._log_cycle(cycle, cfg) + return + + # Pick task + task = self._pick_task(cfg) + if task is None: + cycle.status = "skipped_no_task" + cycle.completed_at = now_iso() + self._log_cycle(cycle, cfg) + return + cycle.task = task + + # Pick model + model_cfg = self._pick_model(task, cfg) + if model_cfg is None: + cycle.status = "skipped_decision_blocked" + cycle.completed_at = now_iso() + self._post_discord_block(task) + self._log_cycle(cycle, cfg) + return + cycle.model_used = { + "provider": str(model_cfg.get("provider", "")), + "model": str(model_cfg.get("model", "")), + } + + # Run delegation (mutually-exclusive with itself) + with self._delegation_lock: + result = self._run_delegation(task, model_cfg) + cycle.delegation_summary = (result.get("summary") or "")[:500] or None + cycle.delegation_api_calls = result.get("api_calls") + cycle.delegation_duration_s = result.get("duration_seconds") + cycle.delegation_status = result.get("status") + cycle.delegation_error = result.get("error") + + # Mark in picked-index for dedupe + self._mark_picked(task, last_cycle_status=cycle.delegation_status or "unknown", cfg=cfg) + + # Quota delta after + cycle.quota_after = self._read_quota() + + # Cycle status & cooldown reset / increment + if cycle.delegation_status == "completed": + cycle.status = "completed" + self._consecutive_errors = 0 + st = load_state() + if (st.cooldown_level or 0) > 0: + st.cooldown_level = 0 + save_state(st) + else: + cycle.status = "error" + self._consecutive_errors += 1 + if self._consecutive_errors >= COOLDOWN_ERROR_THRESHOLD: + self._raise_cooldown() + + cycle.completed_at = now_iso() + self._log_cycle(cycle, cfg) + + # ───────────────────────── config helpers ───────────────────────── + + def _load_cfg(self) -> dict: + try: + from hermes_cli.config import load_config + full = load_config() or {} + except Exception: + logger.debug("afk-worker: load_config failed, treating as disabled") + return {"enabled": False} + return (full.get("afk_worker") or {}) + + def _interval_seconds(self, cfg: dict) -> int: + try: + mins = int(cfg.get("interval_minutes", DEFAULT_INTERVAL_MIN)) + except (TypeError, ValueError): + mins = DEFAULT_INTERVAL_MIN + return max(60, mins * 60) + + def _build_tick_id(self) -> str: + return datetime.now(timezone.utc).strftime("afk-%Y-%m-%d-%H%M") + + # ───────────────────────── quota ────────────────────────────────── + + def _read_quota(self) -> dict: + """Inspect MiniMax Token Plan via the existing tool. Returns a + small dict with the two windows we care about. Unknown values are + None (treated as 'don't block').""" + try: + from tools.minimax_quota import get_minimax_quota + payload = json.loads(get_minimax_quota()) + except Exception as e: + logger.warning("afk-worker: quota read failed: %s", e) + return {"minimax_5h_pct": None, "minimax_weekly_pct": None} + for row in payload.get("models") or []: + if row.get("model") == "MiniMax-M*": + return { + "minimax_5h_pct": row.get("interval_used_pct"), + "minimax_weekly_pct": row.get("weekly_used_pct"), + } + return {"minimax_5h_pct": None, "minimax_weekly_pct": None} + + @staticmethod + def _quota_exceeds(quota: dict, threshold_pct: float) -> bool: + for v in (quota.get("minimax_5h_pct"), quota.get("minimax_weekly_pct")): + if v is None: + continue + try: + if float(v) >= threshold_pct: + return True + except (TypeError, ValueError): + continue + return False + + # ───────────────────────── pick task ────────────────────────────── + + def _pick_task(self, cfg: dict) -> Optional[TaskCandidate]: + files = cfg.get("status_files") or [] + candidates: list[TaskCandidate] = [] + for raw_path in files: + path = Path(os.path.expanduser(str(raw_path))) + if not path.is_file(): + logger.warning("afk-worker: status file not found: %s (OQ5 skip)", path) + continue + try: + candidates.extend(self._scan_status_file(path)) + except OSError as e: + logger.warning("afk-worker: cannot read %s: %s", path, e) + continue + + # Filter excluded + already-completed-in-picked-index + picked = self._load_picked_index(cfg) + eligible = [] + for c in candidates: + if c.is_excluded: + continue + entry = picked.get(c.task_key()) + if entry is None: + eligible.append(c) + continue + last_status = entry.get("last_cycle_status") + if last_status == "completed": + continue # already done — wait for status file edit + if last_status == "error": + # Cooldown N ticks before retry + n_cycles = entry.get("n_cycles", 0) + err_skips = entry.get("error_skips_remaining", 0) + if err_skips > 0: + # Decrement (we observed it again this tick — not picking) + entry["error_skips_remaining"] = err_skips - 1 + self._save_picked_index(picked, cfg) + continue + # Cooldown expired → eligible again + eligible.append(c) + continue + # Any other status → eligible + eligible.append(c) + + if not eligible: + return None + + # Sort: P0 < P1 < P2; tie-break by source-file mtime ascending + def _key(c: TaskCandidate): + try: + mtime = os.path.getmtime(c.source_file) + except OSError: + mtime = 0.0 + return (_PRIORITY_RANK.get(c.priority, 9), mtime) + + eligible.sort(key=_key) + return eligible[0] + + def _scan_status_file(self, path: Path) -> list[TaskCandidate]: + out: list[TaskCandidate] = [] + current_section = "(top)" + with path.open(encoding="utf-8") as f: + for lineno, raw in enumerate(f, start=1): + line = raw.rstrip("\n") + m_h = RE_HEADING.match(line) + if m_h: + current_section = m_h.group(2).strip() + continue + m_t = RE_TASK_LINE.match(line) + if not m_t: + continue + marker = m_t.group(1) + title_full = m_t.group(2) + + # Parse tags + m_pri = RE_PRIORITY.search(title_full) + priority = m_pri.group(1).upper() if m_pri else "P2" + + m_type = RE_AFK_TYPE.search(title_full) + afk_type_raw = m_type.group(1).lower() if m_type else "research" + + excluded = False + reason = None + if RE_AFK_NO.search(title_full): + excluded = True + reason = "[afk:no] tag" + elif RE_BLOCKED.search(title_full): + excluded = True + reason = "[blocked:*] tag" + + # Strip all known tags from title for cleanliness + title_clean = RE_TAG_ANY.sub("", title_full).strip() + # Strip trailing markdown emphasis tokens left over + title_clean = re.sub(r"\s+", " ", title_clean).strip() + + out.append( + TaskCandidate( + source_file=str(path), + section=current_section, + line_number=lineno, + raw_line=line, + title=title_clean or title_full.strip(), + priority=priority, + afk_type=afk_type_raw, + is_excluded=excluded, + exclusion_reason=reason, + ) + ) + return out + + # ───────────────────────── pick model ───────────────────────────── + + def _pick_model(self, task: TaskCandidate, cfg: dict) -> Optional[dict]: + """Lookup task.afk_type in matrix. Returns None for unmapped types + (caller posts Discord block). decision_structurelle is the explicit + sentinel — never mapped.""" + if task.afk_type == "decision_structurelle": + return None + routing = cfg.get("model_routing") or {} + return routing.get(task.afk_type) + + # ───────────────────────── delegation ───────────────────────────── + + def _run_delegation(self, task: TaskCandidate, model_cfg: dict) -> dict: + from tools.delegate_tool import delegate_task + + goal = ( + f"AFK autonomous tick — work on the following task pulled from " + f"AH's backlog:\n\n" + f" Source: {task.source_file} ({task.section})\n" + f" Title: {task.title}\n" + f" AFK type: {task.afk_type} | Priority: {task.priority}\n\n" + f"Constraints:\n" + f"- §9.1 of HERMES.md: do NOT modify hermes-agent code, configs, " + f"or skills. Read-only on those paths.\n" + f"- Publish actions blocked (already enforced at tool layer).\n" + f"- Output a structured summary of what you did, what's left, " + f"and any blockers.\n" + ) + + parent_agent = _get_runtime_parent_agent() + if parent_agent is None: + return { + "status": "error", + "summary": None, + "error": "no parent_agent available at runtime", + "api_calls": 0, + "duration_seconds": 0.0, + } + + # Resolve api_key from env (we do NOT load the value into the log). + api_key_env_name = model_cfg.get("api_key_env") or "" + api_key = os.environ.get(api_key_env_name) if api_key_env_name else None + api_key = api_key or None # empty string → None + + # OQ3 — toolsets from matrix; fallback to safe minimal + toolsets = model_cfg.get("toolsets") or ["file", "todo"] + + try: + result_str = delegate_task( + goal=goal, + context=None, + toolsets=list(toolsets), + role="leaf", + parent_agent=parent_agent, + model_override=model_cfg.get("model") or None, + provider_override=model_cfg.get("provider") or None, + base_url_override=model_cfg.get("base_url") or None, + api_key_override=api_key, + ) + except TypeError as e: + # Likely the extended signature isn't deployed yet — fall back + # silently so the worker keeps running on the default route. + logger.warning( + "afk-worker: delegate_task does not accept overrides (%s); " + "falling back to default credentials", + e, + ) + result_str = delegate_task( + goal=goal, + context=None, + toolsets=list(toolsets), + role="leaf", + parent_agent=parent_agent, + ) + except Exception as e: + return { + "status": "error", + "summary": None, + "error": f"delegate_task raised: {e}", + "api_calls": 0, + "duration_seconds": 0.0, + } + + try: + payload = json.loads(result_str) + except Exception as e: + return { + "status": "error", + "summary": None, + "error": f"delegate_task returned non-JSON: {e}", + "api_calls": 0, + "duration_seconds": 0.0, + } + + if isinstance(payload, dict) and payload.get("error"): + return { + "status": "error", + "summary": None, + "error": str(payload.get("error")), + "api_calls": 0, + "duration_seconds": 0.0, + } + + results = (payload.get("results") if isinstance(payload, dict) else None) or [] + if not results: + return { + "status": "error", + "summary": None, + "error": "no results array in delegate_task payload", + "api_calls": 0, + "duration_seconds": 0.0, + } + r = results[0] + return { + "status": r.get("status"), + "summary": r.get("summary"), + "error": r.get("error"), + "api_calls": r.get("api_calls", 0), + "duration_seconds": r.get("duration_seconds", 0.0), + } + + # ───────────────────────── picked-index dedupe ──────────────────── + + def _picked_path(self, cfg: dict) -> Path: + return Path( + os.path.expanduser(cfg.get("picked_index_path") or DEFAULT_PICKED_INDEX) + ) + + def _load_picked_index(self, cfg: dict) -> dict: + p = self._picked_path(cfg) + if not p.is_file(): + return {} + try: + with p.open(encoding="utf-8") as f: + data = json.load(f) + if isinstance(data, dict): + return data + except (json.JSONDecodeError, OSError) as e: + logger.warning("afk-worker: picked-index unreadable (%s) — resetting", e) + return {} + + def _save_picked_index(self, idx: dict, cfg: dict) -> None: + p = self._picked_path(cfg) + p.parent.mkdir(parents=True, exist_ok=True) + # Atomic write via tempfile + os.replace + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=str(p.parent), + prefix=p.name + ".tmp-", + delete=False, + ) as tmp: + json.dump(idx, tmp, indent=2, sort_keys=True) + tmp_path = Path(tmp.name) + os.replace(tmp_path, p) + + def _mark_picked(self, task: TaskCandidate, last_cycle_status: str, cfg: dict) -> None: + idx = self._load_picked_index(cfg) + key = task.task_key() + entry = idx.get(key) or {} + now = now_iso() + entry.setdefault("first_picked_at", now) + entry["last_picked_at"] = now + entry["n_cycles"] = int(entry.get("n_cycles", 0)) + 1 + entry["last_cycle_status"] = last_cycle_status + if last_cycle_status == "error": + entry["error_skips_remaining"] = ERROR_COOLDOWN_TICKS + else: + entry["error_skips_remaining"] = 0 + idx[key] = entry + self._save_picked_index(idx, cfg) + + # ───────────────────────── cooldown ─────────────────────────────── + + def _raise_cooldown(self) -> None: + st = load_state() + st.cooldown_level = min(3, (st.cooldown_level or 0) + 1) + save_state(st) + skips = COOLDOWN_LEVEL_TICK_SKIPS.get(st.cooldown_level, 0) + if skips == float("inf"): + self._cooldown_skip_remaining = 10**9 + self._post_discord_cooldown_halt(st.cooldown_level) + else: + self._cooldown_skip_remaining = int(skips) + if st.cooldown_level >= 2: + self._post_discord_cooldown_warning(st.cooldown_level, int(skips)) + # Cooldown absorbs the consecutive-error counter + self._consecutive_errors = 0 + + # ───────────────────────── Discord posts ────────────────────────── + + def _select_discord_adapter(self): + if not self.adapters: + return None + _iter = self.adapters.values() if isinstance(self.adapters, dict) else self.adapters + return next( + (a for a in _iter if getattr(a, "name", "").lower() == "discord"), + None, + ) + + def _post_discord(self, text: str) -> None: + """Send a message to the home channel from this background thread. + + Mirrors gateway/run.py:_post_afk_notif_to_discord pattern.""" + if self.loop is None: + logger.debug("afk-worker: no asyncio loop, dropping Discord msg") + return + adapter = self._select_discord_adapter() + if adapter is None: + logger.debug("afk-worker: no Discord adapter") + return + channel_id = os.environ.get("DISCORD_HERMES_CHANNEL_ID") + if not channel_id: + logger.debug("afk-worker: DISCORD_HERMES_CHANNEL_ID empty") + return + try: + import asyncio + fut = asyncio.run_coroutine_threadsafe( + adapter.send(channel_id, text), self.loop + ) + fut.result(timeout=15) + except Exception: + logger.exception("afk-worker: Discord post failed") + + def _post_discord_block(self, task: TaskCandidate) -> None: + self._post_discord( + f"🚫 **AFK skip — décision structurelle**\n" + f"Tâche : `{task.title}`\n" + f"Source : `{task.source_file}` ({task.section})\n" + f"Type AFK : `{task.afk_type}` (non mappé dans la matrice).\n" + f"§9.1 HERMES.md — j'attends ton arbitrage avant d'avancer." + ) + + def _post_discord_cooldown_warning(self, level: int, skip_ticks: int) -> None: + self._post_discord( + f"⚠️ **AFK worker cooldown level {level}**\n" + f"3 erreurs consécutives sur les délégations. Je skip les " + f"{skip_ticks} prochains ticks et je réessaie ensuite.\n" + f"Détails : `~/AFK_LOG.md`." + ) + + def _post_discord_cooldown_halt(self, level: int) -> None: + self._post_discord( + f"🛑 **AFK worker halt — cooldown level {level}**\n" + f"Trop d'erreurs consécutives. Le worker est arrêté jusqu'au " + f"prochain redémarrage du service ou retour en mode normal.\n" + f"Détails : `~/AFK_LOG.md`." + ) + + # ───────────────────────── logging (md + jsonl) ─────────────────── + + def _log_cycle(self, cycle: CycleResult, cfg: dict) -> None: + md_path = Path(os.path.expanduser(cfg.get("log_path") or DEFAULT_LOG_MD)) + jsonl_path = Path( + os.path.expanduser(cfg.get("log_jsonl_path") or DEFAULT_LOG_JSONL) + ) + md_path.parent.mkdir(parents=True, exist_ok=True) + jsonl_path.parent.mkdir(parents=True, exist_ok=True) + try: + self._append_md(md_path, cycle) + except OSError as e: + logger.warning("afk-worker: cannot append %s: %s", md_path, e) + try: + self._append_jsonl(jsonl_path, cycle) + except OSError as e: + logger.warning("afk-worker: cannot append %s: %s", jsonl_path, e) + + @staticmethod + def _append_md(path: Path, cycle: CycleResult) -> None: + block = _render_md_block(cycle) + with path.open("a", encoding="utf-8") as f: + try: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + f.write(block) + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + + @staticmethod + def _append_jsonl(path: Path, cycle: CycleResult) -> None: + line = json.dumps(_render_jsonl_obj(cycle), ensure_ascii=False) + with path.open("a", encoding="utf-8") as f: + try: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + f.write(line + "\n") + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + + +# --------------------------------------------------------------------------- +# Renderers (separate from the class so they're trivially testable) +# --------------------------------------------------------------------------- + + +def _render_jsonl_obj(cycle: CycleResult) -> dict[str, Any]: + obj: dict[str, Any] = { + "tick_id": cycle.tick_id, + "started_at": cycle.started_at, + "completed_at": cycle.completed_at, + "afk_mode": cycle.afk_mode, + "status": cycle.status, + "task": cycle.task.to_jsonable() if cycle.task else None, + "model_used": cycle.model_used, + "delegation": ( + { + "summary": cycle.delegation_summary, + "api_calls": cycle.delegation_api_calls, + "duration_seconds": cycle.delegation_duration_s, + "status": cycle.delegation_status, + "error": cycle.delegation_error, + } + if cycle.delegation_status is not None + else None + ), + "quota_before": cycle.quota_before, + "quota_after": cycle.quota_after, + "notes": list(cycle.notes), + } + return obj + + +def _render_md_block(cycle: CycleResult) -> str: + lines: list[str] = [] + lines.append( + f"## {cycle.started_at} — {cycle.tick_id} — {cycle.afk_mode or '?'}" + ) + lines.append("") + lines.append(f"**Status** : {cycle.status}") + if cycle.task: + lines.append( + f"**Tâche pickée** : `{cycle.task.source_file}` " + f"({cycle.task.section}) — {cycle.task.title}" + ) + lines.append( + f"**Type AFK** : {cycle.task.afk_type} | " + f"**Priority** : {cycle.task.priority}" + ) + if cycle.model_used: + lines.append( + f"**Modèle utilisé** : " + f"{cycle.model_used.get('provider', '?')} / " + f"{cycle.model_used.get('model', '?')}" + ) + if cycle.delegation_status is not None: + lines.append("") + lines.append("**Délégation** :") + lines.append(f"- subagent status : {cycle.delegation_status}") + lines.append(f"- API calls : {cycle.delegation_api_calls}") + if cycle.delegation_duration_s is not None: + lines.append(f"- duration : {cycle.delegation_duration_s:.1f}s") + if cycle.delegation_error: + lines.append(f"- error : `{cycle.delegation_error}`") + if cycle.delegation_summary: + lines.append("") + lines.append("**Résumé du sub-agent** (≤ 500 chars) :") + lines.append("> " + cycle.delegation_summary.replace("\n", "\n> ")) + if cycle.quota_before or cycle.quota_after: + lines.append("") + lines.append("**Quota MiniMax** :") + if cycle.quota_before: + lines.append( + f"- avant : " + f"{cycle.quota_before.get('minimax_5h_pct')}% (5h) / " + f"{cycle.quota_before.get('minimax_weekly_pct')}% (semaine)" + ) + if cycle.quota_after: + lines.append( + f"- après : " + f"{cycle.quota_after.get('minimax_5h_pct')}% (5h) / " + f"{cycle.quota_after.get('minimax_weekly_pct')}% (semaine)" + ) + if cycle.notes: + lines.append("") + lines.append(f"**Notes** : {' | '.join(cycle.notes)}") + lines.append("") + lines.append("---") + lines.append("") + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Helpers usable by other modules (afk_scheduler purge hook — OQ4) +# --------------------------------------------------------------------------- + + +def purge_picked_index(cfg: Optional[dict] = None) -> bool: + """Delete the picked-index file if present. Called by afk_scheduler + when AH transitions out of AFK to normal mode (OQ4).""" + if cfg is None: + try: + from hermes_cli.config import load_config + cfg = (load_config() or {}).get("afk_worker") or {} + except Exception: + cfg = {} + p = Path(os.path.expanduser(cfg.get("picked_index_path") or DEFAULT_PICKED_INDEX)) + try: + if p.is_file(): + p.unlink() + logger.info("afk-worker: picked-index purged (mode flip → normal)") + return True + except OSError as e: + logger.warning("afk-worker: could not purge picked-index: %s", e) + return False diff --git a/gateway/run.py b/gateway/run.py index c2a4b8d22628..d5621fce4b8a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -11350,7 +11350,42 @@ def restart_signal_handler(): name="cron-ticker", ) cron_thread.start() - + + # WS-AUTO-002 — AFK Work Loop background thread. + # Independent from the cron ticker so a long delegation cycle (up to + # delegation.child_timeout_seconds = 600) can never block the 60s + # heartbeat / auto-AFK transition checks. Shares the same stop_event + # so shutdown is coordinated. + from agent.afk_worker import AFKWorker, install_parent_agent_getter + + def _get_afk_parent_agent(): + """Pick the most-recently-started running agent for the AFK worker + to use as parent_agent in delegate_task. Returns None if no agent + is currently active (worker logs and skips that tick).""" + agents = getattr(runner, "_running_agents", None) or {} + if not agents: + return None + ts = getattr(runner, "_running_agents_ts", {}) or {} + if ts: + latest = max(ts, key=ts.get) + if latest in agents: + return agents[latest] + return next(iter(agents.values()), None) + + install_parent_agent_getter(_get_afk_parent_agent) + + afk_worker = AFKWorker( + stop_event=cron_stop, + adapters=runner.adapters, + loop=asyncio.get_running_loop(), + ) + afk_worker_thread = threading.Thread( + target=afk_worker.run, + daemon=True, + name="afk-worker", + ) + afk_worker_thread.start() + # Wait for shutdown await runner.wait_for_shutdown() @@ -11359,9 +11394,10 @@ def restart_signal_handler(): logger.error("Gateway exiting with failure: %s", runner.exit_reason) return False - # Stop cron ticker cleanly + # Stop cron ticker + AFK worker cleanly (they share cron_stop). cron_stop.set() cron_thread.join(timeout=5) + afk_worker_thread.join(timeout=10) # Close MCP server connections try: diff --git a/tests/agent/test_afk_worker.py b/tests/agent/test_afk_worker.py new file mode 100644 index 000000000000..589fab5ab0f2 --- /dev/null +++ b/tests/agent/test_afk_worker.py @@ -0,0 +1,586 @@ +"""Tests for agent.afk_worker — WS-AUTO-002. + +Spec reference: ~/Documents/Projects/ACOS-HERMES/.workspace/WS_AFK_WORK_LOOP_SPEC.md §11 +""" + +import json +import threading +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest + +from agent.afk_state import ( + AFKState, + MODE_AFK_AUTO, + MODE_AFK_MANUAL, + MODE_NORMAL, + MODE_STAND_BY, +) +from agent import afk_worker as W +from agent.afk_worker import ( + AFKWorker, + CycleResult, + TaskCandidate, + _render_jsonl_obj, + _render_md_block, + install_parent_agent_getter, + purge_picked_index, +) + + +# ─── helpers ───────────────────────────────────────────────────────────────── + + +def _afk_state_file(tmp_path: Path, monkeypatch, mode=MODE_AFK_AUTO, **fields): + """Set up an isolated AFK state file for the duration of one test.""" + state_path = tmp_path / "afk_state.json" + monkeypatch.setattr( + "agent.afk_state._default_state_path", lambda: state_path + ) + state = AFKState(mode=mode, **fields) + if state.entered_at is None and mode in (MODE_AFK_AUTO, MODE_AFK_MANUAL): + state.entered_at = "2026-05-03T00:00:00Z" + from agent.afk_state import save_state + save_state(state) + return state_path + + +def _make_worker(tmp_path: Path, monkeypatch, adapters=None, loop=None) -> AFKWorker: + stop = threading.Event() + return AFKWorker(stop_event=stop, adapters=adapters, loop=loop) + + +def _cfg(tmp_path: Path, **overrides) -> dict: + """Default cfg dict used in tests.""" + base = { + "enabled": True, + "interval_minutes": 30, + "status_files": [], + "log_path": str(tmp_path / "AFK_LOG.md"), + "log_jsonl_path": str(tmp_path / "afk_log.jsonl"), + "picked_index_path": str(tmp_path / "picked.json"), + "quota": {"minimax_skip_threshold_pct": 80}, + "stand_by_behavior": "stop", + "model_routing": { + "code_review": { + "provider": "openrouter", + "model": "anthropic/claude-sonnet-4.6", + "base_url": "https://openrouter.ai/api/v1", + "api_key_env": "OPENROUTER_API_KEY", + "toolsets": ["terminal", "file", "todo"], + }, + "research": { + "provider": "minimax", + "model": "MiniMax-M2.7", + "base_url": "https://api.minimax.io/anthropic", + "api_key_env": "", + "toolsets": ["jina", "file", "todo"], + }, + }, + } + base.update(overrides) + return base + + +# ═══ §11.1 unit tests ══════════════════════════════════════════════════════ + + +class TestPickTask: + def _write_status(self, tmp_path: Path, name: str, lines: list[str]) -> Path: + p = tmp_path / name + p.write_text("\n".join(lines), encoding="utf-8") + return p + + def test_priority_order_p0_beats_p1_p2(self, tmp_path, monkeypatch): + f = self._write_status( + tmp_path, "S.md", + [ + "## Section A", + "- [ ] Low task [priority:P2] [afk:research]", + "- [ ] Critical task [priority:P0] [afk:research]", + "- [ ] Medium task [priority:P1] [afk:research]", + ], + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + task = w._pick_task(cfg) + assert task is not None + assert "Critical task" in task.title + assert task.priority == "P0" + + def test_excludes_afk_no(self, tmp_path, monkeypatch): + f = self._write_status( + tmp_path, "S.md", + [ + "## Section", + "- [ ] Khéri-only task [priority:P0] [afk:no]", + "- [ ] Worker-fine task [priority:P1] [afk:research]", + ], + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + task = w._pick_task(cfg) + assert task is not None + assert "Worker-fine" in task.title + + def test_excludes_blocked(self, tmp_path, monkeypatch): + f = self._write_status( + tmp_path, "S.md", + [ + "## Section", + "- [ ] Waiting task [priority:P0] [blocked:waiting-PR] [afk:research]", + "- [ ] Free task [priority:P1] [afk:research]", + ], + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + task = w._pick_task(cfg) + assert task is not None + assert "Free task" in task.title + + def test_no_tags_defaults_p2_research(self, tmp_path, monkeypatch): + f = self._write_status( + tmp_path, "S.md", + [ + "## Section", + "- [ ] Bare task with no tags", + ], + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + task = w._pick_task(cfg) + assert task is not None + assert task.priority == "P2" + assert task.afk_type == "research" + + def test_dedupe_via_picked_index_completed(self, tmp_path, monkeypatch): + f = self._write_status( + tmp_path, "S.md", + [ + "## Section", + "- [ ] Already done [priority:P0] [afk:research]", + "- [ ] Still open [priority:P1] [afk:research]", + ], + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + # Mark first task as completed in picked-index + first = w._scan_status_file(f)[0] + idx = {first.task_key(): {"last_cycle_status": "completed", "n_cycles": 1}} + w._save_picked_index(idx, cfg) + task = w._pick_task(cfg) + assert task is not None + assert "Still open" in task.title + + def test_error_cooldown_skips_then_retries(self, tmp_path, monkeypatch): + f = self._write_status( + tmp_path, "S.md", + ["## Section", "- [ ] Errored task [priority:P0] [afk:research]"], + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + first = w._scan_status_file(f)[0] + idx = { + first.task_key(): { + "last_cycle_status": "error", + "n_cycles": 1, + "error_skips_remaining": 2, + } + } + w._save_picked_index(idx, cfg) + # First call: still in cooldown, skipped, counter decremented + assert w._pick_task(cfg) is None + idx2 = w._load_picked_index(cfg) + assert idx2[first.task_key()]["error_skips_remaining"] == 1 + # Decrement to 0 + assert w._pick_task(cfg) is None + idx3 = w._load_picked_index(cfg) + assert idx3[first.task_key()]["error_skips_remaining"] == 0 + # Now eligible again + task = w._pick_task(cfg) + assert task is not None + assert "Errored task" in task.title + + def test_missing_status_file_logs_warning_skip(self, tmp_path, monkeypatch, caplog): + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(tmp_path / "does-not-exist.md")]) + with caplog.at_level("WARNING"): + assert w._pick_task(cfg) is None + assert any("status file not found" in m for m in caplog.messages) + + def test_in_progress_marker_eligible(self, tmp_path, monkeypatch): + f = self._write_status( + tmp_path, "S.md", + ["## Section", "- [/] In progress task [priority:P0] [afk:research]"], + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + task = w._pick_task(cfg) + assert task is not None + assert "In progress" in task.title + + def test_done_marker_skipped(self, tmp_path, monkeypatch): + f = self._write_status( + tmp_path, "S.md", + [ + "## Section", + "- [x] Done task [priority:P0] [afk:research]", + "- [ ] Open task [priority:P2] [afk:research]", + ], + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + task = w._pick_task(cfg) + assert task is not None + assert "Open task" in task.title + + +class TestPickModel: + def test_matrix_match_code_review(self, tmp_path, monkeypatch): + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) + t = TaskCandidate( + source_file="x", section="y", line_number=1, raw_line="", + title="t", priority="P1", afk_type="code_review", + ) + m = w._pick_model(t, cfg) + assert m is not None + assert m["provider"] == "openrouter" + assert m["model"] == "anthropic/claude-sonnet-4.6" + + def test_decision_structurelle_blocked(self, tmp_path, monkeypatch): + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) + t = TaskCandidate( + source_file="x", section="y", line_number=1, raw_line="", + title="t", priority="P0", afk_type="decision_structurelle", + ) + assert w._pick_model(t, cfg) is None + + def test_unmapped_type_blocked(self, tmp_path, monkeypatch): + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) + t = TaskCandidate( + source_file="x", section="y", line_number=1, raw_line="", + title="t", priority="P0", afk_type="exotic_unknown", + ) + assert w._pick_model(t, cfg) is None + + +class TestQuota: + def test_or_logic_5h_high(self): + q = {"minimax_5h_pct": 85.0, "minimax_weekly_pct": 5.0} + assert AFKWorker._quota_exceeds(q, 80) is True + + def test_or_logic_weekly_high(self): + q = {"minimax_5h_pct": 5.0, "minimax_weekly_pct": 90.0} + assert AFKWorker._quota_exceeds(q, 80) is True + + def test_below_threshold(self): + q = {"minimax_5h_pct": 30.0, "minimax_weekly_pct": 10.0} + assert AFKWorker._quota_exceeds(q, 80) is False + + def test_unknown_does_not_block(self): + q = {"minimax_5h_pct": None, "minimax_weekly_pct": None} + assert AFKWorker._quota_exceeds(q, 80) is False + + +class TestRendering: + def test_jsonl_completed_cycle(self): + cycle = CycleResult( + tick_id="afk-2026-05-15-2230", + started_at="2026-05-15T22:30:00Z", + completed_at="2026-05-15T22:33:04Z", + afk_mode="afk_auto", + status="completed", + task=TaskCandidate( + source_file="/h/SMCP_STATUS.md", section="§3.4", + line_number=42, raw_line="", title="Patch 16", + priority="P1", afk_type="code_refactor", + ), + model_used={"provider": "openrouter", "model": "anthropic/claude-sonnet-4.6"}, + delegation_summary="did the refactor", + delegation_api_calls=12, + delegation_duration_s=184.3, + delegation_status="completed", + quota_before={"minimax_5h_pct": 22.1, "minimax_weekly_pct": 4.0}, + quota_after={"minimax_5h_pct": 22.1, "minimax_weekly_pct": 4.0}, + ) + obj = _render_jsonl_obj(cycle) + assert obj["status"] == "completed" + assert obj["task"]["title"] == "Patch 16" + assert obj["delegation"]["api_calls"] == 12 + # JSON-serialisable + assert json.dumps(obj) + + def test_md_block_contains_key_fields(self): + cycle = CycleResult( + tick_id="afk-x", + started_at="2026-05-15T22:30:00Z", + afk_mode="afk_auto", + status="completed", + task=TaskCandidate( + source_file="/x", section="§3", line_number=1, raw_line="", + title="My Task", priority="P0", afk_type="research", + ), + model_used={"provider": "minimax", "model": "MiniMax-M2.7"}, + ) + block = _render_md_block(cycle) + assert "completed" in block + assert "My Task" in block + assert "minimax" in block + assert block.rstrip().endswith("---") + + +class TestLogAppend: + def test_md_appends_block(self, tmp_path, monkeypatch): + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) + path = Path(cfg["log_path"]) + path.write_text("EXISTING\n", encoding="utf-8") # pre-existing content + cycle = CycleResult( + tick_id="t1", started_at="2026-05-15T22:30:00Z", + afk_mode="afk_auto", status="skipped_no_task", + ) + w._log_cycle(cycle, cfg) + content = path.read_text(encoding="utf-8") + assert content.startswith("EXISTING\n") + assert "skipped_no_task" in content + + def test_jsonl_one_line_per_cycle(self, tmp_path, monkeypatch): + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) + path = Path(cfg["log_jsonl_path"]) + for i in range(3): + cycle = CycleResult( + tick_id=f"t{i}", started_at=f"2026-05-15T22:3{i}:00Z", + afk_mode="afk_auto", status="skipped_no_task", + ) + w._log_cycle(cycle, cfg) + lines = path.read_text(encoding="utf-8").strip().splitlines() + assert len(lines) == 3 + for ln in lines: + obj = json.loads(ln) # all valid JSON + assert obj["status"] == "skipped_no_task" + + +class TestDynamicRecap: + def test_no_jsonl_returns_fallback(self, tmp_path, monkeypatch): + from agent.afk_heartbeat import _build_dynamic_recap + state = AFKState(mode=MODE_AFK_AUTO, entered_at="2026-05-03T00:00:00Z") + cfg = {"log_jsonl_path": str(tmp_path / "missing.jsonl")} + recap = _build_dynamic_recap(state, cfg=cfg) + assert "Aucun cycle" in recap or "absent" in recap + + def test_filters_pre_entered_at(self, tmp_path, monkeypatch): + from agent.afk_heartbeat import _build_dynamic_recap + # Two cycles: one before AFK entered, one after + path = tmp_path / "afk_log.jsonl" + with path.open("w", encoding="utf-8") as f: + json.dump({ + "tick_id": "before", "started_at": "2026-05-01T00:00:00Z", + "status": "completed", "task": {"title": "OLD"}, + }, f); f.write("\n") + json.dump({ + "tick_id": "after", "started_at": "2026-05-04T00:00:00Z", + "status": "completed", "task": {"title": "NEW"}, + }, f); f.write("\n") + state = AFKState(mode=MODE_AFK_AUTO, entered_at="2026-05-03T00:00:00Z") + cfg = {"log_jsonl_path": str(path)} + recap = _build_dynamic_recap(state, cfg=cfg) + assert "NEW" in recap + assert "OLD" not in recap + + +class TestPurgePickedIndex: + def test_purge_removes_existing(self, tmp_path, monkeypatch): + p = tmp_path / "picked.json" + p.write_text("{}", encoding="utf-8") + cfg = {"picked_index_path": str(p)} + assert purge_picked_index(cfg) is True + assert not p.exists() + + def test_purge_missing_no_op(self, tmp_path, monkeypatch): + cfg = {"picked_index_path": str(tmp_path / "ghost.json")} + # Should not raise + result = purge_picked_index(cfg) + assert result is False + + +# ═══ §11.2 integration tests ══════════════════════════════════════════════ + + +class TestTickGates: + def test_normal_mode_silent_skip(self, tmp_path, monkeypatch): + _afk_state_file(tmp_path, monkeypatch, mode=MODE_NORMAL) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) + # No log file should be created (silent skip) + w._tick(cfg) + assert not Path(cfg["log_jsonl_path"]).exists() + + def test_stand_by_silent_skip(self, tmp_path, monkeypatch): + # Q8 stand_by behavior = stop. is_afk() returns False for stand_by + # (AFK_MODES = manual + auto only), so the worker silent-skips just + # like in normal mode. No log spam every 30 min for the duration of + # an indefinite stand_by. + _afk_state_file(tmp_path, monkeypatch, mode=MODE_STAND_BY) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) + w._tick(cfg) + assert not Path(cfg["log_jsonl_path"]).exists() + + def test_quota_high_logs_skipped_quota(self, tmp_path, monkeypatch): + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 85.0, "minimax_weekly_pct": 10.0, + }): + w._tick(cfg) + obj = json.loads( + Path(cfg["log_jsonl_path"]).read_text(encoding="utf-8").splitlines()[0] + ) + assert obj["status"] == "skipped_quota" + + def test_no_eligible_task_logs_skipped_no_task(self, tmp_path, monkeypatch): + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path) # status_files=[] by default + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 5.0, "minimax_weekly_pct": 3.0, + }): + w._tick(cfg) + obj = json.loads( + Path(cfg["log_jsonl_path"]).read_text(encoding="utf-8").splitlines()[0] + ) + assert obj["status"] == "skipped_no_task" + + +class TestTickFullCycle: + def test_decision_blocked_posts_discord(self, tmp_path, monkeypatch): + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + f = tmp_path / "S.md" + f.write_text( + "## Section\n- [ ] X [priority:P0] [afk:decision_structurelle]\n", + encoding="utf-8", + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 5.0, "minimax_weekly_pct": 3.0, + }), patch.object(w, "_post_discord") as post_mock: + w._tick(cfg) + # Discord posted with the block message + assert post_mock.call_count == 1 + msg = post_mock.call_args[0][0] + assert "décision structurelle" in msg + # Cycle logged with the right status + obj = json.loads( + Path(cfg["log_jsonl_path"]).read_text(encoding="utf-8").splitlines()[0] + ) + assert obj["status"] == "skipped_decision_blocked" + + def test_completed_cycle_logs_full(self, tmp_path, monkeypatch): + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + f = tmp_path / "S.md" + f.write_text( + "## Section A\n- [ ] Refactor X [priority:P1] [afk:code_review]\n", + encoding="utf-8", + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + # Stub the parent_agent getter and delegate_task + install_parent_agent_getter(lambda: object()) + fake_payload = json.dumps({ + "results": [{ + "status": "completed", + "summary": "OK done", + "api_calls": 5, + "duration_seconds": 12.3, + }] + }) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 5.0, "minimax_weekly_pct": 3.0, + }), patch("tools.delegate_tool.delegate_task", return_value=fake_payload): + w._tick(cfg) + # Log shows completed + obj = json.loads( + Path(cfg["log_jsonl_path"]).read_text(encoding="utf-8").splitlines()[0] + ) + assert obj["status"] == "completed" + assert obj["delegation"]["api_calls"] == 5 + assert obj["model_used"]["provider"] == "openrouter" + # Picked-index updated + idx = json.loads(Path(cfg["picked_index_path"]).read_text(encoding="utf-8")) + assert any( + entry.get("last_cycle_status") == "completed" + for entry in idx.values() + ) + # Cleanup + install_parent_agent_getter(None) + + +# ═══ §11.4 security ═══════════════════════════════════════════════════════ + + +class TestSecurity: + def test_worker_does_not_modify_status_files(self, tmp_path, monkeypatch): + """The worker must only READ status files, never write to them.""" + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + f = tmp_path / "S.md" + original = "## Section\n- [ ] T [priority:P0] [afk:research]\n" + f.write_text(original, encoding="utf-8") + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + install_parent_agent_getter(lambda: object()) + fake_payload = json.dumps({ + "results": [{ + "status": "completed", "summary": "done", + "api_calls": 1, "duration_seconds": 1.0, + }] + }) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 5.0, "minimax_weekly_pct": 3.0, + }), patch("tools.delegate_tool.delegate_task", return_value=fake_payload): + w._tick(cfg) + # Status file content unchanged + assert f.read_text(encoding="utf-8") == original + install_parent_agent_getter(None) + + def test_disabled_via_config_thread_exits(self, tmp_path, monkeypatch): + """enabled: false → run() returns immediately without ticking.""" + w = _make_worker(tmp_path, monkeypatch) + with patch.object(w, "_load_cfg", return_value={"enabled": False}), \ + patch.object(w, "_tick") as tick_mock: + w.run() + tick_mock.assert_not_called() + + def test_api_key_value_never_in_log(self, tmp_path, monkeypatch): + """The log files must never contain the resolved api_key value.""" + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + f = tmp_path / "S.md" + f.write_text( + "## S\n- [ ] Code thing [priority:P0] [afk:code_review]\n", + encoding="utf-8", + ) + secret = "sk-or-v1-FAKE-SECRET-VALUE-12345" + monkeypatch.setenv("OPENROUTER_API_KEY", secret) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)]) + install_parent_agent_getter(lambda: object()) + fake_payload = json.dumps({ + "results": [{ + "status": "completed", "summary": "done", + "api_calls": 1, "duration_seconds": 1.0, + }] + }) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 5.0, "minimax_weekly_pct": 3.0, + }), patch("tools.delegate_tool.delegate_task", return_value=fake_payload): + w._tick(cfg) + for path_key in ("log_path", "log_jsonl_path"): + content = Path(cfg[path_key]).read_text(encoding="utf-8") + assert secret not in content, f"secret leaked into {path_key}" + install_parent_agent_getter(None) diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index abdec4717fef..1635f8a43954 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -1799,6 +1799,12 @@ def delegate_task( acp_args: Optional[List[str]] = None, role: Optional[str] = None, parent_agent=None, + # WS-AUTO-002 — per-call provider/model overrides for the AFK worker. + # All optional; None = preserve config-resolved credentials. + model_override: Optional[str] = None, + provider_override: Optional[str] = None, + base_url_override: Optional[str] = None, + api_key_override: Optional[str] = None, ) -> str: """ Spawn one or more child agents to handle delegated tasks. @@ -1812,6 +1818,12 @@ def delegate_task( toolset and can spawn its own workers, bounded by delegation.max_spawn_depth. Per-task role beats the top-level one. + Per-call credential overrides (`model_override`, `provider_override`, + `base_url_override`, `api_key_override`) let in-process callers — like + the AFK worker — route a single delegation to a different provider/ + model without mutating ~/.hermes/config.yaml. Any unset override + leaves the config-resolved value in place. + Returns JSON with results array, one entry per task. """ if parent_agent is None: @@ -1871,6 +1883,18 @@ def delegate_task( except ValueError as exc: return tool_error(str(exc)) + # WS-AUTO-002 — apply per-call overrides on top of config-resolved + # creds. Each override is independent; unset (None) overrides leave + # the corresponding config value untouched. + if model_override: + creds["model"] = model_override + if provider_override: + creds["provider"] = provider_override + if base_url_override: + creds["base_url"] = base_url_override + if api_key_override: + creds["api_key"] = api_key_override + # Normalize to task list max_children = _get_max_concurrent_children() if tasks and isinstance(tasks, list): From 6d72eff7c0d4f3d43cfe8a0ffffcdbdf2a97daaf Mon Sep 17 00:00:00 2001 From: Anu Kheru Date: Sun, 3 May 2026 19:19:17 -0300 Subject: [PATCH 2/6] =?UTF-8?q?ah:=20WS-AUTO-002=20follow-up=20=E2=80=94?= =?UTF-8?q?=20Discord=20verbosity=20in=20AFK=20worker?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per Khéri review of Phase 4 Étape 2: the worker was silent during delegation cycles (only AFK_LOG.md/jsonl tracked activity). Added 2 Discord posts per delegated cycle so Khéri can observe the worker working in real-time: - Pre-delegation: "🔄 AFK cycle starting" with task title, source file/section/line, AFK type, priority, model, toolsets, hint that the 600s timeout applies. - Post-delegation: "✅/❌ AFK cycle done" with status, api_calls, duration, error if any, sub-agent summary (≤500 chars), current MiniMax quota, pointer to ~/AFK_LOG.md. Skipped statuses (skipped_quota, skipped_no_task, skipped_cooldown) remain silent to avoid spam — only successful/errored cycles post. The existing decision_blocked + cooldown_halt posts are unchanged. Behavior is gated by afk_worker.verbose in config.yaml. Default is True (the new default — matches Khéri's preference). Set false for silent operation if needed later. 6 new tests cover the verbose flag (default true, explicit on/off, 2 posts on completed cycle, 0 posts when verbose=false, 0 posts on skipped_no_task regardless). Total: 39 tests, all green. --- agent/afk_worker.py | 72 +++++++++++++++++++++++++++++++ tests/agent/test_afk_worker.py | 78 +++++++++++++++++++++++++++++++++- 2 files changed, 149 insertions(+), 1 deletion(-) diff --git a/agent/afk_worker.py b/agent/afk_worker.py index ec562c983dd2..fd0c67649e8c 100644 --- a/agent/afk_worker.py +++ b/agent/afk_worker.py @@ -285,6 +285,10 @@ def _tick(self, cfg: dict) -> None: "model": str(model_cfg.get("model", "")), } + # Verbose pre-delegation Discord post (default ON — see config.verbose) + if self._verbose(cfg): + self._post_discord_cycle_starting(task, model_cfg) + # Run delegation (mutually-exclusive with itself) with self._delegation_lock: result = self._run_delegation(task, model_cfg) @@ -300,6 +304,10 @@ def _tick(self, cfg: dict) -> None: # Quota delta after cycle.quota_after = self._read_quota() + # Verbose post-delegation Discord post (default ON) + if self._verbose(cfg): + self._post_discord_cycle_done(cycle) + # Cycle status & cooldown reset / increment if cycle.delegation_status == "completed": cycle.status = "completed" @@ -705,6 +713,70 @@ def _post_discord(self, text: str) -> None: except Exception: logger.exception("afk-worker: Discord post failed") + @staticmethod + def _verbose(cfg: dict) -> bool: + """Return True if afk_worker.verbose is enabled (default True). + + When True, the worker posts 2 Discord messages per delegated cycle + (pre-delegation + post-delegation) on top of the existing + block/cooldown notifications. Set false for silent operation. + """ + v = cfg.get("verbose") + if v is None: + return True + return bool(v) + + def _post_discord_cycle_starting(self, task: TaskCandidate, model_cfg: dict) -> None: + toolsets = model_cfg.get("toolsets") or ["file", "todo"] + provider = model_cfg.get("provider", "?") + model = model_cfg.get("model", "?") + # Truncate long titles so Discord doesn't fail on > 2000 chars + title = task.title if len(task.title) <= 200 else (task.title[:197] + "...") + self._post_discord( + f"🔄 **AFK cycle starting**\n" + f"• Tâche : `{title}`\n" + f"• Source : `{task.source_file}` ({task.section}, l.{task.line_number})\n" + f"• Type AFK : `{task.afk_type}` | Priority : `{task.priority}`\n" + f"• Modèle : `{provider} / {model}`\n" + f"• Toolsets : {', '.join(f'`{t}`' for t in toolsets)}\n" + f"⏱ Délégation lancée — `delegation.child_timeout_seconds` = 600s max." + ) + + def _post_discord_cycle_done(self, cycle: CycleResult) -> None: + status = cycle.delegation_status or "unknown" + emoji = "✅" if status == "completed" else ("❌" if status == "error" else "⚠️") + title = (cycle.task.title if cycle.task else "?") + if len(title) > 200: + title = title[:197] + "..." + # Build a result block — summary truncated to ~500 chars (already + # capped at cycle build time, but defensive here too). + summary = cycle.delegation_summary or "(no summary)" + if len(summary) > 500: + summary = summary[:497] + "..." + duration = cycle.delegation_duration_s + duration_str = f"{duration:.1f}s" if isinstance(duration, (int, float)) else "?" + api_calls = cycle.delegation_api_calls if cycle.delegation_api_calls is not None else "?" + + msg = ( + f"{emoji} **AFK cycle done — `{status}`**\n" + f"• Tâche : `{title}`\n" + f"• API calls : {api_calls} | Duration : {duration_str}\n" + ) + if cycle.delegation_error: + msg += f"• Error : `{cycle.delegation_error}`\n" + if status == "completed" and summary != "(no summary)": + msg += f"\n**Résumé sub-agent :**\n> {summary.replace(chr(10), chr(10) + '> ')}\n" + # Quota delta info + qa = cycle.quota_after or {} + if qa.get("minimax_5h_pct") is not None or qa.get("minimax_weekly_pct") is not None: + msg += ( + f"\n📊 Quota MiniMax : " + f"{qa.get('minimax_5h_pct', '?')}% (5h) / " + f"{qa.get('minimax_weekly_pct', '?')}% (semaine)" + ) + msg += f"\n_(détails : `~/AFK_LOG.md`)_" + self._post_discord(msg) + def _post_discord_block(self, task: TaskCandidate) -> None: self._post_discord( f"🚫 **AFK skip — décision structurelle**\n" diff --git a/tests/agent/test_afk_worker.py b/tests/agent/test_afk_worker.py index 589fab5ab0f2..0ec20628cc79 100644 --- a/tests/agent/test_afk_worker.py +++ b/tests/agent/test_afk_worker.py @@ -490,7 +490,7 @@ def test_completed_cycle_logs_full(self, tmp_path, monkeypatch): encoding="utf-8", ) w = _make_worker(tmp_path, monkeypatch) - cfg = _cfg(tmp_path, status_files=[str(f)]) + cfg = _cfg(tmp_path, status_files=[str(f)], verbose=False) # silent for log assert # Stub the parent_agent getter and delegate_task install_parent_agent_getter(lambda: object()) fake_payload = json.dumps({ @@ -522,6 +522,82 @@ def test_completed_cycle_logs_full(self, tmp_path, monkeypatch): install_parent_agent_getter(None) +class TestVerbosity: + def test_verbose_default_true(self): + assert AFKWorker._verbose({}) is True + + def test_verbose_explicit_false(self): + assert AFKWorker._verbose({"verbose": False}) is False + + def test_verbose_explicit_true(self): + assert AFKWorker._verbose({"verbose": True}) is True + + def test_verbose_posts_two_messages_on_completed(self, tmp_path, monkeypatch): + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + f = tmp_path / "S.md" + f.write_text( + "## S\n- [ ] My task [priority:P1] [afk:code_review]\n", + encoding="utf-8", + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)], verbose=True) + install_parent_agent_getter(lambda: object()) + fake_payload = json.dumps({ + "results": [{ + "status": "completed", "summary": "got it done", + "api_calls": 7, "duration_seconds": 42.0, + }] + }) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 1.0, "minimax_weekly_pct": 0.5, + }), patch.object(w, "_post_discord") as post_mock, \ + patch("tools.delegate_tool.delegate_task", return_value=fake_payload): + w._tick(cfg) + # Two posts: cycle starting + cycle done + assert post_mock.call_count == 2 + msg_pre = post_mock.call_args_list[0][0][0] + msg_post = post_mock.call_args_list[1][0][0] + assert "AFK cycle starting" in msg_pre + assert "My task" in msg_pre + assert "anthropic/claude-sonnet-4.6" in msg_pre + assert "AFK cycle done" in msg_post + assert "completed" in msg_post + assert "got it done" in msg_post + install_parent_agent_getter(None) + + def test_verbose_false_silent_on_completed(self, tmp_path, monkeypatch): + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + f = tmp_path / "S.md" + f.write_text( + "## S\n- [ ] X [priority:P0] [afk:research]\n", encoding="utf-8", + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)], verbose=False) + install_parent_agent_getter(lambda: object()) + fake_payload = json.dumps({ + "results": [{"status": "completed", "summary": "ok", + "api_calls": 1, "duration_seconds": 1.0}] + }) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 1.0, "minimax_weekly_pct": 0.5, + }), patch.object(w, "_post_discord") as post_mock, \ + patch("tools.delegate_tool.delegate_task", return_value=fake_payload): + w._tick(cfg) + post_mock.assert_not_called() + install_parent_agent_getter(None) + + def test_verbose_skip_no_task_silent(self, tmp_path, monkeypatch): + # skipped_no_task should be silent regardless of verbose flag + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[], verbose=True) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 1.0, "minimax_weekly_pct": 0.5, + }), patch.object(w, "_post_discord") as post_mock: + w._tick(cfg) + post_mock.assert_not_called() + + # ═══ §11.4 security ═══════════════════════════════════════════════════════ From 9c3bbcc2498c1352dfa73704f071cf61a0d9d538 Mon Sep 17 00:00:00 2001 From: Anu Kheru Date: Sun, 3 May 2026 19:36:26 -0300 Subject: [PATCH 3/6] =?UTF-8?q?ah:=20WS-AUTO-002=20fix=20=E2=80=94=20spawn?= =?UTF-8?q?=20transient=20parent=5Fagent=20for=20AFK=20worker?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Étape 3 testing on VPS exposed an OQ1 design gap: the worker assumed runner._running_agents would always have at least one entry to reuse as parent_agent for delegate_task. In practice the gateway is idle when the worker ticks (no Discord conversation underway), so the registry is empty and every cycle errored: delegation: error, duration: 0s, error: "no parent_agent available at runtime" Fix: _get_afk_parent_agent now follows the cron/scheduler.py pattern. If no session agent exists, build a transient AIAgent dedicated to the worker (quiet_mode, skip_context_files, skip_memory, platform= "afk_worker") and cache it for subsequent ticks. Credentials come from _resolve_runtime_agent_kwargs() — same chain the gateway uses. The session-agent path stays as the cheap fast-path when an active conversation is running (free reuse). Transient build only happens when truly idle. --- gateway/run.py | 69 ++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 59 insertions(+), 10 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index d5621fce4b8a..a25984b22c3a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -11358,19 +11358,68 @@ def restart_signal_handler(): # so shutdown is coordinated. from agent.afk_worker import AFKWorker, install_parent_agent_getter + # Cache the transient AFK-worker agent so we don't pay the build cost + # on every tick. Built lazily on first need; reused across ticks. + # Lives for the duration of the gateway process. + _afk_agent_cache: dict = {"agent": None} + def _get_afk_parent_agent(): - """Pick the most-recently-started running agent for the AFK worker - to use as parent_agent in delegate_task. Returns None if no agent - is currently active (worker logs and skips that tick).""" + """Resolve a parent_agent suitable for delegate_task in the AFK worker. + + Strategy: + 1. If a Discord/session agent is currently active, reuse it (free). + 2. Otherwise build a transient AIAgent dedicated to the worker + (same pattern cron/scheduler.py uses for its jobs). This is + needed because the gateway is typically idle when the worker + ticks — _running_agents is empty until a user message arrives. + 3. Cache the transient agent so subsequent ticks reuse it. + + Returns None on credential resolution failure — the worker logs + the cycle as 'error' and retries next tick. + """ + # Strategy 1: active session agent (most-recent) agents = getattr(runner, "_running_agents", None) or {} - if not agents: + if agents: + ts = getattr(runner, "_running_agents_ts", {}) or {} + if ts: + latest = max(ts, key=ts.get) + if latest in agents: + return agents[latest] + return next(iter(agents.values()), None) + + # Strategy 2: cached transient agent + if _afk_agent_cache["agent"] is not None: + return _afk_agent_cache["agent"] + + # Strategy 3: build a fresh transient agent + try: + from run_agent import AIAgent + runtime = _resolve_runtime_agent_kwargs() + gateway_model = _resolve_gateway_model() + agent = AIAgent( + model=gateway_model, + api_key=runtime.get("api_key"), + base_url=runtime.get("base_url"), + provider=runtime.get("provider"), + api_mode=runtime.get("api_mode"), + acp_command=runtime.get("command"), + acp_args=runtime.get("args"), + credential_pool=runtime.get("credential_pool"), + quiet_mode=True, + # No SOUL.md / AGENTS.md context injected (the delegated + # child gets its own context per delegate_task design). + skip_context_files=True, + # Don't let worker delegations corrupt user memory. + skip_memory=True, + platform="afk_worker", + session_id="afk_worker_persistent", + ) + _afk_agent_cache["agent"] = agent + logger.info("afk-worker: built transient parent_agent for delegations") + return agent + except Exception: + logger.exception("afk-worker: cannot build transient parent_agent") return None - ts = getattr(runner, "_running_agents_ts", {}) or {} - if ts: - latest = max(ts, key=ts.get) - if latest in agents: - return agents[latest] - return next(iter(agents.values()), None) install_parent_agent_getter(_get_afk_parent_agent) From 8071adc10442f9da1ad8a37303f4f00b04ba2f58 Mon Sep 17 00:00:00 2001 From: Anu Kheru Date: Mon, 4 May 2026 09:04:37 -0300 Subject: [PATCH 4/6] =?UTF-8?q?ah:=20WS-AUTO-002=20=E2=80=94=20fix=20Bug?= =?UTF-8?q?=202=20+=20debug=20logs=20for=20Bug=201=20OpenRouter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Overnight run produced 3 real completions via MiniMax + 3 false "completed" cycles where Sonnet/OpenRouter exhausted retries: status=completed summary="API call failed after 3 retries: " Bug 2 (fixed here): delegate_task hands back status=completed even when the child's LLM call never returned. The worker took it at face value and marked tasks as done in the picked-index, blocking retry. Fix: in _run_delegation, detect summaries starting with "API call failed" and promote status to error. The picked-index then schedules a 3-tick cooldown and retries instead of locking the task as done. +1 test simulating the prod failure pattern. Bug 1 (still open — debug logs added): direct curl from VPS with the same OPENROUTER_API_KEY → HTTP 200, Sonnet replies. But delegate_task → child agent → openrouter call → 3 retries failed. Override propagation likely loses the api_key somewhere between the worker call and the child runtime. Added INFO logs at three points: 1. afk_worker._run_delegation: provider/model/base_url/api_key_len before calling delegate_task 2. delegate_task: same after applying the 4 overrides to creds 3. _build_child_agent: effective creds vs parent creds for diff All length-only on api_key — no value ever logged (§1). Once we see the chain in journalctl, we'll know exactly where the override falls back to parent_api_key (which is MiniMax for the transient AFK parent, hence the 401-equivalent). --- agent/afk_worker.py | 37 +++++++++++++++++++++++++++++++--- tests/agent/test_afk_worker.py | 35 ++++++++++++++++++++++++++++++++ tools/delegate_tool.py | 28 +++++++++++++++++++++++++ 3 files changed, 97 insertions(+), 3 deletions(-) diff --git a/agent/afk_worker.py b/agent/afk_worker.py index fd0c67649e8c..ee8578c7cb95 100644 --- a/agent/afk_worker.py +++ b/agent/afk_worker.py @@ -536,6 +536,19 @@ def _run_delegation(self, task: TaskCandidate, model_cfg: dict) -> dict: # OQ3 — toolsets from matrix; fallback to safe minimal toolsets = model_cfg.get("toolsets") or ["file", "todo"] + # WS-AUTO-002 Bug 1 debug — trace what we're about to pass to + # delegate_task. Logs key length only, never the value (§1). + logger.info( + "afk-worker delegation overrides: provider=%s model=%s " + "base_url=%s api_key_env=%r api_key_len=%d toolsets=%s", + model_cfg.get("provider"), + model_cfg.get("model"), + model_cfg.get("base_url"), + api_key_env_name, + len(api_key) if api_key else 0, + toolsets, + ) + try: result_str = delegate_task( goal=goal, @@ -602,10 +615,28 @@ def _run_delegation(self, task: TaskCandidate, model_cfg: dict) -> dict: "duration_seconds": 0.0, } r = results[0] + status = r.get("status") + summary = r.get("summary") or "" + error = r.get("error") + + # Bug 2 fix — delegate_task returns status=completed even when the + # child agent's LLM call exhausted retries without ever responding. + # Detect the pattern in the summary and promote to status=error so + # the picked-index marks the task for retry instead of "done". + if status == "completed" and summary.lstrip().lower().startswith( + "api call failed" + ): + logger.warning( + "afk-worker: child reports completed but summary indicates " + "API failure; promoting to status=error" + ) + status = "error" + error = error or summary.strip().splitlines()[0] + return { - "status": r.get("status"), - "summary": r.get("summary"), - "error": r.get("error"), + "status": status, + "summary": summary or None, + "error": error, "api_calls": r.get("api_calls", 0), "duration_seconds": r.get("duration_seconds", 0.0), } diff --git a/tests/agent/test_afk_worker.py b/tests/agent/test_afk_worker.py index 0ec20628cc79..ac6d64eb8083 100644 --- a/tests/agent/test_afk_worker.py +++ b/tests/agent/test_afk_worker.py @@ -586,6 +586,41 @@ def test_verbose_false_silent_on_completed(self, tmp_path, monkeypatch): post_mock.assert_not_called() install_parent_agent_getter(None) + def test_completed_with_api_fail_promoted_to_error(self, tmp_path, monkeypatch): + """Bug 2 — when delegate_task returns status=completed but the summary + starts with 'API call failed', the worker must override to error.""" + _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) + f = tmp_path / "S.md" + f.write_text( + "## S\n- [ ] X [priority:P0] [afk:code_review]\n", encoding="utf-8", + ) + w = _make_worker(tmp_path, monkeypatch) + cfg = _cfg(tmp_path, status_files=[str(f)], verbose=False) + install_parent_agent_getter(lambda: object()) + # Simulate the exact failure pattern observed in prod + fake_payload = json.dumps({ + "results": [{ + "status": "completed", + "summary": "API call failed after 3 retries: ", + "api_calls": 1, "duration_seconds": 8.7, + }] + }) + with patch.object(w, "_read_quota", return_value={ + "minimax_5h_pct": 1.0, "minimax_weekly_pct": 0.5, + }), patch("tools.delegate_tool.delegate_task", return_value=fake_payload): + w._tick(cfg) + # Cycle status should be 'error', NOT 'completed' + obj = json.loads( + Path(cfg["log_jsonl_path"]).read_text(encoding="utf-8").splitlines()[0] + ) + assert obj["status"] == "error" + # Picked-index should record last_cycle_status=error so it's retried + idx = json.loads(Path(cfg["picked_index_path"]).read_text(encoding="utf-8")) + entry = next(iter(idx.values())) + assert entry["last_cycle_status"] == "error" + assert entry["error_skips_remaining"] > 0 + install_parent_agent_getter(None) + def test_verbose_skip_no_task_silent(self, tmp_path, monkeypatch): # skipped_no_task should be silent regardless of verbose flag _afk_state_file(tmp_path, monkeypatch, mode=MODE_AFK_AUTO) diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 1635f8a43954..ef8c02a8785f 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -985,6 +985,20 @@ def _child_thinking(text: str) -> None: effective_base_url = override_base_url or parent_agent.base_url effective_api_key = override_api_key or parent_api_key effective_api_mode = override_api_mode or getattr(parent_agent, "api_mode", None) + + # Bug 1 debug — log what _build_child_agent will instantiate the child + # with. Compare against parent values to spot drift. Length-only on keys. + if any((override_provider, override_base_url, override_api_key)): + logger.info( + "_build_child_agent effective creds: model=%s provider=%s " + "base_url=%s api_key_len=%d api_mode=%s | " + "parent_provider=%s parent_base_url=%s parent_key_len=%d", + effective_model, effective_provider, effective_base_url, + len(effective_api_key or ""), effective_api_mode, + getattr(parent_agent, "provider", None), + getattr(parent_agent, "base_url", None), + len(parent_api_key or ""), + ) effective_acp_command = override_acp_command or getattr( parent_agent, "acp_command", None ) @@ -1895,6 +1909,20 @@ def delegate_task( if api_key_override: creds["api_key"] = api_key_override + # Bug 1 debug — log resolved creds AFTER overrides applied. Length-only + # for api_key (never the value, §1). + if any((model_override, provider_override, base_url_override, api_key_override)): + _ck = creds.get("api_key") or "" + logger.info( + "delegate_task creds (post-override): provider=%s model=%s " + "base_url=%s api_mode=%s api_key_len=%d", + creds.get("provider"), + creds.get("model"), + creds.get("base_url"), + creds.get("api_mode"), + len(_ck), + ) + # Normalize to task list max_children = _get_max_concurrent_children() if tasks and isinstance(tasks, list): From 164141dabdd008007883e20a867435052b32b771 Mon Sep 17 00:00:00 2001 From: Anu Kheru Date: Mon, 4 May 2026 09:28:33 -0300 Subject: [PATCH 5/6] =?UTF-8?q?ah:=20WS-AUTO-002=20=E2=80=94=20add=20uncon?= =?UTF-8?q?ditional=20entry=20log=20in=20delegate=5Ftask?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Suspect the conditional 'if any(...)' debug log isn't firing for some reason (override seems to propagate based on auxiliary_client log, but no INFO from tools.delegate_tool appears in journal). Add an entry log without any condition to confirm code path is executed and override kwargs are received. --- tools/delegate_tool.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index ef8c02a8785f..2661eb5ee34a 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -1840,6 +1840,17 @@ def delegate_task( Returns JSON with results array, one entry per task. """ + # Bug 1 debug — unconditional entry log to confirm this code path runs + # and the new override params are received. Length-only on api_key. + logger.info( + "delegate_task ENTRY: goal_set=%s tasks_set=%s role=%s | " + "model_override=%r provider_override=%r base_url_override=%r " + "api_key_override_len=%d", + bool(goal), bool(tasks), role, + model_override, provider_override, base_url_override, + len(api_key_override or "") if api_key_override else 0, + ) + if parent_agent is None: return tool_error("delegate_task requires a parent agent context.") From 437ae69af8a51b3fca33ae922939cb97dc712cb9 Mon Sep 17 00:00:00 2001 From: Anu Kheru Date: Mon, 4 May 2026 10:37:41 -0300 Subject: [PATCH 6/6] =?UTF-8?q?ah:=20WS-AUTO-002=20=E2=80=94=20remove=20Op?= =?UTF-8?q?enRouter=20Bug=201=20debug=20logs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit These were added to trace where the override propagation lost the api_key for OpenRouter calls. Bug 1 became moot once we switched the AFK matrix away from openrouter to MiniMax (sub-agents) under Codex (main). Keeping the override-application code (4 if-blocks) since the wiring is correct and useful for any future per-call provider override. The afk_worker side log ('afk-worker delegation overrides: ...') stays in place — it's useful in steady-state to see what the worker picks per tick, length-only on api_key. Tests: 40/40 still green. --- tools/delegate_tool.py | 39 --------------------------------------- 1 file changed, 39 deletions(-) diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 2661eb5ee34a..1635f8a43954 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -985,20 +985,6 @@ def _child_thinking(text: str) -> None: effective_base_url = override_base_url or parent_agent.base_url effective_api_key = override_api_key or parent_api_key effective_api_mode = override_api_mode or getattr(parent_agent, "api_mode", None) - - # Bug 1 debug — log what _build_child_agent will instantiate the child - # with. Compare against parent values to spot drift. Length-only on keys. - if any((override_provider, override_base_url, override_api_key)): - logger.info( - "_build_child_agent effective creds: model=%s provider=%s " - "base_url=%s api_key_len=%d api_mode=%s | " - "parent_provider=%s parent_base_url=%s parent_key_len=%d", - effective_model, effective_provider, effective_base_url, - len(effective_api_key or ""), effective_api_mode, - getattr(parent_agent, "provider", None), - getattr(parent_agent, "base_url", None), - len(parent_api_key or ""), - ) effective_acp_command = override_acp_command or getattr( parent_agent, "acp_command", None ) @@ -1840,17 +1826,6 @@ def delegate_task( Returns JSON with results array, one entry per task. """ - # Bug 1 debug — unconditional entry log to confirm this code path runs - # and the new override params are received. Length-only on api_key. - logger.info( - "delegate_task ENTRY: goal_set=%s tasks_set=%s role=%s | " - "model_override=%r provider_override=%r base_url_override=%r " - "api_key_override_len=%d", - bool(goal), bool(tasks), role, - model_override, provider_override, base_url_override, - len(api_key_override or "") if api_key_override else 0, - ) - if parent_agent is None: return tool_error("delegate_task requires a parent agent context.") @@ -1920,20 +1895,6 @@ def delegate_task( if api_key_override: creds["api_key"] = api_key_override - # Bug 1 debug — log resolved creds AFTER overrides applied. Length-only - # for api_key (never the value, §1). - if any((model_override, provider_override, base_url_override, api_key_override)): - _ck = creds.get("api_key") or "" - logger.info( - "delegate_task creds (post-override): provider=%s model=%s " - "base_url=%s api_mode=%s api_key_len=%d", - creds.get("provider"), - creds.get("model"), - creds.get("base_url"), - creds.get("api_mode"), - len(_ck), - ) - # Normalize to task list max_children = _get_max_concurrent_children() if tasks and isinstance(tasks, list):