From a991dfc25daf68994c21d6adcdfbafb1b3dc23cf Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Mon, 3 Aug 2026 12:07:53 -0700 Subject: [PATCH 1/2] docs: document /personality none|default|neutral reset across personality docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reset keywords have existed in both CLI and gateway handlers since June but were undocumented — users couldn't find how to cancel a personality overlay. Adds a 'Resetting to the default' section to the personality feature page and mentions the reset in the CLI guide, slash-command reference (both tables), and messaging command table. --- website/docs/reference/slash-commands.md | 4 ++-- website/docs/user-guide/cli.md | 2 ++ website/docs/user-guide/features/personality.md | 12 ++++++++++++ website/docs/user-guide/messaging/index.md | 2 +- 4 files changed, 17 insertions(+), 3 deletions(-) diff --git a/website/docs/reference/slash-commands.md b/website/docs/reference/slash-commands.md index ae38e858f6a12..860d2c8d1a9eb 100644 --- a/website/docs/reference/slash-commands.md +++ b/website/docs/reference/slash-commands.md @@ -74,7 +74,7 @@ Type `/` in the CLI to open the autocomplete menu. Built-in commands are case-in | `/config` | Show current configuration | | `/model [model-name]` | Show or change the current model. Supports: `/model claude-sonnet-4`, `/model provider:model` (switch providers), `/model custom:model` (custom endpoint), `/model custom:name:model` (named custom provider), `/model custom` (auto-detect from endpoint), and user-defined aliases (`/model fav`, `/model grok` — see [Custom model aliases](#custom-model-aliases)). Flags: `--global` persists the change to config.yaml; `--session` forces session-only; `--once` applies to the next turn only; `--refresh` re-fetches the provider's model list; `--provider ` switches backend (session-only unless `--global`). A plain `/model ` is session-only unless `model.persist_switch_by_default: true` is set. **Note:** `/model` can only switch between already-configured providers. To add a new provider, exit the session and run `hermes model` from your terminal. **Cost note:** switching models mid-conversation resets the prompt cache — the cache key includes the model, so your next turn re-reads the entire conversation at full input price instead of the ~75%-discounted cached rate. Expected and unavoidable, but worth knowing on long sessions. | | `/codex-runtime [auto\|codex_app_server\|on\|off]` | Toggle the optional [Codex app-server runtime](../user-guide/features/codex-app-server-runtime) for OpenAI/Codex models. `auto` (default) uses Hermes' standard chat completions; `codex_app_server` hands turns to a `codex app-server` subprocess for native shell, apply_patch, ChatGPT subscription auth, and migrated Codex plugins. Effective on next session. | -| `/personality` | Set a predefined personality | +| `/personality` | Set a predefined personality. `/personality none` (or `default` / `neutral`) clears the overlay and returns to base behavior. | | `/verbose` | Cycle tool progress display: off → new → all → verbose. Can be [enabled for messaging](#notes) via config. | | `/focus [on\|off\|status]` | Toggle **focus view** — a display-only reduced-output mode showing just your prompt and the final response. Composes with `/verbose`: turning it on snaps tool progress to `off` and remembers your previous mode, and `/focus off` restores it. Each turn ends with a dim recovery line (`⋯ 7 tool lines hidden · /focus off to show`) and a persistent `◉ focus` badge sits in the status bar so you always know you're in the reduced view. Nothing is sent differently to the model — detail is hidden, never discarded. | | `/fast [normal\|fast\|status]` | Toggle fast mode — OpenAI Priority Processing / Anthropic Fast Mode. Options: `normal`, `fast`, `status`. | @@ -226,7 +226,7 @@ The messaging gateway supports the following built-in commands inside Telegram, | `/stop` | Kill all running background processes and interrupt the running agent. | | `/model [provider:model]` | Show or change the model. Supports provider switches (`/model zai:glm-5`), custom endpoints (`/model custom:model`), named custom providers (`/model custom:local:qwen`), auto-detect (`/model custom`), and user-defined aliases (`/model fav`, `/model grok` — see [Custom model aliases](#custom-model-aliases)). Use `--global` to persist the change to config.yaml. **Note:** `/model` can only switch between already-configured providers. To add a new provider or set up API keys, use `hermes model` from your terminal (outside the chat session). **Cost note:** a mid-session model switch resets the prompt cache (the cache key includes the model), so the next message re-reads the whole conversation at full input price. | | `/codex-runtime [auto\|codex_app_server\|on\|off]` | Toggle the optional [Codex app-server runtime](../user-guide/features/codex-app-server-runtime). Persists to `model.openai_runtime` in config.yaml and evicts the cached agent so the next message picks up the new runtime. Effective on next session. | -| `/personality [name]` | Set a personality overlay for the session. | +| `/personality [name]` | Set a personality overlay for the session. `/personality none` (or `default` / `neutral`) clears it. | | `/fast [normal\|fast\|status]` | Toggle fast mode — OpenAI Priority Processing / Anthropic Fast Mode. | | `/retry` | Retry the last message. | | `/undo` | Remove the last exchange. | diff --git a/website/docs/user-guide/cli.md b/website/docs/user-guide/cli.md index 260ac47fc4107..a6b9356c453b2 100644 --- a/website/docs/user-guide/cli.md +++ b/website/docs/user-guide/cli.md @@ -223,6 +223,8 @@ Set a predefined personality to change the agent's tone: Built-in personalities include: `helpful`, `concise`, `technical`, `creative`, `teacher`, `kawaii`, `catgirl`, `pirate`, `shakespeare`, `surfer`, `noir`, `uwu`, `philosopher`, `hype`. +To go back to the default (no overlay), use `/personality none` — `default` and `neutral` work too. + You can also define custom personalities in `~/.hermes/config.yaml`: ```yaml diff --git a/website/docs/user-guide/features/personality.md b/website/docs/user-guide/features/personality.md index 14b26e4451605..cfae4abdf9207 100644 --- a/website/docs/user-guide/features/personality.md +++ b/website/docs/user-guide/features/personality.md @@ -227,6 +227,18 @@ Then switch to it with: /personality codereviewer ``` +## Resetting to the default + +To cancel the active personality overlay and return to base behavior (your `SOUL.md` persona), use any of: + +```text +/personality none +/personality default +/personality neutral +``` + +All three clear the overlay: the saved `agent.system_prompt` is emptied and the change takes effect on your next message. Running `/personality` with no arguments also lists `none` alongside the available presets. + ## Recommended workflow A strong default setup is: diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index 8cf1491ad329b..61c47d5a421ca 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -192,7 +192,7 @@ platform network disconnect as an event-loop failure. |---------|-------------| | `/new` or `/reset` | Start a fresh conversation | | `/model [provider:model]` | Show or change the model (supports `provider:model` syntax) | -| `/personality [name]` | Set a personality | +| `/personality [name]` | Set a personality (`none` to reset) | | `/retry` | Retry the last message | | `/undo` | Remove the last exchange | | `/status` | Show session info | From 197ce4ff3edc8cd7ce553cac93b4428c146de75b Mon Sep 17 00:00:00 2001 From: ChanPark03 Date: Tue, 4 Aug 2026 04:57:28 +0900 Subject: [PATCH 2/2] fix: harden kanban dispatch cleanup and fresh triage selection --- gateway/kanban_watchers.py | 4 +- hermes_cli/kanban_db.py | 188 ++++++++----- hermes_cli/kanban_decompose.py | 13 + tests/hermes_cli/test_kanban_db.py | 304 ++++++++++++++++++++++ tests/hermes_cli/test_kanban_decompose.py | 231 ++++++++++++++++ 5 files changed, 679 insertions(+), 61 deletions(-) diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index dc2c5efd9011a..c69cdb37c4c0d 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -1370,10 +1370,10 @@ def _auto_decompose_tick(auto_decompose_per_tick: int) -> int: try: os.environ["HERMES_KANBAN_BOARD"] = slug try: - triage_ids = _decomp.list_triage_ids() + triage_ids = _decomp.list_auto_decompose_triage_ids() except Exception as exc: logger.debug( - "kanban auto-decompose: list_triage_ids failed on board %s (%s)", + "kanban auto-decompose: fresh triage selection failed on board %s (%s)", slug, exc, ) triage_ids = [] diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 113e34842ec54..77b3e9ec1fefc 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -3365,6 +3365,7 @@ def list_tasks( order_by: Optional[str] = None, workflow_template_id: Optional[str] = None, current_step_key: Optional[str] = None, + exclude_block_provenance: bool = False, ) -> list[Task]: query = "SELECT * FROM tasks WHERE 1=1" params: list[Any] = [] @@ -3388,6 +3389,8 @@ def list_tasks( if current_step_key is not None: query += " AND current_step_key = ?" params.append(current_step_key) + if exclude_block_provenance: + query += " AND block_kind IS NULL AND COALESCE(block_recurrences, 0) = 0" if not include_archived and status != "archived": query += " AND status != 'archived'" if order_by is not None: @@ -4628,6 +4631,12 @@ def reclaim_task( termination = _terminate_reclaimed_worker( row["worker_pid"], prev_lock, signal_fn=signal_fn, ) + if _worker_survived_termination(termination): + _defer_reclaim_for_live_worker( + conn, task_id, prev_lock, int(time.time()), termination, + reason="manual_reclaim_worker_alive", + ) + return False with write_txn(conn): cur = conn.execute( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " @@ -7020,7 +7029,7 @@ def _terminate_reclaimed_worker( *, signal_fn=None, ) -> dict[str, Any]: - """Best-effort host-local worker termination for reclaim paths.""" + """Best-effort host-local worker-group termination for retry paths.""" import signal info: dict[str, Any] = { @@ -7028,6 +7037,8 @@ def _terminate_reclaimed_worker( "host_local": False, "termination_attempted": False, "terminated": False, + "termination_blocked": None, + "process_group": False, "sigkill": False, } if not pid or pid <= 0 or not claim_lock: @@ -7037,10 +7048,58 @@ def _terminate_reclaimed_worker( if not str(claim_lock).startswith(host_prefix): return info info["host_local"] = True + if int(pid) <= 1: + info["termination_blocked"] = "unsafe_pid" + return info kill = signal_fn if signal_fn is not None else ( os.kill if hasattr(os, "kill") else None ) + alive = _pid_alive + process_group = False + if ( + signal_fn is None + and os.name != "nt" + and hasattr(os, "killpg") + and hasattr(os, "getpgid") + and hasattr(os, "getpgrp") + ): + if int(pid) == os.getpgrp(): + info["termination_blocked"] = "dispatcher_process_group" + return info + try: + current_pgid = os.getpgid(int(pid)) + except ProcessLookupError: + current_pgid = int(pid) + except OSError: + info["termination_blocked"] = "pgid_lookup_failed" + return info + if current_pgid != int(pid): + info["termination_blocked"] = "unexpected_pgid" + return info + try: + os.killpg(int(pid), 0) # windows-footgun: ok — POSIX-only branch + except ProcessLookupError: + info["terminated"] = True + return info + except (PermissionError, OSError): + info["termination_blocked"] = "pgid_probe_failed" + return info + + process_group = True + info["process_group"] = True + + def kill(_pid, sig): + os.killpg(int(pid), sig) # windows-footgun: ok — POSIX-only branch + + def alive(_pid): + try: + os.killpg(int(pid), 0) # windows-footgun: ok — POSIX-only branch + return True + except ProcessLookupError: + return False + except (PermissionError, OSError): + return True if kill is None: return info @@ -7057,38 +7116,56 @@ def _terminate_reclaimed_worker( return info for _ in range(10): - if not _pid_alive(pid): + if not alive(pid): info["terminated"] = True return info time.sleep(0.5) - if _pid_alive(pid): + if alive(pid): try: # signal.SIGKILL doesn't exist on Windows; fall back to SIGTERM # (which maps to TerminateProcess via the stdlib shim). _sigkill = getattr(signal, "SIGKILL", signal.SIGTERM) kill(int(pid), _sigkill) info["sigkill"] = True - except (ProcessLookupError, OSError): + except ProcessLookupError: + info["terminated"] = True + return info + except OSError: return info - info["terminated"] = not _pid_alive(pid) + if process_group: + for _ in range(10): + if not alive(pid): + info["terminated"] = True + return info + time.sleep(0.05) + # A successful SIGKILL cannot be handled or ignored. The kernel may + # retain a dead group briefly while zombies are reaped, but none of + # its members can overlap the retry. + info["terminated"] = info["sigkill"] + return info + + info["terminated"] = not alive(pid) return info def _worker_survived_termination(termination: dict) -> bool: - """True when we tried to kill our own host-local worker and it is still alive. + """True when a host-local worker may still execute after termination. Reclaiming in this state would release the claim and let the dispatcher spawn a second worker while the first is still running — the duplication - loop. Only host-local workers we actually signalled count: a non-local - claim lock or a no-op attempt (no ``os.kill`` available) must fall through - to the normal release path, since we cannot manage that worker anyway. + loop. A refused unsafe PID/PGID target also counts so we fail closed rather + than signal a foreign group or overlap it with a retry. Non-local claims + and platforms without a signal primitive preserve the prior release path. """ return bool( - termination.get("termination_attempted") - and termination.get("host_local") + termination.get("host_local") and not termination.get("terminated") + and ( + termination.get("termination_attempted") + or termination.get("termination_blocked") + ) ) @@ -7201,7 +7278,6 @@ def enforce_max_runtime( (same reasoning as ``detect_crashed_workers``). ``signal_fn`` is a test hook; defaults to ``os.kill`` on POSIX. """ - import signal timed_out: list[str] = [] now = int(time.time()) host_prefix = f"{_claimer_id().split(':', 1)[0]}:" @@ -7229,31 +7305,16 @@ def enforce_max_runtime( pid = int(row["worker_pid"]) tid = row["id"] - # SIGTERM then SIGKILL. Keep it simple: 5 s grace. Workers that - # want a cleaner shutdown can install their own SIGTERM handler - # before the grace expires. - killed = False - kill = signal_fn if signal_fn is not None else ( - os.kill if hasattr(os, "kill") else None + termination = _terminate_reclaimed_worker( + pid, row["claim_lock"], signal_fn=signal_fn, ) - if kill is not None: - try: - kill(pid, signal.SIGTERM) - except (ProcessLookupError, OSError): - pass - # Short polling wait — no time.sleep on the write txn. - for _ in range(10): - if not _pid_alive(pid): - break - time.sleep(0.5) - if _pid_alive(pid): - try: - # signal.SIGKILL doesn't exist on Windows. - _sigkill = getattr(signal, "SIGKILL", signal.SIGTERM) - kill(pid, _sigkill) - killed = True - except (ProcessLookupError, OSError): - pass + if _worker_survived_termination(termination): + _defer_reclaim_for_live_worker( + conn, tid, row["claim_lock"], now, termination, + reason="max_runtime_worker_alive", + ) + continue + killed = bool(termination.get("sigkill")) with write_txn(conn): cur = conn.execute( @@ -7545,37 +7606,45 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: """ crashed: list[str] = [] rate_limited: list[str] = [] - # Per-crash details collected inside the main txn, used after it - # closes to run ``_record_task_failure`` (which needs its own + # Per-crash details collected inside per-task txns, used after they + # close to run ``_record_task_failure`` (which needs its own # write_txn so can't nest). ``protocol_violation`` flags the # clean-exit-but-still-running case, which is accounted against its # own bounded violation streak instead of the unified failure # counter (see the post-txn loop below). crash_details: list[tuple[str, int, str, bool, str]] = [] # (task_id, pid, claimer, protocol_violation, error_text) - with write_txn(conn): - rows = conn.execute( - "SELECT id, worker_pid, claim_lock, started_at FROM tasks " - "WHERE status = 'running' AND worker_pid IS NOT NULL" - ).fetchall() - host_prefix = f"{_claimer_id().split(':', 1)[0]}:" - for row in rows: - # Only check liveness for claims owned by this host. - lock = row["claim_lock"] or "" - if not lock.startswith(host_prefix): - continue - # Skip liveness check inside the launch-window grace period - # so a freshly-spawned worker isn't reclaimed before its PID - # is visible on /proc. - started_at = row["started_at"] if "started_at" in row.keys() else None - if started_at is not None: - grace = _resolve_crash_grace_seconds() - if time.time() - started_at < grace: - continue - if _pid_alive(row["worker_pid"]): + rows = conn.execute( + "SELECT id, worker_pid, claim_lock, started_at FROM tasks " + "WHERE status = 'running' AND worker_pid IS NOT NULL" + ).fetchall() + host_prefix = f"{_claimer_id().split(':', 1)[0]}:" + for row in rows: + # Only check liveness for claims owned by this host. + lock = row["claim_lock"] or "" + if not lock.startswith(host_prefix): + continue + # Skip liveness check inside the launch-window grace period + # so a freshly-spawned worker isn't reclaimed before its PID + # is visible on /proc. + started_at = row["started_at"] if "started_at" in row.keys() else None + if started_at is not None: + grace = _resolve_crash_grace_seconds() + if time.time() - started_at < grace: continue + if _pid_alive(row["worker_pid"]): + continue - pid = int(row["worker_pid"]) + pid = int(row["worker_pid"]) + termination = _terminate_reclaimed_worker(pid, row["claim_lock"]) + if _worker_survived_termination(termination): + _defer_reclaim_for_live_worker( + conn, row["id"], row["claim_lock"], int(time.time()), termination, + reason="crashed_worker_group_alive", + ) + continue + + with write_txn(conn): kind, code = _classify_worker_exit(pid) rate_limited_exit = False if kind == "clean_exit": @@ -7638,6 +7707,7 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: if code is not None and kind != "unknown": event_payload["exit_kind"] = kind event_payload["exit_code"] = code + event_payload.update(termination) cur = conn.execute( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " diff --git a/hermes_cli/kanban_decompose.py b/hermes_cli/kanban_decompose.py index d6c248c658a6d..15254a07d8a63 100644 --- a/hermes_cli/kanban_decompose.py +++ b/hermes_cli/kanban_decompose.py @@ -456,6 +456,19 @@ def decompose_task( ) +def list_auto_decompose_triage_ids(*, tenant: Optional[str] = None) -> list[str]: + """Return fresh triage ids eligible for automatic decomposition.""" + with kb.connect_closing() as conn: + rows = kb.list_tasks( + conn, + status="triage", + tenant=tenant, + limit=1000, + exclude_block_provenance=True, + ) + return [row.id for row in rows] + + def list_triage_ids(*, tenant: Optional[str] = None) -> list[str]: """Return task ids currently in the triage column.""" with kb.connect_closing() as conn: diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 7de9b4c0b7c72..ffe787aa15dd9 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -192,6 +192,310 @@ def test_schedule_task_parks_time_delay_without_dispatching(kanban_home): assert any(e.kind == "scheduled" and e.payload == {"reason": "run next week"} for e in events) +def test_unblock_scheduled_rechecks_parent_gate(kanban_home): + with kb.connect() as conn: + parent = kb.create_task(conn, title="parent") + child = kb.create_task(conn, title="child", parents=[parent]) + assert kb.get_task(conn, child).status == "todo" + assert kb.schedule_task(conn, child, reason="wait until tomorrow") is True + + assert kb.unblock_task(conn, child) is True + assert kb.get_task(conn, child).status == "todo" + + kb.complete_task(conn, parent) + assert kb.schedule_task(conn, child, reason="second timer") is True + assert kb.unblock_task(conn, child) is True + assert kb.get_task(conn, child).status == "ready" + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX process-group semantics") +@pytest.mark.live_system_guard_bypass +def test_crashed_worker_kills_surviving_process_group(kanban_home, monkeypatch): + import signal + + worker = subprocess.Popen( + [ + sys.executable, + "-u", + "-c", + """ +import os, signal, subprocess, sys, time + +child = subprocess.Popen( + [sys.executable, "-u", "-c", "import signal,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); print('ready', flush=True); time.sleep(60)"], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, +) +assert child.stdout is not None +assert child.stdout.readline() == "ready\\n" +print(child.pid, flush=True) +os._exit(1) +""", + ], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + start_new_session=True, + ) + assert worker.stdout is not None + child_pid = int(worker.stdout.readline()) + worker.wait(timeout=2) + monkeypatch.setattr(kb, "_resolve_crash_grace_seconds", lambda: 0) + + try: + with kb.connect() as conn: + task_id = kb.create_task(conn, title="crashed worker", assignee="test") + assert kb.claim_task(conn, task_id) is not None + kb._set_worker_pid(conn, task_id, worker.pid) + + assert task_id in kb.detect_crashed_workers(conn) + assert kb.get_task(conn, task_id).status == "ready" + events = kb.list_events(conn, task_id) + assert events[-1].payload["process_group"] is True + assert events[-1].payload["sigkill"] is True + + assert not kb._pid_alive(child_pid) + finally: + try: + os.killpg(worker.pid, signal.SIGKILL) # windows-footgun: ok — POSIX-only test cleanup + except (ProcessLookupError, PermissionError): + pass + try: + os.kill(child_pid, signal.SIGKILL) # windows-footgun: ok — POSIX-only test cleanup + except (ProcessLookupError, PermissionError): + pass + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX process-group semantics") +def test_worker_termination_refuses_dispatcher_process_group(): + claim_lock = f"{kb._claimer_id().split(':', 1)[0]}:self-group-test" + + termination = kb._terminate_reclaimed_worker(os.getpgrp(), claim_lock) + + assert termination["termination_attempted"] is False + assert termination["termination_blocked"] == "dispatcher_process_group" + + +def test_stale_claim_reclaimed(kanban_home, monkeypatch): + import signal + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + killed: list[int] = [] + + def _signal(_pid, sig): + killed.append(sig) + + kb._set_worker_pid(conn, t, 12345) + # Rewind claim_expires so it looks stale. + conn.execute( + "UPDATE tasks SET claim_expires = ? WHERE id = ?", + (int(time.time()) - 3600, t), + ) + # Worker PID has died — exactly the case ``release_stale_claims`` + # should still reclaim (post-#23025: live PIDs are now extended). + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) + reclaimed = kb.release_stale_claims(conn, signal_fn=_signal) + assert reclaimed == 1 + assert kb.get_task(conn, t).status == "ready" + assert killed == [signal.SIGTERM] + + +def test_stale_claim_with_live_pid_extends_instead_of_reclaiming( + kanban_home, monkeypatch, +): + """A stale-by-TTL claim whose worker PID is still alive should be + extended, not reclaimed (#23025). Slow models can spend longer than + ``DEFAULT_CLAIM_TTL_SECONDS`` inside a single tool-free LLM call; + killing those healthy workers produces a respawn loop with zero + progress.""" + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + kb._set_worker_pid(conn, t, 12345) + + old_expires = int(time.time()) - 60 + conn.execute( + "UPDATE tasks SET claim_expires = ? WHERE id = ?", + (old_expires, t), + ) + + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: True) + killed: list[int] = [] + reclaimed = kb.release_stale_claims( + conn, signal_fn=lambda _p, sig: killed.append(sig), + ) + assert reclaimed == 0 + task = kb.get_task(conn, t) + assert task.status == "running" + assert task.claim_expires is not None + assert task.claim_expires > old_expires + assert killed == [] # live worker not killed + + kinds = [ + r["kind"] for r in conn.execute( + "SELECT kind FROM task_events WHERE task_id = ?", (t,), + ).fetchall() + ] + assert "claim_extended" in kinds + assert "reclaimed" not in kinds + + +def test_stale_claim_with_live_pid_uses_env_ttl_override( + kanban_home, monkeypatch, +): + import hermes_cli.kanban_db as _kb + + monkeypatch.setenv("HERMES_KANBAN_CLAIM_TTL_SECONDS", "3600") + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + kb._set_worker_pid(conn, t, 12345) + conn.execute( + "UPDATE tasks SET claim_expires = ? WHERE id = ?", + (int(time.time()) - 60, t), + ) + + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: True) + reclaimed = kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None) + assert reclaimed == 0 + + task = kb.get_task(conn, t) + assert task is not None + assert task.claim_expires is not None + assert task.claim_expires > int(time.time()) + 3000 + + +def test_stale_claim_deferred_when_live_worker_survives_termination( + kanban_home, monkeypatch, +): + """A TTL-expired claim whose worker survives the kill must NOT be released. + + Releasing would let the dispatcher spawn a duplicate beside the still-alive + worker — the runaway seen when a cgroup memory.high throttle parks a worker + in uninterruptible (D) state, where a pending SIGKILL cannot land. The claim + is held (extended) and retried next tick instead. + """ + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + kb._set_worker_pid(conn, t, 12345) + + old_expires = int(time.time()) - 60 + # Heartbeat stale by > 1h so the live-pid EXTEND branch is skipped and + # the terminate path (the wedged-worker case) runs. + conn.execute( + "UPDATE tasks SET claim_expires = ?, last_heartbeat_at = ? " + "WHERE id = ?", + (old_expires, int(time.time()) - 7200, t), + ) + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: True) + monkeypatch.setattr( + _kb, "_terminate_reclaimed_worker", + lambda *a, **k: { + "termination_attempted": True, + "host_local": True, + "terminated": False, + }, + ) + reclaimed = kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None) + assert reclaimed == 0 + + assert kb.get_task(conn, t).status == "running" + worker_pid = conn.execute( + "SELECT worker_pid FROM tasks WHERE id = ?", (t,), + ).fetchone()[0] + assert worker_pid == 12345 # worker not orphaned + claim_expires = conn.execute( + "SELECT claim_expires FROM tasks WHERE id = ?", (t,), + ).fetchone()[0] + assert claim_expires > old_expires # claim held, not released + + kinds = [ + r["kind"] for r in conn.execute( + "SELECT kind FROM task_events WHERE task_id = ?", (t,), + ).fetchall() + ] + assert "reclaim_deferred" in kinds + assert "reclaimed" not in kinds + + +def test_stale_claim_reclaimed_when_termination_succeeds( + kanban_home, monkeypatch, +): + """When the worker is actually killed, the claim is released as before.""" + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + kb._set_worker_pid(conn, t, 12345) + conn.execute( + "UPDATE tasks SET claim_expires = ?, last_heartbeat_at = ? " + "WHERE id = ?", + (int(time.time()) - 60, int(time.time()) - 7200, t), + ) + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) + monkeypatch.setattr( + _kb, "_terminate_reclaimed_worker", + lambda *a, **k: { + "termination_attempted": True, + "host_local": True, + "terminated": True, + }, + ) + reclaimed = kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None) + assert reclaimed == 1 + assert kb.get_task(conn, t).status == "ready" + + +def test_stale_claim_released_when_worker_not_host_local( + kanban_home, monkeypatch, +): + """The defer guard only holds OUR own surviving workers. + + A claim we cannot manage (different host, or no kill attempted) must still + be released, otherwise a foreign-host claim could strand a task forever. + """ + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + kb._set_worker_pid(conn, t, 12345) + conn.execute( + "UPDATE tasks SET claim_expires = ?, last_heartbeat_at = ? " + "WHERE id = ?", + (int(time.time()) - 60, int(time.time()) - 7200, t), + ) + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: True) + monkeypatch.setattr( + _kb, "_terminate_reclaimed_worker", + lambda *a, **k: { + "termination_attempted": False, + "host_local": False, + "terminated": False, + }, + ) + reclaimed = kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None) + assert reclaimed == 1 + assert kb.get_task(conn, t).status == "ready" diff --git a/tests/hermes_cli/test_kanban_decompose.py b/tests/hermes_cli/test_kanban_decompose.py index 31303b8d52e30..5469b18b29fb3 100644 --- a/tests/hermes_cli/test_kanban_decompose.py +++ b/tests/hermes_cli/test_kanban_decompose.py @@ -7,8 +7,10 @@ from __future__ import annotations +import asyncio import json as jsonlib from pathlib import Path +from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest @@ -145,6 +147,235 @@ def test_decompose_fanout_false_invalid_llm_assignee_uses_default(kanban_home): assert task.assignee == "fallback" +def test_decompose_unknown_assignee_falls_back_to_default(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="x", triage=True) + + # Roster only has 'orchestrator' and 'fallback'; LLM picks 'made_up'. + llm_payload = jsonlib.dumps({ + "fanout": True, + "rationale": "test", + "tasks": [ + {"title": "do X", "body": "", "assignee": "made_up", "parents": []}, + ], + }) + + patches = _patch_list_profiles(["orchestrator", "fallback"]) + for p in patches: + p.start() + try: + with patch.dict( + "os.environ", {}, clear=False, + ), _patch_aux_client(llm_payload), _patch_extra_body(), \ + patch( + "hermes_cli.kanban_decompose._load_config", + return_value={ + "kanban": { + "orchestrator_profile": "orchestrator", + "default_assignee": "fallback", + } + }, + ): + outcome = decomp.decompose_task(tid, author="me") + finally: + for p in patches: + p.stop() + + assert outcome.ok, outcome.reason + assert outcome.child_ids and len(outcome.child_ids) == 1 + with kb.connect() as conn: + child = kb.get_task(conn, outcome.child_ids[0]) + # 'made_up' wasn't in roster, so assignee rewritten to 'fallback' + assert child.assignee == "fallback" + + +def test_auto_selection_skips_block_loop_triage_but_manual_decompose_works( + kanban_home, + monkeypatch, +): + with kb.connect_closing() as conn: + fresh_id = kb.create_task(conn, title="fresh idea", triage=True) + escalated_id = kb.create_task( + conn, + title="review gate", + assignee="worker", + ) + assert kb.claim_task(conn, escalated_id, claimer="worker") is not None + assert kb.block_task( + conn, + escalated_id, + reason="first review pause", + kind="needs_input", + ) + assert kb.unblock_task(conn, escalated_id) + assert kb.claim_task(conn, escalated_id, claimer="worker") is not None + assert kb.block_task( + conn, + escalated_id, + reason="still waiting for a reviewer", + kind="needs_input", + ) + escalated = kb.get_task(conn, escalated_id) + + assert escalated is not None + assert escalated.status == "triage" + assert escalated.block_kind == "needs_input" + assert escalated.block_recurrences == kb.BLOCK_RECURRENCE_LIMIT + assert set(decomp.list_triage_ids()) == {fresh_id, escalated_id} + + from gateway.kanban_watchers import GatewayKanbanWatchersMixin + + runner = GatewayKanbanWatchersMixin() + runner._running = True + auto_attempts = [] + + async def _run_inline(fn, *args, **kwargs): + return fn(*args, **kwargs) + + async def _skip_sleep(_delay): + return None + + def _stop_after_tick(_conn, **_kwargs): + runner._running = False + return SimpleNamespace( + spawned=[], + reclaimed=0, + crashed=[], + timed_out=[], + promoted=0, + auto_blocked=[], + ) + + def _record_auto_attempt(task_id, **_kwargs): + auto_attempts.append(task_id) + return decomp.DecomposeOutcome(task_id, True, "selected") + + monkeypatch.setattr( + "hermes_cli.config.load_config", + lambda: { + "kanban": { + "dispatch_in_gateway": True, + "dispatch_interval_seconds": 1, + "auto_decompose": True, + "auto_decompose_per_tick": 3, + } + }, + ) + monkeypatch.setattr( + "gateway.kanban_watchers._acquire_singleton_lock", + lambda _path: (None, "unavailable"), + ) + monkeypatch.setattr("gateway.kanban_watchers.asyncio.to_thread", _run_inline) + monkeypatch.setattr("gateway.kanban_watchers.asyncio.sleep", _skip_sleep) + monkeypatch.setattr(kb, "dispatch_once", _stop_after_tick) + with patch.object(decomp, "decompose_task", side_effect=_record_auto_attempt): + asyncio.run(runner._kanban_dispatcher_watcher()) + + assert auto_attempts == [fresh_id] + + llm_payload = jsonlib.dumps({ + "fanout": False, + "rationale": "manual operator decision", + "title": "Resume reviewed work", + "body": "Proceed after explicit review.", + }) + patches = _patch_list_profiles(["orchestrator", "worker"]) + for p in patches: + p.start() + try: + with _patch_aux_client(llm_payload), _patch_extra_body(): + outcome = decomp.decompose_task(escalated_id, author="operator") + finally: + for p in patches: + p.stop() + + assert outcome.ok, outcome.reason + with kb.connect_closing() as conn: + assert kb.get_task(conn, escalated_id).status == "ready" + + +def test_auto_selection_filters_parked_rows_before_limit(kanban_home): + with kb.connect_closing() as conn: + parked_ids = [ + kb.create_task( + conn, + title=f"parked-{index}", + triage=True, + tenant="tenant-a", + priority=1, + ) + for index in range(998) + ] + conn.executemany( + "UPDATE tasks SET block_kind = 'needs_input', " + "block_recurrences = 2 WHERE id = ?", + ((task_id,) for task_id in parked_ids), + ) + kind_only_id = kb.create_task( + conn, + title="kind-only", + triage=True, + tenant="tenant-a", + priority=1, + ) + conn.execute( + "UPDATE tasks SET block_kind = 'needs_input' WHERE id = ?", + (kind_only_id,), + ) + recurrence_only_id = kb.create_task( + conn, + title="recurrence-only", + triage=True, + tenant="tenant-a", + priority=1, + ) + conn.execute( + "UPDATE tasks SET block_recurrences = 1 WHERE id = ?", + (recurrence_only_id,), + ) + parked_ids.extend((kind_only_id, recurrence_only_id)) + fresh_id = kb.create_task( + conn, + title="fresh", + triage=True, + tenant="tenant-a", + priority=0, + ) + other_tenant_id = kb.create_task( + conn, + title="other tenant", + triage=True, + tenant="tenant-b", + priority=2, + ) + conn.commit() + + assert decomp.list_auto_decompose_triage_ids(tenant="tenant-a") == [fresh_id] + assert decomp.list_auto_decompose_triage_ids(tenant="tenant-b") == [ + other_tenant_id + ] + assert decomp.list_auto_decompose_triage_ids() == [other_tenant_id, fresh_id] + assert set(decomp.list_triage_ids(tenant="tenant-a")) == set(parked_ids) + + +def test_decompose_handles_malformed_llm_json(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="x", triage=True) + + patches = _patch_list_profiles(["orchestrator"]) + for p in patches: + p.start() + try: + with _patch_aux_client("not json at all, sorry"), _patch_extra_body(): + outcome = decomp.decompose_task(tid, author="me") + finally: + for p in patches: + p.stop() + + assert outcome.ok is False + assert "malformed JSON" in outcome.reason + + def test_decompose_returns_false_when_task_not_triage(kanban_home): with kb.connect() as conn: tid = kb.create_task(conn, title="x") # ready, not triage