diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 160da36296..7902293d2b 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -1,22 +1,11 @@ #!/usr/bin/env python3 -""" -Async (background) delegation registry. +"""Async (background) delegation registry behind ``delegate_task(background=true)``. -Backs ``delegate_task(background=true)``: the parent dispatches a subagent on -a module-level daemon executor and returns a handle immediately. When the -child finishes, a ``type="async_delegation"`` event is pushed onto the SHARED -``process_registry.completion_queue``, which the CLI and gateway already drain -while the agent is idle. Reusing that rail means completions surface as a NEW -turn (never spliced mid-turn, so role alternation and the prompt cache stay -intact) and inherit the queue's de-dup, crash-recovery and drain wiring. - -The completion payload carries a self-contained task-source block (goal, -context, toolsets, model, dispatch time, status, result) so a parent deep in -unrelated context can use the result or re-dispatch. - -This module owns ONLY the async lifecycle; the child build + run is injected -as a ``runner`` (``delegate_tool._run_single_child``). -""" +The parent dispatches a subagent on a module-level daemon executor and returns a handle +immediately. On completion a ``type="async_delegation"`` event (self-contained task-source +block) is pushed onto the SHARED ``process_registry.completion_queue`` the CLI/gateway drain +while idle, so results surface as a NEW turn (never mid-turn) and inherit its de-dup and +crash-recovery wiring. Only the async lifecycle lives here; the child run is an injected ``runner``.""" from __future__ import annotations @@ -53,24 +42,20 @@ _DEFAULT_MAX_ASYNC_CHILDREN = 3 _MAX_RETAINED_COMPLETED = 50 _DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60 _MAX_DURABLE_PENDING = 1000 -# Cap retried deliveries so an unroutable row converges to terminal 'dropped' -# instead of replaying on every restart forever. +# Cap retried deliveries so an unroutable row converges to terminal 'dropped'. _MAX_DELIVERY_ATTEMPTS = 8 -# Pending completions older than this are dropped on restart replay instead -# of re-run as a full-context turn; 48h keeps weekend results deliverable. +# Pending completions older than this are dropped on restart replay instead of +# re-run as a full-context turn; 48h keeps weekend results deliverable. _MAX_COMPLETION_REPLAY_AGE_S = 48 * 3600.0 _DB_LOCK = threading.Lock() # ── Stale-delegation detection (progress-based, on by default) ────────────── -# A runner wedged before returning (e.g. stuck in its first API call) never -# reaches its finalizer, so the delegation shows "dispatched" forever. We do -# NOT use a wall-clock timeout (legitimate heavy work must never be killed for -# taking long); a single monitor thread samples per-dispatch PROGRESS via an -# injected ``progress_fn``. Advancing children are left alone; a frozen one is -# interrupted, given a grace window to unwind through the normal finalize path, -# and only force-finalized with a terminal ``stalled`` event if it never returns. -# Thresholds mirror delegate_tool's sync heartbeat monitor: idle stays tight, -# in-tool is much higher so legitimately slow tools can finish. +# A runner wedged before returning never reaches its finalizer, so it would show +# "dispatched" forever. No wall-clock timeout (heavy work must never be killed for +# taking long): one monitor thread samples per-dispatch PROGRESS via an injected +# ``progress_fn``; a frozen child is interrupted, given a grace window to unwind via +# the normal finalize path, and only force-finalized (terminal ``stalled`` event) if +# it never returns. Thresholds mirror delegate_tool's sync heartbeat monitor. _STALE_CHECK_INTERVAL = 30.0 _STALE_IDLE_SECONDS = 450.0 _STALE_IN_TOOL_SECONDS = 1200.0 @@ -86,27 +71,10 @@ _ACTIVE_STATES = ("running", "stalling") # reconstruct a full SessionSource (scope_id drives relay tenant egress). _ROUTING_KEYS = ("scope_id", "user_id", "user_name") # Structured stall metadata — additive, present only on stall finalizations. -_STALL_META_KEYS = ( - "stalled_after_quiet_seconds", "stall_threshold_seconds", "stall_phase", "stall_grace_seconds", -) +_STALL_META_KEYS = ("stalled_after_quiet_seconds", "stall_threshold_seconds", "stall_phase", "stall_grace_seconds") # Private stall bookkeeping on the record -> public field in list_async_delegations(). -_STALL_FIELD_MAP = ( - ("_stall_quiet_seconds", "stalled_after_quiet_seconds"), - ("_stall_threshold_seconds", "stall_threshold_seconds"), - ("_stall_in_tool", "stall_in_tool"), -) - - -def _copy_truthy(src: Dict[str, Any], dst: Dict[str, Any], keys) -> None: - for k in keys: - if src.get(k): - dst[k] = src[k] - - -def _copy_present(src: Dict[str, Any], dst: Dict[str, Any], keys) -> None: - for k in keys: - if k in src: - dst[k] = src[k] +_STALL_FIELD_MAP = (("_stall_quiet_seconds", "stalled_after_quiet_seconds"), + ("_stall_threshold_seconds", "stall_threshold_seconds"), ("_stall_in_tool", "stall_in_tool")) # ── Durable ledger (state.db / async_delegations) ─────────────────────────── @@ -128,13 +96,10 @@ def _connect() -> sqlite3.Connection: def _initialize_schema(conn: sqlite3.Connection) -> None: from hermes_state import apply_durability_barriers - - # Preserve the journal mode SessionDB configured on state.db: forcing WAL - # from every short-lived connection needs an exclusive lock and collides - # with live transcript/FTS writers. The ledger works in WAL or DELETE mode. + # Preserve the journal mode SessionDB configured on state.db: forcing WAL from + # every short-lived connection collides with live transcript/FTS writers. apply_durability_barriers(conn) - conn.execute( - """CREATE TABLE IF NOT EXISTS async_delegations ( + conn.execute("""CREATE TABLE IF NOT EXISTS async_delegations ( delegation_id TEXT PRIMARY KEY, origin_session TEXT NOT NULL, origin_ui_session_id TEXT NOT NULL DEFAULT '', @@ -154,30 +119,20 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: delivery_claim TEXT, delivery_claimed_at REAL, origin_session_id TEXT NOT NULL DEFAULT '' - )""" - ) + )""") columns = {row[1] for row in conn.execute("PRAGMA table_info(async_delegations)")} - for name, sql_type in ( - ("owner_pid", "INTEGER"), - ("owner_started_at", "INTEGER"), - ("task_json", "TEXT"), - ("delivery_claim", "TEXT"), - ("delivery_claimed_at", "REAL"), - # Raw api_server session id of the ORIGINATING request (wake target); - # without it restart-recovered completions are unroutable there. - ("origin_session_id", "TEXT"), - ): + # origin_session_id: raw api_server session id of the ORIGINATING request + # (wake target); without it restart-recovered completions are unroutable there. + for name, sql_type in (("owner_pid", "INTEGER"), ("owner_started_at", "INTEGER"), ("task_json", "TEXT"), + ("delivery_claim", "TEXT"), ("delivery_claimed_at", "REAL"), ("origin_session_id", "TEXT")): if name not in columns: conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}") @contextmanager def _transaction() -> Iterator[sqlite3.Connection]: - """Open a connection, commit/rollback on exit, and ALWAYS close it. - - ``with conn:`` only commits/rolls back; without the explicit close every - durable write leaks a connection (and WAL/SHM fds) until GC. - """ + """Open a connection, commit/rollback on exit, and ALWAYS close it (``with conn:`` + alone leaks the connection and WAL/SHM fds until GC).""" conn = _connect() try: with conn: @@ -187,23 +142,14 @@ def _transaction() -> Iterator[sqlite3.Connection]: def _capture_routing_origin() -> Dict[str, Any]: - """Snapshot scope_id/user_id/user_name from the dispatching turn. - - Captured on the PARENT thread (the daemon worker has no contextvars) and - persisted so a restart-replayed completion can rebuild a SessionSource. - Best-effort: empty values are omitted. - """ - origin: Dict[str, Any] = {} + """Snapshot scope_id/user_id/user_name on the PARENT thread (the daemon worker + has no contextvars) so a restart-replayed completion can rebuild a SessionSource. + Best-effort: empty values are omitted.""" try: from gateway.session_context import get_session_env - - for key in _ROUTING_KEYS: - value = get_session_env(f"HERMES_SESSION_{key.upper()}", "") - if value: - origin[key] = value + return {k: v for k in _ROUTING_KEYS if (v := get_session_env(f"HERMES_SESSION_{k.upper()}", ""))} except Exception: # noqa: BLE001 - routing origin is additive, never fatal - pass - return origin + return {} def _persist_dispatch(record: Dict[str, Any]) -> None: @@ -215,82 +161,54 @@ def _persist_dispatch(record: Dict[str, Any]) -> None: owner_started_at = None task_payload = { key: record.get(key) - for key in ( - "goal", "goals", "context", "toolsets", "role", "model", "is_batch", - *_ROUTING_KEYS, - ) - if key in record - } + for key in ("goal", "goals", "context", "toolsets", "role", "model", "is_batch", *_ROUTING_KEYS) + if key in record} with _DB_LOCK, _transaction() as conn: - conn.execute( - """INSERT OR REPLACE INTO async_delegations + conn.execute("""INSERT OR REPLACE INTO async_delegations (delegation_id, origin_session, origin_ui_session_id, parent_session_id, state, dispatched_at, updated_at, delivery_state, delivery_attempts, owner_pid, owner_started_at, task_json, origin_session_id) VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?, ?)""", - (record["delegation_id"], record.get("session_key", ""), - record.get("origin_ui_session_id", ""), record.get("parent_session_id"), - record["dispatched_at"], now, os.getpid(), - owner_started_at, json.dumps(task_payload), - record.get("origin_session_id", "")), - ) + (record["delegation_id"], record.get("session_key", ""), record.get("origin_ui_session_id", ""), + record.get("parent_session_id"), record["dispatched_at"], now, os.getpid(), owner_started_at, + json.dumps(task_payload), record.get("origin_session_id", ""))) _prune_durable_records() -def _delete_durable_delegation(delegation_id: str) -> None: - with _DB_LOCK, _transaction() as conn: - conn.execute("DELETE FROM async_delegations WHERE delegation_id=?", (delegation_id,)) - - def _prune_durable_records() -> None: """Bound terminal history, preferring delivered records for deletion.""" cutoff = time.time() - _DURABLE_RETENTION_SECONDS with _DB_LOCK, _transaction() as conn: conn.execute( - "DELETE FROM async_delegations WHERE delivery_state='delivered' AND updated_at < ?", - (cutoff,), - ) + "DELETE FROM async_delegations WHERE delivery_state='delivered' AND updated_at < ?", (cutoff,)) terminal_count = conn.execute( - "SELECT COUNT(*) FROM async_delegations WHERE state NOT IN ('running','finalizing')" - ).fetchone()[0] - excess = max(0, terminal_count - _MAX_RETAINED_COMPLETED) - if excess: - conn.execute( - """DELETE FROM async_delegations WHERE delegation_id IN ( + "SELECT COUNT(*) FROM async_delegations WHERE state NOT IN ('running','finalizing')").fetchone()[0] + if terminal_count > _MAX_RETAINED_COMPLETED: + conn.execute("""DELETE FROM async_delegations WHERE delegation_id IN ( SELECT delegation_id FROM async_delegations WHERE state NOT IN ('running','finalizing') ORDER BY CASE delivery_state WHEN 'delivered' THEN 0 ELSE 1 END, updated_at ASC LIMIT ? - )""", - (excess,), - ) - pending_count = conn.execute( - """SELECT COUNT(*) FROM async_delegations - WHERE state NOT IN ('running','finalizing') AND delivery_state='pending'""" - ).fetchone()[0] - overflow = max(0, pending_count - _MAX_DURABLE_PENDING) - if overflow: - conn.execute( - """DELETE FROM async_delegations WHERE delegation_id IN ( + )""", (terminal_count - _MAX_RETAINED_COMPLETED,)) + pending_count = conn.execute("""SELECT COUNT(*) FROM async_delegations + WHERE state NOT IN ('running','finalizing') AND delivery_state='pending'""").fetchone()[0] + if pending_count > _MAX_DURABLE_PENDING: + conn.execute("""DELETE FROM async_delegations WHERE delegation_id IN ( SELECT delegation_id FROM async_delegations WHERE state NOT IN ('running','finalizing') AND delivery_state='pending' ORDER BY updated_at ASC LIMIT ? - )""", - (overflow,), - ) + )""", (pending_count - _MAX_DURABLE_PENDING,)) def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None: now = time.time() with _DB_LOCK, _transaction() as conn: - conn.execute( - """UPDATE async_delegations SET state=?, completed_at=?, updated_at=?, + conn.execute("""UPDATE async_delegations SET state=?, completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?""", (event.get("status", "completed"), event.get("completed_at", now), now, - json.dumps(event), json.dumps(result), event["delegation_id"]), - ) + json.dumps(event), json.dumps(result), event["delegation_id"])) def recover_abandoned_delegations() -> int: @@ -299,85 +217,58 @@ def recover_abandoned_delegations() -> int: from gateway.status import _pid_exists, get_process_start_time except Exception: return 0 - now = time.time() - recovered = 0 + now, recovered = time.time(), 0 with _DB_LOCK, _transaction() as conn: - rows = conn.execute( - """SELECT delegation_id, origin_session, origin_ui_session_id, + rows = conn.execute("""SELECT delegation_id, origin_session, origin_ui_session_id, parent_session_id, dispatched_at, owner_pid, owner_started_at, task_json, origin_session_id - FROM async_delegations WHERE state IN ('running','finalizing')""" - ).fetchall() + FROM async_delegations WHERE state IN ('running','finalizing')""").fetchall() for row in rows: - (delegation_id, session_key, origin_ui, parent_id, dispatched_at, - pid, started, task_json, origin_session_id) = row - live = False - if pid: - live = _pid_exists(int(pid)) - if live and started is not None: - live = get_process_start_time(int(pid)) == int(started) - if live: + delegation_id, session_key, origin_ui, parent_id, dispatched_at, pid, started, task_json, origin_sid = row + if pid and _pid_exists(int(pid)) and (started is None or get_process_start_time(int(pid)) == int(started)): continue task = json.loads(task_json or "{}") event = { - "type": "async_delegation", "delegation_id": delegation_id, - "session_key": session_key, "origin_ui_session_id": origin_ui, - "origin_session_id": origin_session_id or "", - "parent_session_id": parent_id, "goal": task.get("goal", ""), - "goals": task.get("goals"), "context": task.get("context"), - "toolsets": task.get("toolsets"), "role": task.get("role"), + "type": "async_delegation", "delegation_id": delegation_id, "session_key": session_key, + "origin_ui_session_id": origin_ui, "origin_session_id": origin_sid or "", + "parent_session_id": parent_id, "goal": task.get("goal", ""), "goals": task.get("goals"), + "context": task.get("context"), "toolsets": task.get("toolsets"), "role": task.get("role"), "model": task.get("model"), "is_batch": bool(task.get("is_batch")), "status": "unknown", "summary": None, "error": "Delegation owner exited before recording a terminal result; outcome unknown.", "dispatched_at": dispatched_at, "completed_at": now, - } - _copy_truthy(task, event, _ROUTING_KEYS) + **{k: task[k] for k in _ROUTING_KEYS if task.get(k)}} result = {"status": "unknown", "summary": None, "error": event["error"]} - conn.execute( - """UPDATE async_delegations SET state='unknown', completed_at=?, + conn.execute("""UPDATE async_delegations SET state='unknown', completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' - WHERE delegation_id=?""", - (now, now, json.dumps(event), json.dumps(result), delegation_id), - ) + WHERE delegation_id=?""", (now, now, json.dumps(event), json.dumps(result), delegation_id)) recovered += 1 return recovered def restore_undelivered_completions(target_queue) -> int: """Enqueue durable pending completions as fresh turns after process start. - - Restored events are stamped ``restored=True`` in memory only: they came - from a PREVIOUS process, so drains without an ownership filter must leave - them for a consumer that can prove ownership. Rows older than - ``_MAX_COMPLETION_REPLAY_AGE_S`` are terminally dropped instead of - replaying a full-context turn nobody is waiting on. - """ + Restored events are stamped ``restored=True`` in memory only: they came from a PREVIOUS + process, so drains without an ownership filter must leave them for a consumer that can + prove ownership. Rows older than ``_MAX_COMPLETION_REPLAY_AGE_S`` are terminally dropped + instead of replaying a turn nobody is waiting on.""" recover_abandoned_delegations() - now = time.time() - restored = 0 + now, restored = time.time(), 0 with _DB_LOCK, _transaction() as conn: - rows = conn.execute( - """SELECT delegation_id, event_json, completed_at, dispatched_at + rows = conn.execute("""SELECT delegation_id, event_json, completed_at, dispatched_at FROM async_delegations WHERE state != 'running' AND delivery_state='pending' AND event_json IS NOT NULL - ORDER BY completed_at, delegation_id""" - ).fetchall() + ORDER BY completed_at, delegation_id""").fetchall() for delegation_id, payload, completed_at, dispatched_at in rows: age_basis = completed_at or dispatched_at if age_basis and (now - age_basis) > _MAX_COMPLETION_REPLAY_AGE_S: - conn.execute( - """UPDATE async_delegations SET delivery_state='dropped', + conn.execute("""UPDATE async_delegations SET delivery_state='dropped', delivery_claim=NULL, delivery_claimed_at=NULL, updated_at=? - WHERE delegation_id=? AND delivery_state='pending'""", - (now, delegation_id), - ) - logger.warning( - "Async delegation %s: pending completion is %.1fh old " - "(cap %.1fh); terminally dropping the replay (result " - "remains queryable).", - delegation_id, (now - age_basis) / 3600.0, _MAX_COMPLETION_REPLAY_AGE_S / 3600.0, - ) + WHERE delegation_id=? AND delivery_state='pending'""", (now, delegation_id)) + logger.warning("Async delegation %s: pending completion is %.1fh old " + "(cap %.1fh); terminally dropping the replay (result remains queryable).", + delegation_id, (now - age_basis) / 3600.0, _MAX_COMPLETION_REPLAY_AGE_S / 3600.0) continue evt = json.loads(payload) if isinstance(evt, dict): @@ -398,9 +289,7 @@ def mark_completion_delivered(delegation_id: str) -> bool: now = time.time() return _update_delivery( """UPDATE async_delegations SET delivery_state='delivered', delivered_at=?, updated_at=? - WHERE delegation_id=? AND delivery_state!='delivered'""", - (now, now, delegation_id), - ) + WHERE delegation_id=? AND delivery_state!='delivered'""", (now, now, delegation_id)) def claim_completion_delivery(delegation_id: str, claim_id: str) -> bool: @@ -408,26 +297,20 @@ def claim_completion_delivery(delegation_id: str, claim_id: str) -> bool: now = time.time() with _DB_LOCK, _transaction() as conn: row = conn.execute( - "SELECT delivery_state FROM async_delegations WHERE delegation_id=?", - (delegation_id,), - ).fetchone() + "SELECT delivery_state FROM async_delegations WHERE delegation_id=?", (delegation_id,)).fetchone() if row is None: return True # legacy event created before durable dispatch - cur = conn.execute( - """UPDATE async_delegations SET delivery_claim=?, delivery_claimed_at=?, + cur = conn.execute("""UPDATE async_delegations SET delivery_claim=?, delivery_claimed_at=?, delivery_attempts=delivery_attempts+1, updated_at=? WHERE delegation_id=? AND delivery_state='pending' AND (delivery_claim IS NULL OR delivery_claimed_at < ?)""", - (claim_id, now, now, delegation_id, now - 300), - ) + (claim_id, now, now, delegation_id, now - 300)) return cur.rowcount == 1 def claim_event_delivery(evt: Dict[str, Any], consumer: str) -> Optional[str]: """Claim a durable delegation event; non-durable events need no token.""" - if evt.get("type") != "async_delegation": - return "" - delegation_id = str(evt.get("delegation_id") or "") + delegation_id = str(evt.get("delegation_id") or "") if evt.get("type") == "async_delegation" else "" if not delegation_id: return "" claim_id = f"{consumer}:{os.getpid()}:{uuid.uuid4().hex}" @@ -435,106 +318,83 @@ def claim_event_delivery(evt: Dict[str, Any], consumer: str) -> Optional[str]: def release_completion_delivery(delegation_id: str, claim_id: str) -> bool: - """Release a failed delivery claim so another consumer may retry. - - Attempts are counted at claim time; once the budget is exhausted the row - converges to terminal ``dropped`` (only pending rows replay on restart). - """ + """Release a failed delivery claim so another consumer may retry. Attempts are + counted at claim time; once the budget is exhausted the row converges to + terminal ``dropped`` (only pending rows replay on restart).""" now = time.time() with _DB_LOCK, _transaction() as conn: - capped = conn.execute( - """UPDATE async_delegations SET delivery_state='dropped', + capped = conn.execute("""UPDATE async_delegations SET delivery_state='dropped', delivery_claim=NULL, delivery_claimed_at=NULL, updated_at=? WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=? AND delivery_attempts>=?""", - (now, delegation_id, claim_id, _MAX_DELIVERY_ATTEMPTS), - ) + (now, delegation_id, claim_id, _MAX_DELIVERY_ATTEMPTS)) if capped.rowcount == 1: logger.warning("Async delegation %s exhausted its %d delivery attempts; " "marking terminally dropped (result remains queryable).", delegation_id, _MAX_DELIVERY_ATTEMPTS) return True - cur = conn.execute( - """UPDATE async_delegations SET delivery_claim=NULL, + cur = conn.execute("""UPDATE async_delegations SET delivery_claim=NULL, delivery_claimed_at=NULL, updated_at=? WHERE delegation_id=? AND delivery_state='pending' - AND delivery_claim=?""", - (now, delegation_id, claim_id), - ) + AND delivery_claim=?""", (now, delegation_id, claim_id)) return cur.rowcount == 1 def drop_completion_delivery(delegation_id: str, claim_id: str) -> bool: - """Terminally drop a claimed completion whose target is permanently gone. - - Used when the spawning session ended at an explicit user boundary (/new, - reset) rather than a compression rotation. ``dropped`` (not ``delivered``) - keeps the ack honest; (not ``pending``) - keeps restart recovery from replaying it into a fail-closed drop forever. - """ - return _update_delivery( - """UPDATE async_delegations SET delivery_state='dropped', + """Terminally drop a claimed completion whose target is permanently gone (the + spawning session ended at an explicit user boundary such as /new or reset). + ``dropped`` — not ``delivered`` — keeps the ack honest; not ``pending`` keeps + restart recovery from replaying it into a fail-closed drop forever.""" + return _update_delivery("""UPDATE async_delegations SET delivery_state='dropped', updated_at=?, delivery_claim=NULL, delivery_claimed_at=NULL WHERE delegation_id=? AND delivery_state='pending' - AND delivery_claim=?""", - (time.time(), delegation_id, claim_id), - ) + AND delivery_claim=?""", (time.time(), delegation_id, claim_id)) def complete_completion_delivery(delegation_id: str, claim_id: str) -> bool: """Acknowledge acceptance for the consumer holding this claim.""" now = time.time() - return _update_delivery( - """UPDATE async_delegations SET delivery_state='delivered', + return _update_delivery("""UPDATE async_delegations SET delivery_state='delivered', delivered_at=?, updated_at=?, delivery_claim=NULL, delivery_claimed_at=NULL WHERE delegation_id=? AND delivery_state='pending' - AND delivery_claim=?""", - (now, now, delegation_id, claim_id), - ) + AND delivery_claim=?""", (now, now, delegation_id, claim_id)) def complete_event_delivery(evt: Dict[str, Any], claim_id: str) -> None: - if claim_id and evt.get("type") == "async_delegation": - complete_completion_delivery(str(evt.get("delegation_id") or ""), claim_id) + _event_delivery(complete_completion_delivery, evt, claim_id) def release_event_delivery(evt: Dict[str, Any], claim_id: str) -> None: + _event_delivery(release_completion_delivery, evt, claim_id) + + +def _event_delivery(fn, evt: Dict[str, Any], claim_id: str) -> None: if claim_id and evt.get("type") == "async_delegation": - release_completion_delivery(str(evt.get("delegation_id") or ""), claim_id) + fn(str(evt.get("delegation_id") or ""), claim_id) def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: with _DB_LOCK, _transaction() as conn: - row = conn.execute( - """SELECT origin_session, state, dispatched_at, completed_at, + row = conn.execute("""SELECT origin_session, state, dispatched_at, completed_at, result_json, delivery_state, delivery_attempts, origin_session_id - FROM async_delegations WHERE delegation_id=?""", (delegation_id,), - ).fetchone() - if row is None: - return None - return { - "delegation_id": delegation_id, "origin_session": row[0], "state": row[1], - "dispatched_at": row[2], "completed_at": row[3], - "result": json.loads(row[4]) if row[4] else None, - "delivery_state": row[5], "delivery_attempts": row[6], "origin_session_id": row[7] or "", - } + FROM async_delegations WHERE delegation_id=?""", (delegation_id,)).fetchone() + return None if row is None else { + "delegation_id": delegation_id, "origin_session": row[0], "state": row[1], "dispatched_at": row[2], + "completed_at": row[3], "result": json.loads(row[4]) if row[4] else None, "delivery_state": row[5], + "delivery_attempts": row[6], "origin_session_id": row[7] or ""} # ── In-memory registry queries ────────────────────────────────────────────── def _get_executor(max_workers: int) -> ThreadPoolExecutor: - """Lazily create (or grow, never shrink) the shared daemon executor. - - In-flight futures keep running on a replaced pool until it is collected. - """ + """Lazily create (or grow, never shrink) the shared daemon executor; in-flight + futures keep running on a replaced pool until it is collected.""" global _executor, _executor_max_workers with _executor_lock: if _executor is None or max_workers > _executor_max_workers: - _executor = DaemonThreadPoolExecutor( - max_workers=max_workers, thread_name_prefix="async-delegate", - ) + _executor = DaemonThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="async-delegate") _executor_max_workers = max_workers return _executor @@ -546,55 +406,32 @@ def active_count() -> int: def active_task_count() -> int: - """Number of running child subagents (a batch of N contributes N). - - The truthful observability figure, unlike ``active_count()`` slots; a - batch with no goal list falls back to 1. - """ - def _size(r: Dict[str, Any]) -> int: - goals = r.get("goals") if r.get("is_batch") else None - return len(goals) if isinstance(goals, (list, tuple)) and goals else 1 - + """Number of running child subagents (a batch of N contributes N; a batch with + no goal list counts 1) — the truthful observability figure, unlike slots.""" with _records_lock: - return sum(_size(r) for r in _records.values() if r.get("status") in {"running", "finalizing"}) + return sum( + len(r["goals"]) if r.get("is_batch") and isinstance(r.get("goals"), (list, tuple)) and r["goals"] else 1 + for r in _records.values() if r.get("status") in {"running", "finalizing"}) -def _matches_session_selectors( - record: Dict[str, Any], *, session_key: str = "", origin_ui_session_id: str = "", - parent_session_id: str = "", -) -> bool: - return any( - wanted and str(record.get(field) or "") == wanted - for field, wanted in ( - ("origin_ui_session_id", origin_ui_session_id), - ("session_key", session_key), - ("parent_session_id", parent_session_id), - ) - ) +def _session_records(statuses, session_key: str, origin_ui_session_id: str, parent_session_id: str) -> list: + """Records in ``statuses`` owned by a session: any non-empty selector claims the + record — ``origin_ui_session_id`` (TUI tab), ``session_key`` (routing key at + dispatch), or ``parent_session_id`` (spawner's durable id — the right one for + gateway chats, whose session_key survives ``/new`` while the session id rotates).""" + selectors = [(field, wanted) for field, wanted in ( + ("origin_ui_session_id", origin_ui_session_id), ("session_key", session_key), + ("parent_session_id", parent_session_id)) if wanted] + if not selectors: + return [] + with _records_lock: + return [r for r in _records.values() if r.get("status") in statuses + and any(str(r.get(field) or "") == wanted for field, wanted in selectors)] -def _session_records_locked(statuses, session_key, origin_ui_session_id, parent_session_id): - """Records in ``statuses`` matching any selector. Caller holds ``_records_lock``.""" - return [ - r for r in _records.values() - if r.get("status") in statuses - and _matches_session_selectors( - r, - session_key=session_key, - origin_ui_session_id=origin_ui_session_id, - parent_session_id=parent_session_id, - ) - ] - - -def has_live_for_session( - session_key: str = "", origin_ui_session_id: str = "", parent_session_id: str = "", -) -> bool: +def has_live_for_session(session_key: str = "", origin_ui_session_id: str = "", parent_session_id: str = "") -> bool: """Whether a session still owns any live (running/stalling/finalizing) delegation.""" - if not session_key and not origin_ui_session_id and not parent_session_id: - return False - with _records_lock: - return bool(_session_records_locked(_LIVE_STATES, session_key, origin_ui_session_id, parent_session_id)) + return bool(_session_records(_LIVE_STATES, session_key, origin_ui_session_id, parent_session_id)) def _new_delegation_id() -> str: @@ -604,37 +441,28 @@ def _new_delegation_id() -> str: def _prune_completed_locked() -> None: """Drop the oldest completed records beyond the cap. Caller holds ``_records_lock``.""" completed = [(rid, r) for rid, r in _records.items() if r.get("status") != "running"] - if len(completed) <= _MAX_RETAINED_COMPLETED: - return completed.sort(key=lambda kv: kv[1].get("completed_at") or kv[1].get("dispatched_at") or 0) - for rid, _ in completed[: len(completed) - _MAX_RETAINED_COMPLETED]: + for rid, _ in completed[: max(0, len(completed) - _MAX_RETAINED_COMPLETED)]: _records.pop(rid, None) def _current_origin_session_id() -> str: - """Raw session id of the ORIGINATING api_server request, or ``""``. - - ``HERMES_SESSION_ID`` is unsafe here: building the child agent calls - ``set_current_session_id(child.session_id)`` moments before dispatch, so - the wake would self-post into the subagent's own session. The - request-scoped ``HERMES_SESSION_CHAT_ID`` binding (raw X-Hermes-Session-Id - on api_server) survives child construction; on push platforms chat_id is - a chat, not a session, so yield ``""`` there. - """ + """Raw session id of the ORIGINATING api_server request, or ``""``. ``HERMES_SESSION_ID`` + is unsafe here: building the child agent calls ``set_current_session_id(child.session_id)`` + just before dispatch, so the wake would self-post into the subagent's own session. The + request-scoped ``HERMES_SESSION_CHAT_ID`` (raw X-Hermes-Session-Id on api_server) survives + child construction; on push platforms chat_id is a chat, not a session => ``""``.""" try: from gateway.session_context import get_session_env - - if get_session_env("HERMES_SESSION_PLATFORM", "") != "api_server": - return "" - return get_session_env("HERMES_SESSION_CHAT_ID", "") or "" + is_api = get_session_env("HERMES_SESSION_PLATFORM", "") == "api_server" + return (get_session_env("HERMES_SESSION_CHAT_ID", "") or "") if is_api else "" except Exception: return "" # ── Dispatch ──────────────────────────────────────────────────────────────── def _single_crash(error: str, duration: float) -> Dict[str, Any]: - return {"status": "error", "summary": None, "error": error, "api_calls": 0, - "duration_seconds": duration} + return {"status": "error", "summary": None, "error": error, "api_calls": 0, "duration_seconds": duration} def _batch_crash(error: str, duration: float) -> Dict[str, Any]: @@ -644,63 +472,41 @@ def _batch_crash(error: str, duration: float) -> Dict[str, Any]: def _batch_status(combined: Dict[str, Any]) -> str: """Batch status: completed unless every child errored/was interrupted.""" child_results = combined.get("results") or [] - if child_results and all( - (r.get("status") not in ("completed", "success")) for r in child_results - ): - return "error" - return "completed" + ok = ("completed", "success") + return "error" if child_results and all(r.get("status") not in ok for r in child_results) else "completed" def _dispatch( *, delegation_id: str, goal: str, goals: Optional[List[str]], context: Optional[str], toolsets: Optional[List[str]], role: str, model: Optional[str], session_key: str, - parent_session_id: Optional[str], runner: Callable[[], Dict[str, Any]], - origin_ui_session_id: str, origin_session_id: str, - interrupt_fn: Optional[Callable[[], None]], max_async_children: int, + parent_session_id: Optional[str], runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str, + origin_session_id: str, interrupt_fn: Optional[Callable[[], None]], max_async_children: int, progress_fn: Optional[Callable[[], tuple]], capacity_error: str, ) -> Dict[str, Any]: - """Shared dispatch core for single (``goals is None``) and batch units. - - Capacity check + record insert happen under ONE lock hold so concurrent - dispatches can't both pass the check and exceed the cap. At capacity the - dispatch is REJECTED (never queued) so a runaway model can't pile up - unbounded background work. - """ + """Shared dispatch core for single (``goals is None``) and batch units. Capacity check + + record insert happen under ONE lock hold so concurrent dispatches can't both pass the check + and exceed the cap. At capacity the dispatch is REJECTED (never queued) so a runaway model + can't pile up unbounded background work.""" is_batch = goals is not None label = " batch" if is_batch else "" classify = _batch_status if is_batch else (lambda r: r.get("status") or "completed") crash_result = _batch_crash if is_batch else _single_crash dispatched_at = time.time() record: Dict[str, Any] = { - "delegation_id": delegation_id, - "goal": goal, - **({"goals": list(goals)} if is_batch else {}), - "context": context, - "toolsets": list(toolsets) if toolsets else None, - "role": role, - "model": model, - "session_key": session_key, - "origin_ui_session_id": origin_ui_session_id, - "origin_session_id": origin_session_id, - "parent_session_id": parent_session_id, + "delegation_id": delegation_id, "goal": goal, **({"goals": list(goals)} if is_batch else {}), + "context": context, "toolsets": list(toolsets) if toolsets else None, "role": role, "model": model, + "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, + "origin_session_id": origin_session_id, "parent_session_id": parent_session_id, **_capture_routing_origin(), - "status": "running", - "dispatched_at": dispatched_at, - "completed_at": None, - "interrupt_fn": interrupt_fn, - **({"is_batch": True} if is_batch else {}), - "progress_fn": progress_fn, + "status": "running", "dispatched_at": dispatched_at, "completed_at": None, + "interrupt_fn": interrupt_fn, **({"is_batch": True} if is_batch else {}), "progress_fn": progress_fn, # Stale-monitor bookkeeping (see _stale_monitor_loop). - "_progress_token": None, - "_progress_ts": dispatched_at, - "_interrupted_at": None, - } + "_progress_token": None, "_progress_ts": dispatched_at, "_interrupted_at": None} with _records_lock: running = sum(1 for r in _records.values() if r.get("status") in _ACTIVE_STATES) if running >= max_async_children: return {"status": "rejected", "error": capacity_error} _records[delegation_id] = record - _persist_dispatch(record) executor = _get_executor(max_async_children) @@ -713,18 +519,17 @@ def _dispatch( except Exception as exc: # noqa: BLE001 — must never crash the worker logger.exception(f"Async delegation{label} %s crashed", delegation_id) result = crash_result(f"{type(exc).__name__}: {exc}", round(time.time() - dispatched_at, 2)) - status = "error" finally: _finalize(delegation_id, result, status) try: - # Propagate the dispatching profile so the detached child resolves - # get_hermes_home() under the right profile. + # Propagate the dispatching profile so the detached child resolves get_hermes_home() correctly. executor.submit(propagate_context_to_thread(_worker)) except Exception as exc: # pragma: no cover — pool submit failure is rare with _records_lock: _records.pop(delegation_id, None) - _delete_durable_delegation(delegation_id) + with _DB_LOCK, _transaction() as conn: + conn.execute("DELETE FROM async_delegations WHERE delegation_id=?", (delegation_id,)) return {"status": "rejected", "error": f"Failed to schedule async delegation{label}: {exc}"} if progress_fn is not None: _ensure_stale_monitor() @@ -732,37 +537,27 @@ def _dispatch( def dispatch_async_delegation( - *, goal: str, context: Optional[str], toolsets: Optional[List[str]], role: str, - model: Optional[str], session_key: str, parent_session_id: Optional[str] = None, - runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", - origin_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, - max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, - progress_fn: Optional[Callable[[], tuple]] = None, + *, goal: str, context: Optional[str], toolsets: Optional[List[str]], role: str, model: Optional[str], + session_key: str, parent_session_id: Optional[str] = None, runner: Callable[[], Dict[str, Any]], + origin_ui_session_id: str = "", origin_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, + max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, progress_fn: Optional[Callable[[], tuple]] = None, ) -> Dict[str, Any]: """Spawn ``runner`` on the daemon executor and return a handle immediately. - - ``session_key``/``parent_session_id`` are captured on the parent thread - (the worker carries no contextvars) and route the completion back to the - spawning session. ``progress_fn() -> (token, in_tool)`` enables stale - monitoring; omitted = unmonitored. Returns ``{"status": "dispatched", - "delegation_id"}`` or ``{"status": "rejected", "error"}`` at capacity. - """ + ``session_key``/``parent_session_id`` are captured on the parent thread (the worker carries + no contextvars) and route the completion back to the spawning session. + ``progress_fn() -> (token, in_tool)`` enables stale monitoring; omitted = unmonitored. + Returns ``{"status": "dispatched", "delegation_id"}`` or ``{"status": "rejected", "error"}``.""" delegation_id = _new_delegation_id() handle = _dispatch( delegation_id=delegation_id, goal=goal, goals=None, context=context, toolsets=toolsets, role=role, model=model, session_key=session_key, parent_session_id=parent_session_id, runner=runner, origin_ui_session_id=origin_ui_session_id, origin_session_id=origin_session_id, - interrupt_fn=interrupt_fn, max_async_children=max_async_children, - progress_fn=progress_fn, + interrupt_fn=interrupt_fn, max_async_children=max_async_children, progress_fn=progress_fn, capacity_error=( - f"Async delegation capacity reached ({max_async_children} " - f"running). Wait for one to finish (its result will re-enter " - f"the chat), or run this task synchronously " - f"(background=false). Raise delegation.max_concurrent_children in " - f"config.yaml to allow more concurrent background subagents." - ), - ) + f"Async delegation capacity reached ({max_async_children} running). Wait for one to finish " + "(its result will re-enter the chat), or run this task synchronously (background=false). " + "Raise delegation.max_concurrent_children in config.yaml to allow more concurrent background subagents.")) if handle["status"] == "dispatched": logger.info("Dispatched async delegation %s (session_key=%s): %s", delegation_id, session_key or "", (goal or "")[:80]) @@ -770,39 +565,29 @@ def dispatch_async_delegation( def dispatch_async_delegation_batch( - *, goals: List[str], context: Optional[str], toolsets: Optional[List[str]], role: str, - model: Optional[str], session_key: str, parent_session_id: Optional[str] = None, - runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", - origin_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, - max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, - delegation_id: Optional[str] = None, progress_fn: Optional[Callable[[], tuple]] = None, + *, goals: List[str], context: Optional[str], toolsets: Optional[List[str]], role: str, model: Optional[str], + session_key: str, parent_session_id: Optional[str] = None, runner: Callable[[], Dict[str, Any]], + origin_ui_session_id: str = "", origin_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, + max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, delegation_id: Optional[str] = None, + progress_fn: Optional[Callable[[], tuple]] = None, ) -> Dict[str, Any]: - """Dispatch a WHOLE fan-out batch as ONE background unit. - - ``runner`` runs the entire batch and returns the combined - ``{"results": [...], "total_duration_seconds": N}`` dict. The batch - occupies ONE async slot (in-batch parallelism is bounded separately) and - produces a SINGLE completion event carrying the per-task ``results``. - """ + """Dispatch a WHOLE fan-out batch as ONE background unit: ``runner`` runs the + entire batch and returns the combined ``{"results": [...], "total_duration_seconds": N}`` + dict. The batch occupies ONE async slot (in-batch parallelism is bounded + separately) and produces a SINGLE completion event carrying per-task ``results``.""" delegation_id = delegation_id or _new_delegation_id() n = len(goals) - combined_goal = ( - goals[0] if n == 1 else f"{n} parallel subagents: " + "; ".join(g[:40] for g in goals) - ) + combined_goal = goals[0] if n == 1 else f"{n} parallel subagents: " + "; ".join(g[:40] for g in goals) handle = _dispatch( delegation_id=delegation_id, goal=combined_goal, goals=goals, context=context, toolsets=toolsets, role=role, model=model, session_key=session_key, parent_session_id=parent_session_id, runner=runner, origin_ui_session_id=origin_ui_session_id, origin_session_id=origin_session_id, - interrupt_fn=interrupt_fn, max_async_children=max_async_children, - progress_fn=progress_fn, + interrupt_fn=interrupt_fn, max_async_children=max_async_children, progress_fn=progress_fn, capacity_error=( - f"Async delegation capacity reached ({max_async_children} " - f"running). Wait for one to finish (its result will re-enter " - f"the chat), or raise delegation.max_concurrent_children in " - f"config.yaml to allow more concurrent background units." - ), - ) + f"Async delegation capacity reached ({max_async_children} running). Wait for one to finish " + "(its result will re-enter the chat), or raise delegation.max_concurrent_children in " + "config.yaml to allow more concurrent background units.")) if handle["status"] == "dispatched": logger.info("Dispatched async delegation batch %s (%d task(s), session_key=%s)", delegation_id, n, session_key or "") @@ -810,46 +595,33 @@ def dispatch_async_delegation_batch( # ── Finalization + completion events ──────────────────────────────────────── -def _finalize(delegation_id: str, result: Dict[str, Any], status: str) -> None: - """Mark a record complete and push its completion event onto the queue.""" - event_record = _begin_finalization(delegation_id) - if event_record is None: - return - _push_completion_event(event_record, result, status) - _finish_finalization(delegation_id, status) - - -def _begin_finalization(delegation_id: str) -> Optional[Dict[str, Any]]: - """Atomically claim terminal delivery; returns a record snapshot or None.""" +def _finalize(delegation_id: str, result: Any, status: str) -> None: + """Atomically claim terminal delivery, push the completion event, then mark ``status``. + ``result`` is a dict or a callable receiving the record snapshot (stall path). The record + stays active ("finalizing") until durable persistence and queue publication finish; otherwise + process shutdown can kill this daemon worker after status flips but before SQLite commits. + A second call for the same id (late runner return after a forced stall) is a no-op.""" with _records_lock: record = _records.get(delegation_id) if record is None or record.get("status") not in _ACTIVE_STATES: - return None - # Stay active until durable persistence and queue publication finish; - # otherwise process shutdown can kill this daemon worker in the gap - # after status flips but before SQLite is committed. + return record["status"] = "finalizing" record["completed_at"] = time.time() record["interrupt_fn"] = None # drop the closure; child is done record["progress_fn"] = None # stop stale-monitor sampling - return dict(record) - - -def _finish_finalization(delegation_id: str, status: str) -> None: + snapshot = dict(record) + _push_completion_event(snapshot, result(snapshot) if callable(result) else result, status) with _records_lock: - record = _records.get(delegation_id) - if record is not None: - record["status"] = status + if delegation_id in _records: + _records[delegation_id]["status"] = status _prune_completed_locked() def _push_completion_event(record: Dict[str, Any], result: Dict[str, Any], status: str) -> None: - """Push a type='async_delegation' event onto the shared completion queue. - - Batch records (``is_batch``) carry the per-task ``results`` list instead of - a single summary. Best-effort: failure must not crash the worker, but it - WOULD mean a silently-lost result, so we log loudly. - """ + """Push a type='async_delegation' event onto the shared completion queue. Batch records + (``is_batch``) carry the per-task ``results`` list (plus live transcript paths, the + full-fidelity record of each child's run) instead of a single summary. Best-effort: failure + must not crash the worker, but it WOULD mean a silently-lost result, so we log loudly.""" is_batch = bool(record.get("is_batch")) label = " batch" if is_batch else "" try: @@ -858,51 +630,31 @@ def _push_completion_event(record: Dict[str, Any], result: Dict[str, Any], statu logger.error(f"Async delegation{label} %s finished but process_registry import failed; " "result lost: %s", record.get("delegation_id"), exc) return - dispatched_at = record.get("dispatched_at") or time.time() completed_at = record.get("completed_at") or time.time() if is_batch: - # Batch payload: the full per-task results list (the formatter renders a - # consolidated block) plus live transcript paths, which persist after - # completion as the full-fidelity record of each child's run. payload = { - "is_batch": True, - "results": result.get("results") or [], - "live_transcripts": result.get("live_transcripts"), - "error": result.get("error"), - "total_duration_seconds": result.get("total_duration_seconds"), - } + "is_batch": True, "results": result.get("results") or [], + "live_transcripts": result.get("live_transcripts"), "error": result.get("error"), + "total_duration_seconds": result.get("total_duration_seconds")} else: payload = { - "summary": result.get("summary"), - "error": result.get("error"), - "api_calls": result.get("api_calls", 0), - "duration_seconds": result.get( - "duration_seconds", round(completed_at - dispatched_at, 2) - ), - } + "summary": result.get("summary"), "error": result.get("error"), "api_calls": result.get("api_calls", 0), + "duration_seconds": result.get("duration_seconds", round(completed_at - dispatched_at, 2))} evt = { - "type": "async_delegation", - "delegation_id": record.get("delegation_id"), + "type": "async_delegation", "delegation_id": record.get("delegation_id"), # session_key routes back to the originating gateway session; "" => CLI. "session_key": record.get("session_key", ""), "origin_ui_session_id": record.get("origin_ui_session_id", ""), "origin_session_id": record.get("origin_session_id", ""), "parent_session_id": record.get("parent_session_id"), - "goal": record.get("goal", ""), - **({"goals": record.get("goals")} if is_batch else {}), - "context": record.get("context"), - "toolsets": record.get("toolsets"), - "role": record.get("role"), + "goal": record.get("goal", ""), **({"goals": record.get("goals")} if is_batch else {}), + "context": record.get("context"), "toolsets": record.get("toolsets"), "role": record.get("role"), "model": record.get("model") if is_batch else (result.get("model") or record.get("model")), - "status": status, - **payload, - "dispatched_at": dispatched_at, - "completed_at": completed_at, + "status": status, **payload, "dispatched_at": dispatched_at, "completed_at": completed_at, **({} if is_batch else {"exit_reason": result.get("exit_reason")}), - } - _copy_truthy(record, evt, _ROUTING_KEYS) - _copy_present(result, evt, _STALL_META_KEYS) + **{k: record[k] for k in _ROUTING_KEYS if record.get(k)}, + **{k: result[k] for k in _STALL_META_KEYS if k in result}} _persist_completion(evt, result) try: process_registry.completion_queue.put(evt) @@ -913,104 +665,98 @@ def _push_completion_event(record: Dict[str, Any], result: Dict[str, Any], statu # ── Stale monitor ─────────────────────────────────────────────────────────── def _ensure_stale_monitor() -> None: - """Start (once) the stale-delegation monitor thread. - - One daemon thread serves every dispatch; it exits when no monitorable - records remain and is restarted by the next dispatch with a ``progress_fn``. - """ + """Start (once) the stale-delegation monitor thread. One daemon thread serves + every dispatch; it exits when no monitorable records remain and is restarted + by the next dispatch with a ``progress_fn``.""" global _monitor_thread with _monitor_lock: if _monitor_thread is not None and _monitor_thread.is_alive(): return _monitor_stop.clear() _monitor_thread = threading.Thread( - target=_stale_monitor_loop, name="async-delegate-stale-monitor", daemon=True, - ) + target=_stale_monitor_loop, name="async-delegate-stale-monitor", daemon=True) _monitor_thread.start() -def _stale_monitor_loop() -> None: - """Sweep running delegations for stalled progress. +def _sweep_stale_locked(now: float): + """One monitor pass over ``_records``; caller holds ``_records_lock``. Returns + ``(stalled, expired, any_monitorable)``: newly-stalling ``(delegation_id, quiet_for, in_tool)`` + tuples, stalling ids past the grace window, and whether anything is left to monitor.""" + stalled, expired, any_monitorable = [], [], False # (delegation_id, quiet_for, in_tool) / ids past grace + for record in _records.values(): + status = record.get("status") + if status == "stalling": + any_monitorable = True + if now - (record.get("_interrupted_at") or now) >= _STALL_GRACE_SECONDS: + expired.append(record["delegation_id"]) + continue + progress_fn = record.get("progress_fn") + if status != "running" or progress_fn is None: + continue + any_monitorable = True + try: + token, in_tool = progress_fn() + except Exception: + # An unreadable child must not look permanently healthy — + # keep the last timestamp running instead of refreshing it. + token, in_tool = record.get("_progress_token"), False + if token != record.get("_progress_token"): + record.update(_progress_token=token, _progress_ts=now) + continue + quiet_for = now - (record.get("_progress_ts") or now) + limit = _STALE_IN_TOOL_SECONDS if in_tool else _STALE_IDLE_SECONDS + if quiet_for >= limit: + # Stall context feeds the terminal event and status listings. + record.update( + status="stalling", _interrupted_at=now, _stall_quiet_seconds=round(quiet_for, 2), + _stall_threshold_seconds=limit, _stall_in_tool=bool(in_tool)) + stalled.append((record["delegation_id"], quiet_for, in_tool)) + return stalled, expired, any_monitorable - A changed progress token refreshes the record's timestamp. A frozen token - past the idle/in-tool threshold marks the record ``stalling`` and calls - ``interrupt_fn``; a ``stalling`` record still unreturned after the grace - window is force-finalized with a terminal ``stalled`` event (a late - runner return is then ignored by ``_begin_finalization``). - """ + +def _call_interrupt(fn, msg: str, *args) -> bool: + """Invoke an ``interrupt_fn``; True on success, else debug-log ``msg`` (+ exc).""" + if not callable(fn): + return False + try: + fn() + return True + except Exception as exc: + logger.debug(msg, *args, exc) + return False + + +def _stale_monitor_loop() -> None: + """Sweep running delegations for stalled progress. A changed progress token refreshes the + record's timestamp; a frozen token past the idle/in-tool threshold marks the record + ``stalling`` and calls ``interrupt_fn``; a ``stalling`` record still unreturned after the + grace window is force-finalized with a terminal ``stalled`` event.""" while not _monitor_stop.wait(_STALE_CHECK_INTERVAL): now = time.time() - stalled: List[tuple] = [] # (delegation_id, quiet_for, in_tool) - expired: List[str] = [] # stalling past grace → force-finalize - any_monitorable = False with _records_lock: - for record in _records.values(): - status = record.get("status") - if status == "stalling": - any_monitorable = True - interrupted_at = record.get("_interrupted_at") or now - if now - interrupted_at >= _STALL_GRACE_SECONDS: - expired.append(record["delegation_id"]) - continue - if status != "running": - continue - progress_fn = record.get("progress_fn") - if progress_fn is None: - continue - any_monitorable = True - try: - token, in_tool = progress_fn() - except Exception: - # An unreadable child must not look permanently healthy — - # keep the last timestamp running instead of refreshing it. - token, in_tool = record.get("_progress_token"), False - if token != record.get("_progress_token"): - record.update(_progress_token=token, _progress_ts=now) - continue - quiet_for = now - (record.get("_progress_ts") or now) - limit = _STALE_IN_TOOL_SECONDS if in_tool else _STALE_IDLE_SECONDS - if quiet_for >= limit: - # Stall context feeds the terminal event and status listings. - record.update( - status="stalling", _interrupted_at=now, - _stall_quiet_seconds=round(quiet_for, 2), - _stall_threshold_seconds=limit, _stall_in_tool=bool(in_tool), - ) - stalled.append((record["delegation_id"], quiet_for, in_tool)) + stalled, expired, any_monitorable = _sweep_stale_locked(now) for delegation_id, quiet_for, in_tool in stalled: logger.warning("Async delegation %s made no progress for %.0fs " "(in_tool=%s) — interrupting; grace window %.0fs", delegation_id, quiet_for, in_tool, _STALL_GRACE_SECONDS) with _records_lock: - record = _records.get(delegation_id) - fn = record.get("interrupt_fn") if record else None - if callable(fn): - try: - fn() - except Exception as exc: - logger.debug("Async delegation %s stall interrupt failed: %s", delegation_id, exc) + fn = (_records.get(delegation_id) or {}).get("interrupt_fn") + _call_interrupt(fn, "Async delegation %s stall interrupt failed: %s", delegation_id) for delegation_id in expired: - _finalize_stalled(delegation_id) + _finalize(delegation_id, lambda rec, d=delegation_id: _stalled_result(d, rec), "stalled") if not any_monitorable: return -def _finalize_stalled(delegation_id: str) -> None: - """Force-finalize a stalling delegation whose runner never returned.""" - event_record = _begin_finalization(delegation_id) - if event_record is None: - return +def _stalled_result(delegation_id: str, event_record: Dict[str, Any]) -> Dict[str, Any]: + """Synthetic terminal result for a stalling delegation whose runner never returned.""" completed_at = event_record.get("completed_at") or time.time() duration = round(completed_at - (event_record.get("dispatched_at") or completed_at), 2) error = ( - f"Async delegation {delegation_id} stalled: the detached subagent " - "stopped making progress (no new API calls, tool activity, or " - "streamed tokens), did not respond to interruption, and never " - "produced a completion event. The worker may be wedged inside a " - "model API call — this is a known failure mode of long-lived " - "gateway processes (#60203). Re-dispatch the task if it is still " - "needed." - ) + f"Async delegation {delegation_id} stalled: the detached subagent stopped making progress " + "(no new API calls, tool activity, or streamed tokens), did not respond to interruption, and never " + "produced a completion event. The worker may be wedged inside a model API call — this is a known " + "failure mode of long-lived gateway processes (#60203). Re-dispatch the task if it is still needed.") logger.error("Async delegation %s force-finalized as stalled after %.0fs", delegation_id, duration) # Structured stall metadata lets parents/UIs distinguish a stall-monitor # kill from other failures without parsing the error string. @@ -1019,71 +765,54 @@ def _finalize_stalled(delegation_id: str) -> None: "stalled_after_quiet_seconds": event_record.get("_stall_quiet_seconds"), "stall_threshold_seconds": event_record.get("_stall_threshold_seconds"), "stall_phase": "in_tool" if stall_in_tool else "idle" if stall_in_tool is not None else None, - "stall_grace_seconds": _STALL_GRACE_SECONDS, - } + "stall_grace_seconds": _STALL_GRACE_SECONDS} if event_record.get("is_batch"): - result = {**_batch_crash(error, duration), **stall_meta} - else: - result = {**_single_crash(error, duration), "status": "stalled", "exit_reason": "stalled", **stall_meta} - _push_completion_event(event_record, result, "stalled") - _finish_finalization(delegation_id, "stalled") + return {**_batch_crash(error, duration), **stall_meta} + return {**_single_crash(error, duration), "status": "stalled", "exit_reason": "stalled", **stall_meta} # ── Observability + control ───────────────────────────────────────────────── def _children_activity_from_token(token: Any, now: float) -> Optional[List]: - """Parse a progress token into per-child activity dicts (best-effort). - - delegate_tool emits one ``(api_call_count, current_tool, last_activity_ts)`` - tuple per child; foreign token shapes degrade to ``None`` entries. - """ + """Parse a progress token into per-child activity dicts (best-effort): delegate_tool + emits one ``(api_call_count, current_tool, last_activity_ts)`` tuple per child; + foreign token shapes degrade to ``None`` entries.""" try: parts = list(token) except TypeError: return None out: List[Optional[Dict[str, Any]]] = [] for part in parts: - if isinstance(part, (list, tuple)) and len(part) >= 2: - entry: Dict[str, Any] = {"api_calls": part[0], "current_tool": part[1]} - if len(part) >= 3 and isinstance(part[2], (int, float)): - entry["seconds_since_activity"] = round(max(0.0, now - float(part[2])), 1) - out.append(entry) - else: + if not (isinstance(part, (list, tuple)) and len(part) >= 2): out.append(None) + continue + entry: Dict[str, Any] = {"api_calls": part[0], "current_tool": part[1]} + if len(part) >= 3 and isinstance(part[2], (int, float)): + entry["seconds_since_activity"] = round(max(0.0, now - float(part[2])), 1) + out.append(entry) return out def list_async_delegations() -> List[Dict[str, Any]]: - """Snapshot of async delegations (running + recently completed). - - Excludes callables and private monitor bookkeeping; adds computed live - fields for UIs: ``seconds_since_progress``, ``children_activity`` / - ``in_tool`` (sampled from ``progress_fn``), and stall context once the - monitor has tripped. - """ + """Snapshot of async delegations (running + recently completed) without callables or private + monitor bookkeeping; adds computed live fields for UIs (``seconds_since_progress``, + ``children_activity``/``in_tool`` sampled from ``progress_fn``) and stall context once tripped.""" now = time.time() samplers: Dict[str, Callable] = {} with _records_lock: items = [] for r in _records.values(): - item = { - k: v - for k, v in r.items() - if k not in {"interrupt_fn", "progress_fn"} and not k.startswith("_") - } + item = {k: v for k, v in r.items() if k not in {"interrupt_fn", "progress_fn"} and not k.startswith("_")} status = r.get("status") if status in _ACTIVE_STATES: - ts = r.get("_progress_ts") - if ts: - item["seconds_since_progress"] = round(now - ts, 1) - fn = r.get("progress_fn") - if callable(fn): - samplers[r["delegation_id"]] = fn + if r.get("_progress_ts"): + item["seconds_since_progress"] = round(now - r["_progress_ts"], 1) + if callable(r.get("progress_fn")): + samplers[r["delegation_id"]] = r["progress_fn"] if status in ("stalling", "stalled"): for src, dst in _STALL_FIELD_MAP: if r.get(src) is not None: item[dst] = r.get(src) items.append(item) - # Sample OUTSIDE the lock — progress_fn reads child-agent attributes and a # slow/broken sampler must not block every dispatch/finalize. for item in items: @@ -1101,53 +830,33 @@ def list_async_delegations() -> List[Dict[str, Any]]: return items -def _interrupt_records(targets: List[Dict[str, Any]], caller: str) -> int: - """Call ``interrupt_fn`` on each record; returns how many succeeded.""" - count = 0 - for r in targets: - fn = r.get("interrupt_fn") - if callable(fn): - try: - fn() - count += 1 - except Exception as exc: - logger.debug("%s: %s interrupt failed: %s", caller, r.get("delegation_id"), exc) +def _interrupt_records(targets: List[Dict[str, Any]], caller: str, reason: str, msg: str) -> int: + """Call ``interrupt_fn`` on each record; log ``msg`` once; returns how many succeeded.""" + count = sum( + _call_interrupt(r.get("interrupt_fn"), "%s: %s interrupt failed: %s", caller, r.get("delegation_id")) + for r in targets) + if count: + logger.info(msg, count, reason) return count def interrupt_all(reason: str = "shutdown") -> int: - """Signal every running async delegation to stop (``/stop``, shutdown). Returns how many. - - The child still emits a completion event (status='interrupted') via the - normal finalize path. - """ + """Signal every running async delegation to stop (``/stop``, shutdown). Returns how + many. The child still emits a completion event (status='interrupted') via the + normal finalize path.""" with _records_lock: targets = [r for r in _records.values() if r.get("status") in _ACTIVE_STATES] - count = _interrupt_records(targets, "interrupt_all") - if count: - logger.info("Interrupted %d async delegation(s) (%s)", count, reason) - return count + return _interrupt_records(targets, "interrupt_all", reason, "Interrupted %d async delegation(s) (%s)") def interrupt_for_session( - session_key: str = "", origin_ui_session_id: str = "", parent_session_id: str = "", - reason: str = "session_end", + session_key: str = "", origin_ui_session_id: str = "", parent_session_id: str = "", reason: str = "session_end", ) -> int: - """Signal running async delegations owned by ONE ending session to stop. - - Any matching selector claims the record: ``origin_ui_session_id`` (TUI - tab), ``session_key`` (routing key at dispatch), or ``parent_session_id`` - (spawner's durable id — the right one for gateway chats, whose session_key - survives ``/new`` while the session id rotates). Returns how many. - """ - if not session_key and not origin_ui_session_id and not parent_session_id: - return 0 - with _records_lock: - targets = _session_records_locked(_ACTIVE_STATES, session_key, origin_ui_session_id, parent_session_id) - count = _interrupt_records(targets, "interrupt_for_session") - if count: - logger.info("Interrupted %d async delegation(s) for ending session (%s)", count, reason) - return count + """Signal running async delegations owned by ONE ending session to stop (any + selector matches, see ``_session_records``). Returns how many.""" + targets = _session_records(_ACTIVE_STATES, session_key, origin_ui_session_id, parent_session_id) + return _interrupt_records( + targets, "interrupt_for_session", reason, "Interrupted %d async delegation(s) for ending session (%s)") def _reset_for_tests() -> None: @@ -1160,8 +869,7 @@ def _reset_for_tests() -> None: _executor_max_workers = 0 _monitor_stop.set() with _monitor_lock: - thread = _monitor_thread - _monitor_thread = None + thread, _monitor_thread = _monitor_thread, None if thread is not None and thread.is_alive(): thread.join(timeout=2) with _records_lock: diff --git a/tools/clarify_gateway.py b/tools/clarify_gateway.py index 7c3b5225fb..1c6a72b332 100644 --- a/tools/clarify_gateway.py +++ b/tools/clarify_gateway.py @@ -1,15 +1,11 @@ -"""Gateway-side clarify primitive (blocking event-based queue). - -In gateway mode the agent runs on a worker thread while the event loop handles -the user's reply, so a pending clarify is stored module-level (same shape as -``tools.approval``) and the agent thread blocks on an ``Event`` until an -adapter button callback or the gateway text-intercept resolves it, or the -timeout fires. Adapters render either inline buttons (with an "Other" row that -flips the entry into text-capture mode) or a numbered-list text fallback. -""" +"""Gateway-side clarify primitive (blocking event-based queue). The agent runs on a worker +thread while the event loop handles the user's reply, so a pending clarify is stored +module-level (same shape as ``tools.approval``) and the agent thread blocks on an ``Event`` +until an adapter button callback or the gateway text-intercept resolves it, or the timeout +fires. Adapters render inline buttons (an "Other" row flips the entry into text-capture +mode) or a numbered-list text fallback.""" from __future__ import annotations - import json import logging import threading @@ -36,7 +32,7 @@ class _ClarifyEntry: _lock = threading.RLock() _entries: Dict[str, _ClarifyEntry] = {} # clarify_id -> entry (button callbacks) _session_index: Dict[str, List[str]] = {} # session_key -> [clarify_id] FIFO (text intercept, cleanup) -# Per-session notify callbacks (gateway -> adapter bridge); mirrors tools.approval. +# Per-session notify callbacks (gateway -> adapter bridge); mirrors tools.approval. Tests clear it. _notify_cbs: Dict[str, Callable[[_ClarifyEntry], None]] = {} # Outcomes for typed clarify replies. Gateway cancels the pending prompt on @@ -47,25 +43,12 @@ TEXT_REJECTED_SELECTION = "rejected_selection" TEXT_NO_PENDING = "no_pending" -# --- agent-thread side --------------------------------------------------- - -def register( - clarify_id: str, - session_key: str, - question: str, - choices: Optional[List[str]], - multi_select: bool = False, -) -> _ClarifyEntry: - """Register a pending clarify request; caller then blocks on ``wait_for_response``.""" - entry = _ClarifyEntry( - clarify_id=clarify_id, - session_key=session_key, - question=question, - choices=list(choices) if choices else None, - multi_select=bool(multi_select) and bool(choices), - # Open-ended (no choices) -> next message IS the response, no buttons needed. - awaiting_text=not bool(choices), - ) +def register(clarify_id: str, session_key: str, question: str, choices: Optional[List[str]], + multi_select: bool = False) -> _ClarifyEntry: + """Register a pending clarify request; caller then blocks on ``wait_for_response``. + Open-ended (no choices) entries start in text mode: the next message IS the response.""" + entry = _ClarifyEntry(clarify_id, session_key, question, list(choices) if choices else None, + bool(multi_select) and bool(choices), awaiting_text=not bool(choices)) with _lock: _entries[clarify_id] = entry _session_index.setdefault(session_key, []).append(clarify_id) @@ -73,51 +56,35 @@ def register( def wait_for_response(clarify_id: str, timeout: float) -> Optional[str]: - """Block until the entry resolves or ``timeout`` elapses; None on timeout/unknown id. - - Polls in 1s slices so the inactivity heartbeat keeps firing (a single long - ``Event.wait`` would let the gateway watchdog kill a live prompt). - ``timeout <= 0`` means an unlimited wait. - """ + """Block until the entry resolves or ``timeout`` (``<= 0`` = unlimited) elapses; None on + timeout/unknown id. Polls in 1s slices so the inactivity heartbeat keeps firing (a + single long ``Event.wait`` would let the gateway watchdog kill a live prompt).""" with _lock: entry = _entries.get(clarify_id) if entry is None: return None - try: from tools.environments.base import touch_activity_if_due except Exception: # pragma: no cover - optional touch_activity_if_due = None - - unlimited = timeout is None or float(timeout) <= 0.0 - deadline = None if unlimited else time.monotonic() + float(timeout) + deadline = None if timeout is None or float(timeout) <= 0.0 else time.monotonic() + float(timeout) activity_state = {"last_touch": time.monotonic(), "start": time.monotonic()} while True: - if deadline is None: - slice_s = 1.0 - else: - remaining = deadline - time.monotonic() - if remaining <= 0: - break - slice_s = min(1.0, remaining) - if entry.event.wait(timeout=slice_s): + remaining = 1.0 if deadline is None else deadline - time.monotonic() + if remaining <= 0 or entry.event.wait(timeout=min(1.0, remaining)): break if touch_activity_if_due is not None: touch_activity_if_due(activity_state, "waiting for user clarify response") - with _lock: _entries.pop(clarify_id, None) # regardless of outcome - ids = _session_index.get(entry.session_key) - if ids and clarify_id in ids: + ids = _session_index.get(entry.session_key) or [] + if clarify_id in ids: ids.remove(clarify_id) if not ids: _session_index.pop(entry.session_key, None) - return entry.response -# --- gateway / adapter side ----------------------------------------------- - def resolve_gateway_clarify(clarify_id: str, response: str) -> bool: """Unblock the waiter on ``clarify_id``; False if already resolved/expired/unknown.""" with _lock: @@ -129,17 +96,10 @@ def resolve_gateway_clarify(clarify_id: str, response: str) -> bool: return True -def get_pending_for_session( - session_key: str, - *, - include_choice_prompts: bool = False, -) -> Optional[_ClarifyEntry]: - """Oldest pending entry awaiting free text (open-ended, or after "Other"). - - ``include_choice_prompts=True`` returns the oldest unresolved entry of any - kind — for when the user typed directly at an active multi-choice prompt, - so the text resolves it instead of queueing as an unrelated follow-up turn. - """ +def get_pending_for_session(session_key: str, *, include_choice_prompts: bool = False) -> Optional[_ClarifyEntry]: + """Oldest pending entry awaiting free text (open-ended, or after "Other"); + ``include_choice_prompts=True`` returns the oldest unresolved entry of any kind (user + typed at an active choice prompt: resolve it rather than queue a follow-up turn).""" with _lock: for cid in _session_index.get(session_key) or []: entry = _entries.get(cid) @@ -148,18 +108,13 @@ def get_pending_for_session( return None -def _label_matches(text: str, choice: object) -> bool: - """Case-insensitive label match ignoring the '(Recommended)' suffix the - first choice carries by the time it reaches adapters.""" - from tools.clarify_tool import strip_recommended - - return strip_recommended(text).casefold() == strip_recommended(str(choice)).casefold() - - def _match_label(text: str, choices: List[str]) -> Optional[str]: - """Canonical (stripped) choice text matching ``text``, or None.""" + """Stripped choice text matching ``text`` case-insensitively, ignoring the '(Recommended)' + suffix the first choice carries by the time it reaches adapters; None if no match.""" + from tools.clarify_tool import strip_recommended + wanted = strip_recommended(text).casefold() for choice in choices: - if _label_matches(text, choice): + if strip_recommended(str(choice)).casefold() == wanted: return str(choice).strip() return None @@ -169,38 +124,32 @@ def _split_tokens(text: str) -> Optional[List[str]]: if "," in text: return [t.strip() for t in text.split(",") if t.strip()] parts = text.split() - if len(parts) > 1 and all(p.isdigit() for p in parts): - return parts - return None + return parts if len(parts) > 1 and all(p.isdigit() for p in parts) else None -def _selection_attempt_tokens( - text: str, - choices: Optional[List[str]] = None, -) -> Optional[List[str]]: +def _is_int(text: str) -> bool: + try: + int(text) + return True + except ValueError: + return False + + +def _selection_attempt_tokens(text: str, choices: Optional[List[str]] = None) -> Optional[List[str]]: """Tokens when ``text`` looks like a typed selection (bare int, comma list, all-numeric space list); None for free prose so the gateway can release the clarify. Comma-list labels may span up to the longest choice's word count.""" stripped = str(text).strip() if not stripped: return None - tokens = _split_tokens(stripped) - if tokens is not None: - if "," not in stripped or not tokens: - return tokens or None - max_words = max(1, max((len(str(c).split()) for c in choices or []), default=1)) - if all(t.isdigit() or len(t.split()) <= max_words for t in tokens): - return tokens - return None - - if stripped.isdigit() or (stripped.startswith("-") and stripped[1:].isdigit()): - return [stripped] - try: - int(stripped) - return [stripped] - except ValueError: - return None + if tokens is None: + digits = stripped[1:] if stripped.startswith("-") else stripped + return [stripped] if digits.isdigit() or _is_int(stripped) else None + if "," not in stripped or not tokens: + return tokens or None + max_words = max(1, max((len(str(c).split()) for c in choices or []), default=1)) + return tokens if all(t.isdigit() or len(t.split()) <= max_words for t in tokens) else None def _coerce_text_response(entry: _ClarifyEntry, response: str) -> Optional[str]: @@ -208,81 +157,48 @@ def _coerce_text_response(entry: _ClarifyEntry, response: str) -> Optional[str]: return _coerce_text_response_detailed(entry, response)[0] -def _coerce_text_response_detailed( - entry: _ClarifyEntry, - response: str, -) -> tuple[Optional[str], Optional[str]]: - """Map a typed reply to ``(value, None)`` or ``(None, reason)``. - - Reasons: ``"invalid_selection"`` (selection-shaped but out of range / - unrecognised — keep the clarify armed for a retry) or ``"prose"`` (free - text on a native choice prompt — gateway may cancel and route normally so a - redirect-to-steer path cannot deadlock behind the waiting tool). - Open-ended entries and ``awaiting_text`` mode accept any text; numeric - picks and exact labels always resolve; multi-select returns a JSON array - string that ``_parse_multi_select_response`` decodes on the tool side, and - one bad token rejects the whole reply rather than resolving a partial pick. - """ +def _coerce_text_response_detailed(entry: _ClarifyEntry, response: str) -> tuple[Optional[str], Optional[str]]: + """Map a typed reply to ``(value, None)`` or ``(None, reason)``: ``"invalid_selection"`` + (selection-shaped but out of range/unrecognised — keep the clarify armed for a retry) or + ``"prose"`` (free text on a native choice prompt — the gateway may cancel and route normally + so a redirect-to-steer path cannot deadlock behind the waiting tool). Open-ended entries and + ``awaiting_text`` accept any text; numeric picks and exact labels always resolve; multi-select + returns a JSON array string (decoded tool-side); one bad token rejects the whole reply.""" text = str(response).strip() - if not entry.choices: return text, None - if entry.multi_select: coerced = _coerce_multi_select_text(entry, text) - if coerced is not None: - return coerced, None - if entry.awaiting_text: - return text, None - if _selection_attempt_tokens(text, entry.choices) is not None: - return None, "invalid_selection" - return None, "prose" - - try: - idx = int(text) - 1 - is_int = True - except ValueError: - idx = -1 - is_int = False - - if is_int and 0 <= idx < len(entry.choices): - return entry.choices[idx], None - - matched = _match_label(text, entry.choices) - if matched is not None: - return matched, None - + selection_shaped = _selection_attempt_tokens(text, entry.choices) is not None + else: + # Out-of-range / non-canonical integer is a failed selection, not prose. + selection_shaped = _is_int(text) + idx = int(text) - 1 if selection_shaped else -1 + coerced = entry.choices[idx] if 0 <= idx < len(entry.choices) else _match_label(text, entry.choices) + if coerced is not None: + return coerced, None if entry.awaiting_text: return text, None - - # Out-of-range / non-canonical integer is a failed selection, not prose. - return None, "invalid_selection" if is_int else "prose" + return None, "invalid_selection" if selection_shaped else "prose" def _coerce_multi_select_text(entry: _ClarifyEntry, text: str) -> Optional[str]: """Parse "1,3" / "1 3" / "staging, prod" into a JSON array of choice labels; None when any token is out of range or unrecognised (reject the whole reply).""" + choices, selected = entry.choices or [], [] if not text: return None - choices = entry.choices or [] tokens = _split_tokens(text) - if tokens is None: - tokens = [text] - - selected: List[str] = [] - for token in tokens: + for token in [text] if tokens is None else tokens: if token.isdigit(): idx = int(token) - 1 - if not 0 <= idx < len(choices): - return None - label = str(choices[idx]).strip() + label = str(choices[idx]).strip() if 0 <= idx < len(choices) else None else: label = _match_label(token, choices) - if label is None: - return None + if label is None: + return None if label not in selected: selected.append(label) - return json.dumps(selected, ensure_ascii=False) if selected else None @@ -291,15 +207,12 @@ def attempt_text_response_for_session(session_key: str, response: str) -> str: entry = get_pending_for_session(session_key, include_choice_prompts=True) if entry is None: return TEXT_NO_PENDING - coerced, reason = _coerce_text_response_detailed(entry, response) if coerced is None: return TEXT_REJECTED_SELECTION if reason == "invalid_selection" else TEXT_REJECTED_PROSE - if resolve_gateway_clarify(entry.clarify_id, coerced): return TEXT_RESOLVED - # Lost a race with a button/callback resolution — treat as no work left. - return TEXT_NO_PENDING + return TEXT_NO_PENDING # lost a race with a button/callback resolution — no work left def resolve_text_response_for_session(session_key: str, response: str) -> bool: @@ -311,37 +224,28 @@ def mark_awaiting_text(clarify_id: str) -> bool: """Flip an entry into text-capture mode (user picked 'Other'); False if unknown.""" with _lock: entry = _entries.get(clarify_id) - if entry is None: - return False - entry.awaiting_text = True - return True + if entry is not None: + entry.awaiting_text = True + return entry is not None def has_pending(session_key: str) -> bool: """True when this session has at least one pending clarify entry.""" with _lock: - ids = _session_index.get(session_key) or [] - return any(_entries.get(cid) is not None for cid in ids) + return any(_entries.get(cid) is not None for cid in _session_index.get(session_key) or []) def clear_session(session_key: str) -> int: - """Drop every pending clarify for a session; returns how many were cancelled. - - Session-boundary cleanup (``/new``, shutdown, cached-agent eviction) so - blocked agent threads don't outlive their session; cancelled waiters see "". - First-writer-wins: an entry whose event is already set was answered for - real, so it is dropped from the registry but its response is preserved - rather than overwritten with the "" cancellation sentinel. The whole loop - stays inside the lock so a button callback cannot slip between pop and check. - Entries are removed from the registry regardless of state — a cleared - session must not be resurrected by late callbacks. Callers distinguish the - "" sentinel from a real reply only by their own timeout bookkeeping; most - treat any falsy result as "user did not respond". - """ + """Drop every pending clarify for a session (``/new``, shutdown, cached-agent eviction) so + blocked agent threads don't outlive it; returns how many were cancelled. Cancelled waiters + see "" (callers tell it from a real reply only via their own timeout bookkeeping; most treat + any falsy result as no response). First-writer-wins: an already-set entry was answered for + real, so it is dropped but its response preserved. The loop stays inside the lock so a button + callback cannot slip between pop and check; entries go regardless of state so a cleared + session is never resurrected by late callbacks.""" with _lock: - ids = list(_session_index.pop(session_key, []) or []) cancelled = 0 - for entry in (_entries.pop(cid, None) for cid in ids): + for entry in (_entries.pop(cid, None) for cid in list(_session_index.pop(session_key, []) or [])): if entry is None or entry.event.is_set(): continue entry.response = "" @@ -350,13 +254,10 @@ def clear_session(session_key: str) -> int: return cancelled -# --- config --------------------------------------------------------------- - def resolve_clarify_timeout(config: dict) -> int: - """Clarify timeout (seconds) from a loaded config: legacy ``clarify.timeout`` - if explicitly set, else ``agent.clarify_timeout``, else 3600. Single source - of truth for every surface (gateway, CLI, TUI) so the value can't drift. - ``<= 0`` is preserved verbatim (unlimited); non-numeric falls back to 3600.""" + """Clarify timeout (seconds): legacy ``clarify.timeout`` if explicitly set, else + ``agent.clarify_timeout``, else 3600 — the single source of truth for every surface + (gateway, CLI, TUI). ``<= 0`` is kept verbatim (unlimited); non-numeric -> 3600.""" raw = (config.get("clarify") or {}).get("timeout") if raw is None: raw = (config.get("agent") or {}).get("clarify_timeout", 3600) @@ -367,12 +268,9 @@ def resolve_clarify_timeout(config: dict) -> int: def get_clarify_timeout() -> int: - """Read the clarify timeout from config.yaml; 0/negative = unlimited. - - Default 3600: long enough that a user who stepped away still finds a live - entry when they tap, short enough that an abandoned prompt eventually - unblocks the agent thread instead of pinning the running-agent guard. - """ + """Clarify timeout from config.yaml; 0/negative = unlimited. Default 3600: long enough + that a user who stepped away still finds a live entry when they tap, short enough that + an abandoned prompt eventually unblocks the agent thread instead of pinning the guard.""" try: from hermes_cli.config import load_config return resolve_clarify_timeout(load_config() or {}) diff --git a/tools/clarify_tool.py b/tools/clarify_tool.py index 74bc064f4d..e3032ce3d2 100644 --- a/tools/clarify_tool.py +++ b/tools/clarify_tool.py @@ -1,62 +1,42 @@ -#!/usr/bin/env python3 """Clarify tool: structured multiple-choice / open-ended questions to the user. - Schema, validation and a thin dispatcher; the UI lives in a platform-provided -callback (cli.py, gateway/run.py, tui_gateway). -""" +callback (cli.py, gateway/run.py, tui_gateway).""" import inspect import json from typing import Dict, List, Optional, Callable - MAX_CHOICES = 4 # the UI always appends an "Other (type your answer)" row MAX_QUESTIONS = 5 # independent questions per batch call - # Canonical timeout sentinel. The CLI returns this exact text; the batch loop # treats it (like ``None``) as "the user walked away" and aborts remaining questions. -TIMEOUT_RESPONSE = ( - "The user did not provide a response within the time limit. " - "Use your best judgement to make the choice and proceed." -) - +TIMEOUT_RESPONSE = ("The user did not provide a response within the time limit. " + "Use your best judgement to make the choice and proceed.") # Applied to the first choice here (not per-surface) so every adapter renders it identically. RECOMMENDED_LABEL = "(Recommended)" +_UNAVAILABLE = "Clarify tool is not available in this execution context." def _flatten_choice(c) -> str: - """Coerce one choice to display text. - - LLMs sometimes emit dict-shaped choices; ``str(c)`` would leak the dict repr - onto every surface and back as the answer, so normalise once here. Unwrap - order ``label`` > ``description`` > ``text`` > ``title``; ``name``/``value`` - are excluded (component fields carrying raw enums, not labels). A dict with - none of these becomes "" and is dropped — no choice beats a garbage label. - """ - if c is None: - return "" + """Coerce one choice to display text. LLMs sometimes emit dict-shaped choices and ``str(c)`` + would leak the repr onto every surface and back as the answer; unwrap order ``label`` > + ``description`` > ``text`` > ``title`` (``name``/``value`` excluded: raw component enums, + not labels). No match -> "" and dropped: no choice beats a garbage label.""" if isinstance(c, str): return c.strip() if isinstance(c, dict): - for key in ("label", "description", "text", "title"): - v = c.get(key) - if isinstance(v, str) and v.strip(): - return v.strip() - return "" + return next((v.strip() for k in ("label", "description", "text", "title") + if isinstance(v := c.get(k), str) and v.strip()), "") if isinstance(c, (list, tuple)): return " ".join(_flatten_choice(x) for x in c).strip() - return str(c).strip() + return "" if c is None else str(c).strip() def mark_recommended(choices: List[str]) -> List[str]: - """Suffix the first choice (the schema says best-first) with RECOMMENDED_LABEL. - - Idempotent, and a lone choice is left untouched (nothing to prefer it over). - """ - if len(choices) < 2: - return choices - first = str(choices[0]).strip() - if first != strip_recommended(first): + """Suffix the first choice (schema says best-first) with RECOMMENDED_LABEL; idempotent, + and a lone choice is left untouched (nothing to prefer it over).""" + first = str(choices[0]).strip() if choices else "" + if len(choices) < 2 or first != strip_recommended(first): return choices return [f"{first} {RECOMMENDED_LABEL}"] + list(choices[1:]) @@ -70,16 +50,13 @@ def strip_recommended(text: str) -> str: def _accepts_kwarg(callback, name: str) -> bool: - """Signature-inspect (never a TypeError retry, which could re-prompt the user) - whether ``callback`` takes ``name`` or ``**kwargs``. Non-introspectable - callables are conservatively treated as legacy.""" + """Signature-inspect (never a TypeError retry, which could re-prompt the user) whether + ``callback`` takes ``name`` or ``**kwargs``; non-introspectable callables are legacy.""" try: params = inspect.signature(callback).parameters except (TypeError, ValueError): return False - return name in params or any( - p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values() - ) + return name in params or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()) def _invoke_callback(callback, question, choices, multi_select): @@ -89,21 +66,29 @@ def _invoke_callback(callback, question, choices, multi_select): return callback(question, choices) +def _json_as(raw: str, kind): + """``json.loads(raw)`` when it decodes to an instance of ``kind``, else None.""" + try: + parsed = json.loads(raw) + except json.JSONDecodeError: + return None + return parsed if isinstance(parsed, kind) else None + + def _parse_multi_select_response(raw_response) -> List[str]: """Parse a list / JSON array / comma-separated reply into stripped non-empty strings.""" - if isinstance(raw_response, list): - return [str(r).strip() for r in raw_response if str(r).strip()] + items = raw_response + if not isinstance(items, list): + raw = str(items).strip() + items = _json_as(raw, list) if raw.startswith("[") else None + if items is None: + items = raw.split(",") + return [str(r).strip() for r in items if str(r).strip()] - raw = str(raw_response).strip() - if raw.startswith("["): - try: - parsed = json.loads(raw) - if isinstance(parsed, list): - return [str(p).strip() for p in parsed if str(p).strip()] - except json.JSONDecodeError: - pass - return [s.strip() for s in raw.split(",") if s.strip()] +def _clean_answer(raw, multi: bool): + """Strip presentation (the label, multi-select JSON) from a locked answer.""" + return [strip_recommended(r) for r in _parse_multi_select_response(raw)] if multi else strip_recommended(raw) def _clean_choices(choices: list) -> Optional[List[str]]: @@ -116,78 +101,49 @@ def _is_timeout(raw) -> bool: return raw is None or (isinstance(raw, str) and raw.strip() == TIMEOUT_RESPONSE) -# --- batch (multi-question) support ----------------------------------------- - def _normalize_questions(questions) -> tuple: - """Validate the ``questions`` batch param -> ``(normalized, error)``. - - An empty list returns ``(None, None)`` (fall back to the single-question - path). Each entry carries ``qid`` (stable wire id ``q`` surfaces key - answers by; the model's ``id`` is unvalidated text so it is only echoed in - results), ``question``, decorated ``choices``, bare ``choices_offered``, - and ``multi_select`` (honored only with choices). - """ + """Validate the ``questions`` batch param -> ``(normalized, error)``; an empty list gives + ``(None, None)`` (fall back to the single-question path). Entries carry ``qid`` (stable + wire id ``q`` surfaces key answers by; the model's ``id`` is unvalidated text, only + echoed), ``question``, decorated ``choices``, bare ``choices_offered``, ``multi_select``.""" if not isinstance(questions, list): return None, "questions must be an array of question objects." if not questions: return None, None if len(questions) > MAX_QUESTIONS: return None, f"questions supports at most {MAX_QUESTIONS} items." - normalized = [] for index, item in enumerate(questions): - if isinstance(item, str): - # Tolerate bare-string items: LLMs sometimes send ["Q1?", "Q2?"]. + if isinstance(item, str): # tolerate bare-string items: LLMs sometimes send ["Q1?", "Q2?"] item = {"question": item} if not isinstance(item, dict): return None, f"questions[{index}] must be an object with a 'question'." - text = str(item.get("question") or "").strip() if not text: return None, f"questions[{index}].question must be non-empty text." - choices = item.get("choices") if choices is not None: if not isinstance(choices, list): return None, f"questions[{index}].choices must be a list." choices = _clean_choices(choices) - - model_id = str(item.get("id") or "").strip() or None - normalized.append({ - "qid": f"q{index}", - "id": model_id, - "question": text, + "qid": f"q{index}", "id": str(item.get("id") or "").strip() or None, "question": text, "choices": mark_recommended(list(choices)) if choices else None, "choices_offered": list(choices) if choices else None, - "multi_select": bool(item.get("multi_select")) and bool(choices), - }) - + "multi_select": bool(item.get("multi_select")) and bool(choices)}) return normalized, None -def _clean_batch_answer(entry: dict, raw) -> object: - """Strip presentation from one locked answer (label, multi-select JSON).""" - if entry["multi_select"]: - return [strip_recommended(r) for r in _parse_multi_select_response(raw)] - return strip_recommended(raw) - - def _batch_result(normalized: List[dict], answers: dict, timed_out: bool) -> str: - """Batch result JSON; unanswered -> "" — the top-level ``timed_out`` flag - (present only when true) tells the agent whether blanks are deliberate - skips or the user walking away.""" + """Batch result JSON; unanswered -> "". The top-level ``timed_out`` flag (present only when + true) tells the agent whether blanks are deliberate skips or the user walking away.""" responses = [] for entry in normalized: - row = {} - if entry["id"]: - row["id"] = entry["id"] - row["question"] = entry["question"] - row["choices_offered"] = entry["choices_offered"] raw = answers.get(entry["qid"]) - row["user_response"] = _clean_batch_answer(entry, raw) if raw else "" - responses.append(row) - + responses.append({ + **({"id": entry["id"]} if entry["id"] else {}), + "question": entry["question"], "choices_offered": entry["choices_offered"], + "user_response": _clean_answer(raw, entry["multi_select"]) if raw else ""}) result: Dict[str, object] = {"responses": responses} if timed_out: result["timed_out"] = True @@ -195,39 +151,25 @@ def _batch_result(normalized: List[dict], answers: dict, timed_out: bool) -> str def _run_batch(normalized: List[dict], callback, question: str) -> str: - """Dispatch a validated batch to the platform callback. - - Batch-capable callbacks (``questions`` kwarg) get the whole list once and - reply ``{"answers": {qid: raw}, "timed_out"?}`` as a dict or JSON string - (the tui_gateway bridge only carries strings). Legacy callbacks are looped - per question; an empty answer is a skip, a timeout (``None`` or the - sentinel) means the user walked away so the loop aborts instead of pestering - them — answers collected before the abort are kept either way. - """ + """Dispatch a validated batch. Batch-capable callbacks (``questions`` kwarg) get the + whole list once and reply ``{"answers": {qid: raw}, "timed_out"?}`` as a dict or JSON + string (the tui_gateway bridge only carries strings); any other falsy/unparseable reply + is a cancel-all (mirrors the single-question skip). Legacy callbacks are looped per + question: an empty answer is a skip, a timeout (``None`` or the sentinel) means the user + walked away so the loop aborts instead of pestering them; earlier answers are kept.""" answers: dict = {} timed_out = False if _accepts_kwarg(callback, "questions"): raw = callback(question, None, questions=normalized) - if _is_timeout(raw): - timed_out = True - elif isinstance(raw, dict): + timed_out = _is_timeout(raw) + if isinstance(raw, str): + raw = _json_as(raw, dict) # the sentinel is not JSON -> None, timed_out stays True + if isinstance(raw, dict): answers = dict(raw.get("answers") or {}) timed_out = bool(raw.get("timed_out")) - elif isinstance(raw, str) and raw.strip(): - try: - parsed = json.loads(raw) - except json.JSONDecodeError: - parsed = None - if isinstance(parsed, dict): - answers = dict(parsed.get("answers") or {}) - timed_out = bool(parsed.get("timed_out")) - # Any other falsy/unparseable reply is a cancel-all (mirrors the single-question skip). return _batch_result(normalized, answers, timed_out) - for entry in normalized: - raw = _invoke_callback( - callback, entry["question"], entry["choices"], entry["multi_select"], - ) + raw = _invoke_callback(callback, entry["question"], entry["choices"], entry["multi_select"]) if _is_timeout(raw): timed_out = True break @@ -235,72 +177,43 @@ def _run_batch(normalized: List[dict], callback, question: str) -> str: return _batch_result(normalized, answers, timed_out) -def clarify_tool( - question: str, - choices: Optional[List[str]] = None, - multi_select: bool = False, - questions: Optional[List[dict]] = None, - callback: Optional[Callable] = None, -) -> str: - """Ask the user one question (``question``/``choices``/``multi_select``) or a - batch (``questions``, which takes precedence when non-empty). - - ``callback(question, choices, multi_select=False) -> str`` is platform - injected; batch-capable callbacks also accept ``questions=``. Returns the - result JSON (``{"responses": [...]}`` for batches). - """ +def clarify_tool(question: str, choices: Optional[List[str]] = None, multi_select: bool = False, + questions: Optional[List[dict]] = None, callback: Optional[Callable] = None) -> str: + """Ask one question (``question``/``choices``/``multi_select``) or a batch (``questions`` + wins when non-empty). ``callback(question, choices, multi_select=False) -> str`` is + platform injected (batch-capable ones also take ``questions=``). Returns result JSON.""" if questions is not None: normalized, error = _normalize_questions(questions) if error: return tool_error(error) if normalized: if callback is None: - return tool_error( - "Clarify tool is not available in this execution context." - ) + return tool_error(_UNAVAILABLE) try: return _run_batch(normalized, callback, str(question or "").strip()) except Exception as exc: return tool_error(f"Failed to get user input: {exc}") # Empty questions array → fall through to the single-question path. - if not question or not question.strip(): - return tool_error( - "No question provided. Pass questions=[{question: '...', " - "choices?: [...], multi_select?: bool}, ...] — a single question " - "is a one-entry array." - ) - + return tool_error("No question provided. Pass questions=[{question: '...', " + "choices?: [...], multi_select?: bool}, ...] — a single question " + "is a one-entry array.") question = question.strip() - if choices is not None: if not isinstance(choices, list): return tool_error("choices must be a list of strings.") choices = _clean_choices(choices) - if callback is None: - return tool_error("Clarify tool is not available in this execution context.") - + return tool_error(_UNAVAILABLE) # The bare list goes back to the agent; the "(Recommended)" label is presentation only. - offered = choices - if choices is not None: - choices = mark_recommended(choices) - + shown = mark_recommended(choices) if choices is not None else None try: - raw_response = _invoke_callback(callback, question, choices, multi_select) + raw_response = _invoke_callback(callback, question, shown, multi_select) except Exception as exc: return tool_error(f"Failed to get user input: {exc}") - - if multi_select and choices is not None: - user_response = [strip_recommended(r) for r in _parse_multi_select_response(raw_response)] - else: - user_response = strip_recommended(raw_response) - - return json.dumps({ - "question": question, - "choices_offered": offered, - "user_response": user_response, - }, ensure_ascii=False) + return json.dumps({"question": question, "choices_offered": choices, + "user_response": _clean_answer(raw_response, multi_select and choices is not None)}, + ensure_ascii=False) def check_clarify_requirements() -> bool: @@ -366,7 +279,6 @@ CLARIFY_SCHEMA = { }, } - # --- Registry --- from tools.registry import registry, tool_error diff --git a/tools/cronjob_job_args.py b/tools/cronjob_job_args.py index 80ec1ef32c..ee7bdfa217 100644 --- a/tools/cronjob_job_args.py +++ b/tools/cronjob_job_args.py @@ -32,10 +32,8 @@ def _origin_from_env() -> Optional[Dict[str, str]]: "Cron origin captured thread_id=%s for %s:%s", thread_id, origin_platform, origin_chat_id) return { - "platform": origin_platform, - "chat_id": origin_chat_id, - "chat_name": get_session_env("HERMES_SESSION_CHAT_NAME") or None, - "thread_id": thread_id, + "platform": origin_platform, "chat_id": origin_chat_id, + "chat_name": get_session_env("HERMES_SESSION_CHAT_NAME") or None, "thread_id": thread_id, # Lets a delivery mirror resolve the participant's session in per-user-isolated groups. "user_id": get_session_env("HERMES_SESSION_USER_ID") or None, # Workspace/server scope (Slack team, Discord guild...): Slack session keys embed it, @@ -52,7 +50,6 @@ def _local_delivery_notice(job: Dict[str, Any], user_deliver: Optional[str]) -> return None try: from cron.scheduler import _resolve_delivery_targets - if _resolve_delivery_targets(job): return None except Exception: # resolution unavailable — fall back to the origin signal @@ -211,7 +208,6 @@ def _resolve_cron_context_deliver(deliver: Optional[str]) -> Optional[str]: elements pass through. Otherwise the scheduler would guess a home channel.""" from gateway.session_context import get_session_env from utils import is_truthy_value - if not is_truthy_value(get_session_env("HERMES_CRON_SESSION", "")): return deliver @@ -291,7 +287,6 @@ def _validate_cron_script_path(script: Optional[str]) -> Optional[str]: return None from hermes_constants import get_hermes_home - raw = script.strip() if raw.startswith(("/", "~")) or (len(raw) >= 2 and raw[1] == ":"): return ( @@ -300,7 +295,6 @@ def _validate_cron_script_path(script: Optional[str]) -> Optional[str]: f"Place scripts in ~/.hermes/scripts/ and use just the filename.") from tools.path_security import validate_within_dir - scripts_dir = get_hermes_home() / "scripts" scripts_dir.mkdir(parents=True, exist_ok=True) if validate_within_dir(scripts_dir / raw, scripts_dir): @@ -392,7 +386,6 @@ def _gateway_liveness_notice(plural: bool = False) -> dict: on "scheduler active". False -> warning (no gateway process), None -> probe failed.""" try: from hermes_cli.cron import _builtin_gateway_liveness - _gw = _builtin_gateway_liveness() except Exception: return {"gateway_running": None} diff --git a/tools/cronjob_prompt_scan.py b/tools/cronjob_prompt_scan.py index 20ca81fd22..6f8522d8ee 100644 --- a/tools/cronjob_prompt_scan.py +++ b/tools/cronjob_prompt_scan.py @@ -19,8 +19,7 @@ _CRON_THREAT_PATTERNS = [ (r'system\s+prompt\s+override', "sys_prompt_override"), (r'disregard\s+(your|all|any)\s+(instructions|rules|guidelines)', "disregard_rules"), (r'cat\s+[^\n]*(\.env|credentials|\.netrc|\.pgpass|id_rsa|id_ed25519|id_ecdsa)', "read_secrets"), - (r'authorized_keys', "ssh_backdoor"), - (r'/etc/sudoers|visudo', "sudoers_mod"), + (r'authorized_keys', "ssh_backdoor"), (r'/etc/sudoers|visudo', "sudoers_mod"), (r'rm\s+-rf\s+/', "destructive_root_rm"), ] diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 2efe325a64..3c5b6881d8 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -99,7 +99,6 @@ def _api_server_base_url() -> str: """``http://host:port`` of the local api_server, mirroring its bind resolution (extra.host -> API_SERVER_HOST -> 127.0.0.1); a wildcard bind listens on loopback too.""" import os - port_raw = os.getenv("API_SERVER_PORT", "").strip() try: port = int(port_raw) if port_raw else 8642 @@ -119,17 +118,13 @@ def _api_server_base_url() -> str: def _forward_relay_fronted_run(job: Dict[str, Any], extra_prompt: Optional[str] = None) -> Optional[str]: - """Forward a manual run to the gateway when it targets a relay-fronted platform. - - Relay-fronted delivery has no standalone sender — the gateway's live relay adapter is - the only path, reached via ``POST /api/jobs/{id}/run`` (marks the job due for the - gateway ticker; ``extra_prompt`` rides in the body). Returns a JSON result string when - forwarding engages, else None to fall through to the normal in-process run. - """ + """Forward a manual run to the gateway when it targets a relay-fronted platform: such delivery + has no standalone sender — the gateway's live relay adapter is the only path, reached via + ``POST /api/jobs/{id}/run`` (marks the job due; ``extra_prompt`` rides in the body). Returns a + JSON result string when forwarding engages, else None (normal in-process run).""" if not _relay_fronted_delivery_platforms(job): return None from agent.secret_scope import get_secret - key = get_secret("API_SERVER_KEY", "") or "" try: import httpx @@ -175,12 +170,9 @@ _ALREADY_RUNNING_ERROR = ( def _claim_for_manual_run(job_id: str, log_label: str): - """At-most-once claim shared by the sync and background run paths. - - Returns ``(claimed_job, None)`` or ``(None, error_dict)`` in the ``_execute_job_now`` - result shape. A lost claim is labelled precisely: claim_job_for_fire also returns False - for paused / disabled / missing jobs, which must not read as "already being fired". - """ + """At-most-once claim shared by the sync and background run paths: ``(claimed_job, None)`` or + ``(None, error_dict)`` in the ``_execute_job_now`` shape. A lost claim is labelled precisely — + claim_job_for_fire also returns False for paused/disabled/missing jobs, not just in-flight ones.""" try: claimed_job = claim_job_for_fire(job_id, return_job=True) if isinstance(claimed_job, dict): @@ -310,7 +302,6 @@ def _latest_job_output_excerpt(job_id: str, max_chars: int = 2000) -> Optional[s block (parent sees what the job produced). Never raises.""" try: from cron.jobs import get_cron_output_dir - files = sorted((get_cron_output_dir() / job_id).glob("*.md")) text = files[-1].read_text(encoding="utf-8", errors="replace").strip() if files else "" if not text: @@ -329,7 +320,6 @@ def _reap_stale_executions(job_name: str) -> None: Best-effort self-heal: must not block dispatch.""" try: from cron.executions import recover_interrupted_executions - _reclaimed = recover_interrupted_executions() if _reclaimed: logger.warning( @@ -344,7 +334,6 @@ def _background_session_key(session_id: Optional[str]) -> str: cross the pool). Empty string = no durable consumer.""" try: from tools.approval import get_current_session_key - session_key = get_current_session_key(default="") except Exception: session_key = "" @@ -371,35 +360,24 @@ def _manual_run_completion( if excerpt: lines += ["--- JOB OUTPUT ---", excerpt] return { - "status": "completed" if res.get("success") else "error", - "summary": "\n".join(lines), - "error": res.get("error"), - "api_calls": 0, - "duration_seconds": duration, + "status": "completed" if res.get("success") else "error", "summary": "\n".join(lines), + "error": res.get("error"), "api_calls": 0, "duration_seconds": duration, } def _try_dispatch_background_run( job: Dict[str, Any], session_id: Optional[str] = None, extra_prompt: Optional[str] = None, ) -> Optional[Dict[str, Any]]: - """Claim ``job`` now, then fire it on the async-delegation daemon executor. - - A cron job is a full agent run (minutes to hours); inline it made the parent turn - uninterruptible. Dispatches like ``delegate_task``'s background mode: the tool returns a - handle and a ``type="async_delegation"`` completion re-enters the conversation as a - fresh turn (role alternation legal, prompt cache intact). The claim is taken - SYNCHRONOUSLY so unrunnable jobs report immediately. - - Returns None when background delivery is unavailable (caller falls back to the sync - path); ``{"claimed": False, ...}`` on a lost claim; ``{"claimed": True, "dispatched": - True, "delegation_id"}`` when running in the background; ``{"claimed": True, - "dispatched": False, ...}`` when the pool was full and the run executed inline. - """ - # Finite sessions cannot route a detached result back after the turn ends — mirror - # delegate_task's gate and fall back to sync execution. + """Claim ``job`` now (SYNCHRONOUSLY, so unrunnable jobs report immediately), then fire it + on the async-delegation executor like ``delegate_task``'s background mode: the tool returns + a handle and a ``type="async_delegation"`` completion re-enters as a fresh turn (role + alternation legal, prompt cache intact) instead of blocking the parent turn for minutes. + Returns None when background delivery is unavailable (caller runs sync); ``{"claimed": + False}`` on a lost claim; ``{"claimed": True, "dispatched": True, "delegation_id"}``; or + ``{"claimed": True, "dispatched": False, ...}`` when the pool was full and it ran inline.""" + # Finite sessions cannot route a detached result back after the turn ends (delegate_task's gate). try: from gateway.session_context import async_delivery_supported - if not async_delivery_supported(): return None except Exception: @@ -409,18 +387,16 @@ def _try_dispatch_background_run( job_name = str(job.get("name") or job_id) _reap_stale_executions(job_name) - # Routing capture BEFORE the claim: with no routable session there is no durable - # consumer for a detached completion, so we must not claim-and-dispatch. Direct Python - # callers (`hermes cron run`, tests) exit right after the tool returns -> run sync. + # Routing capture BEFORE the claim: no routable session = no durable consumer for a detached + # completion, so don't claim-and-dispatch (direct callers like `hermes cron run` exit right after). session_key = _background_session_key(session_id) if not session_key: return None - # Best-effort early dedupe so a mid-run job reports in THIS tool response instead of - # as a delayed error completion (authoritative check: try_register_running_job). + # Early dedupe so a mid-run job reports in THIS response, not as a delayed error completion + # (authoritative check: try_register_running_job). try: from cron.scheduler import get_running_job_ids - if job_id in get_running_job_ids(): return {"claimed": False, "success": False, "error": _ALREADY_RUNNING_ERROR} except Exception: @@ -435,14 +411,12 @@ def _try_dispatch_background_run( origin_ui_session_id = "" try: from gateway.session_context import get_session_env - origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "") or "" except Exception: pass try: from tools.async_delegation import _current_origin_session_id, dispatch_async_delegation - origin_session_id = _current_origin_session_id() except Exception as e: logger.warning( @@ -453,16 +427,13 @@ def _try_dispatch_background_run( try: from tools.delegate_tool import _get_max_async_children - max_async = _get_max_async_children() except Exception: max_async = 3 started_at = time.time() - # Canonicalize with the scheduler's own normalizer (falsy -> "local", list -> comma - # string), reading the claimed snapshot the run actually executes. + # Scheduler's own normalizer (falsy -> "local", list -> comma string) on the claimed snapshot. from cron.scheduler import _normalize_deliver_value - deliver = _normalize_deliver_value(claimed_job.get("deliver", "local")) def _runner() -> Dict[str, Any]: @@ -471,23 +442,16 @@ def _try_dispatch_background_run( dispatch = dispatch_async_delegation( goal=f"Manual run of cron job '{job_name}' ({job_id})", - context=( - "Triggered via cronjob(action='run'). The job executed in its own " - "fresh cron session; this block reports its outcome."), - toolsets=None, - role="cron_run", - model=job.get("model"), - session_key=session_key, - parent_session_id=str(session_id) if session_id else None, - runner=_runner, - origin_ui_session_id=origin_ui_session_id, - origin_session_id=origin_session_id, + context=("Triggered via cronjob(action='run'). The job executed in its own " + "fresh cron session; this block reports its outcome."), + toolsets=None, role="cron_run", model=job.get("model"), session_key=session_key, + parent_session_id=str(session_id) if session_id else None, runner=_runner, + origin_ui_session_id=origin_ui_session_id, origin_session_id=origin_session_id, max_async_children=max_async) if dispatch.get("status") == "dispatched": return {"claimed": True, "dispatched": True, "delegation_id": dispatch.get("delegation_id")} - # Pool at capacity (or submit failure): the claim is already taken and must not be - # stranded — run inline exactly as the legacy path did. + # Pool at capacity (or submit failure): the claim is already taken and must not be stranded. logger.info( "cronjob run: background pool unavailable (%s); running job '%s' inline.", dispatch.get("error", "rejected"), job_name) @@ -549,53 +513,31 @@ def _action_create(a: Dict[str, Any]) -> str: context_from = _apply_continuity(context_from, a["continuity"]) from cron.scheduler import CronSchedulerRegistrationError, create_job_with_scheduler_registration - try: job = create_job_with_scheduler_registration( - prompt=prompt or "", - schedule=a["schedule"], - name=a["name"], - repeat=a["repeat"], - deliver=_resolve_cron_context_deliver(deliver), - origin=_origin_from_env(), - skills=canonical_skills, - model=_normalize_optional_job_value(a["model"]), - provider=_normalize_optional_job_value(a["provider"]), + prompt=prompt or "", schedule=a["schedule"], name=a["name"], repeat=a["repeat"], + deliver=_resolve_cron_context_deliver(deliver), origin=_origin_from_env(), skills=canonical_skills, + model=_normalize_optional_job_value(a["model"]), provider=_normalize_optional_job_value(a["provider"]), base_url=_normalize_optional_job_value(a["base_url"], strip_trailing_slash=True), - script=_normalize_optional_job_value(script), - context_from=context_from, - enabled_toolsets=a["enabled_toolsets"] or None, - workdir=_normalize_optional_job_value(a["workdir"]), - no_agent=_no_agent, - attach_to_session=a["attach_to_session"], + script=_normalize_optional_job_value(script), context_from=context_from, + enabled_toolsets=a["enabled_toolsets"] or None, workdir=_normalize_optional_job_value(a["workdir"]), + no_agent=_no_agent, attach_to_session=a["attach_to_session"], monitor_script=_normalize_optional_job_value(a["monitor_script"]), monitor_url=_normalize_optional_job_value(a["monitor_url"]), - # CLI-only lane: absent from CRONJOB_SCHEMA and the model dispatch — models - # do not make model-config decisions. + # CLI-only lane: absent from CRONJOB_SCHEMA and the model dispatch (models don't pick models). reasoning_effort=a["reasoning_effort"], failure_deliver=_resolve_cron_context_deliver(_normalize_deliver_param(a["failure_deliver"]))) except CronSchedulerRegistrationError as exc: _partial = exc.to_dict() return tool_error(_partial.pop("error"), success=False, **_partial) - _create_message = f"Cron job '{job['name']}' created." - _local_notice = _local_delivery_notice(job, deliver) - if _local_notice: - _create_message = f"{_create_message} {_local_notice}" - # The builtin ticker lives in the gateway process: a job created with no gateway - # running is stored but never fires — tell the model (the CLI already warns). + _create_message = " ".join(filter(None, (f"Cron job '{job['name']}' created.", _local_delivery_notice(job, deliver)))) + # The builtin ticker lives in the gateway process: with no gateway running the job is stored + # but never fires — tell the model (the CLI already warns). _result = { - "success": True, - "job_id": job["id"], - "name": job["name"], - "skill": job.get("skill"), - "skills": job.get("skills", []), - "schedule": job["schedule_display"], - "repeat": _repeat_display(job), - "deliver": job.get("deliver", "local"), - "next_run_at": job["next_run_at"], - "job": _format_job(job), - "message": _create_message, - **_gateway_liveness_notice(), + "success": True, "job_id": job["id"], "name": job["name"], "skill": job.get("skill"), + "skills": job.get("skills", []), "schedule": job["schedule_display"], "repeat": _repeat_display(job), + "deliver": job.get("deliver", "local"), "next_run_at": job["next_run_at"], "job": _format_job(job), + "message": _create_message, **_gateway_liveness_notice(), } return _dumps(_with_guidance(_result, job, deliver)) @@ -827,13 +769,10 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str: # Actions that need no job_id, and job-bound actions (job resolved first). _JOBLESS_ACTIONS = {"create": _action_create, "list": _action_list} _JOB_ACTIONS = { - "remove": _action_remove, + "remove": _action_remove, "update": _action_update, + "run": _action_run, "run_now": _action_run, "trigger": _action_run, "pause": lambda job, a: _job_state_result(pause_job(job["id"], reason=a["reason"])), "resume": lambda job, a: _job_state_result(resume_job(job["id"])), - "run": _action_run, - "run_now": _action_run, - "trigger": _action_run, - "update": _action_update, } @@ -999,7 +938,6 @@ def check_cronjob_requirements() -> bool: """Available in interactive CLI mode and gateway/messaging platforms (the scheduler is internal; no crontab needed). Flags must be explicitly truthy via ``env_var_enabled``.""" from utils import env_var_enabled - return ( env_var_enabled("HERMES_INTERACTIVE") or env_var_enabled("HERMES_GATEWAY_SESSION") @@ -1012,10 +950,8 @@ def check_cronjob_requirements() -> bool: # create/edit --model`, hand-edited jobs) — the agent must not point unattended spend at a # different model. Programmatic callers of cronjob() itself retain the parameters. _HANDLER_FORWARDED_ARGS = ( - "job_id", "prompt", "schedule", "name", "repeat", "deliver", "failure_deliver", "skill", "skills", - "reason", "script", "context_from", "continuity", "enabled_toolsets", "workdir", - "no_agent", "attach_to_session", -) + "job_id", "prompt", "schedule", "name", "repeat", "deliver", "failure_deliver", "skill", "skills", "reason", + "script", "context_from", "continuity", "enabled_toolsets", "workdir", "no_agent", "attach_to_session") def _cronjob_handler(args, **kw):