mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(state): restore async_delegation.py symbols, keep only journal_mode routing

Browse source 
The previous commit accidentally reverted async_delegation.py to a
pre-origin_session_id snapshot, dropping _MAX_DELIVERY_ATTEMPTS,
_current_origin_session_id, _transaction(), the origin_session_id
column, and drop_completion_delivery() — causing ImportError in
delegate_tool.py and kanban_tools.py.

This restores the upstream main version of async_delegation.py and
re-applies only the journal_mode routing change (db_label update to
'async_delegation.db').

Addresses reviewer feedback on #68912.

Signed-off-by: Jasmine Naderi <jasmine@smfworks.com>
This commit is contained in:
Jasmine Naderi 2026-07-25 09:50:11 -04:00 • committed by Teknium
parent 04ec841462
commit 92914e9b09
1 changed files with 59 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

66
tools/async_delegation.py
View file

@ -78,6 +78,11 @@ _DEFAULT_MAX_ASYNC_CHILDREN = 3
_MAX_RETAINED_COMPLETED = 50
_DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60
_MAX_DURABLE_PENDING = 1000
# A pending completion whose delivery keeps failing is retried across claim
# cycles (and across restarts via restore_undelivered_completions). Cap the
# attempts so an unroutable row converges to a terminal 'dropped' state
# instead of replaying on every restart forever.
_MAX_DELIVERY_ATTEMPTS = 8
_DB_LOCK = threading.Lock()
# ---------------------------------------------------------------------------
@ -152,7 +157,8 @@ def _initialize_schema(conn: sqlite3.Connection) -> None:
owner_started_at INTEGER,
task_json TEXT,
delivery_claim TEXT,
delivery_claimed_at REAL
delivery_claimed_at REAL,
origin_session_id TEXT NOT NULL DEFAULT ''
)"""
)
columns = {row[1] for row in conn.execute("PRAGMA table_info(async_delegations)")}
@ -162,6 +168,11 @@ def _initialize_schema(conn: sqlite3.Connection) -> None:
("task_json", "TEXT"),
("delivery_claim", "TEXT"),
("delivery_claimed_at", "REAL"),
# Raw api_server session id (X-Hermes-Session-Id) of the ORIGINATING
# request — the wake self-post target. Without persisting it,
# completions recovered after a process restart are unroutable on
# api_server (the in-memory record that carried it is gone).
("origin_session_id", "TEXT"),
):
if name not in columns:
conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}")
@ -204,12 +215,13 @@ def _persist_dispatch(record: Dict[str, Any]) -> None:
(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)
VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?)""",
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, __import__("os").getpid(),
owner_started_at, json.dumps(task_payload)),
owner_started_at, json.dumps(task_payload),
record.get("origin_session_id", "")),
)
_prune_durable_records()
@ -290,11 +302,12 @@ def recover_abandoned_delegations() -> int:
rows = conn.execute(
"""SELECT delegation_id, origin_session, origin_ui_session_id,
parent_session_id, dispatched_at, owner_pid,
owner_started_at, task_json
owner_started_at, task_json, origin_session_id
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 = row
(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))
@ -306,6 +319,9 @@ def recover_abandoned_delegations() -> int:
event = {
"type": "async_delegation", "delegation_id": delegation_id,
"session_key": session_key, "origin_ui_session_id": origin_ui,
# Restore the durable wake target so completions recovered
# after a restart remain routable to api_server sessions.
"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"),
@ -482,7 +498,8 @@ 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
result_json, delivery_state, delivery_attempts,
origin_session_id
FROM async_delegations WHERE delegation_id=?""", (delegation_id,),
).fetchone()
if row is None:
@ -492,6 +509,7 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]:
"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 "",
}
@ -575,6 +593,34 @@ def _prune_completed_locked() -> None:
_records.pop(rid, None)
def _current_origin_session_id() -> str:
"""Raw session id of the ORIGINATING api_server request, or ``""``.
The obvious source — ``HERMES_SESSION_ID`` via ``get_session_env`` — is
NOT safe to read at dispatch time: constructing a child agent
(``agent/agent_init.py``) calls ``set_current_session_id(child.session_id)``,
clobbering that ContextVar *and* ``os.environ`` with the subagent's
internal ``{timestamp}_{uuid}`` id moments before the dispatch code reads
it, so the completion wake would self-post into the subagent's own
(unread) session instead of the spawner's.
The request-scoped ``HERMES_SESSION_CHAT_ID`` binding survives child
construction: ``_bind_api_server_session`` binds ``chat_id`` to the raw
``X-Hermes-Session-Id``, and its only writer is ``set_session_vars`` —
``set_current_session_id`` never touches it. Gate on the platform: on
push platforms ``chat_id`` is a chat, not a session, so yield ``""``
there.
"""
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 ""
except Exception:
return ""
def dispatch_async_delegation(
*,
goal: str,
@ -586,6 +632,7 @@ def dispatch_async_delegation(
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,
@ -643,6 +690,7 @@ def dispatch_async_delegation(
"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,
"status": "running",
"dispatched_at": dispatched_at,
@ -789,6 +837,7 @@ def _push_completion_event(
# session; empty string => CLI (single-session) path.
"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", ""),
"context": record.get("context"),
@ -838,6 +887,7 @@ def dispatch_async_delegation_batch(
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,
@ -880,6 +930,7 @@ def dispatch_async_delegation_batch(
"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,
"status": "running",
"dispatched_at": dispatched_at,
@ -991,6 +1042,7 @@ def _push_batch_completion_event(
"delegation_id": event_record.get("delegation_id"),
"session_key": event_record.get("session_key", ""),
"origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
"origin_session_id": event_record.get("origin_session_id", ""),
"parent_session_id": event_record.get("parent_session_id"),
"goal": event_record.get("goal", ""),
"goals": event_record.get("goals"),