feat: add emergency cleanup command

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc

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()

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

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):

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 "