02ed55e572
Every TUI/Desktop poller drains one process-wide queue. A session that cannot prove it owns an async delegation drops its copy while the durable row stays pending, and the sweep then skipped that row for the rest of the process because the offer was recorded at queue insertion. The drop and the TUI's release-after-failed-turn path now return the offer, so the next sweep re-offers the row; the delivery claim still settles it.
1173 lines
65 KiB
Python
1173 lines
65 KiB
Python
#!/usr/bin/env python3
|
|
"""Async (background) delegation registry behind ``delegate_task(background=true)``.
|
|
|
|
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
|
|
|
|
import contextvars
|
|
import json
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import Any, Callable, Dict, List, Optional
|
|
|
|
from hermes_constants import get_hermes_home, hermes_home_key
|
|
from tools.daemon_pool import DaemonThreadPoolExecutor
|
|
from tools.thread_context import propagate_context_to_thread
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Module-level state ──────────────────────────────────────────────────────
|
|
# Persistent daemon executor (never a `with ThreadPoolExecutor()` block, which
|
|
# would join on exit and defeat async); daemon workers can't hang a hard exit.
|
|
_executor: Optional[ThreadPoolExecutor] = None
|
|
_executor_lock = threading.Lock()
|
|
_executor_max_workers: int = 0
|
|
|
|
_records_lock = threading.Lock()
|
|
# delegation_id -> record dict; kept for the run plus a short completed tail.
|
|
_records: Dict[str, Dict[str, Any]] = {}
|
|
|
|
_DEFAULT_MAX_ASYNC_CHILDREN = 3
|
|
# Completed records retained (in memory and in the ledger) for status queries.
|
|
_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'.
|
|
_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.
|
|
_MAX_COMPLETION_REPLAY_AGE_S = 48 * 3600.0
|
|
# A delivery claim older than this is abandoned and may be re-claimed.
|
|
_CLAIM_LEASE_S = 300.0
|
|
_DB_LOCK = threading.Lock()
|
|
|
|
# ── Orphaned-completion sweep ────────────────────────────────────────────────
|
|
# Startup replay runs once per process, so a completion whose owner died while THIS process was
|
|
# already running (a desktop reload) would wait for the next restart (#97202). Delivery loops (gateway
|
|
# watcher, TUI poller) sweep each home they serve at most once per interval.
|
|
ORPHAN_SWEEP_INTERVAL_S = 30.0
|
|
# Idle time before a dead owner's pending row is re-offered; keeps the sweep off a row just touched.
|
|
_ORPHAN_STALE_S = 60.0
|
|
_orphan_lock = threading.Lock()
|
|
# (home key, delegation_id) put on this process's queue by replay or sweep and not re-offered while
|
|
# that copy is alive. A consumer that discards its copy with the row still pending hands it back
|
|
# (``return_completion_offer``); the delivery claim stays the only thing that settles the row.
|
|
_offered: set = set()
|
|
_last_orphan_sweep: Dict[str, float] = {}
|
|
|
|
# ── Stale-delegation detection (progress-based, on by default) ──────────────
|
|
# 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
|
|
_STALL_GRACE_SECONDS = 120.0
|
|
|
|
_monitor_lock = threading.Lock()
|
|
_monitor_thread: Optional[threading.Thread] = None
|
|
_monitor_stop = threading.Event()
|
|
|
|
_LIVE_STATES = {"running", "stalling", "finalizing"}
|
|
_ACTIVE_STATES = ("running", "stalling")
|
|
# Routing origin persisted at dispatch so a restart-recovered completion can
|
|
# 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")
|
|
# 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"))
|
|
|
|
|
|
# ── Durable ledger (state.db / async_delegations) ───────────────────────────
|
|
def _db_path():
|
|
return get_hermes_home() / "state.db"
|
|
|
|
|
|
def _connect() -> sqlite3.Connection:
|
|
from hermes_cli.sqlite_util import open_db
|
|
# Same state.db as hermes_state.SessionDB -- reuse its owner-only (0600)
|
|
# hardening so this writer doesn't create/leave the file (and its WAL
|
|
# sidecars) at the process umask. See hermes_state._secure_state_db_files.
|
|
from hermes_state import _secure_state_db_files
|
|
|
|
path = _db_path()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
_secure_state_db_files(path, create_main=True)
|
|
# wal=False: SessionDB owns state.db's journal mode (_initialize_schema applies the barriers).
|
|
conn = open_db(path, db_label="state.db (async_delegation)", busy_timeout_ms=10_000,
|
|
wal=False, row_factory=None, initialize=_initialize_schema)
|
|
_secure_state_db_files(path)
|
|
return conn
|
|
|
|
|
|
def _initialize_schema(conn: sqlite3.Connection) -> None:
|
|
from hermes_state_repair import apply_durability_barriers
|
|
from hermes_state_schema import reconcile_state_schema
|
|
# 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)
|
|
# Single durable-shape authority: the canonical SCHEMA_SQL drives both
|
|
# table creation and column backfill (reconcile_state_schema replays the
|
|
# canonical DDL and reuses SessionDB's declarative reconciliation). This
|
|
# module previously carried its own CREATE TABLE + ALTER column list,
|
|
# which drifted from SCHEMA_SQL — same-name columns with different
|
|
# nullability/defaults depending on which authority touched the database
|
|
# first (#94691).
|
|
reconcile_state_schema(conn)
|
|
|
|
|
|
def _transaction():
|
|
from hermes_cli.sqlite_util import transaction
|
|
|
|
return transaction(_connect())
|
|
|
|
|
|
def _capture_routing_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
|
|
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
|
|
return {}
|
|
|
|
|
|
def _persist_dispatch(record: Dict[str, Any]) -> None:
|
|
now = time.time()
|
|
try:
|
|
from gateway.status import get_process_start_time
|
|
owner_started_at = get_process_start_time(os.getpid())
|
|
except Exception:
|
|
owner_started_at = None
|
|
task_payload = {
|
|
key: record.get(key)
|
|
for key in ("goal", "goals", "context", "toolsets", "role", "model", "is_batch", "task_indexes", "task_transcripts", *_ROUTING_KEYS)
|
|
if key in record}
|
|
try: # where the children's terminals started; lets recovery add a git-state hint
|
|
task_payload["owner_cwd"] = os.getcwd()
|
|
except OSError:
|
|
pass
|
|
with _DB_LOCK, _transaction() as conn:
|
|
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", "")))
|
|
_prune_durable_records()
|
|
|
|
|
|
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,))
|
|
terminal_count = conn.execute(
|
|
"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 ?
|
|
)""", (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 ?
|
|
)""", (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=?,
|
|
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"]))
|
|
|
|
|
|
def record_unit_child(delegation_id: str, entry: Dict[str, Any]) -> None:
|
|
"""Durably record ONE finished child of a still-running multi-child unit on the unit's own row, so a crash before
|
|
the unit joins loses only the children that had not finished. Stored in ``result_json`` (overwritten by the real
|
|
result at finalize); ``recover_abandoned_delegations`` replays it. Best-effort: a failed write costs recovery
|
|
fidelity, never the live result."""
|
|
try:
|
|
with _DB_LOCK, _transaction() as conn:
|
|
row = conn.execute("SELECT result_json FROM async_delegations WHERE delegation_id=? AND state='running'",
|
|
(delegation_id,)).fetchone()
|
|
if row is None:
|
|
return
|
|
partial = json.loads(row[0] or "{}") or {}
|
|
results = [r for r in partial.get("results") or [] if r.get("task_index") != entry.get("task_index")]
|
|
results.append(entry)
|
|
conn.execute("UPDATE async_delegations SET result_json=?, updated_at=? WHERE delegation_id=? AND state='running'",
|
|
(json.dumps({"results": results, "partial": True}), time.time(), delegation_id))
|
|
except Exception: # noqa: BLE001 — recovery bookkeeping must never fail a live child
|
|
logger.warning("Async delegation %s: could not record finished child %s", delegation_id, entry.get("task_index"), exc_info=True)
|
|
|
|
|
|
def _recovered_results(task: Dict[str, Any], result_json: Optional[str], error: str) -> Optional[List[Dict[str, Any]]]:
|
|
"""Per-task results for an abandoned unit: recorded children as they finished, the rest ``unknown``."""
|
|
partial = json.loads(result_json or "{}") or {}
|
|
if not (task.get("is_batch") and partial.get("partial") and partial.get("results")):
|
|
return None
|
|
recorded = {r["task_index"]: r for r in partial["results"] if isinstance(r.get("task_index"), int)}
|
|
indexes = task.get("task_indexes") or list(range(len(task.get("goals") or [])))
|
|
return [recorded.get(i) or {"task_index": i, "status": "unknown", "summary": None, "error": error} for i in indexes]
|
|
|
|
|
|
def _owner_liveness() -> Optional[Callable[[Any, Any], bool]]:
|
|
"""``alive(owner_pid, owner_started_at)`` over the shared drift-tolerant start-time comparator,
|
|
or None when the liveness probes cannot be imported."""
|
|
try:
|
|
from gateway.status import _pid_exists, get_process_start_time, start_time_fingerprints_match
|
|
except Exception:
|
|
return None
|
|
|
|
def alive(pid, started) -> bool:
|
|
return bool(pid) and _pid_exists(int(pid)) and (
|
|
started is None or start_time_fingerprints_match(started, get_process_start_time(int(pid)) or 0))
|
|
return alive
|
|
|
|
|
|
def recover_abandoned_delegations() -> int:
|
|
"""Classify records whose owning process disappeared as outcome unknown; children a multi-child unit had already
|
|
recorded (``record_unit_child``) are replayed with their real results."""
|
|
alive = _owner_liveness()
|
|
if alive is None:
|
|
return 0
|
|
now, recovered = time.time(), 0
|
|
with _DB_LOCK, _transaction() as conn:
|
|
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, result_json, state
|
|
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_sid, result_json, last_state = row
|
|
if alive(pid, started):
|
|
continue
|
|
task = json.loads(task_json or "{}")
|
|
error = "Delegation owner exited before recording a terminal result; outcome unknown."
|
|
recovered_results = _recovered_results(task, result_json, error)
|
|
if recovered_results:
|
|
done = sum(1 for r in recovered_results if r.get("status") != "unknown")
|
|
error = (f"Delegation owner exited before the unit finished; {done}/{len(recovered_results)} child "
|
|
"results were recorded and are included below, the rest are unknown.")
|
|
diagnostics = {"last_known_status": last_state, "task_transcripts": task.get("task_transcripts") or {}}
|
|
# Verbatim transcript tails + a git snapshot of the owner's cwd, so the parent can
|
|
# continue or re-dispatch from the event alone instead of opening files (#116000).
|
|
from tools.async_delegation_recovery_hints import git_state_hint, transcript_tails
|
|
if tails := transcript_tails(diagnostics["task_transcripts"]):
|
|
diagnostics["transcript_tails"] = tails
|
|
if hint := git_state_hint(task.get("owner_cwd")):
|
|
diagnostics["git_state_hint"] = hint
|
|
event = {
|
|
"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": error, **diagnostics,
|
|
**({"results": recovered_results} if recovered_results else {}),
|
|
"dispatched_at": dispatched_at, "completed_at": now,
|
|
**{k: task[k] for k in _ROUTING_KEYS if task.get(k)}}
|
|
result = {"status": "unknown", "summary": None, "error": event["error"], **diagnostics,
|
|
**({"results": recovered_results} if recovered_results else {})}
|
|
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))
|
|
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 turn nobody is waiting on.
|
|
|
|
Every restored event is stamped ``restored=True`` (in-memory only — the stamp is added after the durable
|
|
payload is deserialized and is never persisted). Restored events originate from a *previous* process, so
|
|
no consumer in THIS process implicitly owns them: drain paths that run without an ownership filter (the
|
|
legacy single-session behavior) must leave them queued for a consumer that can positively prove
|
|
ownership, otherwise a brand-new session adopts a dead session's delegation results seconds after boot
|
|
(#64484).
|
|
"""
|
|
recover_abandoned_delegations()
|
|
now = time.time()
|
|
with _DB_LOCK, _transaction() as conn:
|
|
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()
|
|
return _replay_pending(conn, rows, target_queue, now)
|
|
|
|
|
|
def _replay_pending(conn, rows, target_queue, now: float) -> int:
|
|
"""Put each pending ``(delegation_id, event_json, completed_at, dispatched_at)`` row on ``target_queue``
|
|
stamped ``restored``, or terminally drop it past ``_MAX_COMPLETION_REPLAY_AGE_S``. Records the offer so
|
|
the orphan sweep skips the row until the copy is handed back (``return_completion_offer``)."""
|
|
home, restored = hermes_home_key(get_hermes_home()), 0
|
|
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',
|
|
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)
|
|
continue
|
|
evt = json.loads(payload)
|
|
if isinstance(evt, dict):
|
|
evt["restored"] = True
|
|
target_queue.put(evt)
|
|
with _orphan_lock:
|
|
_offered.add((home, delegation_id))
|
|
restored += 1
|
|
return restored
|
|
|
|
|
|
def sweep_orphaned_completions(target_queue, *, now: Optional[float] = None) -> int:
|
|
"""Offer this home's completions whose owner died after THIS process started (#97202).
|
|
|
|
Startup replay (``restore_undelivered_completions``) covers owners that died before the process
|
|
started; this covers the rest while it runs. Abandoned in-flight rows are first classified by
|
|
``recover_abandoned_delegations``. A terminal row qualifies when it is pending with an event, idle
|
|
past ``_ORPHAN_STALE_S``, not under a live delivery claim, and its owner fails the shared liveness
|
|
check. A row is offered once per live in-memory copy: a consumer that discards the copy with the row
|
|
still pending hands it back for the next sweep. The consumer's ``claim_completion_delivery`` stays
|
|
the atomic cross-process gate, so two processes offering one row never both deliver it. Rows past
|
|
the delivery budget or the replay age converge to ``dropped``. Reads the current profile's ledger:
|
|
callers bind the owning profile first."""
|
|
alive = _owner_liveness()
|
|
if alive is None or not _db_path().exists():
|
|
return 0 # never create a ledger just to sweep it
|
|
recover_abandoned_delegations()
|
|
now = time.time() if now is None else now
|
|
home = hermes_home_key(get_hermes_home())
|
|
with _orphan_lock:
|
|
offered = {delegation_id for key, delegation_id in _offered if key == home}
|
|
with _DB_LOCK, _transaction() as conn:
|
|
rows = conn.execute("""SELECT delegation_id, event_json, completed_at, dispatched_at,
|
|
owner_pid, owner_started_at, delivery_attempts
|
|
FROM async_delegations
|
|
WHERE state NOT IN ('running','finalizing') AND delivery_state='pending'
|
|
AND event_json IS NOT NULL AND updated_at < ?
|
|
AND (delivery_claim IS NULL OR delivery_claimed_at < ?)
|
|
ORDER BY completed_at, delegation_id""", (now - _ORPHAN_STALE_S, now - _CLAIM_LEASE_S)).fetchall()
|
|
orphans = []
|
|
for delegation_id, payload, completed_at, dispatched_at, pid, started, attempts in rows:
|
|
if delegation_id in offered or alive(pid, started):
|
|
continue
|
|
if (attempts or 0) >= _MAX_DELIVERY_ATTEMPTS:
|
|
# Its last claimant died holding the final attempt; converge like release_completion_delivery.
|
|
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 exhausted its %d delivery attempts; "
|
|
"marking terminally dropped (result remains queryable).",
|
|
delegation_id, _MAX_DELIVERY_ATTEMPTS)
|
|
continue
|
|
orphans.append((delegation_id, payload, completed_at, dispatched_at))
|
|
return _replay_pending(conn, orphans, target_queue, now)
|
|
|
|
|
|
def maybe_sweep_orphaned_completions(target_queue, *, now: Optional[float] = None) -> int:
|
|
"""``sweep_orphaned_completions`` at most once per ``ORPHAN_SWEEP_INTERVAL_S`` per home (``now`` is
|
|
monotonic), for delivery loops that tick far more often. Never raises into the loop."""
|
|
home = hermes_home_key(get_hermes_home())
|
|
now = time.monotonic() if now is None else now
|
|
with _orphan_lock:
|
|
last = _last_orphan_sweep.get(home)
|
|
if last is not None and now - last < ORPHAN_SWEEP_INTERVAL_S:
|
|
return 0
|
|
_last_orphan_sweep[home] = now
|
|
try:
|
|
return sweep_orphaned_completions(target_queue)
|
|
except Exception:
|
|
logger.debug("Orphaned async delegation sweep failed", exc_info=True)
|
|
return 0
|
|
|
|
|
|
def _update_delivery(sql: str, params: tuple) -> bool:
|
|
"""Run one UPDATE on the ledger; True iff exactly one row changed."""
|
|
with _DB_LOCK, _transaction() as conn:
|
|
return conn.execute(sql, params).rowcount == 1
|
|
|
|
|
|
def mark_completion_delivered(delegation_id: str) -> bool:
|
|
"""Atomically acknowledge successful injection of a durable completion."""
|
|
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))
|
|
|
|
|
|
def claim_completion_delivery(delegation_id: str, claim_id: str) -> bool:
|
|
"""Claim one pending completion across competing consumers/processes."""
|
|
now = time.time()
|
|
with _DB_LOCK, _transaction() as conn:
|
|
row = conn.execute(
|
|
"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=?,
|
|
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 - _CLAIM_LEASE_S))
|
|
return cur.rowcount == 1
|
|
|
|
|
|
def is_interim_delegation_event(evt: Dict[str, Any]) -> bool:
|
|
"""An early per-task notice for a batch that is still running. It shares the batch's
|
|
``delegation_id`` but is NOT the durable completion: it must never claim, acknowledge or
|
|
dedup against the final result's row (independent review reproduced exactly that loss)."""
|
|
return evt.get("type") == "async_delegation" and bool(evt.get("task_failure_notice"))
|
|
|
|
|
|
def claim_event_delivery(evt: Dict[str, Any], consumer: str) -> Optional[str]:
|
|
"""Claim a durable delegation event; non-durable events (and interim notices) need no token."""
|
|
if is_interim_delegation_event(evt):
|
|
return ""
|
|
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}"
|
|
return claim_id if claim_completion_delivery(delegation_id, claim_id) else None
|
|
|
|
|
|
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)."""
|
|
now = time.time()
|
|
with _DB_LOCK, _transaction() as conn:
|
|
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))
|
|
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,
|
|
delivery_claimed_at=NULL, updated_at=?
|
|
WHERE delegation_id=? AND delivery_state='pending'
|
|
AND delivery_claim=?""", (now, delegation_id, claim_id))
|
|
return cur.rowcount == 1
|
|
|
|
|
|
def defer_completion_delivery(delegation_id: str, claim_id: str) -> bool:
|
|
"""Return an unadmitted completion to pending without spending a delivery attempt."""
|
|
return _update_delivery("""UPDATE async_delegations SET delivery_claim=NULL,
|
|
delivery_claimed_at=NULL, delivery_attempts=MAX(0, delivery_attempts-1),
|
|
updated_at=?
|
|
WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=?""",
|
|
(time.time(), delegation_id, claim_id))
|
|
|
|
|
|
def drop_completion_delivery(delegation_id: str, claim_id: str) -> bool:
|
|
"""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))
|
|
|
|
|
|
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',
|
|
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))
|
|
|
|
|
|
def complete_event_delivery(evt: Dict[str, Any], claim_id: str) -> None:
|
|
_event_delivery(complete_completion_delivery, evt, claim_id)
|
|
|
|
|
|
def release_event_delivery(evt: Dict[str, Any], claim_id: str) -> None:
|
|
"""Release a failed claim for a consumer that discards its copy (the TUI poller): the row is pending
|
|
again, so it must stay eligible for the orphan sweep."""
|
|
_event_delivery(release_completion_delivery, evt, claim_id)
|
|
return_completion_offer(evt)
|
|
|
|
|
|
def return_completion_offer(evt: Dict[str, Any]) -> None:
|
|
"""Hand an offered completion back to the orphan sweep after its in-memory copy was discarded while
|
|
the durable row stays pending, e.g. a TUI session that cannot prove it owns the event drops it (every
|
|
session poller drains one process-wide queue). The next sweep may offer the row again. Delegation ids
|
|
are unique across profiles, so this clears the offer in every home."""
|
|
delegation_id = str(evt.get("delegation_id") or "") if evt.get("type") == "async_delegation" else ""
|
|
if not delegation_id or is_interim_delegation_event(evt):
|
|
return
|
|
with _orphan_lock:
|
|
_offered.difference_update({key for key in _offered if key[1] == delegation_id})
|
|
|
|
|
|
def _event_delivery(fn, evt: Dict[str, Any], claim_id: str) -> None:
|
|
if claim_id and evt.get("type") == "async_delegation":
|
|
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,
|
|
result_json, delivery_state, delivery_attempts,
|
|
origin_session_id
|
|
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 in place, never shrink) the shared daemon executor. Raising
|
|
``_max_workers`` is enough: the next ``submit`` spawns threads up to the new cap."""
|
|
global _executor, _executor_max_workers
|
|
with _executor_lock:
|
|
if _executor is None:
|
|
_executor = DaemonThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="async-delegate")
|
|
_executor_max_workers = max_workers
|
|
elif max_workers > _executor_max_workers:
|
|
_executor._max_workers = max_workers
|
|
_executor_max_workers = max_workers
|
|
return _executor
|
|
|
|
|
|
def active_count() -> int:
|
|
"""Number of live async delegation UNITS (one per completion message: a task group or an ungrouped task)."""
|
|
with _records_lock:
|
|
return sum(1 for r in _records.values() if r.get("status") in _LIVE_STATES)
|
|
|
|
|
|
def active_task_count() -> int:
|
|
"""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(
|
|
len(r.get("task_indexes") or 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 _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 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."""
|
|
return bool(_session_records(_LIVE_STATES, session_key, origin_ui_session_id, parent_session_id))
|
|
|
|
|
|
def _new_delegation_id() -> str:
|
|
return f"deleg_{uuid.uuid4().hex[:8]}"
|
|
|
|
|
|
def _prune_completed_locked() -> None:
|
|
"""Drop the oldest completed records beyond the cap. Caller holds ``_records_lock``.
|
|
``stalling``/``finalizing`` are still live: evicting one makes the late runner return hit
|
|
``_finalize``'s missing-record path and silently drop a real result."""
|
|
completed = [(rid, r) for rid, r in _records.items() if r.get("status") not in _LIVE_STATES]
|
|
completed.sort(key=lambda kv: kv[1].get("completed_at") or kv[1].get("dispatched_at") or 0)
|
|
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)``
|
|
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
|
|
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}
|
|
|
|
|
|
def _batch_crash(error: str, duration: float) -> Dict[str, Any]:
|
|
return {"results": [], "error": error, "total_duration_seconds": duration}
|
|
|
|
|
|
def _batch_status(combined: Dict[str, Any]) -> str:
|
|
"""Batch status: completed unless every child errored/was interrupted."""
|
|
child_results = combined.get("results") or []
|
|
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(**kwargs) -> Dict[str, Any]:
|
|
from hermes_cli.backend_retirement import retirement
|
|
|
|
with retirement.work() as admitted:
|
|
if not admitted:
|
|
return {"status": "rejected", "error": "backend is retiring; reconnect to continue"}
|
|
return _dispatch_admitted(**kwargs)
|
|
|
|
|
|
def _dispatch_admitted(
|
|
*, 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,
|
|
progress_fn: Optional[Callable[[], tuple]], capacity_error: str, slot_key: Optional[str] = None,
|
|
task_indexes: Optional[List[int]] = None,
|
|
task_transcripts: Optional[Dict[str, str]] = None,
|
|
) -> 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. ``slot_key`` names the pool slot the unit occupies
|
|
(default: its own id); the units of one delegate_task call share the first unit's id so
|
|
splitting a call into per-group completions never consumes more capacity than the call did."""
|
|
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,
|
|
**_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,
|
|
"slot_key": slot_key or delegation_id,
|
|
**({"task_transcripts": dict(task_transcripts)} if task_transcripts else {}),
|
|
# Which of the call's ``goals`` this unit runs (None = all of them).
|
|
**({"task_indexes": list(task_indexes)} if task_indexes is not None else {}),
|
|
# The one stale-monitor thread serves every profile and starts with an empty Context;
|
|
# a forced finalization runs under the dispatcher's so it settles the same state.db.
|
|
"_context": contextvars.copy_context(),
|
|
# Stale-monitor bookkeeping (see _stale_monitor_loop).
|
|
"_progress_token": None, "_progress_ts": dispatched_at, "_interrupted_at": None}
|
|
with _records_lock:
|
|
active_slots = {r.get("slot_key") or r["delegation_id"] for r in _records.values() if r.get("status") in _ACTIVE_STATES}
|
|
if record["slot_key"] not in active_slots and len(active_slots) >= max_async_children:
|
|
return {"status": "rejected", "error": capacity_error}
|
|
_records[delegation_id] = record
|
|
live_units = sum(1 for r in _records.values() if r.get("status") in _LIVE_STATES)
|
|
_persist_dispatch(record)
|
|
# Units of one call share a slot, so live units can exceed slots: size the pool by units or a
|
|
# unit queues behind a full pool and the stale monitor kills it before its child ever starts.
|
|
executor = _get_executor(max(max_async_children, live_units))
|
|
|
|
def _worker() -> None:
|
|
result: Dict[str, Any] = {}
|
|
status = "error"
|
|
with _records_lock:
|
|
rec = _records.get(delegation_id)
|
|
if rec is not None:
|
|
# The stall clock starts when the runner starts; a unit queued behind a full pool is not stalled.
|
|
rec.update(_started=True, _progress_ts=time.time())
|
|
try:
|
|
result = runner() or {}
|
|
status = classify(result)
|
|
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))
|
|
finally:
|
|
_finalize(delegation_id, result, status)
|
|
|
|
from hermes_cli.backend_retirement import retirement
|
|
|
|
# The outer dispatch reservation prevents a freeze during this handoff. Retain a worker
|
|
# reservation too: the stall monitor may finalize its registry record before it really exits.
|
|
retirement.acquire()
|
|
try:
|
|
future = executor.submit(propagate_context_to_thread(_worker))
|
|
future.add_done_callback(lambda _: retirement.release())
|
|
except Exception as exc: # pragma: no cover — pool submit failure is rare
|
|
retirement.release()
|
|
with _records_lock:
|
|
_records.pop(delegation_id, None)
|
|
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()
|
|
return {"status": "dispatched", "delegation_id": delegation_id}
|
|
|
|
|
|
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,
|
|
) -> 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"}``."""
|
|
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,
|
|
capacity_error=(
|
|
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 "<cli>", (goal or "")[:80])
|
|
return handle
|
|
|
|
|
|
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, slot_key: Optional[str] = None,
|
|
task_indexes: Optional[List[int]] = None,
|
|
task_transcripts: Optional[Dict[str, str]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Dispatch a fan-out unit (a whole batch, or one ``group`` of a delegate_task call) as ONE
|
|
background unit: ``runner`` runs its tasks and returns the combined ``{"results": [...],
|
|
"total_duration_seconds": N}`` dict. The unit occupies ONE async slot — or joins the slot named
|
|
by ``slot_key`` (in-unit parallelism is bounded separately) — and produces a SINGLE completion
|
|
event carrying per-task ``results``."""
|
|
delegation_id = delegation_id or _new_delegation_id()
|
|
# ``goals`` is the whole call (result task_index indexes it); the unit's own goals label the record.
|
|
unit_goals = [goals[i] for i in task_indexes] if task_indexes is not None else list(goals)
|
|
n = len(unit_goals)
|
|
combined_goal = unit_goals[0] if n == 1 else f"{n} parallel subagents: " + "; ".join(g[:40] for g in unit_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, slot_key=slot_key,
|
|
task_indexes=task_indexes, task_transcripts=task_transcripts,
|
|
capacity_error=(
|
|
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 "<cli>")
|
|
return handle
|
|
|
|
|
|
# ── Finalization + completion events ────────────────────────────────────────
|
|
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
|
|
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
|
|
snapshot = dict(record)
|
|
_push_completion_event(snapshot, result(snapshot) if callable(result) else result, status)
|
|
with _records_lock:
|
|
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 (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:
|
|
from tools.process_registry import process_registry
|
|
except Exception as exc: # pragma: no cover
|
|
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:
|
|
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"),
|
|
**({"group": result["group"]} if result.get("group") is not None else {})}
|
|
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))}
|
|
evt = {
|
|
"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"),
|
|
"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,
|
|
**({} if is_batch else {"exit_reason": result.get("exit_reason")}),
|
|
**{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}}
|
|
try:
|
|
_persist_completion(evt, result)
|
|
except Exception as exc: # noqa: BLE001 — a lost durable row is recoverable; a lost result + leaked slot is not
|
|
logger.error(f"Async delegation{label} %s: durable completion write failed; delivering in-memory "
|
|
"only (a restart may report this unit as unknown): %s", record.get("delegation_id"), exc)
|
|
try:
|
|
process_registry.completion_queue.put(evt)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.error(f"Async delegation{label} %s: failed to enqueue completion event; "
|
|
"result lost: %s", record.get("delegation_id"), exc)
|
|
|
|
|
|
def push_task_failure_notice(delegation_id: str, entry: Dict[str, Any], *, n_tasks: int) -> None:
|
|
"""Surface ONE failed child of a still-running detached batch to the parent now, instead of
|
|
when the slowest sibling finishes. In a 1,393-agent run every wave-1 child died in a 401 storm
|
|
at 08:29 and the parent learned of it at 09:36, when the batch's "unknown outcome" block finally
|
|
arrived: 66 minutes of a dead wave with nothing running. The notice rides the same
|
|
``type="async_delegation"`` event shape as the batch result (so every drain/route/format path
|
|
treats it identically) with ``task_failure_notice=True`` and a single-entry ``results`` list; the
|
|
batch record is NOT finalized and its consolidated result still arrives as before."""
|
|
with _records_lock:
|
|
record = _records.get(delegation_id)
|
|
if record is None or record.get("status") not in _ACTIVE_STATES:
|
|
return
|
|
snapshot = dict(record)
|
|
try:
|
|
from tools.process_registry import process_registry
|
|
except Exception as exc: # pragma: no cover
|
|
logger.error("Async delegation batch %s: task failure notice dropped (process_registry import): %s", delegation_id, exc)
|
|
return
|
|
evt = {
|
|
"type": "async_delegation", "task_failure_notice": True, "is_batch": True, "n_tasks": n_tasks,
|
|
"delegation_id": delegation_id, "results": [entry],
|
|
"session_key": snapshot.get("session_key", ""),
|
|
"origin_ui_session_id": snapshot.get("origin_ui_session_id", ""),
|
|
"origin_session_id": snapshot.get("origin_session_id", ""),
|
|
"parent_session_id": snapshot.get("parent_session_id"),
|
|
"goal": snapshot.get("goal", ""), "goals": snapshot.get("goals"), "context": snapshot.get("context"),
|
|
"toolsets": snapshot.get("toolsets"), "role": snapshot.get("role"), "model": snapshot.get("model"),
|
|
"status": "running", "dispatched_at": snapshot.get("dispatched_at") or time.time(), "completed_at": time.time(),
|
|
**{k: snapshot[k] for k in _ROUTING_KEYS if snapshot.get(k)}}
|
|
try:
|
|
process_registry.completion_queue.put(evt)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.error("Async delegation batch %s: failed to enqueue task failure notice: %s", delegation_id, exc)
|
|
|
|
|
|
# ── 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``."""
|
|
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)
|
|
_monitor_thread.start()
|
|
|
|
|
|
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
|
|
if not record.get("_started"):
|
|
continue # queued behind a full pool: not stalled, but keep the monitor alive for when it starts
|
|
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
|
|
|
|
|
|
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()
|
|
with _records_lock:
|
|
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:
|
|
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:
|
|
with _records_lock:
|
|
ctx = (_records.get(delegation_id) or {}).get("_context") or contextvars.copy_context()
|
|
ctx.run(_finalize, delegation_id, lambda rec, d=delegation_id: _stalled_result(d, rec), "stalled")
|
|
if not any_monitorable:
|
|
return
|
|
|
|
|
|
def _stalled_error_text(event_record: Dict[str, Any]) -> str:
|
|
"""Human wording for a force-finalized stall. This string reaches the user (CLI timeline, Desktop
|
|
async-result card), so it names the task, how long it was silent, and what to do — no issue
|
|
numbers or worker internals (those stay in the log line and the stall_* metadata)."""
|
|
goal = " ".join(str(event_record.get("goal") or "").split())
|
|
label = f'Background task "{goal[:120]}"' if goal else "The background task"
|
|
quiet = float(event_record.get("_stall_quiet_seconds") or 0)
|
|
silence = f" after {round(quiet / 60)} min of no progress" if quiet >= 60 else ""
|
|
return (f"{label} stopped responding{silence} and was cancelled. Nothing else was affected; "
|
|
"ask me to run it again if you still need it.")
|
|
|
|
|
|
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 = _stalled_error_text(event_record)
|
|
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.
|
|
stall_in_tool = event_record.get("_stall_in_tool")
|
|
stall_meta = {
|
|
"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}
|
|
if event_record.get("is_batch"):
|
|
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."""
|
|
try:
|
|
parts = list(token)
|
|
except TypeError:
|
|
return None
|
|
out: List[Optional[Dict[str, Any]]] = []
|
|
for part in parts:
|
|
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) 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.
|
|
|
|
Safe to call from any thread. See #51690.
|
|
"""
|
|
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("_")}
|
|
status = r.get("status")
|
|
if status in _ACTIVE_STATES:
|
|
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:
|
|
fn = samplers.get(item.get("delegation_id"))
|
|
if fn is None:
|
|
continue
|
|
try:
|
|
token, in_tool = fn()
|
|
except Exception:
|
|
continue
|
|
activity = _children_activity_from_token(token, now)
|
|
if activity is not None:
|
|
item["children_activity"] = activity
|
|
item["in_tool"] = bool(in_tool)
|
|
return items
|
|
|
|
|
|
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."""
|
|
with _records_lock:
|
|
targets = [r for r in _records.values() if r.get("status") in _ACTIVE_STATES]
|
|
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",
|
|
) -> int:
|
|
"""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:
|
|
"""Test-only: clear all state and tear down the executor + monitor."""
|
|
global _executor, _executor_max_workers, _monitor_thread
|
|
with _executor_lock:
|
|
if _executor is not None:
|
|
_executor.shutdown(wait=False)
|
|
_executor = None
|
|
_executor_max_workers = 0
|
|
_monitor_stop.set()
|
|
with _monitor_lock:
|
|
thread, _monitor_thread = _monitor_thread, None
|
|
if thread is not None and thread.is_alive():
|
|
thread.join(timeout=2)
|
|
with _records_lock:
|
|
_records.clear()
|
|
with _orphan_lock:
|
|
_offered.clear()
|
|
_last_orphan_sweep.clear()
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
|
|
def active_for_session(origin_ui_session_id: str) -> int:
|
|
"""Number of live async delegations owned by one UI session."""
|
|
if not origin_ui_session_id:
|
|
return 0
|
|
with _records_lock:
|
|
return sum(
|
|
1
|
|
for r in _records.values()
|
|
if r.get("status") in {"running", "stalling", "finalizing"}
|
|
and str(r.get("origin_ui_session_id") or "")
|
|
== origin_ui_session_id
|
|
)
|
|
# ---- END PLUGIN-COMPAT ----
|