From 3c29ce3fdb6b8bed74ff280ff5ecdb991583aa5e Mon Sep 17 00:00:00 2001 From: Adam Moussa Date: Wed, 17 Jun 2026 14:35:51 -0400 Subject: [PATCH] Fix verified P1 findings: denylist bypasses, CAS concurrency, operator CLI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolves three execution-proven verifier findings from the scaffold review. Full suite: 548 passed, 1 skipped (stable across repeated runs); ruff clean. builders denylist (§3.3.2 #2): scan was +++-only and missed header-only sections. Now section-driven off `diff --git a/ b/`, catching the 4 proven bypasses — delete of a denied path, mode-change-only, `copy to` a denied path, out-of-scope delete (regression tests for each). §3.3.1 compare-and-set concurrency: BEGIN IMMEDIATE moved inside guarded retry; each CAS now runs on its own connection (shared sqlite3.Connection cannot hold two transactions, and is unsafe for concurrent use even for reads). connect() stashes the db path on a Connection subclass so the path is derived by a thread-safe attribute read, not a PRAGMA on the shared conn; busy_timeout set before the WAL pragma. Added shared-connection concurrent regression tests (distinct + same question) — previously raised "transaction within a transaction". operator CLI (run-team.py): added the design-named re-deliver and force-resume verbs (were missing); audit now records the attempt BEFORE the mutation and the outcome after, so a ledger mutation can never land without a trail; main() catches OSError instead of leaving an uncaught traceback on audit-write failure. --- agent-team/agent_team/db/schema.py | 120 ++++++++++- agent-team/agent_team/nodes/builders.py | 138 +++++++++---- agent-team/run-team.py | 260 +++++++++++++++++++++--- agent-team/tests/test_builders.py | 64 ++++++ agent-team/tests/test_run_team.py | 140 ++++++++++++- agent-team/tests/test_schema.py | 93 +++++++++ 6 files changed, 728 insertions(+), 87 deletions(-) diff --git a/agent-team/agent_team/db/schema.py b/agent-team/agent_team/db/schema.py index 691ed4c..10f4acb 100644 --- a/agent-team/agent_team/db/schema.py +++ b/agent-team/agent_team/db/schema.py @@ -21,6 +21,7 @@ does not serialize a check-and-set. from __future__ import annotations import sqlite3 +import time from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -107,6 +108,19 @@ CREATE TABLE IF NOT EXISTS schema_meta ( """.strip() +class _Connection(sqlite3.Connection): + """``sqlite3.Connection`` subclass that can carry its backing file path. + + The base ``Connection`` has no ``__dict__``, so a path cannot be stashed on + it. This thin subclass (passed as ``factory=`` to :func:`sqlite3.connect`) + lets :func:`connect` record the db file for a thread-safe attribute lookup by + the compare-and-set, avoiding a ``PRAGMA`` on a connection shared across + threads. + """ + + agent_team_db_path: str = "" + + def connect(db_path: Path) -> sqlite3.Connection: """Open ``db_path`` with WAL, foreign keys, and a busy timeout. @@ -123,11 +137,23 @@ def connect(db_path: Path) -> sqlite3.Connection: str(db_path), isolation_level=None, check_same_thread=False, + factory=_Connection, ) conn.row_factory = sqlite3.Row + # Set the busy timeout FIRST so every subsequent statement — including the + # journal-mode pragma below, which briefly needs the write lock — waits for + # the lock instead of failing immediately when another connection is mid + # -write. (Without this, opening a connection under concurrent writers could + # raise "database is locked" before the timeout was ever applied.) + conn.execute(f"PRAGMA busy_timeout={_BUSY_TIMEOUT_MS}") conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") - conn.execute(f"PRAGMA busy_timeout={_BUSY_TIMEOUT_MS}") + # Record the backing file path so the compare-and-set can derive it via a + # thread-safe attribute read instead of running a PRAGMA on a connection that + # callers share across threads (a sqlite3.Connection is not safe for + # concurrent use — even a read would corrupt its transaction state). Empty + # for an in-memory DB (no file to reopen on a second connection). + conn.agent_team_db_path = "" if str(db_path) == ":memory:" else str(db_path) return conn @@ -262,27 +288,107 @@ def supersede_question( ) +# Bounded retry if the write lock is still contended after ``busy_timeout`` +# elapses, so transient over-timeout contention does not surface as an error to +# the responder / deadline-timer callers. +_CAS_RETRY_ATTEMPTS: int = 3 +_CAS_RETRY_BACKOFF_S: float = 0.05 + + +def _main_db_file(conn: sqlite3.Connection) -> str | None: + """Return the file backing ``conn``'s ``main`` database, or ``None``. + + ``None`` signals an in-memory database (no file to reopen on a second + connection). Prefers the path stashed by :func:`connect` — a thread-safe + attribute read, so it is safe even when callers share ``conn`` across + threads. Falls back to ``PRAGMA database_list`` (rows of ``(seq, name, + file)``, indexed positionally to be ``row_factory``-agnostic) only for a + connection not opened via :func:`connect`; such a connection must not be + shared across threads. + """ + stashed = getattr(conn, "agent_team_db_path", None) + if stashed is not None: + return stashed or None + for row in conn.execute("PRAGMA database_list"): + if row[1] == "main": + return row[2] or None + return None + + def _compare_and_set( conn: sqlite3.Connection, *, sql: str, params: tuple[Any, ...], ) -> bool: - """Run a single compare-and-set UPDATE under ``BEGIN IMMEDIATE``. + """Run a single compare-and-set UPDATE under ``BEGIN IMMEDIATE`` (§3.3.1). - Returns ``True`` iff exactly one row changed. Takes the write lock up front - so concurrent responders cannot both observe ``status='open'`` (SQLite's - default deferred isolation would not serialize them). + Returns ``True`` iff exactly one row changed. The check-and-set takes the + write lock up front so concurrent responders cannot both observe + ``status='open'`` (SQLite's default deferred isolation would not serialize + them). + + **Concurrency safety.** The write runs on a private, short-lived connection + to the same database file — never on the passed ``conn``. A single SQLite + connection cannot hold two explicit transactions at once, so if a caller + shares one ``conn`` across threads (the responder and resume worker do, and + ``connect()`` sets ``check_same_thread=False``), two concurrent + ``BEGIN IMMEDIATE`` statements on it would raise "cannot start a transaction + within a transaction". Giving each call its own connection makes the + compare-and-set safe under that sharing; WAL serializes the writers via the + busy handler. A lock that outlasts ``busy_timeout`` is retried a bounded + number of times before propagating. ``BEGIN IMMEDIATE`` runs inside the + guarded path so its lock error is caught and retried, not raised uncaught. + + For an in-memory database (no file to reopen) the call falls back to the + passed ``conn``; in-memory DBs are single-connection and not the concurrent + production path. + """ + db_file = _main_db_file(conn) + if db_file is None: + return _cas_once(conn, sql, params) + + last_err: sqlite3.OperationalError | None = None + for attempt in range(_CAS_RETRY_ATTEMPTS): + write = connect(Path(db_file)) + try: + return _cas_once(write, sql, params) + except sqlite3.OperationalError as err: + if "locked" not in str(err).lower(): + raise + last_err = err + finally: + write.close() + time.sleep(_CAS_RETRY_BACKOFF_S * (attempt + 1)) + + assert last_err is not None # loop only exits early via return or raise + raise last_err + + +def _cas_once( + conn: sqlite3.Connection, + sql: str, + params: tuple[Any, ...], +) -> bool: + """Execute one ``BEGIN IMMEDIATE`` compare-and-set on ``conn``. + + ``BEGIN IMMEDIATE`` is issued before the try so a lock-acquisition error + propagates to the caller's retry loop with no transaction to unwind; once + the transaction is open, any failure rolls it back (best-effort) and + re-raises. """ conn.execute("BEGIN IMMEDIATE") try: cur = conn.execute(sql, params) changed = cur.rowcount == 1 conn.execute("COMMIT") + return changed except BaseException: - conn.execute("ROLLBACK") + try: + conn.execute("ROLLBACK") + except sqlite3.OperationalError: + pass raise - return changed def _utc_now_iso() -> str: diff --git a/agent-team/agent_team/nodes/builders.py b/agent-team/agent_team/nodes/builders.py index 16fe55b..8340451 100644 --- a/agent-team/agent_team/nodes/builders.py +++ b/agent-team/agent_team/nodes/builders.py @@ -202,14 +202,20 @@ def _render_build_prompt(plan: Mapping[str, Any]) -> str: # Diff parsing + canonicalization # --------------------------------------------------------------------------- -# Matches the "+++ b/" (and "--- a/") target lines of a unified -# diff, plus git "rename to"/"rename from" lines. We read targets from the -# header so the scan sees exactly the paths the patch would create/modify. +# A unified diff is parsed **per file section**, each section delimited by its +# ``diff --git a/ b/`` header. That header carries BOTH the source and +# destination path for every change kind — modify, add, delete, mode-change, +# rename, and copy — so reading it (not only the ``+++ b/`` body line) is what +# lets the scan see deletes, mode-only changes, and ``copy to`` targets that have +# no ``+++`` line or whose ``+++`` is ``/dev/null``. The ``---``/``+++`` and +# rename/copy ``from``/``to`` lines refine the section's source/dest when present. _PLUS_RE = re.compile(r"^\+\+\+ (?:b/)?(.+?)\s*$") _MINUS_RE = re.compile(r"^--- (?:a/)?(.+?)\s*$") _DIFF_GIT_RE = re.compile(r"^diff --git a/(.+?) b/(.+?)\s*$") _RENAME_FROM_RE = re.compile(r"^rename from (.+?)\s*$") _RENAME_TO_RE = re.compile(r"^rename to (.+?)\s*$") +_COPY_FROM_RE = re.compile(r"^copy from (.+?)\s*$") +_COPY_TO_RE = re.compile(r"^copy to (.+?)\s*$") # /dev/null appears as the source of an add or target of a delete; it is never a # real repo path and must not be scanned/scoped as one. @@ -257,58 +263,106 @@ def _canonicalize(raw: str) -> str | None: def iter_diff_target_paths(diff: str) -> list[_DiffTarget]: - """Parse a unified diff into the set of target paths it would write. + """Parse a unified diff into every path it would create/modify/move/remove. - Reads the ``+++ b/`` header lines (the destinations a patch creates or - modifies) and git ``rename to`` lines (carrying the matching ``rename - from``), canonicalizing each. A path that fails :func:`_canonicalize` is - surfaced as an unsafe target (path preserved raw, marked via a sentinel) so - the scan rejects it. Pure header parsing — it never executes the diff. + The diff is walked **per file section**, each delimited by its ``diff --git + a/ b/`` header. Because that header carries the source and the + destination for *every* change kind — including deletes (``+++ /dev/null``), + mode-only changes (no ``+++`` line at all), and ``copy to`` targets — reading + it closes the bypasses that a ``+++``-only scan misses. Within a section the + ``--- a/`` / ``+++ b/`` lines and the rename/copy ``from``/``to`` lines refine + the source/destination when present (a rename/copy ``to`` is authoritative for + the destination and links it to its source). BOTH the section source and + destination are emitted as targets, so removing or moving a file *away from* a + denied/out-of-scope path is flagged too. A path that fails + :func:`_canonicalize` is surfaced as an unsafe target. Pure header parsing — + it never executes the diff. """ targets: list[_DiffTarget] = [] seen: set[tuple[str, str | None]] = set() - pending_rename_from: str | None = None + + # Per-section accumulators; flushed at each new ``diff --git`` and at EOF. + src_raw: str | None = None + dst_raw: str | None = None + move_from: str | None = None + + def flush() -> None: + nonlocal src_raw, dst_raw, move_from + if src_raw is None and dst_raw is None: + return + rf = _canonicalize(move_from) if move_from else None + # Source side: catches deletes and renames/copies away from a denied or + # out-of-scope path. Destination side: catches creates/modifies/mode + # changes/copy targets, linked to its source for a clear violation note. + _emit_target(targets, seen, src_raw) + _emit_target(targets, seen, dst_raw, rename_from=rf) + src_raw = dst_raw = move_from = None for line in diff.splitlines(): - rename_from = _RENAME_FROM_RE.match(line) - if rename_from: - pending_rename_from = _canonicalize(rename_from.group(1)) + git = _DIFF_GIT_RE.match(line) + if git: + flush() + src_raw, dst_raw = git.group(1), git.group(2) continue - rename_to = _RENAME_TO_RE.match(line) - if rename_to: - canon = _canonicalize(rename_to.group(1)) - target = _DiffTarget( - path=canon if canon is not None else rename_to.group(1).strip(), - rename_from=pending_rename_from, - ) - _append_unique(targets, seen, target, unsafe=canon is None) - pending_rename_from = None - continue - - plus = _PLUS_RE.match(line) - if plus: - # Ignore hunk body lines that merely start with "+++"; a real header - # is "+++ b/path" or "+++ /dev/null". _canonicalize maps /dev/null - # to None which we drop (a delete has no created target). - raw = plus.group(1) - if raw.strip() == _DEV_NULL: - continue - canon = _canonicalize(raw) - if canon is None: - # Unsafe (absolute / parent-escaping) target line. - _append_unique( - targets, - seen, - _DiffTarget(path=raw.strip()), - unsafe=True, - ) + for pattern in (_RENAME_TO_RE, _COPY_TO_RE): + m = pattern.match(line) + if m: + dst_raw = m.group(1) # authoritative destination for the section + break + else: + for pattern in (_RENAME_FROM_RE, _COPY_FROM_RE): + m = pattern.match(line) + if m: + move_from = m.group(1) + break else: - _append_unique(targets, seen, _DiffTarget(path=canon), unsafe=False) + minus = _MINUS_RE.match(line) + if minus and minus.group(1).strip() != _DEV_NULL: + src_raw = minus.group(1) + continue + plus = _PLUS_RE.match(line) + # Ignore hunk body lines starting with "+++"; a real header is + # "+++ b/path" or "+++ /dev/null". A /dev/null target means a + # delete, so the destination stays the diff --git path. + if plus and plus.group(1).strip() != _DEV_NULL: + dst_raw = plus.group(1) + flush() return targets +def _emit_target( + targets: list[_DiffTarget], + seen: set[tuple[str, str | None]], + raw: str | None, + *, + rename_from: str | None = None, +) -> None: + """Canonicalize ``raw`` and append it as a target (unsafe paths flagged). + + ``/dev/null`` and empty values are dropped (no real path). A path that fails + :func:`_canonicalize` (absolute / parent-escaping) is appended as an unsafe + target so the scan rejects it rather than letting an indirection bypass the + denylist. + """ + if raw is None: + return + stripped = raw.strip() + if not stripped or stripped == _DEV_NULL: + return + canon = _canonicalize(stripped) + if canon is None: + _append_unique(targets, seen, _DiffTarget(path=stripped), unsafe=True) + else: + _append_unique( + targets, + seen, + _DiffTarget(path=canon, rename_from=rename_from), + unsafe=False, + ) + + # Marks a target whose path could not be safely canonicalized. Stored on the # _DiffTarget via a parallel set keyed by identity is overkill; instead we use a # reserved reason string the scanner recognizes. diff --git a/agent-team/run-team.py b/agent-team/run-team.py index c270e39..0564407 100644 --- a/agent-team/run-team.py +++ b/agent-team/run-team.py @@ -98,7 +98,13 @@ _QUESTION_COLUMNS: tuple[str, ...] = ( ) # Destructive subcommands that require ``--confirm`` and are audit-logged. -_DESTRUCTIVE_ACTIONS: frozenset[str] = frozenset({"expire", "answer", "supersede"}) +# ``force-resume`` is the design-named operator verb (§3.3.1/§6.6); ``supersede`` +# is kept as its lower-level alias. ``redeliver`` is NOT here — it is idempotent +# and non-destructive (it only clears a delivery ref), though it is still +# audit-logged for provenance. +_DESTRUCTIVE_ACTIONS: frozenset[str] = frozenset( + {"expire", "answer", "supersede", "force-resume"} +) def _utc_now_iso() -> str: @@ -131,6 +137,64 @@ def _append_audit(audit_log: Path, entry: dict[str, Any]) -> None: atomic_write(audit_log, existing + line) +def _audit_attempt( + audit_log: Path, + action: str, + *, + question_id: str, + operator: str, + detail: dict[str, Any] | None = None, +) -> None: + """Record the *intent* to perform a destructive action BEFORE it mutates. + + §3.3.1 requires every destructive action to be audit-logged. Writing the + attempt before the ledger mutation closes the "mutation applied with no + audit record" gap: if this append fails (e.g. an unwritable audit path) it + raises before any ledger row is touched, so the action aborts cleanly with + nothing changed. The matching :func:`_audit_outcome` records what happened. + """ + _append_audit( + audit_log, + { + "ts": _utc_now_iso(), + "action": action, + "phase": "attempt", + "question_id": question_id, + "operator": operator, + **(detail or {}), + }, + ) + + +def _audit_outcome( + audit_log: Path, + action: str, + *, + question_id: str, + operator: str, + applied: bool, + detail: dict[str, Any] | None = None, +) -> None: + """Record the *result* of a destructive action AFTER it ran. + + Carries ``applied`` (did the compare-and-set change a row). Pairs with the + :func:`_audit_attempt` record written before the mutation, so even if this + outcome append fails the attempt already proves the action was made. + """ + _append_audit( + audit_log, + { + "ts": _utc_now_iso(), + "action": action, + "phase": "outcome", + "question_id": question_id, + "operator": operator, + "applied": applied, + **(detail or {}), + }, + ) + + def _require_confirm(action: str, *, confirm: bool) -> None: """Raise unless a destructive ``action`` was explicitly confirmed. @@ -216,22 +280,26 @@ def _cmd_show(args: argparse.Namespace, *, out: Any) -> int: def _cmd_expire(args: argparse.Namespace, *, out: Any) -> int: - """Force-expire an ``open`` question (destructive; audit-logged).""" + """Force-expire an ``open`` question (destructive; audit-logged). + + Audits the attempt BEFORE mutating so a mutation can never land without a + trail (§3.3.1); records the outcome after. + """ _require_confirm("expire", confirm=args.confirm) + _audit_attempt( + args.audit_log, "expire", question_id=args.question_id, operator=args.operator + ) conn = connect(args.db) try: changed = expire_question(conn, question_id=args.question_id) finally: conn.close() - _append_audit( + _audit_outcome( args.audit_log, - { - "ts": _utc_now_iso(), - "action": "expire", - "question_id": args.question_id, - "operator": args.operator, - "applied": changed, - }, + "expire", + question_id=args.question_id, + operator=args.operator, + applied=changed, ) if not changed: print( @@ -244,6 +312,65 @@ def _cmd_expire(args: argparse.Namespace, *, out: Any) -> int: return 0 +def _cmd_redeliver(args: argparse.Namespace, *, out: Any) -> int: + """Clear an ``open`` question's ``channel_ref`` so it is re-posted (§3.3.1). + + The design's "re-deliver" operator action. Re-delivery itself is performed + by the transport reconcile loop; clearing ``channel_ref`` makes that loop + re-post and record a fresh ref. Idempotent and non-destructive (the question + stays ``open``), so it needs no ``--confirm`` — but it is audit-logged for + provenance. Returns ``1`` if the question is absent or not ``open``. + """ + _audit_attempt( + args.audit_log, + "redeliver", + question_id=args.question_id, + operator=args.operator, + ) + conn = connect(args.db) + try: + row = _fetch_question(conn, args.question_id) + if row is None: + applied = False + prior_ref = None + status = None + elif row["status"] != "open": + applied = False + prior_ref = row["channel_ref"] + status = row["status"] + else: + prior_ref = row["channel_ref"] + status = "open" + conn.execute( + "UPDATE pending_questions SET channel_ref=NULL WHERE question_id=?", + (args.question_id,), + ) + applied = True + finally: + conn.close() + _audit_outcome( + args.audit_log, + "redeliver", + question_id=args.question_id, + operator=args.operator, + applied=applied, + detail={"prior_channel_ref": prior_ref}, + ) + if not applied: + reason = "absent" if status is None else f"status={status}, not open" + print( + f"redeliver no-op: question {args.question_id} ({reason}); " + "nothing to re-post", + file=sys.stderr, + ) + return 1 + print( + f"cleared channel_ref for {args.question_id}; reconcile loop will re-post", + file=out, + ) + return 0 + + def _cmd_answer(args: argparse.Namespace, *, out: Any) -> int: """Answer a question on a task's behalf (destructive; audit-logged). @@ -254,6 +381,13 @@ def _cmd_answer(args: argparse.Namespace, *, out: Any) -> int: """ _require_confirm("answer", confirm=args.confirm) via = args.via or f"cli:{args.operator}" + _audit_attempt( + args.audit_log, + "answer", + question_id=args.question_id, + operator=args.operator, + detail={"answered_via": via}, + ) conn = connect(args.db) try: changed = answer_question( @@ -264,16 +398,13 @@ def _cmd_answer(args: argparse.Namespace, *, out: Any) -> int: ) finally: conn.close() - _append_audit( + _audit_outcome( args.audit_log, - { - "ts": _utc_now_iso(), - "action": "answer", - "question_id": args.question_id, - "operator": args.operator, - "answered_via": via, - "applied": changed, - }, + "answer", + question_id=args.question_id, + operator=args.operator, + applied=changed, + detail={"answered_via": via}, ) if not changed: print( @@ -288,22 +419,68 @@ def _cmd_answer(args: argparse.Namespace, *, out: Any) -> int: def _cmd_supersede(args: argparse.Namespace, *, out: Any) -> int: """Mark a stale question ``superseded`` (destructive; audit-logged).""" - _require_confirm("supersede", confirm=args.confirm) + return _supersede_impl(args, out=out, action="supersede") + + +def _cmd_force_resume(args: argparse.Namespace, *, out: Any) -> int: + """Force-resume a parked task's question (destructive; audit-logged). + + The design-named operator verb (§3.3.1 "force-resume" / §6.6 "an operator + can force-resume ... a parked task via the CLI"). The resume worker proper + runs in a later phase; here we mark the stale question ``superseded`` via the + committed turn-guard helper so a redelivered/stale resume job is skipped + idempotently, and record the resume intent durably in the audit trail for + the worker's next sweep to consume. + """ + return _supersede_impl( + args, out=out, action="force-resume", detail={"resume_requested": True} + ) + + +def _supersede_impl( + args: argparse.Namespace, + *, + out: Any, + action: str, + detail: dict[str, Any] | None = None, +) -> int: + """Shared body for ``supersede`` and ``force-resume``. + + Both flip a stale ``open``/``answered`` question to ``superseded`` via the + committed compare-and-set; ``force-resume`` additionally records resume + intent. Audits the attempt before mutating and the outcome after. + """ + _require_confirm(action, confirm=args.confirm) + _audit_attempt( + args.audit_log, + action, + question_id=args.question_id, + operator=args.operator, + detail=detail, + ) conn = connect(args.db) try: changed = supersede_question(conn, question_id=args.question_id) finally: conn.close() - _append_audit( + _audit_outcome( args.audit_log, - { - "ts": _utc_now_iso(), - "action": "supersede", - "question_id": args.question_id, - "operator": args.operator, - "applied": changed, - }, + action, + question_id=args.question_id, + operator=args.operator, + applied=changed, + detail=detail, ) + if action == "force-resume": + # Force-resume succeeds as an operator intent even if there was no stale + # row to supersede (the task may already be past the question); the + # intent is recorded for the resume worker either way. + print( + f"recorded force-resume intent for {args.question_id}" + + ("" if changed else " (no open/answered question to supersede)"), + file=out, + ) + return 0 if not changed: print( f"supersede no-op: question {args.question_id} was not " @@ -382,6 +559,13 @@ def build_parser() -> argparse.ArgumentParser: p_show.add_argument("question_id", help="the question_id to show") p_show.set_defaults(func=_cmd_show) + p_redeliver = sub.add_parser( + "redeliver", + help="clear an open question's channel_ref so it is re-posted", + ) + p_redeliver.add_argument("question_id", help="the question_id to re-deliver") + p_redeliver.set_defaults(func=_cmd_redeliver) + p_expire = sub.add_parser( "expire", help="force-expire an open question (destructive)" ) @@ -425,6 +609,18 @@ def build_parser() -> argparse.ArgumentParser: ) p_supersede.set_defaults(func=_cmd_supersede) + p_resume = sub.add_parser( + "force-resume", + help="force-resume a parked task's question (destructive)", + ) + p_resume.add_argument("question_id", help="the question_id to force-resume") + p_resume.add_argument( + "--confirm", + action="store_true", + help="required: confirm this destructive, audit-logged action", + ) + p_resume.set_defaults(func=_cmd_force_resume) + return parser @@ -442,8 +638,16 @@ def main(argv: Sequence[str] | None = None, *, out: Any = None) -> int: try: return int(args.func(args, out=out)) except PermissionError as exc: + # A refused destructive action (no --confirm) or an unwritable audit + # path. The attempt-before-mutate ordering means nothing was mutated. print(f"error: {exc}", file=sys.stderr) return 1 + except OSError as exc: + # Any other audit-log / filesystem failure (e.g. the audit append could + # not be written). Surfaced cleanly instead of as an uncaught traceback; + # if the attempt record was written, the action is on the trail. + print(f"error: audit/IO failure: {exc}", file=sys.stderr) + return 1 if __name__ == "__main__": # pragma: no cover diff --git a/agent-team/tests/test_builders.py b/agent-team/tests/test_builders.py index 9fbbc82..eee70f8 100644 --- a/agent-team/tests/test_builders.py +++ b/agent-team/tests/test_builders.py @@ -236,6 +236,70 @@ def test_rename_into_out_of_scope_is_flagged() -> None: assert v[0].rename_from == "src/app.py" +# --------------------------------------------------------------------------- # +# Deletes / mode-changes / copies cannot bypass the scan +# (regression: header-only sections carry their path in ``diff --git``, not the +# ``+++ b/`` body line, so a ``+++``-only scan missed all four of these) +# --------------------------------------------------------------------------- # + + +def test_delete_of_denied_path_is_flagged() -> None: + diff = ( + "diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n" + "deleted file mode 100644\n" + "--- a/.github/workflows/ci.yml\n" + "+++ /dev/null\n" + "@@ -1 +0,0 @@\n" + "-on: push\n" + ) + v = scan_trust_control_surface(diff, scope=["src", ".github"]) + assert len(v) == 1 + assert v[0].path == ".github/workflows/ci.yml" + assert "workflow" in v[0].reason + + +def test_mode_change_only_on_denied_path_is_flagged() -> None: + # A chmod with no +++ line at all — only the diff --git header exists. + diff = ( + "diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n" + "old mode 100644\n" + "new mode 100755\n" + ) + v = scan_trust_control_surface(diff, scope=["src", ".github"]) + assert len(v) == 1 + assert v[0].path == ".github/workflows/ci.yml" + assert "workflow" in v[0].reason + + +def test_copy_into_denied_path_is_flagged() -> None: + # git ``copy to`` (not ``rename to``) must not launder a forbidden dest. + diff = ( + "diff --git a/src/x.py b/.github/workflows/evil.yml\n" + "similarity index 100%\n" + "copy from src/x.py\n" + "copy to .github/workflows/evil.yml\n" + ) + v = scan_trust_control_surface(diff, scope=["src", ".github"]) + assert [x.path for x in v] == [".github/workflows/evil.yml"] + assert v[0].rename_from == "src/x.py" + assert "workflow" in v[0].reason + + +def test_out_of_scope_delete_is_flagged() -> None: + diff = ( + "diff --git a/secret/key.py b/secret/key.py\n" + "deleted file mode 100644\n" + "--- a/secret/key.py\n" + "+++ /dev/null\n" + "@@ -1 +0,0 @@\n" + "-KEY = 1\n" + ) + v = scan_trust_control_surface(diff, scope=["src"]) + assert len(v) == 1 + assert v[0].path == "secret/key.py" + assert "scope" in v[0].reason + + # --------------------------------------------------------------------------- # # Unsafe paths (absolute / parent-escaping / indirection) # --------------------------------------------------------------------------- # diff --git a/agent-team/tests/test_run_team.py b/agent-team/tests/test_run_team.py index 67c9834..0035e55 100644 --- a/agent-team/tests/test_run_team.py +++ b/agent-team/tests/test_run_team.py @@ -286,12 +286,17 @@ def test_expire_with_confirm_flips_status_and_audits( assert code == 0 assert _status_of(db_path, "q1") == "expired" entries = [json.loads(line) for line in audit_log.read_text().splitlines()] - assert len(entries) == 1 - assert entries[0]["action"] == "expire" - assert entries[0]["question_id"] == "q1" - assert entries[0]["operator"] == "adam" - assert entries[0]["applied"] is True - assert "ts" in entries[0] + # Attempt is recorded BEFORE the mutation, outcome after, so a mutation can + # never land without a trail (§3.3.1). + assert len(entries) == 2 + assert entries[0]["phase"] == "attempt" + assert "applied" not in entries[0] + assert entries[-1]["phase"] == "outcome" + assert entries[-1]["action"] == "expire" + assert entries[-1]["question_id"] == "q1" + assert entries[-1]["operator"] == "adam" + assert entries[-1]["applied"] is True + assert "ts" in entries[-1] def test_answer_with_confirm_flips_status_records_via( @@ -426,10 +431,13 @@ def test_audit_log_appends_across_actions( _insert_question(db_path, question_id="q2", status="open") _run(cli, db_path, audit_log, "expire", "q1", "--confirm") _run(cli, db_path, audit_log, "answer", "q2", "--answer", "y", "--confirm") - lines = audit_log.read_text().splitlines() - assert len(lines) == 2 - actions = [json.loads(line)["action"] for line in lines] - assert actions == ["expire", "answer"] + entries = [json.loads(line) for line in audit_log.read_text().splitlines()] + # Each destructive action writes an attempt + an outcome record (append-only). + assert len(entries) == 4 + actions = [e["action"] for e in entries] + assert actions == ["expire", "expire", "answer", "answer"] + outcomes = [e["action"] for e in entries if e["phase"] == "outcome"] + assert outcomes == ["expire", "answer"] def test_audit_entries_are_valid_json_lines( @@ -469,3 +477,115 @@ def test_answer_requires_answer_flag(cli: ModuleType) -> None: def test_parked_states_derived_from_foundation(cli: ModuleType) -> None: # The parked-context states are exactly the non-open foundation states. assert set(cli._PARKED_STATES) == set(QUESTION_STATES) - {"open"} + + +# --------------------------------------------------------------------------- # +# re-deliver + force-resume (design-named operator verbs, §3.3.1 / §6.6) +# --------------------------------------------------------------------------- # + + +def _set_channel_ref(db_path: Path, question_id: str, ref: str) -> None: + conn = connect(db_path) + try: + conn.execute( + "UPDATE pending_questions SET channel_ref = ? WHERE question_id = ?", + (ref, question_id), + ) + finally: + conn.close() + + +def _channel_ref_of(db_path: Path, question_id: str) -> str | None: + conn = connect(db_path) + try: + row = conn.execute( + "SELECT channel_ref FROM pending_questions WHERE question_id = ?", + (question_id,), + ).fetchone() + finally: + conn.close() + return None if row is None else row["channel_ref"] + + +def test_redeliver_clears_channel_ref_and_audits( + cli: ModuleType, db_path: Path, audit_log: Path +) -> None: + _insert_question(db_path, question_id="q1", status="open") + _set_channel_ref(db_path, "q1", "slack:123.456") + code, out = _run(cli, db_path, audit_log, "redeliver", "q1") + assert code == 0 + assert _channel_ref_of(db_path, "q1") is None + entries = [json.loads(line) for line in audit_log.read_text().splitlines()] + # Non-destructive but audited: attempt + outcome, no --confirm needed. + assert [e["phase"] for e in entries] == ["attempt", "outcome"] + assert entries[-1]["action"] == "redeliver" + assert entries[-1]["applied"] is True + assert entries[-1]["prior_channel_ref"] == "slack:123.456" + + +def test_redeliver_needs_no_confirm( + cli: ModuleType, db_path: Path, audit_log: Path +) -> None: + # redeliver is not in the destructive set, so it runs without --confirm. + assert "redeliver" not in cli._DESTRUCTIVE_ACTIONS + + +def test_redeliver_non_open_is_noop_returns_1( + cli: ModuleType, db_path: Path, audit_log: Path +) -> None: + _insert_question(db_path, question_id="q1", status="answered") + code, _ = _run(cli, db_path, audit_log, "redeliver", "q1") + assert code == 1 + entries = [json.loads(line) for line in audit_log.read_text().splitlines()] + assert entries[-1]["applied"] is False + + +def test_force_resume_supersedes_and_records_intent( + cli: ModuleType, db_path: Path, audit_log: Path +) -> None: + _insert_question(db_path, question_id="q1", status="open") + code, out = _run( + cli, db_path, audit_log, "--operator", "adam", "force-resume", "q1", "--confirm" + ) + assert code == 0 + assert _status_of(db_path, "q1") == "superseded" + entries = [json.loads(line) for line in audit_log.read_text().splitlines()] + assert [e["phase"] for e in entries] == ["attempt", "outcome"] + assert entries[-1]["action"] == "force-resume" + assert entries[-1]["resume_requested"] is True + assert entries[-1]["operator"] == "adam" + + +def test_force_resume_without_confirm_refuses( + cli: ModuleType, db_path: Path, audit_log: Path +) -> None: + _insert_question(db_path, question_id="q1", status="open") + code, _ = _run(cli, db_path, audit_log, "force-resume", "q1") + assert code == 1 + assert _status_of(db_path, "q1") == "open" # unmutated + assert not audit_log.exists() # refused before any audit (confirm-check first) + + +def test_force_resume_is_in_destructive_set(cli: ModuleType) -> None: + assert "force-resume" in cli._DESTRUCTIVE_ACTIONS + + +# --------------------------------------------------------------------------- # +# Audit-before-mutate: an unwritable audit path aborts BEFORE the ledger mutates +# (regression: previously the row was mutated, then the audit append crashed, +# leaving a mutation with no record and an uncaught traceback) +# --------------------------------------------------------------------------- # + + +def test_unwritable_audit_path_aborts_before_mutation( + cli: ModuleType, db_path: Path, tmp_path: Path +) -> None: + _insert_question(db_path, question_id="q1", status="open") + # Point the audit log at a path whose parent is a FILE, so the atomic write + # of the attempt record fails with OSError before the mutation runs. + blocker = tmp_path / "not-a-dir" + blocker.write_text("x") + bad_audit = blocker / "audit.jsonl" + code, _ = _run(cli, db_path, bad_audit, "expire", "q1", "--confirm") + assert code == 1 # clean failure, not an uncaught traceback + assert _status_of(db_path, "q1") == "open" # NOT mutated — no trail, no change diff --git a/agent-team/tests/test_schema.py b/agent-team/tests/test_schema.py index 35e2c01..2924987 100644 --- a/agent-team/tests/test_schema.py +++ b/agent-team/tests/test_schema.py @@ -226,3 +226,96 @@ def test_concurrent_answers_single_winner(tmp_path: Path) -> None: t.join() assert sorted(results) == [False, True] + + +def test_shared_connection_concurrent_distinct_questions(tmp_path: Path) -> None: + """Regression: many threads share ONE connection, answering DISTINCT questions. + + The responder and resume worker hold a single connection and call the CAS + helpers from different threads concurrently (``connect()`` sets + ``check_same_thread=False``). A single connection cannot hold two explicit + ``BEGIN IMMEDIATE`` transactions at once, so the previous implementation + raised "cannot start a transaction within a transaction" for all but one + thread. The CAS now runs each write on its own private connection, so every + distinct question is answered with no error. + """ + db = tmp_path / "db.sqlite" + init_db(db) + shared = connect(db) + n = 8 + try: + for i in range(n): + _insert_open_question(shared, f"q{i}") + + barrier = threading.Barrier(n) + lock = threading.Lock() + wins: list[bool] = [] + errors: list[BaseException] = [] + + def worker(qid: str) -> None: + try: + barrier.wait() + won = answer_question( + shared, question_id=qid, answer_json='{"v":1}', answered_via="t" + ) + with lock: + wins.append(won) + except BaseException as exc: # noqa: BLE001 - record for assertion + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(f"q{i}",)) for i in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + finally: + shared.close() + + assert errors == [], f"shared-connection CAS raised: {errors!r}" + assert wins == [True] * n + + +def test_shared_connection_concurrent_same_question_single_winner( + tmp_path: Path, +) -> None: + """Regression: shared connection, many threads racing the SAME question. + + Exactly one first-answer-wins, the rest no-op (rowcount 0), and no thread + raises a transaction-nesting or lock error. + """ + db = tmp_path / "db.sqlite" + init_db(db) + shared = connect(db) + n = 8 + try: + _insert_open_question(shared, "race") + + barrier = threading.Barrier(n) + lock = threading.Lock() + wins: list[bool] = [] + errors: list[BaseException] = [] + + def worker(via: str) -> None: + try: + barrier.wait() + won = answer_question( + shared, question_id="race", answer_json='{"v":1}', answered_via=via + ) + with lock: + wins.append(won) + except BaseException as exc: # noqa: BLE001 - record for assertion + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(f"c{i}",)) for i in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + finally: + shared.close() + + assert errors == [], f"shared-connection CAS raised: {errors!r}" + assert sum(wins) == 1 + assert wins.count(False) == n - 1