Merge branch 'simp/r2-kanban' into simp/integration2
This commit is contained in:
Teknium
+417
-5697
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,514 @@
|
||||
"""Notification subscriptions consumed by the gateway kanban-notifier: per-(task, platform, chat, thread) rows with delivery metadata, unseen-event cursors and purge of stale done-task subs.
|
||||
|
||||
Split out of ``hermes_cli.kanban_db``; every name is re-exported there, and
|
||||
origin-resident helpers are reached late-bound via ``_kb`` so monkeypatching
|
||||
``kanban_db.<name>`` keeps working.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Iterable
|
||||
from typing import Mapping
|
||||
from typing import Optional
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from hermes_cli.kanban_db import Event
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Notification subscriptions (used by the gateway kanban-notifier)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# How the gateway kanban-notifier reacts to a terminal event for a
|
||||
# subscription:
|
||||
# "notify" -> passive ``adapter.send`` only (default)
|
||||
# "notify+wake" -> passive send AND wake the destination gateway agent
|
||||
# "wake" -> wake the agent only; no passive message is sent
|
||||
_NOTIFY_DELIVERY_MODES = ("notify", "notify+wake", "wake")
|
||||
|
||||
|
||||
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
|
||||
if not clean:
|
||||
return None
|
||||
return json.dumps(clean, sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
def _decode_notify_delivery_metadata(raw: Any) -> dict[str, Any]:
|
||||
if isinstance(raw, Mapping):
|
||||
return dict(raw)
|
||||
if not raw:
|
||||
return {}
|
||||
try:
|
||||
data = json.loads(str(raw))
|
||||
except Exception:
|
||||
return {}
|
||||
if not isinstance(data, dict):
|
||||
return {}
|
||||
return {
|
||||
str(key): value
|
||||
for key, value in data.items()
|
||||
if isinstance(value, (str, int, float, bool))
|
||||
}
|
||||
|
||||
|
||||
def add_notify_sub(
|
||||
conn: sqlite3.Connection,
|
||||
*,
|
||||
task_id: str,
|
||||
platform: str,
|
||||
chat_id: str,
|
||||
thread_id: Optional[str] = None,
|
||||
user_id: Optional[str] = None,
|
||||
user_id_alt: Optional[str] = None,
|
||||
chat_type: Optional[str] = None,
|
||||
notifier_profile: Optional[str] = None,
|
||||
delivery_mode: Optional[str] = None,
|
||||
delivery_metadata: Optional[Mapping[str, Any]] = None,
|
||||
) -> None:
|
||||
"""Register a gateway source that wants terminal-state notifications
|
||||
for ``task_id``. Idempotent on (task, platform, chat, thread).
|
||||
|
||||
``user_id_alt`` records the originating source's platform-specific stable
|
||||
alt ID (Signal UUID, Feishu union_id, ...) alongside ``user_id``. Active-wake
|
||||
replay must reproduce it so the woken turn's ``build_session_key`` matches
|
||||
the original event's — ``build_session_key`` prefers ``user_id_alt`` over
|
||||
``user_id`` (gateway/session.py), so replaying only ``user_id`` would key a
|
||||
wake into a different session whenever the two diverge for this source.
|
||||
|
||||
``chat_type`` records the originating source's chat type; the active-wake
|
||||
delivery modes replay it so the woken turn resolves the operator's real
|
||||
channel. ``None`` keeps an existing row's value.
|
||||
|
||||
``delivery_mode`` (see ``_NOTIFY_DELIVERY_MODES``) selects how the
|
||||
kanban-notifier reacts to a terminal event for this subscription. ``None``
|
||||
leaves an existing row's mode untouched (and inserts the ``"notify"``
|
||||
default for a fresh row); an explicit value is last-write-wins, so an
|
||||
operator can intentionally re-subscribe to change the mode (e.g.
|
||||
``notify`` -> ``wake``). An unknown value falls back to ``"notify"``.
|
||||
New subscriptions start "caught up": ``last_event_id`` snaps to the task's
|
||||
current ``MAX(task_events.id)`` instead of the schema default 0, which made
|
||||
the notifier replay every historical terminal event on its next tick (a
|
||||
boot-time burst of 100+ messages with many stale subs). Subscribers only
|
||||
want events AFTER they subscribe; auto-subscribe at task creation snapshots
|
||||
0 anyway.
|
||||
"""
|
||||
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 on that path (see gateway/wake.py and
|
||||
# test_kanban_notifier_apiserver_wake). A plain-'notify' default
|
||||
# would leave those subscriptions with no delivery mechanism at
|
||||
# all, regressing the pre-delivery_mode behavior where a task
|
||||
# carrying a session_id always woke. Explicit modes still win.
|
||||
"notify+wake" if platform == "api_server" else "notify"
|
||||
)
|
||||
insert_chat_type = chat_type or "dm"
|
||||
now = int(time.time())
|
||||
metadata_json = _encode_notify_delivery_metadata(delivery_metadata)
|
||||
with _kb.write_txn(conn):
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO kanban_notify_subs
|
||||
(task_id, platform, chat_id, thread_id, user_id, user_id_alt,
|
||||
chat_type, notifier_profile, delivery_mode, delivery_metadata,
|
||||
created_at, last_event_id)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
|
||||
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,
|
||||
),
|
||||
)
|
||||
# Re-subscribe semantics per column: chat_type / delivery_mode /
|
||||
# delivery_metadata are last-write-wins; user_id_alt and
|
||||
# notifier_profile only self-heal legacy rows that never had 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_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,
|
||||
(value, *key),
|
||||
)
|
||||
|
||||
|
||||
def _notify_profile_filter(
|
||||
notifier_profiles: Optional[Iterable[str]],
|
||||
*,
|
||||
include_unowned: bool,
|
||||
) -> tuple[str, list[str]]:
|
||||
"""Build an optional SQL predicate for notification profile ownership."""
|
||||
if notifier_profiles is None:
|
||||
return "", []
|
||||
|
||||
profiles = sorted(
|
||||
{
|
||||
str(profile).strip()
|
||||
for profile in notifier_profiles
|
||||
if str(profile).strip()
|
||||
}
|
||||
)
|
||||
clauses: list[str] = []
|
||||
params: list[str] = []
|
||||
if 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 = ''")
|
||||
if not clauses:
|
||||
return "0", []
|
||||
return "(" + ") OR (".join(clauses) + ")", params
|
||||
|
||||
|
||||
def list_notify_subs(
|
||||
conn: sqlite3.Connection,
|
||||
task_id: Optional[str] = None,
|
||||
*,
|
||||
notifier_profiles: Optional[Iterable[str]] = None,
|
||||
include_unowned: bool = False,
|
||||
) -> list[dict]:
|
||||
"""List subscriptions, optionally restricted to notifier profile owners.
|
||||
|
||||
Passing no ``notifier_profiles`` preserves the historical all-subscriptions
|
||||
result. Gateway notifier processes pass the profiles whose adapters they
|
||||
own so they cannot claim another gateway's events. ``include_unowned`` is
|
||||
used by the dispatch owner for legacy rows created before profile stamping.
|
||||
"""
|
||||
owner_where, owner_params = _notify_profile_filter(
|
||||
notifier_profiles, include_unowned=include_unowned,
|
||||
)
|
||||
where: list[str] = []
|
||||
params: list[Any] = []
|
||||
if task_id is not None:
|
||||
where.append("task_id = ?")
|
||||
params.append(task_id)
|
||||
if owner_where:
|
||||
where.append(owner_where)
|
||||
params.extend(owner_params)
|
||||
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:
|
||||
item = dict(row)
|
||||
if "delivery_metadata" in item:
|
||||
item["delivery_metadata"] = _decode_notify_delivery_metadata(
|
||||
item.get("delivery_metadata")
|
||||
)
|
||||
out.append(item)
|
||||
return out
|
||||
|
||||
|
||||
def count_notify_subs(
|
||||
db_path: Optional[Path] = None,
|
||||
*,
|
||||
board: Optional[str] = None,
|
||||
notifier_profiles: Optional[Iterable[str]] = None,
|
||||
include_unowned: bool = False,
|
||||
platform: Optional[str] = None,
|
||||
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 gateway notifier's zero-subscription early exit:
|
||||
unlike :func:`connect`, this never creates the DB file, never runs
|
||||
schema init/migration, and never opens the database writable (no
|
||||
write locks, no checkpoints — though a read-only open of a WAL
|
||||
database may still create the ``-shm``/``-wal`` sidecars, it cannot
|
||||
write table content). Rows in a not-yet-checkpointed WAL are
|
||||
visible, so a freshly added subscription is never missed. A missing
|
||||
DB, or a legacy DB that predates the subscriptions table, counts as
|
||||
zero. When ``notifier_profiles`` is supplied, only subscriptions owned
|
||||
by those profiles are counted; ``include_unowned`` also includes legacy
|
||||
rows without an owner stamp. Optional platform/chat/thread filters narrow
|
||||
the probe to one notification owner without changing the unfiltered count.
|
||||
Platform matching is case-insensitive, matching notifier routing; chat and
|
||||
thread identifiers are exact. Path resolution matches :func:`connect`
|
||||
(explicit ``db_path``, else ``board`` via :func:`kanban_db_path`). Raises
|
||||
:class:`sqlite3.Error` when the DB exists but cannot be read
|
||||
(locked, corrupt); callers choose 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
|
||||
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():
|
||||
return 0
|
||||
raise
|
||||
return int(row[0]) if row else 0
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def remove_notify_sub(
|
||||
conn: sqlite3.Connection,
|
||||
*,
|
||||
task_id: str,
|
||||
platform: str,
|
||||
chat_id: str,
|
||||
thread_id: Optional[str] = None,
|
||||
) -> 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 ""),
|
||||
)
|
||||
return cur.rowcount > 0
|
||||
|
||||
|
||||
def purge_stale_done_notify_subs(
|
||||
conn: sqlite3.Connection,
|
||||
*,
|
||||
max_age_days: int = 30,
|
||||
) -> int:
|
||||
"""Delete notify subscriptions whose task has sat in ``done`` or
|
||||
``blocked`` untouched for longer than ``max_age_days``.
|
||||
|
||||
The notifier keeps subscriptions alive through ``done`` because a
|
||||
completed task can be reopened (review corrections, continuation) and
|
||||
the reopened cycle must still notify its origin session. On boards
|
||||
that never archive, that retention would otherwise accumulate
|
||||
subscription rows forever — each one scanned every notifier tick.
|
||||
This GC bounds that: a task that has been ``done`` with no new events
|
||||
for the retention window is treated as settled and its subscriptions
|
||||
are purged. ``blocked`` tasks (circuit-breaker trips, dead workers)
|
||||
are reaped on the same clock — they are abandoned, not idle, unlike a
|
||||
``backlog``/``ready`` card that is merely waiting for pickup (#100955).
|
||||
Age is measured from the task's most recent event
|
||||
(falling back to ``completed_at`` then ``created_at``), so ANY
|
||||
activity — including a reopen, which also moves the task off
|
||||
``done`` — resets or exempts it.
|
||||
|
||||
``max_age_days <= 0`` disables the sweep entirely. Returns the number
|
||||
of subscription rows deleted.
|
||||
"""
|
||||
try:
|
||||
days = int(max_age_days)
|
||||
except (TypeError, ValueError):
|
||||
days = 30
|
||||
if days <= 0:
|
||||
return 0
|
||||
cutoff = int(time.time()) - days * 86400
|
||||
with _kb.write_txn(conn):
|
||||
cur = conn.execute(
|
||||
"DELETE FROM kanban_notify_subs WHERE task_id IN ("
|
||||
" SELECT t.id FROM tasks t"
|
||||
" WHERE t.status IN ('done', 'blocked')"
|
||||
" AND COALESCE("
|
||||
" (SELECT MAX(e.created_at) FROM task_events e"
|
||||
" WHERE e.task_id = t.id),"
|
||||
" t.completed_at, t.created_at, 0"
|
||||
" ) < ?)",
|
||||
(cutoff,),
|
||||
)
|
||||
return int(cur.rowcount or 0)
|
||||
|
||||
|
||||
def _notify_cursor(
|
||||
conn: sqlite3.Connection, task_id: str, platform: str, chat_id: str, thread_id: Optional[str],
|
||||
) -> 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 ""),
|
||||
).fetchone()
|
||||
return None if row is None else int(row["last_event_id"])
|
||||
|
||||
|
||||
def unseen_events_for_sub(
|
||||
conn: sqlite3.Connection,
|
||||
*,
|
||||
task_id: str,
|
||||
platform: str,
|
||||
chat_id: str,
|
||||
thread_id: Optional[str] = None,
|
||||
kinds: Optional[Iterable[str]] = None,
|
||||
) -> tuple[int, list[Event]]:
|
||||
"""Return ``(new_cursor, events)`` for a given subscription.
|
||||
|
||||
Only events with ``id > last_event_id`` are returned. The subscription's
|
||||
cursor is NOT advanced here; call :func:`advance_notify_cursor` after
|
||||
the gateway has successfully delivered the notifications.
|
||||
"""
|
||||
cursor = _notify_cursor(conn, task_id, platform, chat_id, thread_id)
|
||||
if cursor is None:
|
||||
return 0, []
|
||||
kind_list = list(kinds) if kinds else None
|
||||
q = (
|
||||
"SELECT * FROM task_events WHERE task_id = ? AND id > ? "
|
||||
+ ("AND kind IN (" + ",".join("?" * len(kind_list)) + ") " if kind_list else "")
|
||||
+ "ORDER BY id ASC"
|
||||
)
|
||||
params: list[Any] = [task_id, cursor]
|
||||
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"]))
|
||||
return max_id, out
|
||||
|
||||
|
||||
def claim_unseen_events_for_sub(
|
||||
conn: sqlite3.Connection,
|
||||
*,
|
||||
task_id: str,
|
||||
platform: str,
|
||||
chat_id: str,
|
||||
thread_id: Optional[str] = None,
|
||||
kinds: Optional[Iterable[str]] = None,
|
||||
) -> tuple[int, int, list[Event]]:
|
||||
"""Atomically claim unseen notification events for one subscription.
|
||||
|
||||
Returns ``(old_cursor, new_cursor, events)``. When events are returned,
|
||||
``kanban_notify_subs.last_event_id`` has already been advanced to
|
||||
``new_cursor`` inside a ``BEGIN IMMEDIATE`` transaction. That makes the
|
||||
notifier's read/claim step single-owner across multiple gateway watcher
|
||||
processes pointed at the same board DB: concurrent watchers serialize on
|
||||
SQLite's writer lock, and only the first process sees and claims a given
|
||||
event range.
|
||||
|
||||
Callers should send the claimed events, then either leave the cursor at
|
||||
``new_cursor`` on success or call :func:`rewind_notify_cursor` if delivery
|
||||
failed before any terminal unsubscribe removed the row.
|
||||
"""
|
||||
with _kb.write_txn(conn):
|
||||
old_cursor = _notify_cursor(conn, task_id, platform, chat_id, thread_id)
|
||||
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,
|
||||
)
|
||||
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)),
|
||||
)
|
||||
return old_cursor, new_cursor, events
|
||||
|
||||
|
||||
def advance_notify_cursor(
|
||||
conn: sqlite3.Connection,
|
||||
*,
|
||||
task_id: str,
|
||||
platform: str,
|
||||
chat_id: str,
|
||||
thread_id: Optional[str] = None,
|
||||
new_cursor: int,
|
||||
) -> 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 ""),
|
||||
)
|
||||
|
||||
|
||||
def rewind_notify_cursor(
|
||||
conn: sqlite3.Connection,
|
||||
*,
|
||||
task_id: str,
|
||||
platform: str,
|
||||
chat_id: str,
|
||||
thread_id: Optional[str] = None,
|
||||
claimed_cursor: int,
|
||||
old_cursor: int,
|
||||
) -> bool:
|
||||
"""Undo a notification claim when delivery fails.
|
||||
|
||||
The CAS guard only rewinds if no later notifier advanced the row after our
|
||||
claim. This keeps retry behavior for transient send failures without
|
||||
clobbering 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),
|
||||
),
|
||||
)
|
||||
return cur.rowcount > 0
|
||||
|
||||
|
||||
# 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
|
||||
@@ -0,0 +1,670 @@
|
||||
"""Task workspace lifecycle: scratch/dir/worktree resolution (incl. git worktree creation), post-completion cleanup with containment guards, worker tmux teardown and the first-use scratch-workspace tip.
|
||||
|
||||
Split out of ``hermes_cli.kanban_db``; every name is re-exported there, and
|
||||
origin-resident helpers are reached late-bound via ``_kb`` so monkeypatching
|
||||
``kanban_db.<name>`` keeps working.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sqlite3
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from typing import TYPE_CHECKING
|
||||
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")
|
||||
|
||||
|
||||
def _managed_scratch_path_info(p: Path) -> tuple[bool, Optional[str]]:
|
||||
"""Return whether *p* is managed scratch storage and the matching board."""
|
||||
try:
|
||||
p_abs = p.resolve(strict=False)
|
||||
except OSError:
|
||||
return False, None
|
||||
roots: list[tuple[Path, Optional[str]]] = []
|
||||
override = os.environ.get("HERMES_KANBAN_WORKSPACES_ROOT", "").strip()
|
||||
if override:
|
||||
with contextlib.suppress(OSError):
|
||||
roots.append((Path(override).expanduser().resolve(strict=False), None))
|
||||
try:
|
||||
home = _kb.kanban_home()
|
||||
except OSError:
|
||||
home = None
|
||||
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:
|
||||
roots.append(((entry / "workspaces").resolve(strict=False), entry.name))
|
||||
except OSError:
|
||||
continue
|
||||
for root, board in roots:
|
||||
if p_abs == root:
|
||||
continue
|
||||
try:
|
||||
if p_abs.is_relative_to(root):
|
||||
return True, board
|
||||
except ValueError:
|
||||
continue
|
||||
return False, None
|
||||
|
||||
|
||||
def _scratch_workspace(conn: sqlite3.Connection, task_id: str) -> Optional[Path]:
|
||||
"""Expanded ``workspace_path`` when the task uses a scratch workspace, else ``None``."""
|
||||
row = conn.execute(
|
||||
"SELECT workspace_kind, workspace_path FROM tasks WHERE id = ?",
|
||||
(task_id,),
|
||||
).fetchone()
|
||||
if not row or row["workspace_kind"] != "scratch" or not row["workspace_path"]:
|
||||
return None
|
||||
return Path(row["workspace_path"]).expanduser()
|
||||
|
||||
|
||||
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 — never the
|
||||
broader kanban home, a board root, or sibling subtrees like ``logs/`` or
|
||||
``boards/<slug>/`` itself. Allowed roots:
|
||||
|
||||
* ``HERMES_KANBAN_WORKSPACES_ROOT`` when set (worker-side override
|
||||
injected by the dispatcher).
|
||||
* ``<kanban_home>/kanban/workspaces`` — legacy default-board scratch root.
|
||||
* ``<kanban_home>/kanban/boards/<slug>/workspaces`` for each board slug
|
||||
that currently exists on disk.
|
||||
|
||||
The check requires strict descendancy: a path equal to one of these
|
||||
roots is NOT managed (deleting the workspaces root would wipe every
|
||||
task's scratch dir at once), and a path that resolves to ``<kanban_home>
|
||||
/kanban`` itself, ``<kanban_home>/kanban/logs``, or
|
||||
``<kanban_home>/kanban/boards/<slug>`` is rejected because those
|
||||
subtrees hold Hermes' own DB, metadata, and logs, not task workspaces.
|
||||
|
||||
Used by :func:`_cleanup_workspace` to refuse to ``shutil.rmtree`` paths
|
||||
outside Hermes-managed storage. A board ``default_workdir`` pointing at a
|
||||
real source tree can otherwise pair with ``workspace_kind='scratch'`` and
|
||||
cause task completion to delete user data (#28818).
|
||||
"""
|
||||
is_managed, _board = _managed_scratch_path_info(p)
|
||||
return is_managed
|
||||
|
||||
|
||||
def _cleanup_workspace(conn: sqlite3.Connection, task_id: str) -> None:
|
||||
"""Remove a task's scratch workspace dir and kill its stale tmux session.
|
||||
|
||||
Called from :func:`complete_task` after the DB transaction commits.
|
||||
Best-effort — any error is swallowed so cleanup never blocks task completion.
|
||||
``scratch`` workspaces are removed; ``worktree`` workspaces are removed only
|
||||
when provably free of work (clean tree, every commit reachable from a
|
||||
remote-tracking ref); ``dir`` workspaces are intentionally preserved.
|
||||
"""
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT workspace_kind, workspace_path, branch_name FROM tasks WHERE id = ?",
|
||||
(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:
|
||||
# This task's own workspace isn't a removable scratch dir, but its
|
||||
# completion may still unblock a deferred parent scratch cleanup
|
||||
# (e.g. a 'dir' child whose scratch parent was waiting on it). #33774
|
||||
_try_cleanup_parent_workspaces(conn, task_id)
|
||||
return
|
||||
# Check if this task has children that still need the workspace.
|
||||
# If any child is not yet done/archived, defer cleanup so the
|
||||
# child can read handoff artifacts from the workspace (#33774).
|
||||
_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:
|
||||
_kb._log.debug(
|
||||
"Deferring %s workspace cleanup for task %s: "
|
||||
"active children still need workspace at %s",
|
||||
kind, task_id, path,
|
||||
)
|
||||
return
|
||||
if kind == "worktree":
|
||||
# Kill the (dead) tmux worker session BEFORE removing the
|
||||
# worktree so a lingering worker never has its cwd deleted out
|
||||
# from under it. Both steps stay best-effort.
|
||||
_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 (#28818): a board's ``default_workdir`` can
|
||||
# pair ``workspace_kind='scratch'`` with a user-supplied path
|
||||
# pointing at a real source tree. Without this check, task
|
||||
# completion would unconditionally ``shutil.rmtree`` that path
|
||||
# and silently delete the user's source data.
|
||||
if _is_managed_scratch_path(wp):
|
||||
shutil.rmtree(wp, ignore_errors=True)
|
||||
_kb._log.debug("Removed scratch workspace: %s", wp)
|
||||
else:
|
||||
_kb._log.warning(
|
||||
"Refusing to remove out-of-scratch workspace for task %s: %s "
|
||||
"(workspace_kind='scratch' but path is outside any "
|
||||
"kanban-managed workspaces root)",
|
||||
task_id, wp,
|
||||
)
|
||||
# Also kill the tmux session for the worker that owned this task,
|
||||
# if the tmux session is now dead (worker process exited).
|
||||
_cleanup_worker_tmux(conn, task_id)
|
||||
# After cleaning up this task's workspace, check if any parent
|
||||
# tasks now have all children done — their deferred cleanup can
|
||||
# proceed (#33774).
|
||||
_try_cleanup_parent_workspaces(conn, task_id)
|
||||
except Exception:
|
||||
pass # best-effort — never block completion
|
||||
|
||||
|
||||
def _cleanup_worktree_workspace(
|
||||
task_id: str, path: str, branch_name: Optional[str] = None
|
||||
) -> None:
|
||||
"""Remove a finished task's linked git worktree when it holds no work.
|
||||
|
||||
Mirrors the safety judgment of the CLI startup pruner
|
||||
(``cli._prune_stale_worktrees``): removal requires a clean working tree
|
||||
AND every commit reachable from a remote-tracking ref. Any doubt — dirty
|
||||
files, unpushed commits, unresolvable repo, failing git — preserves the
|
||||
worktree. The task's auto-generated ``wt/<task-id>`` branch is deleted
|
||||
with it; custom branches are kept. Best-effort like the scratch path.
|
||||
"""
|
||||
try:
|
||||
from cli import _worktree_has_unpushed_commits, _worktree_is_dirty
|
||||
except Exception:
|
||||
return # CLI safety predicates unavailable — preserve
|
||||
try:
|
||||
wp = Path(path).expanduser()
|
||||
if not wp.is_dir():
|
||||
return
|
||||
common = _git_common_dir(wp)
|
||||
if common is None or common.name != ".git":
|
||||
return # not a linked worktree of a normal repo — never guess
|
||||
repo_root = common.parent
|
||||
if wp.resolve(strict=False) == repo_root.resolve(strict=False):
|
||||
return # never remove the main checkout
|
||||
if _worktree_is_dirty(str(wp)) or _worktree_has_unpushed_commits(str(wp)):
|
||||
_kb._log.info(
|
||||
"Preserving worktree for task %s: dirty or unpushed work at %s",
|
||||
task_id, wp,
|
||||
)
|
||||
return
|
||||
# No --force: the dirty/unpushed checks above run before removal, so
|
||||
# git's own dirty guard re-verifies at removal time. If the tree
|
||||
# became dirty between our check and the removal (TOCTOU), removal
|
||||
# fails safe and the worktree is preserved.
|
||||
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,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
_kb._log.warning(
|
||||
"git worktree remove failed for task %s at %s: %s",
|
||||
task_id, wp, (result.stderr or result.stdout or "").strip(),
|
||||
)
|
||||
return
|
||||
_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,
|
||||
)
|
||||
except Exception:
|
||||
pass # best-effort — never block completion
|
||||
|
||||
|
||||
def _try_cleanup_parent_workspaces(conn: sqlite3.Connection, task_id: str) -> None:
|
||||
"""Clean up parent scratch workspaces now that *task_id* completed.
|
||||
|
||||
When a parent task's cleanup was deferred because it had active children,
|
||||
this function is called after each child completes. If all children of a
|
||||
parent are now done/archived/failed/cancelled, the parent's scratch
|
||||
workspace is removed (#33774).
|
||||
"""
|
||||
try:
|
||||
parents = conn.execute(
|
||||
"SELECT parent_id FROM task_links WHERE child_id = ?",
|
||||
(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()
|
||||
if (
|
||||
not row
|
||||
or row["workspace_kind"] not in ("scratch", "worktree")
|
||||
or not row["workspace_path"]
|
||||
):
|
||||
continue
|
||||
# Check if ALL children of this parent are terminal
|
||||
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
|
||||
# All children done — safe to clean up parent workspace
|
||||
if row["workspace_kind"] == "worktree":
|
||||
_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)
|
||||
_kb._log.debug("Deferred cleanup: removed parent %s scratch workspace: %s", parent_id, wp)
|
||||
except Exception:
|
||||
pass # best-effort
|
||||
|
||||
|
||||
def _cleanup_worker_tmux(conn: sqlite3.Connection, task_id: str) -> None:
|
||||
"""Kill the tmux session associated with a task's assignee, if dead."""
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT assignee FROM tasks WHERE id = ?", (task_id,)
|
||||
).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}"
|
||||
# Check if session exists and pane is dead before killing
|
||||
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,
|
||||
)
|
||||
_kb._log.debug("Killed stale tmux session: %s", session)
|
||||
except Exception:
|
||||
pass # best-effort — never block completion
|
||||
|
||||
|
||||
_SCRATCH_TIP_SENTINEL_NAME = ".scratch_tip_shown"
|
||||
|
||||
|
||||
_SCRATCH_TIP_MESSAGE = (
|
||||
"scratch workspaces are ephemeral — they're deleted when the task "
|
||||
"completes. Use --workspace worktree: (git worktree) or "
|
||||
"--workspace dir:/abs/path (existing dir) to preserve worker output."
|
||||
)
|
||||
|
||||
|
||||
def _scratch_tip_sentinel_path() -> Path:
|
||||
"""Path to the per-install scratch-workspace-tip sentinel file."""
|
||||
return _kb.kanban_home() / _SCRATCH_TIP_SENTINEL_NAME
|
||||
|
||||
|
||||
def _scratch_tip_shown() -> bool:
|
||||
"""True iff the scratch-workspace tip has already been emitted on this
|
||||
install. Best-effort — any error means we re-emit, which is the safer
|
||||
failure mode for a help message."""
|
||||
try:
|
||||
return _scratch_tip_sentinel_path().exists()
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _mark_scratch_tip_shown() -> None:
|
||||
"""Touch the sentinel so future scratch workspaces stay silent.
|
||||
|
||||
Best-effort: a failure here just means the tip might appear once more,
|
||||
which is preferable to crashing dispatch over a help message.
|
||||
"""
|
||||
try:
|
||||
path = _scratch_tip_sentinel_path()
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.touch(exist_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _maybe_emit_scratch_tip(
|
||||
conn: sqlite3.Connection,
|
||||
task_id: str,
|
||||
workspace_kind: Optional[str],
|
||||
) -> None:
|
||||
"""Emit the first-use scratch-workspace tip exactly once per install.
|
||||
|
||||
Called from the dispatcher right after a scratch workspace is
|
||||
materialized. No-op for ``worktree`` / ``dir`` workspaces (they're
|
||||
preserved by design) and no-op after the sentinel exists.
|
||||
"""
|
||||
if (workspace_kind or "scratch") != "scratch":
|
||||
return
|
||||
if _scratch_tip_shown():
|
||||
return
|
||||
try:
|
||||
_kb._log.warning("kanban: %s (task %s)", _SCRATCH_TIP_MESSAGE, task_id)
|
||||
with _kb.write_txn(conn):
|
||||
_kb._append_event(
|
||||
conn, task_id, "tip_scratch_workspace",
|
||||
{"message": _SCRATCH_TIP_MESSAGE},
|
||||
)
|
||||
except Exception:
|
||||
# Best-effort — never block the spawn loop over a help message.
|
||||
pass
|
||||
finally:
|
||||
_mark_scratch_tip_shown()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Workspace resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _git_toplevel(path: Path) -> Optional[Path]:
|
||||
"""Return the git toplevel containing ``path``, or ``None`` if not in a repo."""
|
||||
out = _kb._git_out(path, "rev-parse", "--show-toplevel")
|
||||
if out is None:
|
||||
return None
|
||||
try:
|
||||
return Path(out).expanduser().resolve()
|
||||
except Exception:
|
||||
return Path(out).expanduser()
|
||||
|
||||
|
||||
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,
|
||||
)
|
||||
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")
|
||||
return Path(out).expanduser().resolve(strict=False) if out else None
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
def _git_current_branch(path: Path) -> Optional[str]:
|
||||
return _kb._git_out(path, "branch", "--show-current")
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
def _nearest_existing_path(path: Path) -> Path:
|
||||
current = path
|
||||
while not current.exists() and current != current.parent:
|
||||
current = current.parent
|
||||
return current
|
||||
|
||||
|
||||
def _repo_root_for_worktree_target(path: Path) -> Optional[Path]:
|
||||
current = _nearest_existing_path(path).resolve(strict=False)
|
||||
while True:
|
||||
repo_root = _git_toplevel(current)
|
||||
if repo_root is not None:
|
||||
return repo_root
|
||||
if current == current.parent:
|
||||
return None
|
||||
current = current.parent
|
||||
|
||||
|
||||
def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> None:
|
||||
"""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
|
||||
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]
|
||||
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,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
stderr = (result.stderr or result.stdout or "").strip()
|
||||
raise RuntimeError(
|
||||
f"git worktree add failed for {target} on branch {branch_name}: {stderr}"
|
||||
)
|
||||
|
||||
|
||||
def _resolve_worktree_workspace(
|
||||
task: Task, *, board: Optional[str] = None
|
||||
) -> tuple[Path, str]:
|
||||
"""Resolve + materialize a linked git worktree for ``task``.
|
||||
|
||||
When ``task.workspace_path`` is unset, the anchor is the board's
|
||||
``default_workdir`` (a persistent project checkout). This keeps every
|
||||
worktree task under a meaningful, board-owned repo — ``<repo>/.worktrees/
|
||||
<task-id>`` — instead of silently landing under the dispatcher's current
|
||||
working directory (which is whatever directory the gateway happened to be
|
||||
launched from, e.g. the Hermes checkout). If no anchor is configured
|
||||
anywhere, we fail loudly rather than guess.
|
||||
"""
|
||||
branch_name = (task.branch_name or "").strip() or f"wt/{task.id}"
|
||||
if not task.workspace_path:
|
||||
# Anchor on the board's configured default_workdir, not Path.cwd().
|
||||
# The dispatcher's CWD is incidental (gateway launch dir) and using it
|
||||
# scatters worktrees under whatever repo the gateway started in.
|
||||
board_slug = board if board else _kb.get_current_board()
|
||||
board_default = (_kb.read_board_metadata(board_slug).get("default_workdir") or "").strip()
|
||||
if not board_default:
|
||||
raise ValueError(
|
||||
f"task {task.id} has workspace_kind=worktree but no workspace_path, "
|
||||
f"and board {board_slug!r} has no default_workdir set. Set a board "
|
||||
"default workdir (a git repo) or create the task with "
|
||||
"--workspace worktree:<absolute-repo-path>."
|
||||
)
|
||||
anchor = Path(board_default).expanduser()
|
||||
if not anchor.is_absolute():
|
||||
raise ValueError(
|
||||
f"board {board_slug!r} default_workdir {board_default!r} is not "
|
||||
"absolute; use an absolute path to a git repo"
|
||||
)
|
||||
repo_root = _git_toplevel(anchor)
|
||||
if repo_root is None:
|
||||
raise ValueError(
|
||||
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
|
||||
|
||||
requested = Path(task.workspace_path).expanduser()
|
||||
if not requested.is_absolute():
|
||||
raise ValueError(
|
||||
f"task {task.id} has non-absolute worktree path "
|
||||
f"{task.workspace_path!r}; use an absolute path"
|
||||
)
|
||||
requested_resolved = requested.resolve(strict=False)
|
||||
|
||||
if requested.exists() and _is_linked_worktree_checkout(requested):
|
||||
actual_branch = _git_current_branch(requested)
|
||||
if actual_branch == branch_name:
|
||||
return requested_resolved, actual_branch
|
||||
# The requested path is an existing checkout of a DIFFERENT
|
||||
# task's branch. Decompose children inherit the root's
|
||||
# workspace_path verbatim, so siblings all point here; reusing
|
||||
# the checkout as-is would run this task on the other task's
|
||||
# branch — silent cross-task provenance corruption, and unsafe
|
||||
# when siblings run concurrently. Fall back to a fresh worktree
|
||||
# of our own under the same repo.
|
||||
fallback_root = _repo_root_for_worktree_target(requested.parent)
|
||||
if fallback_root is not None:
|
||||
fallback = fallback_root / ".worktrees" / task.id
|
||||
if fallback.resolve(strict=False) != requested_resolved:
|
||||
_ensure_git_worktree(fallback_root, fallback, branch_name)
|
||||
return fallback.resolve(strict=False), branch_name
|
||||
# No repo to anchor a fallback on (or the occupied path IS this
|
||||
# task's own canonical worktree): keep the legacy reuse rather
|
||||
# than failing dispatch.
|
||||
return requested_resolved, actual_branch or branch_name
|
||||
|
||||
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
|
||||
|
||||
repo_root = _repo_root_for_worktree_target(requested.parent)
|
||||
if repo_root is None:
|
||||
raise ValueError(
|
||||
f"task {task.id} worktree path {task.workspace_path!r} is not inside a git repo "
|
||||
"and does not point at a git repo root"
|
||||
)
|
||||
_ensure_git_worktree(repo_root, requested, branch_name)
|
||||
return requested, branch_name
|
||||
|
||||
|
||||
def resolve_workspace(task: Task, *, board: Optional[str] = None) -> Path:
|
||||
"""Resolve (and create if needed) the workspace for a task.
|
||||
|
||||
- ``scratch``: a fresh dir under ``<board-root>/workspaces/<id>/``,
|
||||
where ``<board-root>`` is the active board's root. The path is the
|
||||
same for the dispatcher and every profile worker, so handoff is
|
||||
path-stable.
|
||||
- ``dir:<path>``: the path stored in ``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 instead of a meaningful
|
||||
root. Users who want a kanban-root-relative workspace should
|
||||
compute the absolute path themselves.
|
||||
- ``worktree``: a real linked git worktree. If ``workspace_path`` names
|
||||
a repo root, Hermes treats it as an anchor and materializes a linked
|
||||
worktree at ``<repo>/.worktrees/<task-id>``. If ``workspace_path`` names
|
||||
a concrete target path, Hermes creates/reuses that linked worktree. With
|
||||
no ``workspace_path``, Hermes anchors on the board's ``default_workdir``
|
||||
and materializes ``<repo>/.worktrees/<task-id>`` per task; if no
|
||||
``default_workdir`` is configured it raises rather than guessing from the
|
||||
dispatcher's CWD. When ``branch_name`` is empty, Hermes uses
|
||||
``wt/<task-id>``.
|
||||
|
||||
Persist the resolved path back to the task row via ``set_workspace_path``
|
||||
so subsequent runs reuse the same directory.
|
||||
"""
|
||||
kind = task.workspace_kind or "scratch"
|
||||
if kind == "scratch":
|
||||
if task.workspace_path:
|
||||
# Legacy scratch tasks that were set to an explicit path get the
|
||||
# same absolute-path guard as dir: — consistent with the
|
||||
# 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:
|
||||
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(
|
||||
f"task {task.id} has non-absolute workspace_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}")
|
||||
|
||||
|
||||
def set_workspace_path(
|
||||
conn: sqlite3.Connection, task_id: str, path: Path | str
|
||||
) -> None:
|
||||
with _kb.write_txn(conn):
|
||||
conn.execute(
|
||||
"UPDATE tasks SET workspace_path = ? WHERE id = ?",
|
||||
(str(path), 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),
|
||||
)
|
||||
|
||||
|
||||
# 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
|
||||
@@ -385,10 +385,10 @@ def test_every_dispatcher_kanban_var_is_identity_gated():
|
||||
|
||||
Fails loudly if a new dispatcher var is added without registering it.
|
||||
"""
|
||||
import hermes_cli.kanban_db as kanban_db
|
||||
import hermes_cli.kanban_db_dispatch as kanban_db_dispatch
|
||||
from agent.delegation_context import KANBAN_ENV_KEYS
|
||||
|
||||
source = ast.parse(open(kanban_db.__file__, encoding="utf-8").read())
|
||||
source = ast.parse(open(kanban_db_dispatch.__file__, encoding="utf-8").read())
|
||||
spawn = next(
|
||||
node for node in ast.walk(source)
|
||||
if isinstance(node, ast.FunctionDef) and node.name == "_default_spawn"
|
||||
|
||||
@@ -1091,6 +1091,8 @@ def test_add_column_if_missing_is_idempotent_on_race(kanban_home):
|
||||
"""
|
||||
import sqlite3
|
||||
|
||||
from hermes_cli.sqlite_util import add_column_if_missing as _add_column_if_missing
|
||||
|
||||
conn = sqlite3.connect(":memory:")
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute(
|
||||
@@ -1098,14 +1100,14 @@ def test_add_column_if_missing_is_idempotent_on_race(kanban_home):
|
||||
)
|
||||
|
||||
# First call adds the column — returns True.
|
||||
added = kb._add_column_if_missing(conn, "tasks", "extra_col", "extra_col TEXT")
|
||||
added = _add_column_if_missing(conn, "tasks", "extra_col", "extra_col TEXT")
|
||||
assert added is True
|
||||
cols = {row["name"] for row in conn.execute("PRAGMA table_info(tasks)")}
|
||||
assert "extra_col" in cols
|
||||
|
||||
# Second call on same connection — column already exists — must return
|
||||
# False without raising, simulating the race the dispatcher hits.
|
||||
added_again = kb._add_column_if_missing(
|
||||
added_again = _add_column_if_missing(
|
||||
conn, "tasks", "extra_col", "extra_col TEXT"
|
||||
)
|
||||
assert added_again is False
|
||||
|
||||
@@ -196,7 +196,7 @@ class TestSigkillFallback:
|
||||
@pytest.mark.parametrize(
|
||||
"module_path, line_pattern",
|
||||
[
|
||||
("hermes_cli.kanban_db", 'getattr(signal, "SIGKILL", signal.SIGTERM)'),
|
||||
("hermes_cli.kanban_db_dispatch", 'getattr(signal, "SIGKILL", signal.SIGTERM)'),
|
||||
],
|
||||
)
|
||||
def test_module_uses_getattr_fallback(self, module_path, line_pattern):
|
||||
@@ -541,7 +541,7 @@ class TestTuiGatewayEntrySignalGuards:
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# hermes_cli/kanban_db.py waitpid guard
|
||||
# hermes_cli/kanban_db_dispatch.py waitpid guard
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@@ -551,7 +551,7 @@ class TestKanbanWaitpidWindowsGuard:
|
||||
|
||||
def test_source_gates_waitpid_loop(self):
|
||||
root = Path(__file__).resolve().parents[2]
|
||||
source = (root / "hermes_cli" / "kanban_db.py").read_text(encoding="utf-8")
|
||||
source = (root / "hermes_cli" / "kanban_db_dispatch.py").read_text(encoding="utf-8")
|
||||
# Find the waitpid call and confirm it's inside a POSIX gate.
|
||||
idx = source.find("os.waitpid(-1, os.WNOHANG)")
|
||||
assert idx > 0, "waitpid call must exist"
|
||||
|
||||
Reference in New Issue
Block a user