ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-20-c' into simp/r3-20

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:06:28 -07:00
parent d41c04afc5 b6838e8135
commit fea6f2f7f9
4 changed files with 1014 additions and 1400 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/kanban_db_connect.py
+186 -286
View File
@@ -9,7 +9,6 @@ from __future__ import annotations
import contextlib
import hashlib
import logging
import random
import re
import secrets
@@ -24,39 +23,26 @@ from pathlib import Path
from typing import Any
from typing import Optional
# Log-record parity with the origin module.
_log = logging.getLogger("hermes_cli.kanban_db")
# ---------------------------------------------------------------------------
# Connection helpers
# ---------------------------------------------------------------------------
_INITIALIZED_PATHS: set[str] = set()
_INIT_LOCK = threading.RLock()
_SQLITE_HEADER = b"SQLite format 3\x00"
DEFAULT_BUSY_TIMEOUT_MS = 120_000
# Cap on ``<db>.corrupt.<hash>.bak`` quarantines per board: content-addressing
# dedupes identical bytes, but mutating corruption mints a new fingerprint each
# time (one user hit 124). Oldest-by-mtime beyond the cap are pruned after each
# new backup.
_CORRUPT_BACKUP_RETENTION = 10
# Bounded init-lock acquire: a bare blocking flock let a wedged holder block the
# dispatcher's next-tick connect forever. Poll non-blocking until the deadline,
# then proceed without the lock (in-process _INIT_LOCK + idempotent init backstop).
_INIT_LOCK_TIMEOUT_SECONDS = 10.0
_INIT_LOCK_POLL_SECONDS = 0.05
@@ -174,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
@@ -227,11 +208,7 @@ def _dispatch_tick_lock(db_path: Path):
# ``journal_size_limit`` (set at connection init) on the writer's natural
# post-checkpoint reset. Best-effort, keyed per resolved DB path.
_WAL_CHECKPOINT_INTERVAL_SECONDS = 300.0
_LAST_WAL_CHECKPOINT: dict[str, float] = {}
_WAL_CHECKPOINT_LOCK = threading.Lock()
@@ -266,9 +243,7 @@ def _looks_like_tls_record_at(data: bytes, offset: int) -> bool:
"""Return True for a TLS record header at ``data[offset:]``."""
if len(data) < offset + 5:
return False
content_type = data[offset]
major = data[offset + 1]
minor = data[offset + 2]
content_type, major, minor = data[offset], data[offset + 1], data[offset + 2]
length = int.from_bytes(data[offset + 3:offset + 5], "big")
return (
content_type in {0x14, 0x15, 0x16, 0x17}
@@ -285,22 +260,17 @@ def _validate_sqlite_header(path: Path) -> None:
collapsed into a generic PRAGMA error and the gateway's corrupt-board
handling can identify the board by fingerprint."""
try:
stat = path.stat()
except FileNotFoundError:
return
if path.stat().st_size == 0:
return
except OSError:
return
if stat.st_size == 0:
return
# Byte-level probe: must run BEFORE any connection to this path exists
# (read_header_bytes_preopen refuses once one is live, because the close()
# would cancel this process's POSIX locks).
from hermes_cli.sqlite_safe_read import read_header_bytes_preopen
head = read_header_bytes_preopen(path, length=64)
if head is None:
return
if head.startswith(_SQLITE_HEADER):
if head is None or head.startswith(_SQLITE_HEADER):
return
signature = ""
if head.startswith(b"SQLit") and _looks_like_tls_record_at(head, 5):
@@ -322,16 +292,17 @@ class KanbanDbCorruptError(RuntimeError):
self.db_path = db_path
self.backup_path = backup_path
self.reason = reason
backup_str = str(backup_path) if backup_path is not None else "<backup failed>"
super().__init__(
f"Refusing to open corrupt kanban DB at {db_path}: {reason}. "
f"Original preserved; backup at {backup_str}."
f"Original preserved; backup at {_backup_label(backup_path)}."
)
def _prune_corrupt_backups(
parent: Path, base_name: str, keep: Optional[Path] = None,
) -> None:
def _backup_label(backup_path: Optional[Path]) -> str:
return str(backup_path) if backup_path is not None else "<backup failed>"
def _prune_corrupt_backups(parent: Path, base_name: str, keep: Optional[Path] = None) -> None:
"""Keep only the ``_CORRUPT_BACKUP_RETENTION`` newest (by mtime)
``<db>.corrupt.<hash>.bak`` files plus their ``-wal``/``-shm`` copies.
``keep`` (the just-created backup) is never pruned regardless of mtime —
@@ -346,8 +317,7 @@ def _prune_corrupt_backups(
]
except OSError:
return
budget = _kb._CORRUPT_BACKUP_RETENTION - (1 if keep is not None else 0)
budget = max(budget, 0)
budget = max(_kb._CORRUPT_BACKUP_RETENTION - (1 if keep is not None else 0), 0)
if len(backups) <= budget:
return
@@ -359,11 +329,7 @@ def _prune_corrupt_backups(
backups.sort(key=_mtime, reverse=True)
for stale in backups[budget:]:
for victim in (
stale,
stale.with_name(stale.name + "-wal"),
stale.with_name(stale.name + "-shm"),
):
for victim in (stale, stale.with_name(stale.name + "-wal"), stale.with_name(stale.name + "-shm")):
with contextlib.suppress(OSError):
victim.unlink(missing_ok=True)
@@ -402,8 +368,7 @@ def _backup_corrupt_db(path: Path) -> Optional[Path]:
digest.update(chunk)
except OSError:
return None
token = digest.hexdigest()[:16]
candidate = parent / f"{base_name}.corrupt.{token}.bak"
candidate = parent / f"{base_name}.corrupt.{digest.hexdigest()[:16]}.bak"
# Defensive: candidate must still be inside parent after construction.
if candidate.parent != parent:
return None
@@ -417,10 +382,11 @@ def _backup_corrupt_db(path: Path) -> Optional[Path]:
_prune_corrupt_backups(parent, base_name, keep=candidate)
for suffix in ("-wal", "-shm"):
sidecar = parent / (base_name + suffix)
if sidecar.parent != parent or not sidecar.exists():
continue
sidecar_backup = parent / (candidate.name + suffix)
if sidecar_backup.parent != parent or sidecar_backup.exists():
if (
sidecar.parent != parent or not sidecar.exists()
or sidecar_backup.parent != parent or sidecar_backup.exists()
):
continue
with contextlib.suppress(OSError):
shutil.copy2(sidecar, sidecar_backup)
@@ -464,7 +430,6 @@ def _repairable_index_names(messages: list[str]) -> Optional[list[str]]:
(caller fails closed; also ``None`` for no messages). First-appearance
order is preserved so the REINDEX pass is deterministic."""
names: list[str] = []
saw_any = False
for raw in messages:
message = (raw or "").strip()
if not message:
@@ -475,18 +440,13 @@ def _repairable_index_names(messages: list[str]) -> Optional[list[str]]:
break
else:
return None
saw_any = True
name = match.group("index").strip()
if name and name not in names:
names.append(name)
if not saw_any or not names:
return None
return names
return names or None
def _attempt_index_reindex_repair(
path: Path, index_names: list[str],
) -> tuple[bool, list[str]]:
def _attempt_index_reindex_repair(path: Path, index_names: list[str]) -> tuple[bool, list[str]]:
"""REINDEX the named indexes (per-index first; bare ``REINDEX`` fallback if
a parsed name is an internal/auto index), then re-run integrity_check.
Returns ``(clean, post_repair_messages)``; never raises. Callers must hold
@@ -511,6 +471,30 @@ def _attempt_index_reindex_repair(
return _integrity_messages_ok(messages), messages
def _missing_or_empty(resolved: Path) -> bool:
"""True for a missing / zero-byte / unstat-able DB file — nothing to probe."""
try:
return not resolved.exists() or resolved.stat().st_size == 0
except OSError:
return True
def _probe_for_corruption(resolved: Path) -> tuple[Optional[list[str]], Optional[str]]:
"""``(messages, reason)`` from an integrity probe; ``reason`` is ``None``
when healthy and ``messages`` is ``None`` when sqlite refused to open the
file at all. ``OperationalError`` (lock/busy) is NOT corruption and
propagates raw so a locked healthy DB is never quarantined."""
try:
messages = _probe_integrity(resolved)
except sqlite3.OperationalError:
raise
except sqlite3.DatabaseError as exc:
return None, f"sqlite refused to open file: {exc}"
if _integrity_messages_ok(messages):
return messages, None
return messages, f"integrity_check returned {messages[0] if messages else '<no row>'!r}"
def _guard_existing_db_is_healthy(path: Path) -> None:
"""Run ``PRAGMA integrity_check`` on an existing non-empty DB file.
@@ -528,47 +512,27 @@ def _guard_existing_db_is_healthy(path: Path) -> None:
resolved = path.resolve()
except OSError:
return
try:
if not resolved.exists() or resolved.stat().st_size == 0:
return
except OSError:
if _missing_or_empty(resolved) or str(resolved) in _INITIALIZED_PATHS:
return
if str(resolved) in _INITIALIZED_PATHS:
return
reason: Optional[str] = None
messages: list[str] = []
try:
messages = _probe_integrity(resolved)
if not _integrity_messages_ok(messages):
reason = (
f"integrity_check returned "
f"{messages[0] if messages else '<no row>'!r}"
)
except sqlite3.OperationalError:
# Lock contention / busy — not corruption. Propagate.
raise
except sqlite3.DatabaseError as exc:
reason = f"sqlite refused to open file: {exc}"
messages, reason = _probe_for_corruption(resolved)
if reason is None:
return
# Quarantine FIRST — both the repair and fail-closed paths preserve the
# pre-touch bytes before anything mutates the file.
backup = _backup_corrupt_db(resolved)
index_names = _repairable_index_names(messages)
index_names = _repairable_index_names(messages or [])
if index_names:
_kb._log.warning(
"kanban DB %s failed integrity_check with index-only errors "
"(%s); pre-repair backup at %s — attempting REINDEX auto-repair.",
resolved, ", ".join(index_names),
backup if backup is not None else "<backup failed>",
resolved, ", ".join(index_names), _backup_label(backup),
)
repaired, post = _attempt_index_reindex_repair(resolved, index_names)
if repaired:
_kb._log.warning(
"kanban DB %s auto-repaired via REINDEX (%s); "
"integrity_check now clean. Pre-repair copy kept at %s.",
resolved, ", ".join(index_names),
backup if backup is not None else "<backup failed>",
resolved, ", ".join(index_names), _backup_label(backup),
)
return
reason = (
@@ -593,11 +557,7 @@ class RepairResult:
reindexed: list[str] = field(default_factory=list)
def repair_db(
db_path: Optional[Path] = None,
*,
board: Optional[str] = None,
) -> RepairResult:
def repair_db(db_path: Optional[Path] = None, *, board: Optional[str] = None) -> RepairResult:
"""Probe a kanban DB and apply the narrow index-REINDEX repair if needed.
Same policy as :func:`_guard_existing_db_is_healthy` (quarantine BEFORE
any mutation; REINDEX under the init flock; anything non-index stays
@@ -610,41 +570,26 @@ def repair_db(
resolved = path.resolve()
except OSError:
resolved = path
try:
if not resolved.exists() or resolved.stat().st_size == 0:
return RepairResult(status="missing", db_path=resolved)
except OSError:
if _missing_or_empty(resolved):
return RepairResult(status="missing", db_path=resolved)
with _kb._cross_process_init_lock(resolved):
messages: list[str] = []
try:
messages = _probe_integrity(resolved)
except sqlite3.OperationalError:
# Locked/busy — not corruption; let the caller report it raw.
raise
except sqlite3.DatabaseError as exc:
messages, reason = _probe_for_corruption(resolved)
if messages is None:
# Same quarantine the connect-time guard takes when sqlite
# refuses to open the file at all.
return RepairResult(
status="corrupt",
db_path=resolved,
messages=[f"sqlite refused to open file: {exc}"],
status="corrupt", db_path=resolved, messages=[str(reason)],
backup_path=_backup_corrupt_db(resolved),
)
if _integrity_messages_ok(messages):
if reason is None:
return RepairResult(status="ok", db_path=resolved, messages=messages)
# Quarantine FIRST — identical policy to the connect-time guard.
backup = _backup_corrupt_db(resolved)
index_names = _repairable_index_names(messages)
if not index_names:
return RepairResult(
status="corrupt",
db_path=resolved,
messages=messages,
backup_path=backup,
)
return RepairResult(status="corrupt", db_path=resolved, messages=messages, backup_path=backup)
repaired, post = _attempt_index_reindex_repair(resolved, index_names)
# The file changed on disk; force the next connect() in this process
# to re-probe instead of trusting the stale healthy-path cache.
@@ -710,11 +655,7 @@ def _open_configured(path: Path, under_lock) -> tuple[sqlite3.Connection, Any]:
return conn, out
def connect(
db_path: Optional[Path] = None,
*,
board: Optional[str] = None,
) -> sqlite3.Connection:
def connect(db_path: Optional[Path] = None, *, board: Optional[str] = None) -> sqlite3.Connection:
"""Open (and initialize if needed) the kanban DB. WAL is (re)enabled on
every connection so a re-created file stays robust; the first connection
per path auto-runs :func:`init_db`, later ones skip via
@@ -771,11 +712,7 @@ def connect(
@contextlib.contextmanager
def connect_closing(
db_path: Optional[Path] = None,
*,
board: Optional[str] = None,
):
def connect_closing(db_path: Optional[Path] = None, *, board: Optional[str] = None):
"""Open a kanban DB connection and guarantee it is closed on exit. Use
instead of ``with kb.connect() as conn:`` — sqlite3's context manager only
commits/rolls back, it does NOT close the fd, so long-lived processes
@@ -788,21 +725,16 @@ def connect_closing(
conn.close()
def init_db(
db_path: Optional[Path] = None,
*,
board: Optional[str] = None,
) -> Path:
def init_db(db_path: Optional[Path] = None, *, board: Optional[str] = None) -> Path:
"""Create the schema if it doesn't exist; return the path used. Unlike
:func:`connect`'s cached first-time auto-init, this always re-runs the
migration pass — callers that know the on-disk schema may have drifted
(tests writing legacy event kinds, external upgrades) use it to force it."""
path = db_path if db_path is not None else _kb.kanban_db_path(board=board)
path.parent.mkdir(parents=True, exist_ok=True)
resolved = str(path.resolve())
# Clear the cache entry so connect() re-runs schema + migrations.
with _INIT_LOCK:
_INITIALIZED_PATHS.discard(resolved)
_INITIALIZED_PATHS.discard(str(path.resolve()))
with contextlib.closing(_kb.connect(path)):
pass
return path
@@ -818,7 +750,6 @@ _EARLY_TASK_COLUMNS = (
("idempotency_key", "idempotency_key TEXT"),
)
# (new column, ddl, legacy source column, copy statement) — see the
# RENAME-avoidance note in ``_migrate_add_optional_columns``.
_RENAMED_TASK_COLUMNS = (
@@ -833,7 +764,6 @@ _RENAMED_TASK_COLUMNS = (
),
)
# NULL / 0 defaults below reproduce the behaviour existing rows had before the
# column existed.
_LATER_TASK_COLUMNS = (
@@ -859,10 +789,31 @@ _LATER_TASK_COLUMNS = (
("block_recurrences", "block_recurrences INTEGER NOT NULL DEFAULT 0"),
)
_NOTIFY_SUB_COLUMNS = (
("notifier_profile", "notifier_profile TEXT"),
("delivery_mode", "delivery_mode TEXT NOT NULL DEFAULT 'notify'"),
("chat_type", "chat_type TEXT"),
# Platform-specific stable alt ID (Signal UUID, Feishu union_id, ...)
# so an active-wake replay reconstructs the SAME ``build_session_key``
# (which prefers ``user_id_alt``). NULL is inert.
("user_id_alt", "user_id_alt TEXT"),
("delivery_metadata", "delivery_metadata TEXT"),
)
def _column_names(conn: sqlite3.Connection, table: str) -> set[str]:
return {row["name"] for row in conn.execute(f"PRAGMA table_info({table})")}
def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
return conn.execute(
f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table}'"
).fetchone() is not None
def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None:
"""Add columns introduced after v1 to legacy DBs (called via ``init_db``)."""
cols = {row["name"] for row in conn.execute("PRAGMA table_info(tasks)")}
cols = _column_names(conn, "tasks")
for name, ddl in _EARLY_TASK_COLUMNS:
if name not in cols:
_add_column_if_missing(conn, "tasks", name, ddl)
@@ -870,7 +821,7 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None:
# Re-snapshot: DBs partially migrated by older releases may already carry
# later columns (e.g. ``consecutive_failures``), keeping the legacy-column
# migration idempotent.
cols = {row["name"] for row in conn.execute("PRAGMA table_info(tasks)")}
cols = _column_names(conn, "tasks")
# Legacy renames via ADD-then-copy rather than ``RENAME COLUMN``: very old
# DBs may lack the legacy column entirely (RENAME raises "no such column"),
@@ -894,42 +845,20 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None:
# init on legacy boards before the ALTER TABLE pass runs. ``IF NOT EXISTS``
# keeps re-running here cheap and correct on fresh DBs.
conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_tenant ON tasks(tenant)")
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_tasks_idempotency ON tasks(idempotency_key)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_tasks_session_id ON tasks(session_id)"
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_idempotency ON tasks(idempotency_key)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_session_id ON tasks(session_id)")
# task_events.run_id back-fills as NULL for historical events (they predate
# runs and can't be attributed).
ev_cols = {row["name"] for row in conn.execute("PRAGMA table_info(task_events)")}
if "run_id" not in ev_cols:
if "run_id" not in _column_names(conn, "task_events"):
_add_column_if_missing(conn, "task_events", "run_id", "run_id INTEGER")
# Same ordering rule as the ``tasks`` indexes above: index after column.
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_events_run "
"ON task_events(run_id, id)"
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_events_run ON task_events(run_id, id)")
notify_table_exists = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='kanban_notify_subs'"
).fetchone() is not None
if notify_table_exists:
notify_cols = {
row["name"] for row in conn.execute("PRAGMA table_info(kanban_notify_subs)")
}
for name, ddl in (
("notifier_profile", "notifier_profile TEXT"),
("delivery_mode", "delivery_mode TEXT NOT NULL DEFAULT 'notify'"),
("chat_type", "chat_type TEXT"),
# Platform-specific stable alt ID (Signal UUID, Feishu union_id, ...)
# so an active-wake replay reconstructs the SAME ``build_session_key``
# (which prefers ``user_id_alt``). NULL is inert.
("user_id_alt", "user_id_alt TEXT"),
("delivery_metadata", "delivery_metadata TEXT"),
):
if _table_exists(conn, "kanban_notify_subs"):
notify_cols = _column_names(conn, "kanban_notify_subs")
for name, ddl in _NOTIFY_SUB_COLUMNS:
if name in notify_cols:
continue
_add_column_if_missing(conn, "kanban_notify_subs", name, ddl)
@@ -944,57 +873,8 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None:
"WHERE platform != 'tui'"
)
# One-shot backfill: tasks 'running' before runs existed carried
# claim_lock / claim_expires / worker_pid on the task row; synthesize a
# matching task_runs row so end-run / heartbeat have something to write.
# write_txn serializes against concurrent dispatchers, and the per-row
# UPDATE uses ``current_run_id IS NULL`` as a CAS guard so a racing claim
# can't produce an orphaned row.
runs_exist = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='task_runs'"
).fetchone() is not None
if runs_exist:
with write_txn(conn):
inflight = conn.execute(
"SELECT id, assignee, claim_lock, claim_expires, worker_pid, "
" max_runtime_seconds, last_heartbeat_at, started_at "
"FROM tasks "
"WHERE status = 'running' AND current_run_id IS NULL"
).fetchall()
for row in inflight:
started = row["started_at"] or int(time.time())
cur = conn.execute(
"""
INSERT INTO task_runs (
task_id, profile, status,
claim_lock, claim_expires, worker_pid,
max_runtime_seconds, last_heartbeat_at,
started_at
) VALUES (?, ?, 'running', ?, ?, ?, ?, ?, ?)
""",
(
row["id"], row["assignee"], row["claim_lock"],
row["claim_expires"], row["worker_pid"],
row["max_runtime_seconds"], row["last_heartbeat_at"],
started,
),
)
# CAS: only install the pointer if nothing claimed the task
# since our SELECT (belt-and-suspenders under write_txn). On
# failure mark the orphan run row reclaimed so it doesn't
# look in-flight.
upd = conn.execute(
"UPDATE tasks SET current_run_id = ? "
"WHERE id = ? AND current_run_id IS NULL",
(cur.lastrowid, row["id"]),
)
if upd.rowcount != 1:
conn.execute(
"UPDATE task_runs SET status = 'reclaimed', "
" outcome = 'reclaimed', ended_at = ? "
"WHERE id = ?",
(int(time.time()), cur.lastrowid),
)
if _table_exists(conn, "task_runs"):
_backfill_legacy_inflight_runs(conn)
# One-shot event-kind rename: old names still worked but were awkward on
# the wire. Fires once per DB — after the UPDATE no rows match.
@@ -1008,17 +888,61 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None:
_rebuild_drifted_tables(conn)
# 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.
def _backfill_legacy_inflight_runs(conn: sqlite3.Connection) -> None:
"""One-shot backfill: tasks 'running' before runs existed carried
claim_lock / claim_expires / worker_pid on the task row; synthesize a
matching task_runs row so end-run / heartbeat have something to write.
write_txn serializes against concurrent dispatchers, and the per-row
UPDATE uses ``current_run_id IS NULL`` as a CAS guard so a racing claim
can't produce an orphaned row."""
with write_txn(conn):
inflight = conn.execute(
"SELECT id, assignee, claim_lock, claim_expires, worker_pid, "
" max_runtime_seconds, last_heartbeat_at, started_at "
"FROM tasks "
"WHERE status = 'running' AND current_run_id IS NULL"
).fetchall()
for row in inflight:
started = row["started_at"] or int(time.time())
cur = conn.execute(
"""
INSERT INTO task_runs (
task_id, profile, status,
claim_lock, claim_expires, worker_pid,
max_runtime_seconds, last_heartbeat_at,
started_at
) VALUES (?, ?, 'running', ?, ?, ?, ?, ?, ?)
""",
(
row["id"], row["assignee"], row["claim_lock"],
row["claim_expires"], row["worker_pid"],
row["max_runtime_seconds"], row["last_heartbeat_at"],
started,
),
)
# CAS: only install the pointer if nothing claimed the task
# since our SELECT (belt-and-suspenders under write_txn). On
# failure mark the orphan run row reclaimed so it doesn't
# look in-flight.
upd = conn.execute(
"UPDATE tasks SET current_run_id = ? "
"WHERE id = ? AND current_run_id IS NULL",
(cur.lastrowid, row["id"]),
)
if upd.rowcount != 1:
conn.execute(
"UPDATE task_runs SET status = 'reclaimed', "
" outcome = 'reclaimed', ended_at = ? "
"WHERE id = ?",
(int(time.time()), cur.lastrowid),
)
# 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 ("
@@ -1083,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:
@@ -1105,24 +1026,19 @@ def _rebuild_drifted_tables(conn: sqlite3.Connection) -> None:
_kb._log.info("kanban migration: rebuilding %s to match current schema", table)
conn.execute(f"ALTER TABLE {table} RENAME TO {table}_legacy")
conn.execute(create_sql)
new_cols = {c["name"] for c in conn.execute(f"PRAGMA table_info({table})")}
new_cols = _column_names(conn, table)
if table == "kanban_notify_subs":
# Cast the legacy TEXT cursor to INTEGER; NULL / non-numeric → 0.
shared = [c for c in old_cols if c in new_cols and c != "last_event_id"]
cols_csv = ", ".join(shared)
conn.execute(
f"INSERT INTO {table} ({cols_csv}, last_event_id) "
f"SELECT {cols_csv}, COALESCE(CAST(last_event_id AS INTEGER), 0) "
f"FROM {table}_legacy"
)
drop, extra_cols = "last_event_id", ", last_event_id"
extra_select = ", COALESCE(CAST(last_event_id AS INTEGER), 0)"
else:
# Drop the legacy TEXT id; AUTOINCREMENT reassigns it.
shared = [c for c in old_cols if c in new_cols and c != "id"]
cols_csv = ", ".join(shared)
conn.execute(
f"INSERT INTO {table} ({cols_csv}) "
f"SELECT {cols_csv} FROM {table}_legacy"
)
drop, extra_cols, extra_select = "id", "", ""
cols_csv = ", ".join(c for c in old_cols if c in new_cols and c != drop)
conn.execute(
f"INSERT INTO {table} ({cols_csv}{extra_cols}) "
f"SELECT {cols_csv}{extra_select} FROM {table}_legacy"
)
conn.execute(f"DROP TABLE {table}_legacy")
for index_sql in index_sqls:
conn.execute(index_sql)
@@ -1155,8 +1071,7 @@ def _check_file_length_invariant(conn: sqlite3.Connection) -> None:
if journal_mode == "wal":
return
ok = file_length_matches_header(conn)
if ok is False:
if file_length_matches_header(conn) is False:
raise sqlite3.DatabaseError(
"torn-extend detected: the database file is shorter than its "
"header page count claims"
@@ -1164,19 +1079,12 @@ 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
@@ -1200,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):
@@ -1227,11 +1131,9 @@ def write_txn(conn: sqlite3.Connection, *, allow_nested: bool = False):
try:
yield conn
except Exception:
try:
with contextlib.suppress(sqlite3.OperationalError):
conn.execute(f"ROLLBACK TO {savepoint}")
conn.execute(f"RELEASE {savepoint}")
except sqlite3.OperationalError:
pass
raise
else:
conn.execute(f"RELEASE {savepoint}")
@@ -1241,12 +1143,10 @@ def write_txn(conn: sqlite3.Connection, *, allow_nested: bool = False):
try:
yield conn
except Exception:
try:
# SQLite may already have auto-rolled-back (EIO, contention, corruption);
# don't let this secondary failure shadow the real one.
with contextlib.suppress(sqlite3.OperationalError):
conn.execute("ROLLBACK")
except sqlite3.OperationalError:
# SQLite already auto-rolled-back (EIO, contention, corruption);
# don't let this secondary failure shadow the real one.
pass
raise
else:
try:
@@ -1261,6 +1161,6 @@ def write_txn(conn: sqlite3.Connection, *, allow_nested: bool = False):
_kb._check_file_length_invariant(conn)
# Late-bound origin namespace (see module docstring). Imported LAST so this
# Late-bound origin namespace (see module docstring); imported LAST so this
# module is fully populated before ``kanban_db`` re-exports from it.
from hermes_cli import kanban_db as _kb # noqa: E402
hermes_cli/kanban_db_dispatch.py
+619 -777
View File
File diff suppressed because it is too large Load Diff
hermes_cli/kanban_db_notify.py
+94 -153
View File
@@ -21,27 +21,30 @@ if TYPE_CHECKING:
from hermes_cli.kanban_db import Event
# ---------------------------------------------------------------------------
# Notification subscriptions (used by the gateway kanban-notifier)
# ---------------------------------------------------------------------------
# Notifier reaction to a terminal event: "notify" = passive adapter.send only
# (default); "notify+wake" = send AND wake the destination agent; "wake" = wake only.
_NOTIFY_DELIVERY_MODES = ("notify", "notify+wake", "wake")
_SCALAR_TYPES = (str, int, float, bool)
def _encode_notify_delivery_metadata(
metadata: Optional[Mapping[str, Any]],
) -> Optional[str]:
# Subscription primary key predicate; every per-row statement below binds
# ``(task_id, platform, chat_id, thread_id or "")`` against it.
_SUB_KEY_WHERE = "WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ?"
def _sub_key(task_id: str, platform: str, chat_id: str, thread_id: Optional[str]) -> tuple:
return (task_id, platform, chat_id, thread_id or "")
def _encode_notify_delivery_metadata(metadata: Optional[Mapping[str, Any]]) -> Optional[str]:
"""Serialize platform send metadata stored on notification subscriptions."""
if not isinstance(metadata, Mapping):
return None
clean: dict[str, Any] = {}
for key, value in metadata.items():
if value is None:
continue
if isinstance(value, (str, int, float, bool)):
clean[str(key)] = value
clean = {
str(key): value
for key, value in metadata.items()
if value is not None and isinstance(value, _SCALAR_TYPES)
}
if not clean:
return None
return json.dumps(clean, sort_keys=True, separators=(",", ":"))
@@ -58,11 +61,7 @@ def _decode_notify_delivery_metadata(raw: Any) -> dict[str, Any]:
return {}
if not isinstance(data, dict):
return {}
return {
str(key): value
for key, value in data.items()
if isinstance(value, (str, int, float, bool))
}
return {str(key): value for key, value in data.items() if isinstance(value, _SCALAR_TYPES)}
def add_notify_sub(
@@ -80,30 +79,23 @@ 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.
"""
insert_mode = delivery_mode if delivery_mode in _NOTIFY_DELIVERY_MODES else (
# api_server is stateless: the adapter has no send(), the wake self-post
# IS the delivery. A plain 'notify' default would leave those subs with
# no delivery mechanism at all. Explicit modes still win.
"notify+wake" if platform == "api_server" else "notify"
)
insert_chat_type = chat_type or "dm"
now = int(time.time())
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
# the delivery. A plain 'notify' default would leave those subs with no
# delivery mechanism at all. Explicit modes still win.
insert_mode = valid_mode or ("notify+wake" if platform == "api_server" else "notify")
metadata_json = _encode_notify_delivery_metadata(delivery_metadata)
key = _sub_key(task_id, platform, chat_id, thread_id)
with _kb.write_txn(conn):
conn.execute(
"""
@@ -115,37 +107,24 @@ def add_notify_sub(
COALESCE((SELECT MAX(id) FROM task_events WHERE task_id = ?), 0))
""",
(
task_id,
platform,
chat_id,
thread_id or "",
user_id,
user_id_alt,
insert_chat_type,
notifier_profile,
insert_mode,
metadata_json,
now,
task_id,
*key, user_id, user_id_alt, chat_type or "dm", notifier_profile,
insert_mode, metadata_json, int(time.time()), task_id,
),
)
# chat_type / delivery_mode / delivery_metadata are last-write-wins;
# user_id_alt and notifier_profile only self-heal legacy rows lacking one.
key = (task_id, platform, chat_id, thread_id or "")
for column, value, fill_only in (
("chat_type", chat_type, False),
("user_id_alt", user_id_alt, True),
("notifier_profile", notifier_profile, True),
("delivery_mode", delivery_mode if delivery_mode in _NOTIFY_DELIVERY_MODES else None, False),
("delivery_mode", valid_mode, False),
("delivery_metadata", metadata_json, False),
):
if not value:
continue
guard = f" AND ({column} IS NULL OR {column} = '')" if fill_only else ""
conn.execute(
f"UPDATE kanban_notify_subs SET {column} = ? "
"WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ?"
+ guard,
f"UPDATE kanban_notify_subs SET {column} = ? " + _SUB_KEY_WHERE + guard,
(value, *key),
)
@@ -159,19 +138,11 @@ def _notify_profile_filter(
if notifier_profiles is None:
return "", []
profiles = sorted(
{
str(profile).strip()
for profile in notifier_profiles
if str(profile).strip()
}
)
profiles = sorted({str(p).strip() for p in notifier_profiles if str(p).strip()})
clauses: list[str] = []
params: list[str] = []
if profiles:
clauses.append(
"notifier_profile IN (" + ",".join("?" for _ in profiles) + ")"
)
clauses.append("notifier_profile IN (" + ",".join("?" for _ in profiles) + ")")
params.extend(profiles)
if include_unowned:
clauses.append("notifier_profile IS NULL OR notifier_profile = ''")
@@ -207,14 +178,11 @@ def list_notify_subs(
sql = "SELECT * FROM kanban_notify_subs"
if where:
sql += " WHERE " + " AND ".join(f"({clause})" for clause in where)
rows = conn.execute(sql, params).fetchall()
out: list[dict] = []
for row in rows:
for row in conn.execute(sql, params).fetchall():
item = dict(row)
if "delivery_metadata" in item:
item["delivery_metadata"] = _decode_notify_delivery_metadata(
item.get("delivery_metadata")
)
item["delivery_metadata"] = _decode_notify_delivery_metadata(item.get("delivery_metadata"))
out.append(item)
return out
@@ -229,44 +197,39 @@ 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():
return 0
owner_where, owner_params = _notify_profile_filter(
notifier_profiles, include_unowned=include_unowned,
)
clauses: list[str] = []
params: list[Any] = []
if owner_where:
clauses.append(f"({owner_where})")
params.extend(owner_params)
for clause, value in (
("LOWER(platform) = LOWER(?)", platform),
("chat_id = ?", chat_id),
("thread_id = ?", thread_id),
):
if value is not None:
clauses.append(clause)
params.append(value)
query = "SELECT COUNT(*) FROM kanban_notify_subs"
if clauses:
query += " WHERE " + " AND ".join(clauses)
conn = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)
try:
try:
owner_where, owner_params = _notify_profile_filter(
notifier_profiles, include_unowned=include_unowned,
)
clauses: list[str] = []
params: list[Any] = []
if owner_where:
clauses.append(f"({owner_where})")
params.extend(owner_params)
if platform is not None:
clauses.append("LOWER(platform) = LOWER(?)")
params.append(platform)
if chat_id is not None:
clauses.append("chat_id = ?")
params.append(chat_id)
if thread_id is not None:
clauses.append("thread_id = ?")
params.append(thread_id)
query = "SELECT COUNT(*) FROM kanban_notify_subs"
if clauses:
query += " WHERE " + " AND ".join(clauses)
row = conn.execute(query, params).fetchone()
except sqlite3.OperationalError as exc:
if "no such table" in str(exc).lower():
@@ -287,29 +250,21 @@ def remove_notify_sub(
) -> bool:
with _kb.write_txn(conn):
cur = conn.execute(
"DELETE FROM kanban_notify_subs WHERE task_id = ? "
"AND platform = ? AND chat_id = ? AND thread_id = ?",
(task_id, platform, chat_id, thread_id or ""),
"DELETE FROM kanban_notify_subs " + _SUB_KEY_WHERE,
_sub_key(task_id, platform, chat_id, thread_id),
)
return cur.rowcount > 0
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``.
def purge_stale_done_notify_subs(conn: sqlite3.Connection, *, max_age_days: int = 30) -> int:
"""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)
@@ -338,9 +293,8 @@ def _notify_cursor(
) -> Optional[int]:
"""``last_event_id`` of one subscription row, or ``None`` when unsubscribed."""
row = conn.execute(
"SELECT last_event_id FROM kanban_notify_subs "
"WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ?",
(task_id, platform, chat_id, thread_id or ""),
"SELECT last_event_id FROM kanban_notify_subs " + _SUB_KEY_WHERE,
_sub_key(task_id, platform, chat_id, thread_id),
).fetchone()
return None if row is None else int(row["last_event_id"])
@@ -370,11 +324,8 @@ def unseen_events_for_sub(
if kind_list:
params.extend(kind_list)
rows = conn.execute(q, params).fetchall()
out: list[Event] = []
max_id = cursor
for r in rows:
out.append(_kb.Event.from_row(r))
max_id = max(max_id, int(r["id"]))
out = [_kb.Event.from_row(r) for r in rows]
max_id = max([cursor, *(int(r["id"]) for r in rows)])
return max_id, out
@@ -401,24 +352,23 @@ def claim_unseen_events_for_sub(
if old_cursor is None:
return 0, 0, []
new_cursor, events = unseen_events_for_sub(
conn,
task_id=task_id,
platform=platform,
chat_id=chat_id,
thread_id=thread_id,
kinds=kinds,
conn, task_id=task_id, platform=platform, chat_id=chat_id,
thread_id=thread_id, kinds=kinds,
)
if not events:
return old_cursor, old_cursor, []
conn.execute(
"UPDATE kanban_notify_subs SET last_event_id = ? "
"WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? "
"AND last_event_id = ?",
(int(new_cursor), task_id, platform, chat_id, thread_id or "", int(old_cursor)),
)
_cas_cursor(conn, _sub_key(task_id, platform, chat_id, thread_id), new_cursor, old_cursor)
return old_cursor, new_cursor, events
def _cas_cursor(conn: sqlite3.Connection, key: tuple, new_cursor: int, expected: int) -> sqlite3.Cursor:
"""Move ``last_event_id`` only if it still equals ``expected``."""
return conn.execute(
"UPDATE kanban_notify_subs SET last_event_id = ? " + _SUB_KEY_WHERE + " AND last_event_id = ?",
(int(new_cursor), *key, int(expected)),
)
def advance_notify_cursor(
conn: sqlite3.Connection,
*,
@@ -430,9 +380,8 @@ def advance_notify_cursor(
) -> None:
with _kb.write_txn(conn):
conn.execute(
"UPDATE kanban_notify_subs SET last_event_id = ? "
"WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ?",
(int(new_cursor), task_id, platform, chat_id, thread_id or ""),
"UPDATE kanban_notify_subs SET last_event_id = ? " + _SUB_KEY_WHERE,
(int(new_cursor), *_sub_key(task_id, platform, chat_id, thread_id)),
)
@@ -450,18 +399,10 @@ def rewind_notify_cursor(
notifier advanced the row, so retries never clobber newer progress.
"""
with _kb.write_txn(conn):
cur = conn.execute(
"UPDATE kanban_notify_subs SET last_event_id = ? "
"WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? "
"AND last_event_id = ?",
(
int(old_cursor), task_id, platform, chat_id, thread_id or "",
int(claimed_cursor),
),
)
cur = _cas_cursor(conn, _sub_key(task_id, platform, chat_id, thread_id), old_cursor, claimed_cursor)
return cur.rowcount > 0
# Late-bound origin namespace (see module docstring). Imported LAST so this
# Late-bound origin namespace (see module docstring); imported LAST so this
# module is fully populated before ``kanban_db`` re-exports from it.
from hermes_cli import kanban_db as _kb # noqa: E402
hermes_cli/kanban_db_workspace.py
+115 -184
View File
@@ -7,8 +7,8 @@ origin-resident helpers are reached late-bound via ``_kb`` so monkeypatching
from __future__ import annotations
import logging
import os
import shutil
import sqlite3
import subprocess
from pathlib import Path
@@ -19,8 +19,32 @@ import contextlib
if TYPE_CHECKING:
from hermes_cli.kanban_db import Task
# Log-record parity with the origin module.
_log = logging.getLogger("hermes_cli.kanban_db")
_REMOVABLE_KINDS = ("scratch", "worktree")
# Statuses after which a child no longer needs its parent's workspace artifacts.
_ACTIVE_CHILDREN_SQL = (
"SELECT 1 FROM task_links l "
"JOIN tasks t ON t.id = l.child_id "
"WHERE l.parent_id = ? AND t.status NOT IN ('done', 'archived', 'failed', 'cancelled') "
"LIMIT 1"
)
_WORKSPACE_ROW_SQL = "SELECT workspace_kind, workspace_path, branch_name FROM tasks WHERE id = ?"
def _git(repo_root: Path, *args: str, timeout: int) -> subprocess.CompletedProcess:
"""``git -C repo_root args``; never raises on a non-zero exit."""
return subprocess.run(
["git", "-C", str(repo_root), *args],
capture_output=True,
text=True, encoding='utf-8', errors='replace',
timeout=timeout,
check=False,
)
def _has_active_children(conn: sqlite3.Connection, task_id: str) -> bool:
return conn.execute(_ACTIVE_CHILDREN_SQL, (task_id,)).fetchone() is not None
def _managed_scratch_path_info(p: Path) -> tuple[bool, Optional[str]]:
@@ -41,25 +65,13 @@ def _managed_scratch_path_info(p: Path) -> tuple[bool, Optional[str]]:
if home is not None:
with contextlib.suppress(OSError):
roots.append(((home / "kanban" / "workspaces").resolve(strict=False), _kb.DEFAULT_BOARD))
try:
boards_parent = (home / "kanban" / "boards").resolve(strict=False)
except OSError:
boards_parent = None
if boards_parent is not None:
try:
entries = list(boards_parent.iterdir())
except OSError:
entries = []
for entry in entries:
try:
if not entry.is_dir():
continue
except OSError:
continue
try:
entries: list[Path] = []
with contextlib.suppress(OSError):
entries = list((home / "kanban" / "boards").resolve(strict=False).iterdir())
for entry in entries:
with contextlib.suppress(OSError):
if entry.is_dir():
roots.append(((entry / "workspaces").resolve(strict=False), entry.name))
except OSError:
continue
for root, board in roots:
if p_abs == root:
continue
@@ -83,22 +95,17 @@ 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_home>/kanban/workspaces`` (legacy default board), or
``<kanban_home>/kanban/boards/<slug>/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_home>/kanban``,
``.../logs`` or ``.../boards/<slug>`` 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_home>/kanban/workspaces``,
or ``<kanban_home>/kanban/boards/<slug>/workspaces``). A path equal to a
root is not managed (deleting it would wipe every task's scratch dir);
``<kanban_home>/kanban``, ``.../logs`` and ``.../boards/<slug>`` 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.
"""
is_managed, _board = _managed_scratch_path_info(p)
return is_managed
return _managed_scratch_path_info(p)[0]
def _cleanup_workspace(conn: sqlite3.Connection, task_id: str) -> None:
@@ -108,43 +115,32 @@ def _cleanup_workspace(conn: sqlite3.Connection, task_id: str) -> None:
only when provably free of work (clean tree, every commit reachable from a
remote-tracking ref); ``dir`` is intentionally preserved."""
try:
row = conn.execute(
"SELECT workspace_kind, workspace_path, branch_name FROM tasks WHERE id = ?",
(task_id,),
).fetchone()
row = conn.execute(_WORKSPACE_ROW_SQL, (task_id,)).fetchone()
if not row:
return
kind: Optional[str] = row["workspace_kind"]
path: Optional[str] = row["workspace_path"]
if kind not in ("scratch", "worktree") or not path:
if kind not in _REMOVABLE_KINDS or not path:
# Not removable itself, but completing may still unblock a deferred
# parent scratch cleanup (e.g. a 'dir' child of a scratch parent).
_try_cleanup_parent_workspaces(conn, task_id)
return
# Defer while any child is not yet terminal so it can still read
# handoff artifacts from this workspace.
_active_children = conn.execute(
"SELECT 1 FROM task_links l "
"JOIN tasks t ON t.id = l.child_id "
"WHERE l.parent_id = ? AND t.status NOT IN ('done', 'archived', 'failed', 'cancelled') "
"LIMIT 1",
(task_id,),
).fetchone()
if _active_children:
if _has_active_children(conn, task_id):
_kb._log.debug(
"Deferring %s workspace cleanup for task %s: "
"active children still need workspace at %s",
kind, task_id, path,
)
return
# Kill the (dead) tmux worker session BEFORE removing a worktree so a
# lingering worker never has its cwd deleted from under it.
if kind == "worktree":
# Kill the (dead) tmux worker session BEFORE removing the worktree
# so a lingering worker never has its cwd deleted from under it.
_cleanup_worker_tmux(conn, task_id)
_cleanup_worktree_workspace(task_id, path, row["branch_name"])
_try_cleanup_parent_workspaces(conn, task_id)
return
import shutil
wp = Path(path)
if wp.is_dir():
# Containment guard: a board's ``default_workdir`` can pair
@@ -199,13 +195,7 @@ def _cleanup_worktree_workspace(
return
# No --force: git's own dirty guard re-verifies at removal time, so if
# the tree became dirty since our check (TOCTOU) removal fails safe.
result = subprocess.run(
["git", "-C", str(repo_root), "worktree", "remove", str(wp)],
capture_output=True,
text=True, encoding='utf-8', errors='replace',
timeout=60,
check=False,
)
result = _git(repo_root, "worktree", "remove", str(wp), timeout=60)
if result.returncode != 0:
_kb._log.warning(
"git worktree remove failed for task %s at %s: %s",
@@ -215,13 +205,7 @@ def _cleanup_worktree_workspace(
_kb._log.debug("Removed worktree workspace: %s", wp)
branch = (branch_name or "").strip() or f"wt/{task_id}"
if branch.startswith("wt/"):
subprocess.run(
["git", "-C", str(repo_root), "branch", "-D", branch],
capture_output=True,
text=True, encoding='utf-8', errors='replace',
timeout=30,
check=False,
)
_git(repo_root, "branch", "-D", branch, timeout=30)
except Exception:
pass # best-effort — never block completion
@@ -236,31 +220,17 @@ def _try_cleanup_parent_workspaces(conn: sqlite3.Connection, task_id: str) -> No
(task_id,),
).fetchall()
for (parent_id,) in parents:
row = conn.execute(
"SELECT workspace_kind, workspace_path, branch_name FROM tasks WHERE id = ?",
(parent_id,),
).fetchone()
row = conn.execute(_WORKSPACE_ROW_SQL, (parent_id,)).fetchone()
if (
not row
or row["workspace_kind"] not in ("scratch", "worktree")
or row["workspace_kind"] not in _REMOVABLE_KINDS
or not row["workspace_path"]
or _has_active_children(conn, parent_id)
):
continue
active = conn.execute(
"SELECT 1 FROM task_links l "
"JOIN tasks t ON t.id = l.child_id "
"WHERE l.parent_id = ? AND t.status NOT IN ('done', 'archived', 'failed', 'cancelled') "
"LIMIT 1",
(parent_id,),
).fetchone()
if active:
continue # still has active children
if row["workspace_kind"] == "worktree":
_cleanup_worktree_workspace(
parent_id, row["workspace_path"], row["branch_name"]
)
_cleanup_worktree_workspace(parent_id, row["workspace_path"], row["branch_name"])
continue
import shutil
wp = Path(row["workspace_path"])
if wp.is_dir() and _is_managed_scratch_path(wp):
shutil.rmtree(wp, ignore_errors=True)
@@ -277,18 +247,14 @@ def _cleanup_worker_tmux(conn: sqlite3.Connection, task_id: str) -> None:
).fetchone()
if not row or not row["assignee"]:
return
assignee: str = row["assignee"]
# Workers named swarm1-12 use tmux sessions named swarm-swarm1 etc.
session = f"swarm-{assignee}"
session = f"swarm-{row['assignee']}"
out = subprocess.run(
["tmux", "list-panes", "-t", session, "-F", "#{pane_dead}"],
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5,
)
if out.stdout.strip() == "1":
subprocess.run(
["tmux", "kill-session", "-t", session],
capture_output=True, timeout=5,
)
subprocess.run(["tmux", "kill-session", "-t", session], capture_output=True, timeout=5)
_kb._log.debug("Killed stale tmux session: %s", session)
except Exception:
pass # best-effort — never block completion
@@ -337,9 +303,7 @@ def _maybe_emit_scratch_tip(
"""Emit the first-use scratch-workspace tip once per install, right after a
scratch workspace is materialized. No-op for ``worktree``/``dir`` (preserved
by design) and once the sentinel exists."""
if (workspace_kind or "scratch") != "scratch":
return
if _scratch_tip_shown():
if (workspace_kind or "scratch") != "scratch" or _scratch_tip_shown():
return
try:
_kb._log.warning("kanban: %s (task %s)", _SCRATCH_TIP_MESSAGE, task_id)
@@ -372,26 +336,23 @@ def _git_toplevel(path: Path) -> Optional[Path]:
def _git_branch_exists(repo_root: Path, branch_name: str) -> bool:
try:
result = subprocess.run(
["git", "-C", str(repo_root), "show-ref", "--verify", f"refs/heads/{branch_name}"],
capture_output=True,
text=True, encoding='utf-8', errors='replace',
timeout=30,
check=False,
)
result = _git(repo_root, "show-ref", "--verify", f"refs/heads/{branch_name}", timeout=30)
except Exception:
return False
return result.returncode == 0
def _git_common_dir(path: Path) -> Optional[Path]:
out = _kb._git_out(path, "rev-parse", "--path-format=absolute", "--git-common-dir")
def _git_abs_path(path: Path, flag: str) -> Optional[Path]:
out = _kb._git_out(path, "rev-parse", "--path-format=absolute", flag)
return Path(out).expanduser().resolve(strict=False) if out else None
def _git_common_dir(path: Path) -> Optional[Path]:
return _git_abs_path(path, "--git-common-dir")
def _git_dir(path: Path) -> Optional[Path]:
out = _kb._git_out(path, "rev-parse", "--path-format=absolute", "--git-dir")
return Path(out).expanduser().resolve(strict=False) if out else None
return _git_abs_path(path, "--git-dir")
def _git_current_branch(path: Path) -> Optional[str]:
@@ -401,9 +362,7 @@ def _git_current_branch(path: Path) -> Optional[str]:
def _is_linked_worktree_checkout(path: Path) -> bool:
git_dir = _git_dir(path)
common_dir = _git_common_dir(path)
if git_dir is None or common_dir is None:
return False
return git_dir != common_dir
return git_dir is not None and common_dir is not None and git_dir != common_dir
def _nearest_existing_path(path: Path) -> Path:
@@ -428,25 +387,14 @@ def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> Non
"""Materialize ``target`` as a linked git worktree under ``repo_root``."""
target = target.expanduser()
repo_common = _git_common_dir(repo_root)
if target.exists() and repo_common is not None:
target_common = _git_common_dir(target)
if target_common == repo_common:
return
if target.exists() and repo_common is not None and _git_common_dir(target) == repo_common:
return
target.parent.mkdir(parents=True, exist_ok=True)
if _git_branch_exists(repo_root, branch_name):
cmd = ["git", "-C", str(repo_root), "worktree", "add", str(target), branch_name]
args = ["worktree", "add", str(target), branch_name]
else:
cmd = [
"git", "-C", str(repo_root), "worktree", "add", "-b", branch_name,
str(target), "HEAD",
]
result = subprocess.run(
cmd,
capture_output=True,
text=True, encoding='utf-8', errors='replace',
timeout=60,
check=False,
)
args = ["worktree", "add", "-b", branch_name, str(target), "HEAD"]
result = _git(repo_root, *args, timeout=60)
if result.returncode != 0:
stderr = (result.stderr or result.stdout or "").strip()
raise RuntimeError(
@@ -454,9 +402,14 @@ def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> Non
)
def _resolve_worktree_workspace(
task: Task, *, board: Optional[str] = None
) -> tuple[Path, str]:
def _anchored_worktree(repo_root: Path, task_id: str, branch_name: str) -> tuple[Path, str]:
"""Materialize the canonical ``<repo>/.worktrees/<task-id>`` worktree."""
target = repo_root / ".worktrees" / task_id
_ensure_git_worktree(repo_root, target, branch_name)
return target, branch_name
def _resolve_worktree_workspace(task: Task, *, board: Optional[str] = None) -> tuple[Path, str]:
"""Resolve + materialize a linked git worktree for ``task``. With no
``task.workspace_path`` the anchor is the board's ``default_workdir`` so
every worktree lands under a board-owned repo (``<repo>/.worktrees/<id>``)
@@ -485,9 +438,7 @@ def _resolve_worktree_workspace(
f"task {task.id} has workspace_kind=worktree but board "
f"{board_slug!r} default_workdir {board_default!r} is not inside a git repo"
)
target = repo_root / ".worktrees" / task.id
_ensure_git_worktree(repo_root, target, branch_name)
return target, branch_name
return _anchored_worktree(repo_root, task.id, branch_name)
requested = Path(task.workspace_path).expanduser()
if not requested.is_absolute():
@@ -518,9 +469,7 @@ def _resolve_worktree_workspace(
repo_root = _git_toplevel(requested)
if repo_root is not None and requested_resolved == repo_root:
target = repo_root / ".worktrees" / task.id
_ensure_git_worktree(repo_root, target, branch_name)
return target, branch_name
return _anchored_worktree(repo_root, task.id, branch_name)
repo_root = _repo_root_for_worktree_target(requested.parent)
if repo_root is None:
@@ -535,40 +484,31 @@ def _resolve_worktree_workspace(
def resolve_workspace(task: Task, *, board: Optional[str] = None) -> Path:
"""Resolve (and create if needed) the workspace for a task.
- ``scratch``: fresh dir under ``<board-root>/workspaces/<id>/`` — the
same path for the dispatcher and every profile worker, so handoff is
path-stable.
- ``dir:<path>``: ``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 ``<repo>/.worktrees/<task-id>``);
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/<task-id>``.
Persist the resolved path via ``set_workspace_path`` so later runs reuse it.
``scratch``: ``<board-root>/workspaces/<id>/`` — 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 ``<repo>/.worktrees/<id>``,
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 == "scratch":
if task.workspace_path:
# Legacy explicit-path scratch tasks get the same absolute-path
# guard as dir: — same threat model.
p = Path(task.workspace_path).expanduser()
if not p.is_absolute():
raise ValueError(
f"task {task.id} has non-absolute workspace_path "
f"{task.workspace_path!r}; workspace paths must be absolute"
)
else:
p = _kb.workspaces_root(board=board) / task.id
p.mkdir(parents=True, exist_ok=True)
return p
if kind == "dir":
if not task.workspace_path:
if kind == "worktree":
return _resolve_worktree_workspace(task, board=board)[0]
if kind == "scratch" and not task.workspace_path:
p = _kb.workspaces_root(board=board) / task.id
elif kind == "scratch":
# Legacy explicit-path scratch tasks get the same absolute-path guard
# as dir: — same threat model.
p = Path(task.workspace_path).expanduser()
if not p.is_absolute():
raise ValueError(
f"task {task.id} has workspace_kind=dir but no workspace_path"
f"task {task.id} has non-absolute workspace_path "
f"{task.workspace_path!r}; workspace paths must be absolute"
)
elif kind == "dir":
if not task.workspace_path:
raise ValueError(f"task {task.id} has workspace_kind=dir but no workspace_path")
p = Path(task.workspace_path).expanduser()
if not p.is_absolute():
raise ValueError(
@@ -576,34 +516,25 @@ def resolve_workspace(task: Task, *, board: Optional[str] = None) -> Path:
f"{task.workspace_path!r}; use an absolute path "
f"(relative paths are ambiguous against the dispatcher's CWD)"
)
p.mkdir(parents=True, exist_ok=True)
return p
if kind == "worktree":
p, _branch_name = _resolve_worktree_workspace(task, board=board)
return p
raise ValueError(f"unknown workspace_kind: {kind}")
else:
raise ValueError(f"unknown workspace_kind: {kind}")
p.mkdir(parents=True, exist_ok=True)
return p
def set_workspace_path(
conn: sqlite3.Connection, task_id: str, path: Path | str
) -> None:
def _set_task_column(conn: sqlite3.Connection, task_id: str, column: str, value: str) -> None:
with _kb.write_txn(conn):
conn.execute(
"UPDATE tasks SET workspace_path = ? WHERE id = ?",
(str(path), task_id),
)
conn.execute(f"UPDATE tasks SET {column} = ? WHERE id = ?", (value, task_id))
def set_branch_name(
conn: sqlite3.Connection, task_id: str, branch_name: str
) -> None:
with _kb.write_txn(conn):
conn.execute(
"UPDATE tasks SET branch_name = ? WHERE id = ?",
(str(branch_name), task_id),
)
def set_workspace_path(conn: sqlite3.Connection, task_id: str, path: Path | str) -> None:
_set_task_column(conn, task_id, "workspace_path", str(path))
# Late-bound origin namespace (see module docstring). Imported LAST so this
def set_branch_name(conn: sqlite3.Connection, task_id: str, branch_name: str) -> None:
_set_task_column(conn, task_id, "branch_name", str(branch_name))
# Late-bound origin namespace (see module docstring); imported LAST so this
# module is fully populated before ``kanban_db`` re-exports from it.
from hermes_cli import kanban_db as _kb # noqa: E402