diff --git a/hermes_cli/kanban_db_connect.py b/hermes_cli/kanban_db_connect.py index 44afe3fc95..9817c2eaf5 100644 --- a/hermes_cli/kanban_db_connect.py +++ b/hermes_cli/kanban_db_connect.py @@ -160,20 +160,15 @@ def _cross_process_init_lock(path: Path): @contextlib.contextmanager def _dispatch_tick_lock(db_path: Path): """Non-blocking single-writer guard around one dispatcher tick; yields - ``True`` if this process holds the board's dispatch lock, else ``False`` - (caller skips the tick). + ``True`` if this process holds the board's ``.dispatch.lock``, else + ``False`` (caller skips the tick). - An orphan gateway (``gateway run --replace`` / restart escaping a systemd - or launchd service cgroup) becomes a second long-lived writer; two - dispatchers both pass ``busy_timeout`` and race on WAL frames — the root - cause of multi-writer corruption. ``_guard_supervised_gateway_conflict`` - blocks the common birth of an orphan; this lock is defense-in-depth. - - **Non-blocking** on purpose: the gateway's async watcher must never stall; - a loser retries next interval while the winner makes progress. Board-scoped - via a ``.dispatch.lock`` sibling of ``kanban.db``. Without - ``fcntl``/``msvcrt`` it degrades to a no-op (yields ``True``) — the orphan - scenario is specific to POSIX service managers. + Two dispatchers (e.g. an orphan gateway escaping its service cgroup) both + pass ``busy_timeout`` and race on WAL frames — the root cause of + multi-writer corruption; this is defense-in-depth behind + ``_guard_supervised_gateway_conflict``. Non-blocking on purpose: the + gateway's async watcher must never stall; the loser retries next interval. + Without ``fcntl``/``msvcrt`` it degrades to a no-op (yields ``True``). """ lock_path = db_path.with_name(db_path.name + ".dispatch.lock") handle = None @@ -943,17 +938,11 @@ def _backfill_legacy_inflight_runs(conn: sqlite3.Connection) -> None: ) -# Legacy DBs defined these tables with a ``TEXT PRIMARY KEY`` id (or a nullable -# ``TEXT last_event_id`` for ``kanban_notify_subs``); the current schema uses -# ``INTEGER PRIMARY KEY AUTOINCREMENT`` / ``INTEGER NOT NULL DEFAULT 0``. -# ``CREATE TABLE IF NOT EXISTS`` skips existing tables and -# ``_add_column_if_missing`` only adds columns, so a drifted column type -# requires a rebuild. -# -# Each entry pairs the canonical CREATE TABLE with the CREATE INDEX statements -# DROP TABLE would take down with it (incl. ``idx_events_run`` from the additive -# pass). ``test_rebuilt_schema_matches_fresh`` guards this list against -# drifting from SCHEMA_SQL. +# Legacy DBs used a ``TEXT PRIMARY KEY`` id (nullable ``TEXT last_event_id`` +# for ``kanban_notify_subs``); the additive migrations can't change a column +# type, so drift requires a rebuild. Each entry pairs the canonical CREATE +# TABLE with the indexes DROP TABLE takes down with it. +# ``test_rebuilt_schema_matches_fresh`` guards this against SCHEMA_SQL drift. _REBUILD_SPECS = { "task_events": ( "CREATE TABLE task_events (" @@ -1018,15 +1007,12 @@ def _table_has_drifted(conn: sqlite3.Connection, table: str) -> bool: def _rebuild_drifted_tables(conn: sqlite3.Connection) -> None: """Rebuild any kanban table whose column types drifted from SCHEMA_SQL. - Old boards crash the gateway notifier (``int(None)`` on a NULL id in - ``unseen_events_for_sub``) and never match ``id > cursor``, so every - notification is silently lost. Rebuild is CREATE new → INSERT shared - columns → DROP old → RENAME, recreating indexes (DROP TABLE takes them - down). Legacy TEXT ids are dropped; AUTOINCREMENT assigns fresh ones and - ``last_event_id`` cursors reset to 0, so the first post-migration tick - replays a task's event history once — the safe failure mode for a feature - that was already fully broken. One transaction so an interruption can't - leave a table half-renamed, under ``connect()``'s init locks. Idempotent. + Drifted boards crash the gateway notifier (``int(None)`` on a NULL id) and + never match ``id > cursor``, silently losing every notification. Legacy + TEXT ids are dropped (AUTOINCREMENT reassigns) and cursors reset to 0, so + the first post-migration tick replays history once — safe for a feature + that was already fully broken. One transaction under ``connect()``'s init + locks so an interruption can't leave a table half-renamed. Idempotent. """ drifted = [t for t in _REBUILD_SPECS if _table_has_drifted(conn, t)] if not drifted: @@ -1093,13 +1079,10 @@ def _check_file_length_invariant(conn: sqlite3.Connection) -> None: # SQLite's busy_timeout backoff is near-deterministic, so stampeding writers -# re-collide in lockstep; a jittered retry on the transaction boundary breaks -# the convoy (mirrors state.db's _execute_write: 20-150ms band, the 20ms floor -# prevents busy-spinning back into the collision). Only BEGIN IMMEDIATE and -# COMMIT are retried — idempotent re-issues that touch no transaction body, so -# a CAS inside write_txn is never replayed. Fewer retries than state.db (5 vs -# 15): the 120s busy_timeout absorbs most waits; this is the backstop for the -# tail where SQLite returns BUSY immediately. +# re-collide in lockstep; a jittered 20-150ms retry on the transaction boundary +# breaks the convoy (mirrors state.db). Only BEGIN IMMEDIATE and COMMIT are +# retried — idempotent re-issues, so a CAS inside write_txn is never replayed. +# 5 retries (not state.db's 15): the 120s busy_timeout absorbs most waits. _BUSY_MAX_RETRIES = 5 _BUSY_RETRY_MIN_S = 0.020 # 20ms _BUSY_RETRY_MAX_S = 0.150 # 150ms @@ -1125,18 +1108,14 @@ def _execute_boundary_with_retry(conn: sqlite3.Connection, sql: str) -> None: @contextlib.contextmanager def write_txn(conn: sqlite3.Connection, *, allow_nested: bool = False): - """Context manager for an IMMEDIATE write transaction; a claim CAS inside - is atomic — at most one concurrent writer succeeds. + """IMMEDIATE write transaction; a claim CAS inside is atomic — at most one + concurrent writer succeeds. - Nesting is an explicit opt-in (``allow_nested=True`` → savepoint instead - of a second ``BEGIN IMMEDIATE``; otherwise a loud ``RuntimeError``). Only - composition primitives graph builders run under one outer commit - (``create_task``, ``add_comment``) opt in — helpers with post-commit side - effects (``complete_task`` & co.) must never run under an open outer - transaction, because those side effects (workspace cleanup, ready - recomputation, failure-counter clears) would fire while the outer - transaction can still roll back. The ROLLBACK on exception is wrapped so a - SQLite auto-rollback doesn't shadow the original exception. + Nesting is an explicit opt-in (``allow_nested=True`` → savepoint; otherwise + a loud ``RuntimeError``). Only composition primitives (``create_task``, + ``add_comment``) opt in — helpers with post-commit side effects + (``complete_task`` & co.) must never run under an open outer transaction, + since those side effects would fire while the outer txn can still roll back. """ _kb._assert_not_delegated_child_mutation() if getattr(conn, "in_transaction", False): diff --git a/hermes_cli/kanban_db_dispatch.py b/hermes_cli/kanban_db_dispatch.py index 2f5d7b75c1..c83b1c4188 100644 --- a/hermes_cli/kanban_db_dispatch.py +++ b/hermes_cli/kanban_db_dispatch.py @@ -154,16 +154,11 @@ def _record_worker_exit(pid: int, raw_status: int) -> None: def _classify_worker_exit(pid: int) -> "tuple[str, Optional[int]]": - """Return ``(kind, code)`` for a reaped worker PID. - - ``clean_exit`` (status 0 while still ``running`` = protocol violation: - no ``kanban_complete``/``kanban_block``, so auto-block — retrying just - loops); ``rate_limited`` (``KANBAN_RATE_LIMIT_EXIT_CODE`` — released to - ``ready`` WITHOUT counting a failure so a long quota window can't trip the - breaker); ``nonzero_exit``; ``signaled`` (OOM/SIGKILL; ``code`` is the - signal); ``unknown`` (pid not in the reap registry — fall back to the - crashed-counter path; ``code`` is None). - """ + """``(kind, code)`` for a reaped worker PID: ``clean_exit`` (rc 0 while + still ``running`` = protocol violation), ``rate_limited`` + (``KANBAN_RATE_LIMIT_EXIT_CODE``, never counts as a failure), + ``nonzero_exit``, ``signaled`` (``code`` is the signal), ``unknown`` (pid + not in the reap registry; ``code`` None).""" entry = _recent_worker_exits.get(int(pid)) if entry is None: return ("unknown", None) @@ -387,18 +382,12 @@ def heartbeat_worker( """ now = int(time.time()) with _kb.write_txn(conn): - if expected_run_id is None: - cur = conn.execute( - "UPDATE tasks SET last_heartbeat_at = ? " - "WHERE id = ? AND status = 'running'", - (now, task_id), - ) - else: - cur = conn.execute( - "UPDATE tasks SET last_heartbeat_at = ? " - "WHERE id = ? AND status = 'running' AND current_run_id = ?", - (now, task_id, int(expected_run_id)), - ) + sql = "UPDATE tasks SET last_heartbeat_at = ? WHERE id = ? AND status = 'running'" + params: tuple = (now, task_id) + if expected_run_id is not None: + sql += " AND current_run_id = ?" + params += (int(expected_run_id),) + cur = conn.execute(sql, params) if cur.rowcount != 1: return False run_id = ( @@ -517,15 +506,12 @@ def detect_stale_running( """Reclaim ``running`` tasks with no heartbeat progress; returns their ids. Stale = running longer than ``stale_timeout_seconds`` (active run's - ``started_at``, else ``tasks.started_at``) AND ``last_heartbeat_at`` older - than ``_STALE_HEARTBEAT_GAP_SECONDS`` or NULL. Task returns to its source - phase, run closes ``outcome='stale'``, live host-local worker is terminated. - ``stale_timeout_seconds=0`` disables the check. ``signal_fn`` is a test hook. - - Deliberately does NOT call ``_record_task_failure``: stale reclaim is - dispatcher-side detection of an absent heartbeat, not a worker failure; - counting it would let legitimately long-running tasks (>4h, no explicit - heartbeat) trip the breaker. The ``stale`` event is the audit surface. + ``started_at``, else ``tasks.started_at``) AND ``last_heartbeat_at`` NULL or + older than ``_STALE_HEARTBEAT_GAP_SECONDS``. Task returns to its source + phase, run closes ``outcome='stale'``, a live host-local worker is killed. + ``0`` disables the check; ``signal_fn`` is a test hook. Deliberately NOT + counted via ``_record_task_failure``: an absent heartbeat is not a worker + failure, and counting it would let long-running tasks trip the breaker. """ if stale_timeout_seconds <= 0: return [] @@ -938,18 +924,12 @@ def _account_crashes(conn: sqlite3.Connection, crash_details: list) -> list[str] def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: """Reclaim ``running`` tasks whose worker PID is no longer alive. - Appends ``crashed`` and restores the task's source phase; unlike - ``release_stale_claims`` it checks liveness immediately rather than waiting - for the claim TTL. Only tasks claimed by *this host* — PIDs from other hosts - are meaningless, and ``_default_spawn`` always runs workers locally. - - Clean exit (rc=0) while still ``running`` is a protocol violation (no - ``kanban_complete``/``kanban_block``); it gets a bounded violation-only retry - budget (``_protocol_violation_streak``). ``KANBAN_RATE_LIMIT_EXIT_CODE`` is a - quota wall, NOT a failure: released WITHOUT counting a failure and stamped - with a quota-blocker error so ``check_respawn_guard`` defers respawn. Those - ids surface via the ``_last_rate_limited`` function attribute; the return - stays crashed-only. + Restores the source phase immediately (no waiting for the claim TTL), for + tasks claimed by *this host* only — other hosts' PIDs are meaningless. + Clean exit while ``running`` is a protocol violation with a bounded + violation-only retry budget; ``KANBAN_RATE_LIMIT_EXIT_CODE`` is a quota + wall, released WITHOUT counting a failure and surfaced via the + ``_last_rate_limited`` attribute (the return stays crashed-only). """ sweep = _reclaim_dead_workers(conn) # Outside the main txn: account each crash and maybe trip the breaker. @@ -986,20 +966,17 @@ def _record_task_failure( end_run: bool = False, event_payload_extra: Optional[dict] = None, ) -> bool: - """Record a non-success outcome (spawn_failed / crashed / timed_out) and - maybe trip the circuit breaker. Every non-success path funnels through - here so ``consecutive_failures`` stays consistent. Returns True when the - task was auto-blocked. + """Record a non-success outcome and maybe trip the circuit breaker; every + non-success path funnels through here so ``consecutive_failures`` stays + consistent. Returns True when the task was auto-blocked. - ``release_claim=True, end_run=True`` is the spawn-failure path (task still - running with an open run: restore source phase — or ``blocked`` on trip — - release the claim, close the run). ``release_claim=False, end_run=False`` - is the timeout/crash path (caller ALREADY restored the phase and closed the - run; only the counter moves, a trip re-transitions to ``blocked`` with - ``gave_up``). Threshold: per-task ``max_retries`` (nothing overrides it), - then ``failure_limit``, then ``DEFAULT_FAILURE_LIMIT``. ``force_trip=True`` - trips unconditionally — the caller already applied its own bounded-retry - policy; the failure is still counted. + ``release_claim=True, end_run=True``: spawn-failure path (task still + running with an open run — restore source phase or ``blocked``, release + claim, close run). Both False: timeout/crash path (caller already restored + the phase and closed the run; only the counter moves, a trip flips to + ``blocked`` + ``gave_up``). Threshold: per-task ``max_retries`` > + ``failure_limit`` > ``DEFAULT_FAILURE_LIMIT``. ``force_trip`` trips + unconditionally (caller applied its own bounded-retry policy). """ if failure_limit is None: failure_limit = DEFAULT_FAILURE_LIMIT @@ -1138,31 +1115,17 @@ def check_respawn_guard( ) -> Optional[str]: """Return a guard reason if ``task_id`` should NOT be re-spawned, else None. - Called per ready/review task in ``dispatch_once`` before any claim attempt; - a reason defers the spawn this tick. The review lane skips - ``recent_success`` and ``active_pr``: a recent PR comment / completed run - is the *precondition* of a review handoff, not a duplicate-work signal. - - Checks in priority order: - - ``"rate_limit_cooldown"`` — latest run ended ``rate_limited`` within - ``_resolve_rate_limit_cooldown_seconds()``. Checked BEFORE - ``blocker_auth`` because the rate-limit requeue stamps a quota-flavored - ``last_failure_error`` that would otherwise match the auth regex and - park the task forever (that path never increments - ``consecutive_failures``, so the breaker can't free it). - ``"blocker_auth"`` — ``last_failure_error`` matches a quota/auth pattern. - ``consecutive_failures`` still trips the breaker eventually, so a - persistent auth error blocks while a transient 429 gets a few ticks. - ``"recent_success"`` — completed run within ``_RESPAWN_GUARD_SUCCESS_WINDOW``. - Bypassed when a re-queue event (status, promote, unblock, reclaim) - arrives AFTER that completion — a deliberate re-run. - ``"active_pr"`` — GitHub PR URL in a comment within - ``_RESPAWN_GUARD_PR_WINDOW``; re-spawning risks a duplicate PR. - - Stale / dead claim locks are NOT a guard reason — ``release_stale_claims`` - and ``detect_crashed_workers`` reset those only after verifying the lock - is genuinely dead. + Called per ready/review row before any claim attempt. Priority order: + ``"rate_limit_cooldown"`` (latest run ``rate_limited`` within the cooldown; + checked BEFORE ``blocker_auth`` because the requeue stamps a quota-flavored + ``last_failure_error`` that would otherwise park the task forever — that + path never increments ``consecutive_failures``), ``"blocker_auth"`` + (quota/auth pattern; the breaker still trips eventually), then for the + ready lane only ``"recent_success"`` (completed run within the window, unless + a re-queue event arrived after it — a deliberate re-run) and ``"active_pr"`` + (PR URL in a recent comment; re-spawning risks a duplicate PR). The review + lane skips the last two: they are the *inputs* to a review handoff. Stale / + dead claim locks are NOT a guard reason — the reclaim passes own those. """ row = conn.execute( "SELECT last_failure_error FROM tasks WHERE id = ?", @@ -1292,16 +1255,11 @@ def review_dispatch_enabled() -> bool: return True -# --------------------------------------------------------------------------- -# Memory-aware dispatch guard -# -# With no ``kanban.max_in_progress`` on a busy board the dispatcher fanned out -# ~30 workers on a 1 GiB VM and OOM'd the host. Two safeguards: a memory-DERIVED -# default cap when the operator never set one (``resolve_max_in_progress``), and -# a live memory-PRESSURE guard inside the tick (``_memory_pressure_level``) -# because a static cap can't see other tenants. Both fail open: non-Linux or -# any read error → empty sample → no cap / "unknown". -# --------------------------------------------------------------------------- +# Memory-aware dispatch guard: an uncapped board once OOM'd a 1 GiB host. Two +# safeguards — a memory-DERIVED default cap when none is configured +# (``resolve_max_in_progress``) and a live memory-PRESSURE guard inside the +# tick (``_memory_pressure_level``) because a static cap can't see other +# tenants. Both fail open: non-Linux / read error → no cap / "unknown". # Assumed per-worker footprint for the derived cap; deliberately conservative # so the cap errs toward fewer workers on small VMs. @@ -1777,14 +1735,11 @@ def _dispatch_once_locked( max_in_progress_per_profile: Optional[int] = None, reconcile_orphans: bool = True, ) -> DispatchResult: - """Run one dispatcher tick: reclaim stale (TTL / heartbeat) and crashed - (dead host-local PID) running tasks, promote todo -> ready where all - parents are done, then for each ready task with an assignee atomically - claim and call ``spawn_fn(task, workspace_path, board) -> Optional[int]``, - recording the PID as ``worker_pid`` so later ticks catch crashes before - the TTL expires. After ``failure_limit`` consecutive failures the task is - auto-blocked. Cap semantics: :func:`_tick_spawn_budget`. - """ + """One dispatcher tick: reclaim stale/crashed running tasks, promote + todo -> ready, then atomically claim each spawnable ready/review row and + call ``spawn_fn(task, workspace_path, board) -> Optional[int]``, recording + the PID so later ticks catch crashes before the TTL. Cap semantics: + :func:`_tick_spawn_budget`.""" result = DispatchResult() _run_reclaim_phase( conn, result, stale_timeout_seconds=stale_timeout_seconds, @@ -1985,16 +1940,12 @@ def _hermes_path_argv(path: str) -> list[str]: def _resolve_hermes_argv() -> list[str]: - """Resolve the ``hermes`` invocation as argv parts for ``Popen``. - - Tries in order: ``$HERMES_BIN`` (path-like values normalized to absolute; - bare names keep PATH semantics, never a same-directory file first); - ``shutil.which("hermes")`` normalized to absolute (on Windows ``which`` can - return a relative ``.\\hermes.CMD`` when cwd is on PATH, and batch shims are - unsafe with task-derived argv, so those fall back to the module form); - ``sys.executable -m hermes_cli.main`` for setups where the shim is not on - the dispatcher's ``$PATH`` (cron, systemd ``User=`` services, launchd). - Mirrors ``gateway.run._resolve_hermes_bin``; kept local because + """Resolve the ``hermes`` invocation as argv for ``Popen``: ``$HERMES_BIN`` + (path-like -> absolute; bare names keep PATH semantics, never a + same-directory file), then ``which("hermes")`` (Windows: safe PATH search, + batch shims fall back to the module form), then ``sys.executable -m + hermes_cli.main`` for shim-less environments (cron, systemd ``User=``, + launchd). Mirrors ``gateway.run._resolve_hermes_bin``; local because ``hermes_cli`` sits below ``gateway`` in the dependency order. """ import shutil diff --git a/hermes_cli/kanban_db_notify.py b/hermes_cli/kanban_db_notify.py index 5365ebb5d0..755d6ca495 100644 --- a/hermes_cli/kanban_db_notify.py +++ b/hermes_cli/kanban_db_notify.py @@ -79,20 +79,15 @@ def add_notify_sub( delivery_metadata: Optional[Mapping[str, Any]] = None, ) -> None: """Register a gateway source wanting terminal-state notifications for - ``task_id``. Idempotent on (task, platform, chat, thread). + ``task_id``; idempotent on (task, platform, chat, thread). - ``user_id_alt`` (Signal UUID, Feishu union_id, ...) must be replayed on - active wake: ``build_session_key`` prefers it over ``user_id``, so replaying - only ``user_id`` would key the wake into a different session when they - diverge. ``chat_type`` is likewise replayed so the woken turn resolves the - operator's real channel; ``None`` keeps an existing row's value. - - ``delivery_mode`` (``_NOTIFY_DELIVERY_MODES``): ``None`` leaves an existing - row untouched (fresh rows get ``"notify"``); an explicit value is - last-write-wins so re-subscribing can change the mode; unknown values fall - back to ``"notify"``. New subs start "caught up" (``last_event_id`` = - current ``MAX(task_events.id)``, not 0) — otherwise the notifier replays - every historical terminal event on its next tick (boot-time bursts). + ``user_id_alt`` (Signal UUID, Feishu union_id, ...) and ``chat_type`` are + replayed on active wake: ``build_session_key`` prefers the alt id, so + omitting it would key the wake into a different session. ``None`` keeps an + existing row's value. ``delivery_mode``: ``None`` leaves an existing row + untouched, an explicit valid value is last-write-wins, unknown falls back + to ``"notify"``. New subs start caught up (``last_event_id`` = + ``MAX(task_events.id)``) so the notifier never replays history at boot. """ valid_mode = delivery_mode if delivery_mode in _NOTIFY_DELIVERY_MODES else None # api_server is stateless: the adapter has no send(), the wake self-post IS @@ -202,17 +197,13 @@ def count_notify_subs( chat_id: Optional[str] = None, thread_id: Optional[str] = None, ) -> int: - """Count ``kanban_notify_subs`` rows via a read-only connection. - - Cheap probe for the notifier's zero-subscription early exit: unlike - :func:`connect` it never creates the DB file, runs schema init/migration, - or opens writable (a read-only WAL open may still create ``-shm``/``-wal`` - sidecars but cannot write table content). Rows in a not-yet-checkpointed - WAL are visible, so a fresh subscription is never missed. A missing DB or - a legacy DB without the table counts as zero. Platform matching is - case-insensitive (matching notifier routing); chat/thread are exact. - Path resolution matches :func:`connect`. Raises :class:`sqlite3.Error` - when the DB exists but cannot be read; callers choose their own fallback. + """Count ``kanban_notify_subs`` rows via a read-only connection — the + notifier's cheap zero-subscription early exit. Unlike :func:`connect` it + never creates the file, runs init/migration or opens writable; WAL rows are + still visible so a fresh sub is never missed. Missing DB / missing table + counts as zero; platform matches case-insensitively (as notifier routing), + chat/thread exactly. Raises :class:`sqlite3.Error` if the DB exists but is + unreadable — callers pick their own fallback. """ path = db_path if db_path is not None else _kb.kanban_db_path(board=board) if not path.exists(): @@ -266,17 +257,14 @@ def remove_notify_sub( def purge_stale_done_notify_subs(conn: sqlite3.Connection, *, max_age_days: int = 30) -> int: - """Delete notify subs whose task has sat in ``done``/``blocked`` untouched - for longer than ``max_age_days``. + """Delete notify subs whose task sat in ``done``/``blocked`` untouched for + longer than ``max_age_days`` (``<= 0`` disables); returns rows deleted. - Subs survive ``done`` because a completed task can be reopened and must - still notify its origin; on never-archiving boards that would accumulate - rows forever, each scanned every notifier tick. ``blocked`` tasks are - reaped on the same clock — they are abandoned, not merely waiting like - ``backlog``/``ready``. Age is measured from the most recent event - (falling back to ``completed_at`` then ``created_at``), so ANY activity — - including a reopen — resets or exempts it. ``max_age_days <= 0`` disables - the sweep. Returns the number of rows deleted. + Subs survive ``done`` because a reopened task must still notify its origin, + which accumulates forever on never-archiving boards. ``blocked`` is + abandoned (unlike ``backlog``/``ready``) so it reaps on the same clock. Age + = latest event, else ``completed_at``, else ``created_at`` — any activity, + including a reopen, exempts the sub. """ try: days = int(max_age_days) diff --git a/hermes_cli/kanban_db_workspace.py b/hermes_cli/kanban_db_workspace.py index 2511c2d1e2..e38ee463a9 100644 --- a/hermes_cli/kanban_db_workspace.py +++ b/hermes_cli/kanban_db_workspace.py @@ -95,19 +95,15 @@ def _scratch_workspace(conn: sqlite3.Connection, task_id: str) -> Optional[Path] def _is_managed_scratch_path(p: Path) -> bool: - """Return True iff *p* is a strict descendant of a kanban-managed scratch root. - - A managed root is exclusively a ``workspaces/`` directory: - ``HERMES_KANBAN_WORKSPACES_ROOT`` (dispatcher-injected worker override), - ``/kanban/workspaces`` (legacy default board), or - ``/kanban/boards//workspaces`` per on-disk board. - Strict descendancy: a path EQUAL to a root is not managed (deleting it - would wipe every task's scratch dir), and ``/kanban``, - ``.../logs`` or ``.../boards/`` are rejected because they hold - Hermes' own DB, metadata and logs. :func:`_cleanup_workspace` uses this to - refuse ``shutil.rmtree`` outside managed storage — a board - ``default_workdir`` pointing at a real source tree can otherwise pair with - ``workspace_kind='scratch'`` and make task completion delete user data. + """True iff *p* is a STRICT descendant of a kanban-managed ``workspaces/`` + root (``HERMES_KANBAN_WORKSPACES_ROOT``, ``/kanban/workspaces``, + or ``/kanban/boards//workspaces``). A path equal to a + root is not managed (deleting it would wipe every task's scratch dir); + ``/kanban``, ``.../logs`` and ``.../boards/`` hold + Hermes' own DB and metadata. :func:`_cleanup_workspace` refuses + ``rmtree`` outside managed storage — a board ``default_workdir`` on a real + source tree paired with ``workspace_kind='scratch'`` would otherwise make + task completion delete user data. """ return _managed_scratch_path_info(p)[0] @@ -488,19 +484,13 @@ def _resolve_worktree_workspace(task: Task, *, board: Optional[str] = None) -> t def resolve_workspace(task: Task, *, board: Optional[str] = None) -> Path: """Resolve (and create if needed) the workspace for a task. - - ``scratch``: fresh dir under ``/workspaces//`` — the - same path for the dispatcher and every profile worker, so handoff is - path-stable. - - ``dir:``: ``workspace_path``, created if missing. MUST be absolute - — relative paths are rejected to prevent confused-deputy traversal where - ``../../../tmp/attacker`` resolves against the dispatcher's CWD. - - ``worktree``: a real linked git worktree. A ``workspace_path`` naming a - repo root is an anchor (materializes ``/.worktrees/``); - a concrete target path is created/reused; with none, the board's - ``default_workdir`` anchors it and raises if unset rather than guessing - from the dispatcher's CWD. Empty ``branch_name`` -> ``wt/``. - - Persist the resolved path via ``set_workspace_path`` so later runs reuse it. + ``scratch``: ``/workspaces//`` — path-stable across the + dispatcher and every profile worker. ``dir``: ``workspace_path``, created + if missing; MUST be absolute (relative paths would resolve against the + dispatcher's CWD — confused-deputy traversal). ``worktree``: a linked git + worktree; a repo-root ``workspace_path`` anchors ``/.worktrees/``, + a concrete path is created/reused, none -> the board's ``default_workdir`` + (raises if unset rather than guessing). Persist via ``set_workspace_path``. """ kind = task.workspace_kind or "scratch" if kind == "worktree": diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 6933f609a2..501072a24c 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -949,8 +949,7 @@ def _read_git_revision_fingerprint(repo_root: Path) -> str | None: common_dir = (git_dir / rel).resolve() except OSError: pass - head_file = git_dir / "HEAD" - head = head_file.read_text(encoding="utf-8", errors="replace").strip() + head = (git_dir / "HEAD").read_text(encoding="utf-8", errors="replace").strip() if head.startswith("ref:"): ref = head.split(":", 1)[1].strip() # Loose refs may live in the worktree gitdir OR the common dir @@ -1035,6 +1034,44 @@ def _termux_should_prefetch_update_check() -> bool: return os.environ.get("HERMES_TERMUX_PREFETCH_UPDATES") == "1" +def _dotenv_has_provider_key(env_file: Path, provider_env_vars: set) -> bool: + """True if ~/.hermes/.env assigns a non-empty value to any provider key.""" + if not env_file.exists(): + return False + try: + for line in env_file.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if line.startswith("#") or "=" not in line: + continue + if line.startswith("export "): + line = line[7:] + key, _, val = line.partition("=") + if key.strip() in provider_env_vars and val.strip().strip("'\""): + return True + except Exception: + pass + return False + + +def _auth_store_logged_in(auth_file: Path, registry, strict_profile_scope: bool) -> bool: + """True if auth.json's active provider is logged in (api_key providers ignored under strict scope).""" + from hermes_cli.auth import get_auth_status + + if not auth_file.exists(): + return False + try: + auth = json.loads(auth_file.read_text(encoding="utf-8-sig")) + active = auth.get("active_provider") + active_config = registry.get(str(active or "").strip().lower()) + if active and not ( + strict_profile_scope and active_config and active_config.auth_type == "api_key" + ): + return bool(get_auth_status(active).get("logged_in")) + except Exception: + pass + return False + + def _has_any_provider_configured(*, strict_profile_scope: bool = False) -> bool: """Check if at least one inference provider is usable. @@ -1043,13 +1080,8 @@ def _has_any_provider_configured(*, strict_profile_scope: bool = False) -> bool: env and host-wide fallbacks (gh auth, Claude Code credentials) must not make it appear ready. Unscoped callers keep the legacy behavior. """ - from hermes_cli.config import get_env_path, get_hermes_home, load_config - from hermes_cli.auth import get_auth_status - - # "Explicitly configured" = model differs from the hardcoded default; gates - # external tool credentials (Claude Code) so they don't skip setup on a - # fresh install. - from hermes_cli.config import DEFAULT_CONFIG + from hermes_cli.config import DEFAULT_CONFIG, get_env_path, get_hermes_home, load_config + from hermes_cli.auth import PROVIDER_REGISTRY, get_auth_status cfg = load_config() model_cfg = cfg.get("model") @@ -1060,12 +1092,12 @@ def _has_any_provider_configured(*, strict_profile_scope: bool = False) -> bool: from hermes_cli.config import split_model_config_default _model_name, _ = split_model_config_default(_model_name) _model_name = str(_model_name).strip() + # "Explicitly configured" = model differs from the hardcoded default; gates + # Claude Code credentials so they don't skip setup on a fresh install. _has_hermes_config = _model_name and _model_name != DEFAULT_CONFIG.get("model", "") # Env vars (.env or shell). OPENAI_BASE_URL alone counts — local models # (vLLM, llama.cpp) often need no API key. - from hermes_cli.auth import PROVIDER_REGISTRY - provider_env_vars = { "OPENROUTER_API_KEY", "OPENAI_API_KEY", @@ -1084,60 +1116,31 @@ def _has_any_provider_configured(*, strict_profile_scope: bool = False) -> bool: read_provider_env = os.getenv if any(read_provider_env(v) for v in provider_env_vars): return True - - # Check .env file for keys - env_file = get_env_path() - if env_file.exists(): - try: - for line in env_file.read_text(encoding="utf-8").splitlines(): - line = line.strip() - if line.startswith("#") or "=" not in line: - continue - if line.startswith("export "): - line = line[7:] - key, _, val = line.partition("=") - val = val.strip().strip("'\"") - if key.strip() in provider_env_vars and val: - return True - except Exception: - pass + if _dotenv_has_provider_key(get_env_path(), provider_env_vars): + return True # Cheap on-disk checks (auth.json, config.yaml) first: the PROVIDER_REGISTRY # sweep below spawns subprocesses (gh) and can take 15-20s — long enough # that desktop setup.status calls time out. - auth_file = get_hermes_home() / "auth.json" - if auth_file.exists(): - try: - auth = json.loads(auth_file.read_text(encoding="utf-8-sig")) - active = auth.get("active_provider") - active_config = PROVIDER_REGISTRY.get(str(active or "").strip().lower()) - if active and not ( - strict_profile_scope and active_config and active_config.auth_type == "api_key" - ): - status = get_auth_status(active) - if status.get("logged_in"): - return True - except Exception: - pass + if _auth_store_logged_in(get_hermes_home() / "auth.json", PROVIDER_REGISTRY, strict_profile_scope): + return True # model as a dict with provider/base_url/api_key means setup ran (fresh # installs have a plain string); also covers custom endpoints kept in config. - if isinstance(model_cfg, dict): - cfg_provider = (model_cfg.get("provider") or "").strip() - cfg_base_url = (model_cfg.get("base_url") or "").strip() - cfg_api_key = (model_cfg.get("api_key") or "").strip() - if cfg_provider or cfg_base_url or cfg_api_key: - return True + if isinstance(model_cfg, dict) and any( + (model_cfg.get(k) or "").strip() for k in ("provider", "base_url", "api_key") + ): + return True # Provider-specific auth fallbacks (e.g. Copilot via gh auth). if not strict_profile_scope: try: - for provider_id, pconfig in PROVIDER_REGISTRY.items(): - if pconfig.auth_type != "api_key": - continue - status = get_auth_status(provider_id) - if status.get("logged_in"): - return True + if any( + get_auth_status(pid).get("logged_in") + for pid, pconfig in PROVIDER_REGISTRY.items() + if pconfig.auth_type == "api_key" + ): + return True except Exception: pass @@ -1210,22 +1213,15 @@ def _confirm_startup_expensive_model_override(args) -> None: # Intentionally independent of --yolo / --accept-hooks: those approve local # command risk, not paid aggregator spend or a surprising provider route. is_interactive = sys.stdin.isatty() - allow_unattended_data_training = ( - security_cfg.get("allow_data_training_tiers_noninteractive") is True - ) - if not is_interactive and allow_unattended_data_training: - acknowledged = [ - warning for warning in warnings if warning.kind == "data_policy" - ] + if not is_interactive and security_cfg.get("allow_data_training_tiers_noninteractive") is True: + acknowledged = [w for w in warnings if w.kind == "data_policy"] if acknowledged: sys.stderr.write(combined_message(acknowledged) + "\n") sys.stderr.write( "Proceeding in non-interactive mode because " "security.allow_data_training_tiers_noninteractive is true.\n" ) - warnings = [ - warning for warning in warnings if warning.kind != "data_policy" - ] + warnings = [w for w in warnings if w.kind != "data_policy"] if not warnings: return @@ -1541,6 +1537,47 @@ def _resolve_continue_arg(args, *, use_tui: bool) -> None: sys.exit(1) +def _apply_in_dir(args) -> None: + """--in DIR: chdir first so workspace-scoped lookups key off DIR; pins the session there.""" + in_dir = getattr(args, "in_dir", None) + if not in_dir: + return + # Git Bash / MSYS hands us POSIX-style paths (`--in ~` → `/c/Users/x`); + # translate drive-root spellings to native Windows form. No-op elsewhere. + from tools.environments.local import _msys_to_windows_path + + _target_dir = os.path.abspath(os.path.expanduser(_msys_to_windows_path(in_dir))) + if not os.path.isdir(_target_dir): + print(f"Error: --in directory not found: {in_dir}") + sys.exit(1) + try: + os.chdir(_target_dir) + except OSError as e: + print(f"Error: cannot enter --in directory {in_dir}: {e}") + sys.exit(1) + args.no_restore_cwd = True + + +def _import_foreign_resume(args) -> None: + """--resume @claude / @codex: import a foreign session and resume it.""" + _resume_foreign = getattr(args, "resume", None) + if not (isinstance(_resume_foreign, str) and _resume_foreign.strip().lower() in ("@claude", "@codex")): + return + from hermes_cli.foreign_sessions import import_foreign_session, pick_foreign_session + + _picked = pick_foreign_session(_resume_foreign.strip().lower().lstrip("@")) + if _picked is None: + sys.exit(1) + try: + _imported_id = import_foreign_session(_picked.source, _picked.path) + except ValueError as e: + print(f"Error: {e}") + sys.exit(1) + print(f"✓ Imported as {_imported_id} — resuming it now.") + print(f" (later: hermes --resume {_imported_id})") + args.resume = _imported_id + + def _resolve_chat_session_args(args, use_tui: bool) -> None: """Normalize --in / --resume / --continue on ``args`` before agent init. @@ -1551,24 +1588,7 @@ def _resolve_chat_session_args(args, use_tui: bool) -> None: cd back into a resumed session's recorded cwd (best-effort, opt-out via --no-restore-cwd, skipped under --worktree). """ - in_dir = getattr(args, "in_dir", None) - if in_dir: - # Git Bash / MSYS hands us POSIX-style paths (`--in ~` → `/c/Users/x`); - # translate drive-root spellings to native Windows form. No-op elsewhere. - from tools.environments.local import _msys_to_windows_path - - _target_dir = os.path.abspath( - os.path.expanduser(_msys_to_windows_path(in_dir)) - ) - if not os.path.isdir(_target_dir): - print(f"Error: --in directory not found: {in_dir}") - sys.exit(1) - try: - os.chdir(_target_dir) - except OSError as e: - print(f"Error: cannot enter --in directory {in_dir}: {e}") - sys.exit(1) - args.no_restore_cwd = True + _apply_in_dir(args) # --resume latest: same resolution as bare `-c`. The keyword wins over a # session literally titled "latest" (still reachable by ID or `-c latest`). @@ -1585,29 +1605,7 @@ def _resolve_chat_session_args(args, use_tui: bool) -> None: _resolve_continue_arg(args, use_tui=use_tui) - # --resume @claude / @codex: import a foreign session and resume it. - _resume_foreign = getattr(args, "resume", None) - if isinstance(_resume_foreign, str) and _resume_foreign.strip().lower() in ( - "@claude", - "@codex", - ): - from hermes_cli.foreign_sessions import ( - import_foreign_session, - pick_foreign_session, - ) - - _foreign_source = _resume_foreign.strip().lower().lstrip("@") - _picked = pick_foreign_session(_foreign_source) - if _picked is None: - sys.exit(1) - try: - _imported_id = import_foreign_session(_picked.source, _picked.path) - except ValueError as e: - print(f"Error: {e}") - sys.exit(1) - print(f"✓ Imported as {_imported_id} — resuming it now.") - print(f" (later: hermes --resume {_imported_id})") - args.resume = _imported_id + _import_foreign_resume(args) resume_val = getattr(args, "resume", None) if resume_val: @@ -2185,7 +2183,7 @@ _LAZY_COMMAND_EXPORTS = { "_stop_process_trees", "_sync_with_upstream_if_needed", "_update_node_dependencies", "_update_via_zip", "_upgrade_pip_before_lazy_refresh", "_validate_critical_files_syntax", "_validate_critical_modules_import", "_venv_launcher_ancestors", - "_wait_for_windows_update_gateway_exit", + "_wait_for_windows_update_gateway_exit", "_warn_orphaned_update_autostashes", "_warn_pending_fleet_restart_on_startup", "_write_marker_file", "_write_update_incomplete_marker", ), @@ -2350,16 +2348,13 @@ def _finalize_update_receipt(code: int, reason: str) -> None: pass -def cmd_update(args): - """Update Hermes Agent: hangup protection + update lock around ``_cmd_update_impl``.""" - from hermes_cli.config import ( - is_managed, - managed_error, - ) +def _update_preflight_handled(args) -> bool: + """Managed-install refusal, --plan, admission gate, --check. True = nothing more to do.""" + from hermes_cli.config import is_managed, managed_error if is_managed(): managed_error("update Hermes Agent") - return + return True # --plan is read-only and deployment-kind aware, so it runs BEFORE the # docker/nix/apt refusal gates: on an image/package-managed install the @@ -2371,7 +2366,7 @@ def cmd_update(args): ) print_update_plan(collect_runtime_inventory()) - return + return True # Image/package-managed admission gate: baked provenance marker first # (fail-closed on malformed), then docker/nix/apt heuristics. Records a @@ -2394,8 +2389,14 @@ def cmd_update(args): branch=branch, branch_explicit=bool(getattr(args, "branch", None)), ) - return + return True + return False + +def cmd_update(args): + """Update Hermes Agent: hangup protection + update lock around ``_cmd_update_impl``.""" + if _update_preflight_handled(args): + return gateway_mode = getattr(args, "gateway", False) _update_io_state = _install_hangup_protection(gateway_mode=gateway_mode)