"""Trusted dispatcher claims enter the ordinary owner's durable admission queue.""" import asyncio import contextlib from contextlib import closing from dataclasses import asdict, replace import threading import json from pathlib import Path import sqlite3 from hermes_state_runtime import RuntimeStoreError def build_kanban_policy(connection, params, config): from hermes_cli import kanban_db as kb from hermes_cli.profiles import resolve_profile_env from gateway.session_policy import build_policy authority, actor = connection.authority, connection.actor if not connection.native_owner or 'session:create' not in actor.capabilities: raise RuntimeStoreError('permission_denied') if actor.profile_id != authority.profile_id: raise RuntimeStoreError('profile_mismatch') if (set(params) not in ({'board', 'task_id', 'run_id', 'claim_lock'}, {'board', 'task_id', 'run_id', 'claim_lock', 'db'}) or any(not isinstance(params[k], str) or not params[k] for k in ('board', 'task_id', 'claim_lock')) or type(params['run_id']) is not int): raise RuntimeStoreError('invalid_params') try: board = kb._require_slug(params['board']) path = Path(params['db']).resolve(strict=True) if 'db' in params else kb.kanban_db_path(board=board).resolve() with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True)) as conn: conn.row_factory = sqlite3.Row task = kb.get_task(conn, params['task_id']) run = conn.execute('SELECT * FROM task_runs WHERE id=?', (params['run_id'],)).fetchone() if (task is None or run is None or task.status != 'running' or run['status'] != 'running' or task.current_run_id != params['run_id'] or run['task_id'] != task.id or task.claim_lock != params['claim_lock'] or run['claim_lock'] != task.claim_lock): raise RuntimeStoreError('stale_kanban_claim') if Path(resolve_profile_env(task.assignee)).resolve() != Path(authority.profile_id).resolve(): raise RuntimeStoreError('profile_mismatch') bound = conn.execute("SELECT payload FROM task_events WHERE task_id=? AND run_id=? AND kind='owner_admitted' ORDER BY id DESC LIMIT 1", (task.id, task.current_run_id)).fetchone() if bound and json.loads(bound[0])['request_id'] != 'kanban:' + json.dumps([board, task.id, task.current_run_id], separators=(',', ':')): raise RuntimeStoreError('stale_kanban_claim') context = kb.build_worker_context(conn, task.id) skills = list(task.skills or ()) if kb._retry_status_for_run(conn, task.id, task.current_run_id) == 'review': # Same rule as the dispatcher's review lane: the reviewer always loads sdlc-review. skills = list(dict.fromkeys([*skills, 'sdlc-review'])) except (ValueError, OSError, sqlite3.Error) as exc: raise RuntimeStoreError('invalid_kanban_claim') from exc cwd = task.workspace_path params_policy = {'source': 'cli', 'cwd': cwd} for name, value in (('model', task.model_override), ('provider', task.provider_override), ('reasoning', task.reasoning_effort)): if value: params_policy[name] = value private = {} policy = build_policy(params_policy, config, private_secrets=private) if policy.model is None: from gateway.run import _resolve_gateway_model policy = replace(policy, model=_resolve_gateway_model(policy.config())) context = dict(params, workspace=cwd, branch=task.branch_name, tenant=task.tenant, profile=task.assignee, workspaces_root=str(kb.workspaces_root(board=board).resolve()), skills=skills, goal_mode=task.goal_mode, goal_max_turns=task.goal_max_turns, goal_text='\n\n'.join(p for p in (task.title, task.body) if p), context=context, max_runtime_seconds=task.max_runtime_seconds, accept_hooks=True) context['db'] = str(path) request = json.loads(policy.request_json) request.update(source='kanban', board=board, task_id=task.id, run_id=task.current_run_id) return replace(policy, source='kanban', platform='cli', toolsets=tuple(sorted(set(policy.toolsets) | {'kanban'})), request_json=json.dumps(request, sort_keys=True), kanban_json=json.dumps(context)), private async def run_task(connection, params): if not connection.native_owner or 'session:create' not in connection.actor.capabilities: raise RuntimeStoreError('permission_denied') from gateway.run import _load_gateway_config from gateway.session_local import create_local_session from gateway.session_contract import Submission from gateway.session_local_recovery import local_identity from hermes_state_local import local_receipt from gateway.session_policy import restore_policy authority = connection.authority # Resolved off-loop up front: no await separates the saved-claim check from session # creation. The Kanban claim check and its marker share one off-loop transaction; a # concurrent retry of the same attempt meets that marker (same session) or a stale claim. owner_db = str(await asyncio.to_thread(Path(authority.db.db_path).resolve)) # The dispatcher attempt is the durable request identity, including after the # task closes. Retry checks the saved claim, never reinterprets updated cards. request_id = 'kanban:' + json.dumps([params.get('board'), params.get('task_id'), params.get('run_id')], separators=(',', ':')) sid = local_identity(authority.profile_id, connection.actor.subject, request_id) saved = local_receipt(authority.db, sid) if authority.db.get_session(sid) is not None else None if saved: policy = restore_policy(saved['policy']) context = json.loads(policy.kanban_json or '{}') if (not connection.native_owner or set(params) not in ({'board', 'task_id', 'run_id', 'claim_lock'}, {'board', 'task_id', 'run_id', 'claim_lock', 'db'}) or any(params[k] != context.get(k) for k in params)): raise RuntimeStoreError('permission_denied') from gateway.session_local_recovery import restore_local_session ref = restore_local_session(authority, sid) else: policy, private = build_kanban_policy(connection, params, _load_gateway_config()) ref = create_local_session(authority, connection.actor, {'request_id': request_id}, trusted_policy=policy, trusted_secrets=private) from hermes_cli.kanban_db_connect import connect_closing from hermes_cli import kanban_db as kb import os import psutil from hermes_cli.process_identity import process_birth context = json.loads(policy.kanban_json) def admit_claim(): # The Kanban DB is the cross-profile dispatch bus (busy_timeout up to 120 s): the claim # check and its marker share one write transaction, off the owner loop. with connect_closing(Path(context['db'])) as conn, kb.write_txn(conn): # Either this exact attempt was already admitted to this session (retry after a # crash between session creation and admission), or the claim is still current # now; a reclaimed/closed task never reaches resume/submit. marker = conn.execute("SELECT payload FROM task_events WHERE task_id=? AND run_id=? AND kind='owner_admitted' ORDER BY id DESC LIMIT 1", (params['task_id'], params['run_id'])).fetchone() if marker: if json.loads(marker[0])['session_id'] != ref.session_id: raise RuntimeStoreError('stale_kanban_claim') else: task = kb.get_task(conn, params['task_id']) if not (task and task.status == 'running' and task.current_run_id == params['run_id'] and task.claim_lock == params['claim_lock']): raise RuntimeStoreError('stale_kanban_claim') kb._append_event(conn, task.id, 'owner_admitted', {'db': owner_db, 'session_id': ref.session_id, 'request_id': request_id, 'pid': os.getpid(), 'birth': process_birth(psutil.Process())}, run_id=task.current_run_id) from gateway.session_runtime_workers import tracked_write await tracked_write(authority, admit_claim) await connection.resume(ref, {}) # Finite like the `chat -q` worker it replaced: nobody can answer an approval or clarify in a # card (approvals.single_query_mode applies). Never unattended: that is `-z`'s auto-approval. # A retry of an attempt admitted before cards were finite resubmits its committed payload, # so the store returns the original receipt instead of admission_conflict. from hermes_state_runtime import find_session_admission committed = None if not saved else await asyncio.to_thread(find_session_admission, authority.db, principal_id=connection.actor.subject, session_id=ref.session_id, request_id=request_id) payload: dict = {'text': f'work kanban task {params["task_id"]}'} if committed is None or 'finite' in committed['payload']: payload['finite'] = True # The dispatcher's trusted producer marker (#134889 KB-8): nobody waits on a card attempt, so one # lost across an owner restart settles interrupted instead of pausing the card session. An # attempt main committed without it resubmits unmarked (same digest, original receipt). card = None if committed is None or 'kanban_card_v1' in committed['payload']: card = {'board': context.get('board'), 'task_id': params['task_id'], 'run_id': params['run_id']} receipt = await authority.submit(connection.actor, Submission(request_id, ref, payload, 'queue'), _kanban_card=card) return {'session_id': ref.session_id, 'receipt': asdict(receipt)} def bind_worker_context(frame): """Only the adopted private exec calls this, before tool discovery imports.""" import os context = json.loads(frame['policy'].get('kanban_json') or 'null') if context is None: return if frame['policy']['source'] != 'kanban': raise ValueError('invalid_kanban_policy') env = {'HERMES_KANBAN_TASK': context['task_id'], 'HERMES_KANBAN_BOARD': context['board'], 'HERMES_KANBAN_DB': context['db'], 'HERMES_KANBAN_RUN_ID': str(context['run_id']), 'HERMES_KANBAN_CLAIM_LOCK': context['claim_lock'], 'HERMES_KANBAN_WORKSPACE': context['workspace'], 'HERMES_KANBAN_WORKSPACES_ROOT': context['workspaces_root'], 'HERMES_PROFILE': context['profile'], 'HERMES_SESSION_SOURCE': 'kanban'} for key, value in (('HERMES_KANBAN_BRANCH', context['branch']), ('HERMES_TENANT', context['tenant'])): if value: env[key] = value if context['goal_mode']: env['HERMES_KANBAN_GOAL_MODE'] = '1' from hermes_cli.kanban_db_dispatch import _worker_terminal_timeout_env for key in ('TERMINAL_TIMEOUT', 'TERMINAL_MAX_FOREGROUND_TIMEOUT'): value = _worker_terminal_timeout_env(context['max_runtime_seconds'], os.environ.get(key)) if value: env[key] = value from hermes_cli.kanban_db_connect import connect_closing from hermes_cli import kanban_db as kb with connect_closing(Path(context['db'])) as conn: with kb.write_txn(conn): task = kb.get_task(conn, context['task_id']) if (task is None or task.status != 'running' or task.current_run_id != context['run_id'] or task.claim_lock != context['claim_lock']): raise RuntimeStoreError('stale_kanban_claim') # The dispatcher's spawned worker stays the run's pid (reclaim, timeout and operator # kills target it; ``_watch_submitter`` ties this interpreter's life to it). The bound # interpreter is recorded beside it so a sweep can tell when the attempt is truly gone. from hermes_cli.kanban_db_dispatch import UNVERIFIED_WORKER_FINGERPRINT, _process_fingerprint fingerprint = _process_fingerprint(os.getpid()) or UNVERIFIED_WORKER_FINGERPRINT kb._append_event(conn, task.id, 'worker_bound', {'pid': os.getpid(), 'claim_lock': context['claim_lock'], 'started_at': fingerprint}, run_id=context['run_id']) claimed = conn.execute("SELECT payload, created_at FROM task_events WHERE task_id=? AND run_id=? AND kind='claimed' ORDER BY id DESC LIMIT 1", (task.id, context['run_id'])).fetchone() # Heartbeats extend the claim by the TTL the dispatcher claimed it with (a spawned worker # inherited it from the dispatcher's environment); this daemon's own default is not that. expires = json.loads(claimed[0]).get('expires') if claimed else None if isinstance(expires, int) and claimed[1]: env['HERMES_KANBAN_CLAIM_TTL_SECONDS'] = str(max(1, expires - int(claimed[1]))) _watch_submitter(Path(context['db']), context['run_id']) os.environ.update(env) os.chdir(context['workspace']) _TURNS_DONE = threading.Event() def _watch_submitter(db, run_id): """The dispatcher's spawned worker is this attempt's handle: killing it (an operator, a timeout, an OOM kill) must end the attempt, as it did when the worker ran the agent itself. Exiting hard leaves no ``worker_result``, so the dead-worker sweep books this run's crash. The run row is re-read each tick: the dispatcher records the pid only after its spawn returns.""" import os from hermes_cli.kanban_db_dispatch import _worker_alive def watch(): with closing(sqlite3.connect(db.resolve().as_uri() + '?mode=ro', uri=True)) as conn: while not _TURNS_DONE.wait(1.0): try: row = conn.execute('SELECT worker_pid, worker_started_at FROM task_runs WHERE id=?', (run_id,)).fetchone() except sqlite3.Error: continue if row and row[0] and row[0] != os.getpid() and not _worker_alive(row[0], row[1]): os._exit(1) threading.Thread(target=watch, name='kanban-submitter-watch', daemon=True).start() def run_worker_turns(agent, frame, history): from gateway.session_managed_worker import worker_turn_scope from gateway.session_surface import arm_surface_turn with worker_turn_scope(frame): # The agent consumes explicit one-turn attributes, not the ContextVar: arm them exactly as # the in-process runner does (voice route marker + surface note on the user message). arm_surface_turn(agent) return _run_worker_turns(agent, frame, history) @contextlib.contextmanager def _worker_workspace(frame): """A managed worker's chat turn in its session's workspace, as the in-process turn runner binds it (#140665): the session id as the terminal task id, the route as the session key (persistent Docker keys the workspace container from it) and the frozen cwd as the session's mount. Yields the task id, or None for a frame with no session scope (nothing is bound then).""" session_id = (frame.get('scope') or {}).get('session_id') if not session_id: yield None return from gateway.session_authorities import served_profile_name from gateway.session_context import clear_session_vars, set_session_vars from gateway.session_policy import bind_turn_workspace, restore_policy from hermes_state_keys import profile_from_session_key policy = restore_policy(frame['policy']) route = frame.get('route') or session_id # The profile its in-process turns bind, so persistent Docker keys the container they and the # session's side agents use (#140665): a served secondary's name, which its routes carry # (``agent::``); none for the launch home, whatever its name (``hermes -p beta gateway`` # keys ``agent:main:``, and only the default profile runs the multiplexer). profile = served_profile_name(frame.get('home')) tokens = set_session_vars(session_key=route, session_id=session_id, profile=profile if profile_from_session_key(route) == profile else '') try: bind_turn_workspace(policy, session_id) yield session_id finally: clear_session_vars(tokens) def _run_worker_turns(agent, frame, history): context = json.loads(frame['policy'].get('kanban_json') or 'null') if context is None: from gateway.session_display import worker_display_kwargs author = frame.get('turn_author') with _worker_workspace(frame) as task_id: return agent.run_conversation(frame['text'], conversation_history=history, **worker_display_kwargs(), **({'task_id': task_id} if task_id else {}), **({'turn_author': author} if author is not None else {})) from hermes_cli.turn_exit import credential_failure_flags, turn_exit_code code, last_output = 1, '' try: result = _run_task_turns(agent, frame, history, context) code = turn_exit_code(result, kanban_worker=True) last_output = str(result.get('final_response') or '')[-500:] if isinstance(result, dict) else '' return result except Exception as exc: code = turn_exit_code(None, kanban_worker=True, **credential_failure_flags(exc)) raise finally: _record_worker_result(context, code, last_output) def record_start_failure(frame, exc): """The worker died before any turn ran (agent construction resolves credentials): book the one-shot CLI's credential exit (75 quota wall / 78 re-login / 1) instead of a bare crash.""" context = json.loads(frame['policy'].get('kanban_json') or 'null') if context is None: return from hermes_cli.turn_exit import credential_failure_flags, turn_exit_code _record_worker_result(context, turn_exit_code(None, kanban_worker=True, **credential_failure_flags(exc)), '') def _record_worker_result(context, code, last_output): _TURNS_DONE.set() import os from hermes_cli.kanban_db_connect import connect_closing from hermes_cli import kanban_db as kb with connect_closing(Path(context['db'])) as conn, kb.write_txn(conn): if _bound_worker_matches(conn, context): # Closing a run clears its claim/PID and replaces metadata; keep the # result in the immutable attempt event stream instead. kb._append_event(conn, context['task_id'], 'worker_result', {'pid': os.getpid(), 'claim_lock': context['claim_lock'], 'exit_code': code, 'last_output': last_output}, run_id=context['run_id']) def _bound_worker_matches(conn, context): import os row = conn.execute("SELECT payload FROM task_events WHERE task_id=? AND run_id=? AND kind='worker_bound' ORDER BY id DESC LIMIT 1", (context['task_id'], context['run_id'])).fetchone() bound = json.loads(row[0]) if row else {} return (bound.get('pid'), bound.get('claim_lock')) == (os.getpid(), context['claim_lock']) def _bound_claim_is_live(context): """Retry proof for a managed worker: the frame's exact run/claim lease is live and unexpired, and THIS interpreter is the one ``bind_worker_context`` bound to it. The run's worker_pid is the dispatcher's spawned submitter here, so pid identity comes from ``worker_bound`` instead.""" from agent.kanban_turn_recovery import worker_claim_is_live if not worker_claim_is_live(task_id=context['task_id'], db_path=context['db'], run_id=str(context['run_id']), claim_lock=context['claim_lock'], verify_pid=False): return False try: with closing(sqlite3.connect(Path(context['db']).resolve().as_uri() + '?mode=ro', uri=True)) as conn: return _bound_worker_matches(conn, context) except sqlite3.Error: return False def worker_result(path, params): """Read only the dispatcher's exact attempt; never infer success from admission settlement.""" with closing(sqlite3.connect(Path(path).resolve().as_uri() + '?mode=ro', uri=True)) as conn: row = conn.execute("SELECT payload FROM task_events WHERE run_id=? AND task_id=? AND kind='worker_result' ORDER BY id DESC LIMIT 1", (params['run_id'], params['task_id'])).fetchone() result = json.loads(row[0]) if row else {} return result if result.get('claim_lock') == params['claim_lock'] else {} def worker_exit_code(path, params): return worker_result(path, params).get('exit_code', 1) def _run_task_turns(agent, frame, history, context): from agent.skill_commands import build_preloaded_skills_prompt skills, loaded, missing = build_preloaded_skills_prompt(context['skills'], task_id=agent.session_id) if missing and not loaded: raise ValueError('kanban_skills_unavailable') prompt = '\n\n'.join(p for p in (skills, context['context'], frame['text']) if p) result = agent.run_conversation(prompt, conversation_history=history) def recover(nudge): nonlocal result prior = result.get('messages') if isinstance(result, dict) else None result = agent.run_conversation(nudge, conversation_history=prior or agent._session_db.get_messages_as_conversation(agent.session_id)) # Same in-place retry as the one-shot CLI worker (agent/kanban_turn_recovery.py), but the # carrier is this frame's bound claim, never the owner process's environment. from agent.kanban_turn_recovery import recover_failed_kanban_turns recover_failed_kanban_turns(recover, lambda: result, task_id=context['task_id'], claim_check=lambda: _bound_claim_is_live(context)) from hermes_cli.turn_exit import turn_exit_code # Only a settled turn may continue the goal loop: an exhausted or denied recovery reaches # the honest exit path without re-entering the model under an unprovable claim. if not context['goal_mode'] or turn_exit_code(result, kanban_worker=True) != 0: return result from hermes_cli.goals import run_kanban_goal_loop, DEFAULT_MAX_TURNS from hermes_cli import kanban_db as kb from hermes_cli.kanban_db_connect import connect_closing def status(): with connect_closing() as conn: return kb.goal_run_status(conn, context['task_id'], context['run_id']) def block(reason): with connect_closing() as conn: kb.block_task(conn, context['task_id'], reason=reason, expected_run_id=context['run_id']) def turn(prompt): nonlocal result result = agent.run_conversation(prompt, conversation_history=agent._session_db.get_messages_as_conversation(agent.session_id)) return result.get('final_response', '') run_kanban_goal_loop(task_id=context['task_id'], goal_text=context['goal_text'], run_turn=turn, task_status_fn=status, block_fn=block, max_turns=context['goal_max_turns'] or DEFAULT_MAX_TURNS, first_response=result.get('final_response', '')) return result