From 82cfe3da5a65708e48d9776abbc671ffc173c50c Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 25 Aug 2026 15:21:50 +0200 Subject: [PATCH] feat: add emergency cleanup command Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc --- README.md | 25 +++ SCOPE.md | 8 +- WORK-RECORDS.md | 4 + src/tamq/cleanup.py | 220 ++++++++++++++++++++ src/tamq/cli.py | 47 ++++- src/tamq/service.py | 7 +- src/tamq/store.py | 61 ++++++ tests/test_cleanup.py | 187 +++++++++++++++++ tests/test_purge.py | 23 ++ workplans/TAMQ-WP-0013-emergency-cleanup.md | 82 ++++++++ 10 files changed, 660 insertions(+), 4 deletions(-) create mode 100644 src/tamq/cleanup.py create mode 100644 tests/test_cleanup.py create mode 100644 workplans/TAMQ-WP-0013-emergency-cleanup.md diff --git a/README.md b/README.md index e424cdc..32d0d04 100644 --- a/README.md +++ b/README.md @@ -227,6 +227,31 @@ tamq stop tmux kill-session -t tamq ``` +For a runaway or stale installation, use the emergency cleanup command from a +terminal outside the managed session. It is a dry run unless `--yes` is given: + +```bash +tamq cleanup +tamq cleanup --yes +``` + +Confirmed cleanup stops only the verified tamq broker, closes only a tmux +session carrying tamq's management marker, disconnects transient endpoints, +clears leases, removes the configured socket/PID/lock files, deletes only +tamq-generated `@repo` shims, and removes owned stale `tamq-*` tmux sockets. +Durable message history and unrelated tmux sessions or files are preserved. +The command is idempotent; an ownership mismatch is reported instead of being +removed. + +History remains a separate explicit operation. A reflected delivery incident +can be selected by durable receipt direction, dry-run, and then removed without +age-based purging: + +```bash +tamq purge --feedback-chain m-... +tamq purge --feedback-chain m-... --yes +``` + Uninstall the user tool from the checkout with `make uninstall`, or from anywhere with `uv tool uninstall tmux-amq`. diff --git a/SCOPE.md b/SCOPE.md index 88d235c..0e371d3 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -31,6 +31,9 @@ tamq does not choose or infer them. adapter. - Policy profiles, safety-gated retries, local diagnostics, tests, packaging, shell completion, and operator documentation. +- Dry-run emergency cleanup for verified services, tamq-marked sessions, + transient registrations/leases, generated shims, and owned stale sockets; + durable history remains separately controlled. ## Out of Scope @@ -53,9 +56,10 @@ transport. | Intent capability | State | Evidence and remaining gap | | --- | --- | --- | | Direct repository addressing | Implemented for local alpha | Exact `gita` validation, per-session `@repo`/`@repo:` shell commands, tapped `#repo:` agent-input routing, a bare `@` composer with latest-counterparty default and Tab recipient cycling, and long-form parsing are covered. | -| Durable, inspectable local queue | Implemented | SQLite history, manual inbox, leases, endpoint records, inspect/history, JSONL export/replay, acknowledgement, and purge are present. | +| Durable, inspectable local queue | Implemented | SQLite history, manual inbox, leases, endpoint records, inspect/history, JSONL export/replay, acknowledgement, age/size purge, and exact reflected-chain purge are present. | | Local socket service | Implemented | Peer-credential checks and structured ping/register/send/history/ack/endpoints/disconnect operations are tested. | | Neutral tmux session lifecycle | Implemented for local alpha | Repository-first startup opens untouched shells at exact gita paths, exports per-window identity, and runs no initial command unless `--command` is explicit. Stable reuse, service restart, and cleanup are covered by the installed-package test. | +| Emergency local cleanup | Implemented for local alpha | `tamq cleanup` dry-runs by default; confirmed cleanup verifies ownership before stopping the broker or marked session, clears only transient DB state, removes configured runtime files/generated shims/owned stale tmux sockets, and preserves history. | | Safe manual messaging | Implemented for local alpha | Normal endpoints write one sanitized comment above a stable shell input row without injecting stdin; conservative fallback handles the first row and alternate screens. Messages remain pending until acknowledgement. Inbox-only manual mode is explicit with `--no-display`. | | Experimental pushy delivery | Explicit opt-in | `--mode pushy --command ...` observes outbound `#repo:`/`@repo:` lines and submits a non-routable `# from sender:` envelope plus Enter through a checked tmux command. Startup rejects stale framing capabilities, and the tap recognizes durable legacy delivery receipts as a feedback circuit breaker. It marks successful delivery `injected`, but cannot identify pane occupants or protect input already being edited. | | Full-duplex input observation | Explicit opt-in | `--tap --command ...` enables the PTY integration path. It is absent from neutral startup and remains covered for geometry, resize, raw mouse input, and lifecycle behavior. | @@ -98,7 +102,7 @@ Not yet suitable: and stronger process-supervision evidence. - Cross-host messaging or use as a general-purpose broker. -The suite currently has 126 passing tests and 77% statement coverage. Coverage +The suite currently has 131 passing tests and 78% statement coverage. Coverage is strongest in durable storage and registry handling, and weakest in the PTY tap and CLI orchestration; PTY statement coverage increased from 23% to 33%, while subprocess behavior is primarily proven by the real-tmux test. The diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 888807f..dc3b916 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -22,6 +22,7 @@ | workplan | TAMQ-WP-0010 | finished | — | workplans/TAMQ-WP-0010-experimental-pushy-delivery.md | | workplan | TAMQ-WP-0011 | finished | — | workplans/TAMQ-WP-0011-hash-routing-for-pushy-agents.md | | workplan | TAMQ-WP-0012 | finished | — | workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md | +| workplan | TAMQ-WP-0013 | finished | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | | task | TAMQ-WP-ADHOC-2026-08-24-T01 | done | — | workplans/ADHOC-2026-08-24.md | | task | TAMQ-WP-ADHOC-2026-08-25-T01 | done | — | workplans/ADHOC-2026-08-25.md | | task | TAMQ-WP-0001-T01 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | @@ -69,3 +70,6 @@ | task | TAMQ-WP-0012-T01 | done | — | workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md | | task | TAMQ-WP-0012-T02 | done | — | workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md | | task | TAMQ-WP-0012-T03 | done | — | workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md | +| task | TAMQ-WP-0013-T01 | done | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | +| task | TAMQ-WP-0013-T02 | done | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | +| task | TAMQ-WP-0013-T03 | done | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | diff --git a/src/tamq/cleanup.py b/src/tamq/cleanup.py new file mode 100644 index 0000000..fb96dc0 --- /dev/null +++ b/src/tamq/cleanup.py @@ -0,0 +1,220 @@ +from __future__ import annotations + +import asyncio +import os +import socket as socket_module +import stat +from collections.abc import Callable +from pathlib import Path + +from .config import db_path, lock_path, pid_path, socket_path, state_dir +from .service import ping, request +from .store import Store +from .tmux import TmuxError, TmuxManager + + +SHIM_MARKER = "# tamq-address-command v1" + + +def _read_pid() -> int | None: + try: + return int(pid_path().read_text(encoding="utf-8")) + except (FileNotFoundError, OSError, ValueError): + return None + + +def _is_tamq_service(pid: int) -> bool: + try: + command = Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0") + except OSError: + return False + return ( + b"-m" in command + and b"tamq.cli" in command + and b"serve" in command + ) or ( + any(Path(item.decode(errors="ignore")).name == "tamq" for item in command if item) + and b"serve" in command + ) + + +def _generated_shims(directory: Path) -> list[Path]: + try: + candidates = list(directory.iterdir()) + except OSError: + return [] + shims: list[Path] = [] + for candidate in candidates: + if not candidate.name.startswith("@") or candidate.is_symlink(): + continue + try: + if candidate.is_file() and SHIM_MARKER in candidate.read_text( + encoding="utf-8" + ): + shims.append(candidate) + except OSError: + continue + return sorted(shims) + + +def _remove_owned_path(path: Path) -> bool: + try: + mode = path.lstat().st_mode + except FileNotFoundError: + return False + if stat.S_ISDIR(mode): + return False + path.unlink() + return True + + +def _stale_tamq_tmux_sockets() -> list[Path]: + socket_root = Path(os.environ.get("TMUX_TMPDIR", "/tmp")) / f"tmux-{os.getuid()}" + try: + if socket_root.stat().st_uid != os.getuid(): + return [] + candidates = list(socket_root.glob("tamq-*")) + except OSError: + return [] + stale: list[Path] = [] + for candidate in candidates: + try: + metadata = candidate.lstat() + if metadata.st_uid != os.getuid() or not stat.S_ISSOCK(metadata.st_mode): + continue + client = socket_module.socket(socket_module.AF_UNIX) + client.settimeout(0.05) + try: + client.connect(str(candidate)) + except (ConnectionRefusedError, FileNotFoundError): + stale.append(candidate) + except OSError: + pass + finally: + client.close() + except OSError: + continue + return sorted(stale) + + +def cleanup_runtime( + *, + apply: bool, + session: str = "tamq", + manager: TmuxManager | None = None, + stop_service: Callable[[], int | None], +) -> dict[str, object]: + """Plan or perform conservative cleanup of tamq-owned runtime state.""" + manager = manager or TmuxManager(session) + running = asyncio.run(ping()) + pidfile_pid = _read_pid() + reported_pid: int | None = None + if running: + try: + value = asyncio.run(request({"op": "ping"})).get("pid") + reported_pid = int(value) if value is not None else None + except (OSError, TypeError, ValueError): + reported_pid = None + service_verified = not running or ( + pidfile_pid is not None + and ( + reported_pid == pidfile_pid + or (reported_pid is None and _is_tamq_service(pidfile_pid)) + ) + ) + + session_error: str | None = None + try: + identity = manager._existing_session_identity() + except TmuxError as exc: + identity = None + session_error = str(exc) + managed_session = identity is not None + command_dir = manager.command_dir + if managed_session: + configured = manager._run( + "show-options", + "-v", + "-t", + session, + "@tamq_command_dir", + check=False, + ) + if configured: + command_dir = Path(configured) + shims = _generated_shims(command_dir) + stale_tmux_sockets = _stale_tamq_tmux_sockets() + runtime_files = [ + path + for path in (socket_path(), pid_path(), lock_path()) + if path.exists() or path.is_symlink() + ] + + store = Store(db_path()) + try: + messages = len(store.list()) + endpoints = len(store.endpoints()) + leases = int(store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0]) + report: dict[str, object] = { + "apply": apply, + "service_running": running, + "service_pid": reported_pid or pidfile_pid, + "service_verified": service_verified, + "managed_session": managed_session, + "session": session, + "session_error": session_error, + "active_endpoints": endpoints, + "leases": leases, + "history_preserved": messages, + "runtime_files": [str(path) for path in runtime_files], + "generated_shims": [str(path) for path in shims], + "stale_tmux_socket_count": len(stale_tmux_sockets), + "actions": [], + "errors": [], + } + if not apply: + return report + actions = report["actions"] + errors = report["errors"] + assert isinstance(actions, list) + assert isinstance(errors, list) + if running: + if not service_verified: + errors.append("refusing to stop an unverified service PID") + return report + stopped_pid = stop_service() + actions.append(f"stopped service {stopped_pid}") + if asyncio.run(ping()): + errors.append("service remained live after stop request") + return report + if managed_session: + manager._run("kill-session", "-t", session) + actions.append(f"closed managed tmux session {session}") + elif session_error: + errors.append(f"left tmux session untouched: {session_error}") + cleared_endpoints, cleared_leases = store.clear_runtime_state() + actions.append( + f"cleared {cleared_endpoints} endpoint(s) and {cleared_leases} lease(s)" + ) + for path in runtime_files: + if _remove_owned_path(path): + actions.append(f"removed {path}") + for shim in shims: + if _remove_owned_path(shim): + actions.append(f"removed generated shim {shim}") + removed_stale_sockets = sum( + 1 for path in stale_tmux_sockets if _remove_owned_path(path) + ) + if removed_stale_sockets: + actions.append( + f"removed {removed_stale_sockets} stale tamq tmux socket(s)" + ) + try: + if command_dir != state_dir() and command_dir.is_relative_to(state_dir()): + command_dir.rmdir() + actions.append(f"removed empty command directory {command_dir}") + except OSError: + pass + return report + finally: + store.close() diff --git a/src/tamq/cli.py b/src/tamq/cli.py index 256dc87..dfe9473 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -25,6 +25,7 @@ from .broker import BrokerIdentity, InputBroker from .ptytap import PtyTap from .control import ControlModeClient from .composer import ComposerError, compose_message, recipient_order +from .cleanup import cleanup_runtime from .diagnostics import configure from .policy import load_profile from .registry import RegistryError, validate_targets @@ -32,7 +33,7 @@ from .terminal import format_comment SUBCOMMANDS = frozenset( - "start attach serve stop status ping inbox history inspect ack send reply export replay purge completion db-version config tap".split() + "start attach serve stop cleanup status ping inbox history inspect ack send reply export replay purge completion db-version config tap".split() ) START_OPTIONS = frozenset({"--command", "--cmd", "--tap", "--detach", "--no-service", "--no-display", "--mode"}) GLOBAL_FLAGS = frozenset({"--orwell", "--verbose"}) @@ -194,6 +195,16 @@ manual messaging from a managed shell: subparsers.add_parser("serve", help="run the local service in the foreground") subparsers.add_parser("stop", help="stop the local service") + cleanup = subparsers.add_parser( + "cleanup", + help="dry-run emergency shutdown and tamq-owned runtime cleanup", + ) + cleanup.add_argument( + "--yes", + action="store_true", + help="stop the broker, close the managed session, and apply cleanup", + ) + cleanup.add_argument("--session", default="tamq", help="managed tmux session") subparsers.add_parser("status", help="show endpoint health") subparsers.add_parser("ping", help="check local service liveness") history = subparsers.add_parser("history", help="inspect local message history") @@ -229,6 +240,11 @@ manual messaging from a managed shell: purge = subparsers.add_parser("purge", help="dry-run or remove history") purge.add_argument("--before", default=None) purge.add_argument("--max-size", default=None) + purge.add_argument( + "--feedback-chain", + metavar="MESSAGE_ID", + help="select one reflected-delivery chain by its durable root message", + ) purge.add_argument("--yes", action="store_true") purge.add_argument("--orwell", action="store_true") completion = subparsers.add_parser("completion", help="emit shell completion script") @@ -290,6 +306,14 @@ def main(argv: list[str] | None = None) -> int: return 0 print("tamq service is not running", file=sys.stderr) return 1 + if args.command == "cleanup": + report = cleanup_runtime( + apply=args.yes, + session=args.session, + stop_service=stop_service_process, + ) + print(json.dumps(report, sort_keys=True)) + return 1 if report["errors"] else 0 if args.command == "serve": try: asyncio.run(Service().run()) except KeyboardInterrupt: return 0 @@ -561,6 +585,27 @@ def main(argv: list[str] | None = None) -> int: return 2 print(json.dumps({"batch_id": batch_id, "count": count})); return 0 if args.command == "purge": + if args.feedback_chain: + if args.before or args.max_size: + print( + "tamq: --feedback-chain cannot be combined with --before or --max-size", + file=sys.stderr, + ) + return 2 + matching = store.feedback_chain(args.feedback_chain) + if not matching: + print("tamq: feedback-chain root not found", file=sys.stderr) + return 1 + if not args.yes: + print( + f"Dry run: {len(matching)} message(s) match feedback chain rooted at {args.feedback_chain}; re-run with --yes to purge." + ) + return 0 + count = store.delete_messages( + row["message_id"] for row in matching + ) + print(f"Purged {count} feedback-chain message(s).") + return 0 before_text = args.before or setting("purge_before", "365d") max_size_text = args.max_size or setting("purge_max_size", "100MB") before = time.time() - int(before_text[:-1]) * 86400 if before_text.endswith("d") else None diff --git a/src/tamq/service.py b/src/tamq/service.py index bcbb323..f2dca71 100644 --- a/src/tamq/service.py +++ b/src/tamq/service.py @@ -154,7 +154,12 @@ class Service: if protocol.split(".")[0] != PROTOCOL_VERSION.split(".")[0]: response = {"ok": False, "error": f"incompatible protocol: {protocol}", "protocol": PROTOCOL_VERSION} elif op == "ping": - response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": SERVICE_CAPABILITIES} + response = { + "ok": True, + "protocol": PROTOCOL_VERSION, + "pid": os.getpid(), + "capabilities": SERVICE_CAPABILITIES, + } elif op == "register": required = ("endpoint_id", "pid", "session", "repos") if any(key not in request for key in required): diff --git a/src/tamq/store.py b/src/tamq/store.py index c403555..86ab7f9 100644 --- a/src/tamq/store.py +++ b/src/tamq/store.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import re import sqlite3 import time from pathlib import Path @@ -8,6 +9,10 @@ from typing import Iterable from uuid import uuid4 SCHEMA_VERSION = 3 +RECEIPT_SUFFIX = re.compile( + r"\s+\[(m-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\]$", + re.IGNORECASE, +) class Store: @@ -104,6 +109,22 @@ class Store: self.db.execute("UPDATE endpoints SET disconnected_at=strftime('%s','now') WHERE disconnected_at IS NULL") self.db.commit() + def clear_runtime_state(self) -> tuple[int, int]: + """Clear transient leases and disconnect live endpoint registrations.""" + leases = int(self.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0]) + endpoints = int( + self.db.execute( + "SELECT COUNT(*) FROM endpoints WHERE disconnected_at IS NULL" + ).fetchone()[0] + ) + with self.db: + self.db.execute("DELETE FROM leases") + self.db.execute( + "UPDATE endpoints SET disconnected_at=strftime('%s','now') " + "WHERE disconnected_at IS NULL" + ) + return endpoints, leases + def endpoints(self) -> list[sqlite3.Row]: return list( self.db.execute( @@ -158,6 +179,46 @@ class Store: "SELECT * FROM messages WHERE message_id=?", (message_id,) ).fetchone() + def feedback_chain(self, root_id: str) -> list[sqlite3.Row]: + """Resolve a reflected-delivery chain using receipt and direction evidence.""" + rows = self.list() + root = next((row for row in rows if row["message_id"] == root_id), None) + if root is None: + return [] + chain = [root] + by_id = {root_id: root} + for row in rows: + if row["message_id"] == root_id: + continue + receipt = RECEIPT_SUFFIX.search(row["body"]) + if receipt is None: + continue + parent = by_id.get(receipt.group(1)) + if ( + parent is not None + and row["created_at"] >= parent["created_at"] + and row["sender_repo"] == parent["target_repo"] + and row["target_repo"] == parent["sender_repo"] + ): + chain.append(row) + by_id[row["message_id"]] = row + return chain + + def delete_messages(self, message_ids: Iterable[str]) -> int: + ids = tuple(dict.fromkeys(message_ids)) + if not ids: + return 0 + placeholders = ",".join("?" for _ in ids) + count = self.db.execute( + f"SELECT COUNT(*) FROM messages WHERE message_id IN ({placeholders})", + ids, + ).fetchone()[0] + with self.db: + self.db.executemany( + "DELETE FROM messages WHERE message_id=?", ((item,) for item in ids) + ) + return int(count) + def latest_counterparty(self, target: str) -> str | None: row = self.db.execute( "SELECT sender_repo FROM messages " diff --git a/tests/test_cleanup.py b/tests/test_cleanup.py new file mode 100644 index 0000000..a2e1608 --- /dev/null +++ b/tests/test_cleanup.py @@ -0,0 +1,187 @@ +from pathlib import Path +import os +import socket + +from tamq.cleanup import SHIM_MARKER, cleanup_runtime +from tamq.cli import main +from tamq.store import Store + + +class FakeManager: + def __init__(self, command_dir: Path): + self.command_dir = command_dir + self.identity = (42, "tmux-amq-42-test") + self.commands = [] + + def _existing_session_identity(self): + return self.identity + + def _run(self, *args, check=True): + self.commands.append(args) + if args[0] == "show-options": + return str(self.command_dir) + if args[0] == "kill-session": + self.identity = None + return "" + + +def configure_paths(tmp_path, monkeypatch): + state = tmp_path / "state" + runtime = tmp_path / "runtime" + runtime.mkdir() + monkeypatch.setenv("TAMQ_STATE_DIR", str(state)) + monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime)) + monkeypatch.setenv("TAMQ_SOCKET", str(runtime / "tamq.sock")) + monkeypatch.setenv("TMUX_TMPDIR", str(tmp_path / "tmux")) + return state, runtime + + +def test_cleanup_is_dry_run_then_removes_only_owned_runtime( + tmp_path, monkeypatch +): + state, runtime = configure_paths(tmp_path, monkeypatch) + command_dir = state / "commands" / "tamq" + command_dir.mkdir(parents=True) + generated = command_dir / "@audit-core" + generated.write_text(f"#!/bin/sh\n{SHIM_MARKER}\n", encoding="utf-8") + unrelated = command_dir / "keep-me" + unrelated.write_text("operator file\n", encoding="utf-8") + for name in ("tamq.sock", "tamq.pid", "tamq.lock"): + (runtime / name).write_text("stale\n", encoding="utf-8") + tmux_socket_dir = tmp_path / "tmux" / f"tmux-{os.getuid()}" + tmux_socket_dir.mkdir(parents=True) + stale_tmux_socket = tmux_socket_dir / "tamq-old-test" + socket_handle = socket.socket(socket.AF_UNIX) + socket_handle.bind(str(stale_tmux_socket)) + socket_handle.close() + live_tmux_socket = tmux_socket_dir / "tamq-live-test" + live_socket_handle = socket.socket(socket.AF_UNIX) + live_socket_handle.bind(str(live_tmux_socket)) + live_socket_handle.listen() + + store = Store(state / "tamq.sqlite3") + message_id = store.add("a", "b", "preserve me") + store.register_endpoint("ep", 42, "tamq", ["a", "b"]) + assert store.claim(message_id, "ep") is not None + store.close() + manager = FakeManager(command_dir) + + async def not_running(): + return False + + monkeypatch.setattr("tamq.cleanup.ping", not_running) + stopped = [] + dry_run = cleanup_runtime( + apply=False, + manager=manager, + stop_service=lambda: stopped.append(True), + ) + assert dry_run["managed_session"] is True + assert dry_run["active_endpoints"] == 1 + assert dry_run["leases"] == 1 + assert dry_run["history_preserved"] == 1 + assert dry_run["stale_tmux_socket_count"] == 1 + assert generated.exists() + assert (runtime / "tamq.sock").exists() + assert stale_tmux_socket.exists() + assert manager.identity is not None + + applied = cleanup_runtime( + apply=True, + manager=manager, + stop_service=lambda: stopped.append(True), + ) + assert applied["errors"] == [] + assert stopped == [] + assert manager.identity is None + assert not generated.exists() + assert unrelated.exists() + assert not any((runtime / name).exists() for name in ("tamq.sock", "tamq.pid", "tamq.lock")) + assert not stale_tmux_socket.exists() + assert live_tmux_socket.exists() + reopened = Store(state / "tamq.sqlite3") + assert len(reopened.list()) == 1 + assert reopened.endpoints() == [] + assert reopened.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0] == 0 + reopened.close() + + repeated = cleanup_runtime( + apply=True, + manager=manager, + stop_service=lambda: stopped.append(True), + ) + assert repeated["errors"] == [] + assert unrelated.exists() + live_socket_handle.close() + live_tmux_socket.unlink() + + +def test_cleanup_refuses_unverified_running_service(tmp_path, monkeypatch): + state, runtime = configure_paths(tmp_path, monkeypatch) + (runtime / "tamq.pid").write_text("41", encoding="utf-8") + manager = FakeManager(state / "commands" / "tamq") + + async def running(): + return True + + async def request(_payload): + return {"ok": True, "pid": 42} + + monkeypatch.setattr("tamq.cleanup.ping", running) + monkeypatch.setattr("tamq.cleanup.request", request) + stopped = [] + report = cleanup_runtime( + apply=True, + manager=manager, + stop_service=lambda: stopped.append(True), + ) + + assert report["errors"] == ["refusing to stop an unverified service PID"] + assert stopped == [] + assert manager.identity is not None + + +def test_cleanup_stops_verified_service_before_removing_runtime( + tmp_path, monkeypatch +): + state, runtime = configure_paths(tmp_path, monkeypatch) + (runtime / "tamq.pid").write_text("42", encoding="utf-8") + manager = FakeManager(state / "commands" / "tamq") + liveness = iter((True, False)) + + async def ping(): + return next(liveness) + + async def request(_payload): + return {"ok": True, "pid": 42} + + monkeypatch.setattr("tamq.cleanup.ping", ping) + monkeypatch.setattr("tamq.cleanup.request", request) + stopped = [] + report = cleanup_runtime( + apply=True, + manager=manager, + stop_service=lambda: stopped.append(42) or 42, + ) + + assert report["errors"] == [] + assert stopped == [42] + assert manager.identity is None + assert not (runtime / "tamq.pid").exists() + + +def test_cleanup_cli_is_dry_run_by_default(monkeypatch, capsys): + calls = [] + + def cleanup_runtime(**kwargs): + calls.append(kwargs) + return {"errors": [], "apply": kwargs["apply"]} + + monkeypatch.setattr("tamq.cli.cleanup_runtime", cleanup_runtime) + assert main(["cleanup", "--session", "test-session"]) == 0 + assert calls[0]["apply"] is False + assert calls[0]["session"] == "test-session" + assert '"apply": false' in capsys.readouterr().out + + assert main(["cleanup", "--yes"]) == 0 + assert calls[1]["apply"] is True diff --git a/tests/test_purge.py b/tests/test_purge.py index 31f316c..312f103 100644 --- a/tests/test_purge.py +++ b/tests/test_purge.py @@ -27,3 +27,26 @@ def test_purge_confirmed_deletes(tmp_path, monkeypatch, capsys): assert main(["purge", "--yes"]) == 0 assert "Purged" in capsys.readouterr().out assert len(Store(state / "tamq.sqlite3").list()) == 0 + + +def test_feedback_chain_purge_selects_only_reflected_direction( + tmp_path, monkeypatch, capsys +): + state = tmp_path / "state" + monkeypatch.setenv("TAMQ_STATE_DIR", str(state)) + store = Store(state / "tamq.sqlite3") + root = store.add("a", "b", "body") + child = store.add("b", "a", f"body [{root}]") + grandchild = store.add("a", "b", f"body [{root}] [{child}]") + unrelated = store.add("a", "b", f"not reflected [{root}]") + store.close() + + args = ["purge", "--feedback-chain", root] + assert main(args) == 0 + assert "Dry run: 3 message(s)" in capsys.readouterr().out + assert len(Store(state / "tamq.sqlite3").list()) == 4 + + assert main([*args, "--yes"]) == 0 + assert "Purged 3" in capsys.readouterr().out + remaining = Store(state / "tamq.sqlite3").list() + assert [row["message_id"] for row in remaining] == [unrelated] diff --git a/workplans/TAMQ-WP-0013-emergency-cleanup.md b/workplans/TAMQ-WP-0013-emergency-cleanup.md new file mode 100644 index 0000000..e0b2279 --- /dev/null +++ b/workplans/TAMQ-WP-0013-emergency-cleanup.md @@ -0,0 +1,82 @@ +--- +id: TAMQ-WP-0013 +type: workplan +title: "Emergency shutdown and owned-artifact cleanup" +domain: communication +repo: tmux-amq +status: finished +owner: codex +topic_slug: coulomb-social +planning_priority: P0 +planning_order: 18 +created: "2026-08-25" +updated: "2026-08-25" +--- + +# Emergency shutdown and owned-artifact cleanup + +Provide a conservative operator escape hatch for runaway or stale local tamq +state, and remove the preserved WP0012 feedback chain exactly. + +## Purge the incident chain precisely + +```task +id: TAMQ-WP-0013-T01 +status: done +priority: critical +``` + +Resolve descendants of root message +`m-a0570ba1-7088-40f3-892f-8c981b4bb69e` only when their receipt identity, +chronology, and direction match the preceding durable delivery. Prove the +82-record dry run, delete that exact set, and retain all unrelated history. + +## Add dry-run emergency cleanup + +```task +id: TAMQ-WP-0013-T02 +status: done +priority: high +``` + +Add `tamq cleanup`, dry-run by default. Its confirmed form stops a verified +broker, closes only the tamq-marked tmux session, clears transient endpoints +and leases, removes configured runtime socket/PID/lock files, and removes only +generated tamq address shims. Preserve durable history and unrelated tmux +sessions/files. + +## Prove, install, and document recovery + +```task +id: TAMQ-WP-0013-T03 +status: done +priority: high +``` + +Cover ownership checks, stale/missing state, idempotence, dry-run behavior, +feedback-chain selection, and isolated installed cleanup. Update help and +operator documentation, install, and leave the live broker stopped. + +## Completion evidence + +- `tamq purge --feedback-chain m-a0570ba1-7088-40f3-892f-8c981b4bb69e` + dry-ran at exactly 82 records. The broker was stopped again after an external + restart, then the confirmed purge deleted exactly 82 records; the root no + longer exists and 25 unrelated records, including 14 pending, remain. +- `tamq cleanup` reports service identity, managed-session ownership, transient + DB counts, runtime files, generated shims, stale socket count, and preserved + history without mutation. `--yes` refuses unverified service PIDs and + unmarked tmux sessions. +- Confirmed cleanup stops the verified broker and verifies it is down, closes + the marked session, disconnects endpoints, clears leases, removes exact + configured runtime files and marker-owned shims, and removes only owned + `tamq-*` Unix sockets that refuse connections. It preserves live sockets, + unrelated files/sessions, and all durable history. +- The installed isolated smoke proved dry-run non-mutation and confirmed + cleanup, then read the preserved message after broker/session/runtime removal. + It also removed the accumulated dead `tamq-*` tmux sockets from prior test + runs; live and default tmux sockets were excluded. +- `make check`: 131 tests passed. Coverage is 78% overall and 83% for the new + cleanup module. `make install` refreshed `tmux-amq==0.1.0`. +- The operator broker remains stopped. This workplan creates no reliability + residual beyond `TAMQ-WP-0003`.