style: ruff-format branch-added python files

CI lint's format-check on added files (base-range dependent) failed on
files that predate the format gate's effective coverage. Formatting-only:
no semantic changes; integration tests 15/15, ruff check clean, ratchet
PASS, daemon convergence re-verified.

bridge/daemon/completion_file.py

@@ -29,6 +29,7 @@ WHY NOT POLLING:
   that fires exactly when the filesystem changes — the latency tracks
   the real completion event, not a configured interval.
 """
+
 from __future__ import annotations
 
 import json
@@ -38,18 +39,19 @@ import struct
 import time
 from pathlib import Path
 
+
 def _wr_bridge() -> Path:
     # WR_ROOT = daemon/infra convention; RUNE_WR_ROOT = repo/scripts
     # convention. Both honored (WR_ROOT wins) so the daemon, the waiter,
     # and any isolated test root always resolve the SAME directory.
-    root = (os.environ.get("WR_ROOT")
-            or os.environ.get("RUNE_WR_ROOT")
-            or "/mnt/workdrive/wr-bridge")
+    root = os.environ.get("WR_ROOT") or os.environ.get("RUNE_WR_ROOT") or "/mnt/workdrive/wr-bridge"
     return Path(root)
 
+
 def _resumes_dir() -> Path:
     return _wr_bridge() / "resumes"
 
+
 # inotify constants (Linux)
 IN_CREATE = 0x100
 IN_CLOSE_WRITE = 0x8
@@ -115,16 +117,14 @@ def _inotify_read(fd: int, timeout_secs: float | None = None) -> list[str]:
         # struct: wd(i32), mask(u32), cookie(u32), name_len(u32)
         wd, mask, cookie, name_len = struct.unpack_from("iIII", buf, pos)
         pos += 16
-        name = buf[pos:pos + name_len].rstrip(b"\x00").decode("utf-8", "replace")
+        name = buf[pos : pos + name_len].rstrip(b"\x00").decode("utf-8", "replace")
         pos += name_len
         if mask & (IN_CREATE | IN_MOVED_TO):
             events.append(name)
     return events
 
 
-def wait_for_completion(operation_id: str, *,
-                        timeout_secs: float | None = None,
-                        poll_callback=None) -> dict | None:
+def wait_for_completion(operation_id: str, *, timeout_secs: float | None = None, poll_callback=None) -> dict | None:
     """Block until a completion file exists for this operation.
 
     Uses inotify for zero-latency, zero-CPU wake. Falls back to
@@ -161,6 +161,7 @@ def wait_for_completion(operation_id: str, *,
 
         # 3. block in kernel on inotify — zero CPU, exact latency
         import select as _select
+
         poll = _select.poll()
         poll.register(fd, _select.POLLIN)
         deadline = (time.monotonic() + timeout_secs) if timeout_secs else None

bridge/daemon/daemon.py

@@ -32,6 +32,7 @@ The resume command is configurable (WR_RESUME_CMD, JSON list with
 {session_id}/{msg} placeholders) so the test harness can use a stub; the
 default is the real mechanism: `mimo run -s <sid> -m <model> <msg>`.
 """
+
 from __future__ import annotations
 import fcntl, json, os, signal, subprocess, sys, threading, time
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -58,8 +59,7 @@ WATCH_SECS = int(os.environ.get("WR_WATCH_SECS", "600"))
 # progress is an orphaned resume flow (thread died between lease acquire
 # and release). Must exceed the longest legitimate in-flight watch
 # (WATCH_SECS); default 3x. Env-overridable for deterministic tests.
-ORPHAN_GRACE_SECS = int(os.environ.get("WR_ORPHAN_GRACE_SECS",
-                                       str(3 * WATCH_SECS)))
+ORPHAN_GRACE_SECS = int(os.environ.get("WR_ORPHAN_GRACE_SECS", str(3 * WATCH_SECS)))
 TERM_GRACE_SECS = int(os.environ.get("WR_TERM_GRACE_SECS", "10"))
 MAX_ATTEMPTS = int(os.environ.get("WR_MAX_ATTEMPTS", "3"))
 _WR_ROOT = Path(os.environ.get("WR_ROOT", "") or Path(__file__).resolve().parent)
@@ -74,25 +74,23 @@ def log(msg: str) -> None:
         f.write(f"{time.strftime('%H:%M:%S')} {msg}\n")
 
 
-def resume_cmd(session_id: str, msg: str, wait_id: str = "",
-               override: list | None = None) -> list[str]:
+def resume_cmd(session_id: str, msg: str, wait_id: str = "", override: list | None = None) -> list[str]:
     """T15 §15D: per-wait override wins; placeholders {session_id}/{msg}/
     {wait_id}. A synthetic controller session (Scaffolding stage driver)
     is not a MiMo session — the global default cannot resume it, which is
     exactly why fixed-sleep polling crept back into T12-T14."""
     if override:
-        return [c.replace("{session_id}", session_id)
-                 .replace("{msg}", msg)
-                 .replace("{wait_id}", wait_id) for c in override]
+        return [
+            c.replace("{session_id}", session_id).replace("{msg}", msg).replace("{wait_id}", wait_id) for c in override
+        ]
     tmpl = os.environ.get("WR_RESUME_CMD")
     cmd = json.loads(tmpl) if tmpl else DEFAULT_RESUME_CMD
-    return [c.replace("{session_id}", session_id)
-             .replace("{msg}", msg).replace("{wait_id}", wait_id)
-             for c in cmd]
+    return [c.replace("{session_id}", session_id).replace("{msg}", msg).replace("{wait_id}", wait_id) for c in cmd]
 
 
 # ── process-tree evidence & ownership ─────────────────────────────────────
 
+
 def pgid_members(pgid: int) -> list[int]:
     out = []
     for d in Path("/proc").iterdir():
@@ -142,8 +140,7 @@ def stray_worker_pids(session_id: str, exclude_pgids: set[int] = ()) -> list[int
         if not d.name.isdigit():
             continue
         try:
-            parts = [p.decode(errors="replace")
-                     for p in (d / "cmdline").read_bytes().split(b"\0") if p]
+            parts = [p.decode(errors="replace") for p in (d / "cmdline").read_bytes().split(b"\0") if p]
         except OSError:
             continue
         if not parts:
@@ -164,6 +161,7 @@ def stray_worker_pids(session_id: str, exclude_pgids: set[int] = ()) -> list[int
 
 # ── verification → transition (unchanged authority model) ─────────────────
 
+
 def _authoritative_verify(rec: dict) -> tuple[str, dict]:
     """Authoritative verification with a deterministic test seam.
 
@@ -183,11 +181,10 @@ def _authoritative_verify(rec: dict) -> tuple[str, dict]:
         verdict = spec.get(key, spec.get("default", "pending"))
         return verdict, {"stub": True, "key": key}
     import verify
+
     if rec.get("wait_kind") == "RELEASE_DEPLOY":
-        return verify.verify_release_deploy(
-            rec.get("sha", ""), rec.get("repo_root", ""), rec.get("deployed_link", ""))
-    return verify.verify_run(rec.get("run_ref", ""), rec.get("sha", ""),
-                             rec.get("expected", "success"))
+        return verify.verify_release_deploy(rec.get("sha", ""), rec.get("repo_root", ""), rec.get("deployed_link", ""))
+    return verify.verify_run(rec.get("run_ref", ""), rec.get("sha", ""), rec.get("expected", "success"))
 
 
 def try_transition(rec: dict, event: dict | None = None) -> dict:
@@ -207,11 +204,8 @@ def try_transition(rec: dict, event: dict | None = None) -> dict:
         if exit_code is None:
             return rec  # malformed event: stay WAITING (15E hostile case)
         verdict = "success" if exit_code == 0 else "failure"
-        evidence = {"process_event": True,
-                    "exit_code": exit_code,
-                    "operation_id": event.get("run_ref", "")}
-        log(f"verify {rec['wait_id']}: {verdict} "
-            f"{json.dumps(evidence)[:120]}")
+        evidence = {"process_event": True, "exit_code": exit_code, "operation_id": event.get("run_ref", "")}
+        log(f"verify {rec['wait_id']}: {verdict} {json.dumps(evidence)[:120]}")
         rec["status"] = "READY_TO_RESUME"
         rec["verdict"] = verdict
         if verdict == "failure":
@@ -237,24 +231,29 @@ def try_transition(rec: dict, event: dict | None = None) -> dict:
 def continuation_msg(rec: dict) -> str:
     c = rec["continuation"]
     if rec.get("verdict") == "success":
-        return (f"External wait satisfied (wait {rec['wait_id']}).\n"
-                f"Original objective: {c.get('objective','')}\n"
-                f"Waited for: {c.get('waited_for','')}\n"
-                f"Authoritative result: workflow success verified via Gitea API.\n"
-                f"SHA/run: {rec.get('sha','')}/{rec.get('run_ref','')}\n"
-                f"Verify: {c.get('verify_when_woken','')}\n"
-                f"Continue from: {c.get('on_success','')}")
+        return (
+            f"External wait satisfied (wait {rec['wait_id']}).\n"
+            f"Original objective: {c.get('objective', '')}\n"
+            f"Waited for: {c.get('waited_for', '')}\n"
+            f"Authoritative result: workflow success verified via Gitea API.\n"
+            f"SHA/run: {rec.get('sha', '')}/{rec.get('run_ref', '')}\n"
+            f"Verify: {c.get('verify_when_woken', '')}\n"
+            f"Continue from: {c.get('on_success', '')}"
+        )
     ev = json.dumps(rec.get("failure_evidence", {}))[:400]
-    return (f"External wait finished with FAILURE (wait {rec['wait_id']}).\n"
-            f"Original objective: {c.get('objective','')}\n"
-            f"Authoritative result: workflow FAILURE verified via Gitea API.\n"
-            f"Failed job evidence: {ev}\n"
-            f"SHA/run: {rec.get('sha','')}/{rec.get('run_ref','')}\n"
-            f"Do NOT treat this as success. Continue from: {c.get('on_failure','')}")
+    return (
+        f"External wait finished with FAILURE (wait {rec['wait_id']}).\n"
+        f"Original objective: {c.get('objective', '')}\n"
+        f"Authoritative result: workflow FAILURE verified via Gitea API.\n"
+        f"Failed job evidence: {ev}\n"
+        f"SHA/run: {rec.get('sha', '')}/{rec.get('run_ref', '')}\n"
+        f"Do NOT treat this as success. Continue from: {c.get('on_failure', '')}"
+    )
 
 
 # ── resume driver (single-flight, tree-owned) ─────────────────────────────
 
+
 def resume(rec: dict) -> None:
     """Run one continuation to completion. Runs on a dedicated worker thread.
     Preconditions: caller verified no inflight worker for this session."""
@@ -268,9 +267,11 @@ def resume(rec: dict) -> None:
     # stale relative to the waiter's terminal stamp.
     fresh = waitstore.load(wid)
     if fresh and fresh.get("status") in waitstore.TERMINAL:
-        log(f"resume {wid}: skipped — operation already terminal "
+        log(
+            f"resume {wid}: skipped — operation already terminal "
             f"({fresh['status']}); delivery worker never overrides "
-            f"operation truth")
+            f"operation truth"
+        )
         return
     rec = fresh or rec
     msg = continuation_msg(rec)
@@ -299,10 +300,13 @@ def resume(rec: dict) -> None:
     waitstore.save(rec)
 
     try:
-        proc = subprocess.Popen(resume_cmd(sid, msg, wid,
-                                           rec.get("resume_cmd")),
-                                stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-                                text=True, start_new_session=True)
+        proc = subprocess.Popen(
+            resume_cmd(sid, msg, wid, rec.get("resume_cmd")),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            start_new_session=True,
+        )
     except Exception as e:
         waitstore.release_lease(sid, wid)
         rec["status"] = "READY_TO_RETRY"  # spawn failed pre-delivery → retryable
@@ -312,8 +316,7 @@ def resume(rec: dict) -> None:
         return
 
     pgid = proc.pid  # start_new_session ⇒ pgid == child pid
-    waitstore.update_lease(sid, wait_id=wid, pid=proc.pid, pgid=pgid,
-                           pid_starttime=waitstore._pid_starttime(proc.pid))
+    waitstore.update_lease(sid, wait_id=wid, pid=proc.pid, pgid=pgid, pid_starttime=waitstore._pid_starttime(proc.pid))
     log(f"launched {wid}: pid={proc.pid} pgid={pgid} attempt={rec['resume_attempts']}")
 
     timed_out = False
@@ -337,12 +340,12 @@ def resume(rec: dict) -> None:
         # D+E: timeout is NOT success. Tree terminated; retry truthfully.
         if rec["resume_attempts"] >= MAX_ATTEMPTS:
             rec["status"] = "RESUME_UNVERIFIED"
-            rec["resume_note"] = (f"watch timeout x{rec['resume_attempts']}; "
-                                  f"process group {pgid} terminated; attempts exhausted")
+            rec["resume_note"] = (
+                f"watch timeout x{rec['resume_attempts']}; process group {pgid} terminated; attempts exhausted"
+            )
         else:
             rec["status"] = "READY_TO_RETRY"
-            rec["resume_note"] = (f"watch timeout; process group {pgid} terminated "
-                                  f"(TERM→KILL, verified); will retry")
+            rec["resume_note"] = f"watch timeout; process group {pgid} terminated (TERM→KILL, verified); will retry"
         log(f"timeout {wid}: tree {pgid} terminated, rc={rc}")
     elif rc == 0:
         # COMPLETION TRUTH (constraint 4): rc==0 means the headless worker
@@ -352,7 +355,8 @@ def resume(rec: dict) -> None:
         terminal = "COMPLETED" if rec.get("verdict") == "success" else "FAILED"
         rec["status"] = terminal
         rec["resume_evidence"] = {
-            "kind": "worker_exited_rc0", "rc": 0,
+            "kind": "worker_exited_rc0",
+            "rc": 0,
             "output_tail": (stdout or "")[-400:],
             "finished_at": waitstore.now(),
         }
@@ -361,7 +365,8 @@ def resume(rec: dict) -> None:
         # Nonzero rc: delivery/completion unproven. Honest, not schedulable.
         rec["status"] = "RESUME_UNVERIFIED"
         rec["resume_evidence"] = {
-            "kind": "worker_exited_nonzero", "rc": rc,
+            "kind": "worker_exited_nonzero",
+            "rc": rc,
             "output_tail": ((stdout or "") + (stderr or ""))[-400:],
             "finished_at": waitstore.now(),
         }
@@ -372,6 +377,7 @@ def resume(rec: dict) -> None:
 
 # ── startup recovery (invariant F) ────────────────────────────────────────
 
+
 def startup_recovery() -> None:
     for rec in waitstore.all_waits():
         if rec.get("status") != "RESUMING":
@@ -388,8 +394,9 @@ def startup_recovery() -> None:
             survivors = terminate_tree(pgid) if pgid_members(pgid) else []
             waitstore.release_lease(sid, wid)
             rec["status"] = "RESUME_UNVERIFIED"
-            rec["resume_note"] = ("adopted post-restart; outcome unverifiable"
-                                  + (f"; survivors killed: {survivors}" if survivors else ""))
+            rec["resume_note"] = "adopted post-restart; outcome unverifiable" + (
+                f"; survivors killed: {survivors}" if survivors else ""
+            )
             waitstore.save(rec)
             log(f"recover {wid}: adopted worker ended → RESUME_UNVERIFIED")
         else:
@@ -419,9 +426,11 @@ def drive_ready() -> None:
         # target_kind="headless" (disposable/owned sessions) auto-dispatch.
         if rec.get("target_kind") != "headless":
             if not rec.get("hold_reason"):
-                rec["hold_reason"] = ("INTERACTIVE_AUTO_RESUME = NOT YET SAFE: "
-                                      "no supported MiMo busy/idle signal; "
-                                      "queued until operator releases")
+                rec["hold_reason"] = (
+                    "INTERACTIVE_AUTO_RESUME = NOT YET SAFE: "
+                    "no supported MiMo busy/idle signal; "
+                    "queued until operator releases"
+                )
                 waitstore.save(rec)
             continue
         if rec.get("ext", {}).get("resume_mode") == "self":
@@ -443,8 +452,7 @@ def drive_ready() -> None:
             if strays:
                 log(f"drive: {rec['wait_id']} held, session busy {strays}")
                 continue
-            th = threading.Thread(target=resume, args=(rec,), daemon=True,
-                                  name=f"resume-{rec['wait_id']}")
+            th = threading.Thread(target=resume, args=(rec,), daemon=True, name=f"resume-{rec['wait_id']}")
             _inflight[sid] = th
             # start INSIDE the lock: a just-created thread is not yet
             # is_alive() until start() runs — starting outside the lock let
@@ -461,6 +469,7 @@ def reconcile_once() -> None:
         created = rec.get("created_at", "")
         try:
             import calendar
+
             age = time.time() - calendar.timegm(time.strptime(created, "%Y-%m-%dT%H:%M:%SZ"))
         except ValueError:
             continue
@@ -482,13 +491,13 @@ def orphan_sweep() -> int:
     (delivery unverifiable), never auto-relaunch.
     """
     import calendar as _cal
+
     healed = 0
     for rec in waitstore.all_waits():
         if rec.get("status") != "RESUMING":
             continue
         try:
-            updated = _cal.timegm(time.strptime(
-                rec.get("last_updated", ""), "%Y-%m-%dT%H:%M:%SZ"))
+            updated = _cal.timegm(time.strptime(rec.get("last_updated", ""), "%Y-%m-%dT%H:%M:%SZ"))
         except ValueError:
             continue
         if time.time() - updated < ORPHAN_GRACE_SECS:
@@ -496,8 +505,7 @@ def orphan_sweep() -> int:
         wid, sid = rec["wait_id"], rec["session_id"]
         lease = waitstore.read_lease(sid)
         owner_dead = lease is not None and not waitstore._owner_alive(lease)
-        daemon_owned = (lease is not None
-                        and lease.get("pid") == os.getpid())
+        daemon_owned = lease is not None and lease.get("pid") == os.getpid()
         if lease is None or owner_dead or daemon_owned:
             if lease is not None:
                 waitstore.release_lease(sid, wid)
@@ -508,9 +516,9 @@ def orphan_sweep() -> int:
             rec["resume_note"] = (
                 "orphaned mid-run: resume flow died between lease acquire "
                 "and delivery ("
-                + ("lease owner dead" if owner_dead else
-                   "resume thread lost" if daemon_owned else
-                   "no lease") + "); orphan sweep recovered")
+                + ("lease owner dead" if owner_dead else "resume thread lost" if daemon_owned else "no lease")
+                + "); orphan sweep recovered"
+            )
             waitstore.save(rec)
             log(f"orphan-sweep {wid}: {rec['status']} — {rec['resume_note']}")
             healed += 1
@@ -546,6 +554,7 @@ def reconciler() -> None:
 
 # ── webhook receiver (wake signal only) ───────────────────────────────────
 
+
 def wake_sweep(event: dict, eid: str) -> dict:
     """Sweep WAITING_EXTERNAL waits through authoritative verification.
 
@@ -568,8 +577,7 @@ def wake_sweep(event: dict, eid: str) -> dict:
             if rec["status"] != "WAITING_EXTERNAL":
                 continue
             try:
-                age = time.time() - calendar.timegm(
-                    time.strptime(rec.get("created_at", ""), "%Y-%m-%dT%H:%M:%SZ"))
+                age = time.time() - calendar.timegm(time.strptime(rec.get("created_at", ""), "%Y-%m-%dT%H:%M:%SZ"))
             except ValueError:
                 continue
             if age < WAKE_MIN_AGE_SECS:
@@ -584,8 +592,10 @@ def wake_sweep(event: dict, eid: str) -> dict:
         return verified, advanced, pending
 
     verified, advanced, pending = sweep()
-    log(f"local wake {eid}: verified={verified} advanced={advanced} "
-        f"pending={pending} sweep_ms={int((time.time() - arrival) * 1000)}")
+    log(
+        f"local wake {eid}: verified={verified} advanced={advanced} "
+        f"pending={pending} sweep_ms={int((time.time() - arrival) * 1000)}"
+    )
     for attempt in range(1, WAKE_RETRIES + 1):
         # Retry while the sweep didn't conclusively advance anything:
         # pending verdicts (ack-window) AND fresh-wait skips (age gate).
@@ -594,9 +604,11 @@ def wake_sweep(event: dict, eid: str) -> dict:
             break
         time.sleep(WAKE_RETRY_SECS)
         verified, advanced, pending = sweep()
-        log(f"local wake {eid}: retry{attempt} verified={verified} "
+        log(
+            f"local wake {eid}: retry{attempt} verified={verified} "
             f"advanced={advanced} pending={pending} "
-            f"total_ms={int((time.time() - arrival) * 1000)}")
+            f"total_ms={int((time.time() - arrival) * 1000)}"
+        )
     return {"event_id": eid}
 
 
@@ -617,7 +629,9 @@ class Handler(BaseHTTPRequestHandler):
             try:
                 payload = json.loads(body) if body else {}
             except json.JSONDecodeError:
-                self.send_response(400); self.end_headers(); return
+                self.send_response(400)
+                self.end_headers()
+                return
             event = {
                 "provider": str(payload.get("provider", "builder")),
                 "event_type": str(payload.get("event_type", "builder_exit")),
@@ -643,13 +657,14 @@ class Handler(BaseHTTPRequestHandler):
                     # completion files (R10 idempotence by construction).
                     # completed_at is volatile and lives in the durable
                     # event log; the file is a WAKE SIGNAL, never truth.
-                    _stable = {k: event.get(k) for k in
-                               ("provider", "event_type", "run_ref",
-                                "sha", "exit_code") if event.get(k) is not None}
+                    _stable = {
+                        k: event.get(k)
+                        for k in ("provider", "event_type", "run_ref", "sha", "exit_code")
+                        if event.get(k) is not None
+                    }
                     completion_file.write_completion(_op_id, _stable)
                 except Exception as e:  # noqa: BLE001 — delivery best-effort; wait-record transition + watchdog recovery still cover the lost file
-                    log(f"process-event {eid}: completion-file write "
-                        f"failed: {e}")
+                    log(f"process-event {eid}: completion-file write failed: {e}")
             matched = 0
             for rec in waitstore.pending():
                 if rec["status"] == "WAITING_EXTERNAL" and waitstore.match(rec, event):
@@ -661,14 +676,12 @@ class Handler(BaseHTTPRequestHandler):
             # single-flight lease + this guard together make resume
             # idempotent.
             for rec in waitstore.pending():
-                if rec["status"] == "READY_TO_RESUME" \
-                        and rec.get("resume_cmd") \
-                        and rec.get("resume_attempts", 0) == 0:
-                    threading.Thread(target=resume, args=(rec,), daemon=True,
-                                     name=f"resume-{rec['wait_id']}").start()
+                if rec["status"] == "READY_TO_RESUME" and rec.get("resume_cmd") and rec.get("resume_attempts", 0) == 0:
+                    threading.Thread(target=resume, args=(rec,), daemon=True, name=f"resume-{rec['wait_id']}").start()
                     matched += 1
             try:
-                self.send_response(200); self.end_headers()
+                self.send_response(200)
+                self.end_headers()
                 self.wfile.write(json.dumps({"event_id": eid, "matched": matched}).encode())
             except (BrokenPipeError, ConnectionResetError):
                 pass
@@ -679,7 +692,9 @@ class Handler(BaseHTTPRequestHandler):
             try:
                 payload = json.loads(body) if body else {}
             except json.JSONDecodeError:
-                self.send_response(400); self.end_headers(); return
+                self.send_response(400)
+                self.end_headers()
+                return
             event = {
                 "provider": "gitea-local-runner",
                 "task_id": str(payload.get("task_id", "")) or None,
@@ -693,24 +708,30 @@ class Handler(BaseHTTPRequestHandler):
             # fire-and-forget (curl -m 2) — a broken pipe on write is
             # harmless and must not propagate.
             try:
-                threading.Thread(target=wake_sweep, args=(event, eid), daemon=True,
-                                 name=f"wake-{payload.get('task_id','?')}").start()
+                threading.Thread(
+                    target=wake_sweep, args=(event, eid), daemon=True, name=f"wake-{payload.get('task_id', '?')}"
+                ).start()
             except Exception as e:
                 log(f"wake spawn error: {e}")
             try:
-                self.send_response(200); self.end_headers()
+                self.send_response(200)
+                self.end_headers()
                 self.wfile.write(json.dumps({"queued": True, "event_id": eid}).encode())
             except (BrokenPipeError, ConnectionResetError):
                 pass
             return
         if self.path != "/webhook":
-            self.send_response(404); self.end_headers(); return
+            self.send_response(404)
+            self.end_headers()
+            return
         length = int(self.headers.get("Content-Length", 0))
         body = self.rfile.read(length)
         try:
             payload = json.loads(body)
         except json.JSONDecodeError:
-            self.send_response(400); self.end_headers(); return
+            self.send_response(400)
+            self.end_headers()
+            return
         action = payload.get("action") or payload.get("event_type") or ""
         wf = payload.get("workflow_run") or {}
         event = {
@@ -727,15 +748,18 @@ class Handler(BaseHTTPRequestHandler):
             if rec["status"] == "WAITING_EXTERNAL" and waitstore.match(rec, event):
                 matched += 1
                 try_transition(rec)
-        self.send_response(200); self.end_headers()
+        self.send_response(200)
+        self.end_headers()
         self.wfile.write(json.dumps({"event_id": eid, "matched": matched}).encode())
 
     def do_GET(self):
         if self.path == "/health":
-            self.send_response(200); self.end_headers()
+            self.send_response(200)
+            self.end_headers()
             self.wfile.write(b'{"ok":true}')
             return
-        self.send_response(404); self.end_headers()
+        self.send_response(404)
+        self.end_headers()
 
     def log_message(self, *a):
         pass

bridge/daemon/verify.py

@@ -6,6 +6,7 @@ references. Returns (verdict, evidence). verdict ∈ {success, failure,
 pending, unknown}. Only this module may move a wait out of
 WAITING_EXTERNAL.
 """
+
 from __future__ import annotations
 import json, subprocess
 from pathlib import Path
@@ -13,37 +14,51 @@ from pathlib import Path
 HOST = "https://gitea.thingscouldgetdicey.online"
 REPO = "conductor/rune"
 
+
 def _token() -> str:
     # Canonical ladder without importing Rune: env → token file → git store
     import os
+
     tok = os.environ.get("GITEA_TOKEN", "").strip()
-    if tok: return tok
+    if tok:
+        return tok
     tf = os.path.expanduser("~/.config/opencode/secrets/gitea-token")
-    if os.path.isfile(tf): return open(tf).read().strip()
-    r = subprocess.run(["git", "credential", "fill"],
+    if os.path.isfile(tf):
+        return open(tf).read().strip()
+    r = subprocess.run(
+        ["git", "credential", "fill"],
         input="protocol=https\nhost=gitea.thingscouldgetdicey.online\nusername=conductor\n\n",
-        capture_output=True, text=True, timeout=10)
+        capture_output=True,
+        text=True,
+        timeout=10,
+    )
     for line in r.stdout.splitlines():
-        if line.startswith("password="): return line[9:]
+        if line.startswith("password="):
+            return line[9:]
     raise RuntimeError("no gitea credential available")
 
+
 def _api(path: str) -> tuple[int, object]:
     import urllib.request
-    req = urllib.request.Request(f"{HOST}/api/v1/repos/{REPO}{path}",
-        headers={"Authorization": f"token {_token()}"})
+
+    req = urllib.request.Request(f"{HOST}/api/v1/repos/{REPO}{path}", headers={"Authorization": f"token {_token()}"})
     try:
         with urllib.request.urlopen(req, timeout=20) as r:
             return r.status, json.loads(r.read())
     except urllib.error.HTTPError as e:
         return e.code, {"error": e.read().decode()[:200]}
 
+
 def _jobs_list(run_ref: str):
     """GET jobs for a run, unwrapping the {'jobs': [...]} envelope."""
     s, data = _api(f"/actions/runs/{run_ref}/jobs")
-    if s != 200: return s, None
-    if isinstance(data, dict): data = data.get("jobs", [])
+    if s != 200:
+        return s, None
+    if isinstance(data, dict):
+        data = data.get("jobs", [])
     return s, data if isinstance(data, list) else None
 
+
 def _resolve_run_id(run_ref: str, sha: str) -> str:
     """Task-list ids are NOT run ids. If run_ref lookup 404s, resolve the
     real run id by head SHA via /actions/runs (T15: run 6907 vs task 15743)."""
@@ -66,6 +81,7 @@ def _resolve_run_id(run_ref: str, sha: str) -> str:
         return fallback
     return ""
 
+
 def verify_run(run_ref: str, sha: str, expected: str) -> tuple[str, dict]:
     """run_ref = actions run id (or task id — resolved via SHA lookup).
 
@@ -83,7 +99,7 @@ def verify_run(run_ref: str, sha: str, expected: str) -> tuple[str, dict]:
             conclusions = {j.get("name"): j.get("conclusion") for j in jobs}
             statuses = {j.get("status") for j in jobs}
             if statuses <= {"completed"}:
-                verdict = "success" if set(conclusions.values()) <= {"success","skipped"} else "failure"
+                verdict = "success" if set(conclusions.values()) <= {"success", "skipped"} else "failure"
                 return verdict, {"run_ref": run_ref, "jobs": conclusions}
             return "pending", {"run_ref": run_ref, "statuses": sorted(statuses)}
     if sha:
@@ -91,19 +107,28 @@ def verify_run(run_ref: str, sha: str, expected: str) -> tuple[str, dict]:
         if s == 200 and isinstance(st, dict):
             state = st.get("state", "unknown")  # success|error|failure|pending|warning
             rid = _resolve_run_id("", sha)  # attach run id so failure evidence works
-            if state == "success": return "success", {"sha": sha, "state": state, "run_ref": rid}
-            if state in ("error", "failure"): return "failure", {"sha": sha, "state": state, "run_ref": rid}
+            if state == "success":
+                return "success", {"sha": sha, "state": state, "run_ref": rid}
+            if state in ("error", "failure"):
+                return "failure", {"sha": sha, "state": state, "run_ref": rid}
             return "pending", {"sha": sha, "state": state}
     return "unknown", {"run_ref": run_ref, "sha": sha}
 
+
 def failed_job_evidence(run_ref: str, sha: str = "") -> dict:
     """For failure continuations: exact failing job + conclusion."""
     run_ref = _resolve_run_id(run_ref, sha)
-    if not run_ref: return {}
+    if not run_ref:
+        return {}
     s, jobs = _jobs_list(run_ref)
-    if s != 200 or not isinstance(jobs, list): return {}
+    if s != 200 or not isinstance(jobs, list):
+        return {}
     bad = [j for j in jobs if j.get("conclusion") not in ("success", "skipped", None)]
-    return {"failed_jobs": [{ "name": j.get("name"), "conclusion": j.get("conclusion"), "url": j.get("url","")} for j in bad]}
+    return {
+        "failed_jobs": [
+            {"name": j.get("name"), "conclusion": j.get("conclusion"), "url": j.get("url", "")} for j in bad
+        ]
+    }
 
 
 # ── RELEASE_DEPLOY verification (W/R v0.1, overnight handoff Part 2) ──────
@@ -114,10 +139,11 @@ def failed_job_evidence(run_ref: str, sha: str = "") -> dict:
 #   source SHA → source runs terminal → release successor commit (descendant)
 #   → successor runs success → deployed/current identity converges.
 
+
 def _git(args: list[str], repo_root: str) -> str:
     import subprocess
-    r = subprocess.run(["git"] + args, capture_output=True, text=True,
-                       cwd=repo_root or None, timeout=15)
+
+    r = subprocess.run(["git"] + args, capture_output=True, text=True, cwd=repo_root or None, timeout=15)
     if r.returncode != 0:
         raise RuntimeError(f"git {' '.join(args[:2])}: {r.stderr.strip()[:120]}")
     return r.stdout
@@ -147,8 +173,7 @@ def _sha_verdict_from_runs(runs: list[dict]) -> tuple[str, dict]:
     return "success", {"runs": conclusions}
 
 
-def verify_release_deploy(source_sha: str, repo_root: str = "",
-                          deployed_link: str = "") -> tuple[str, dict]:
+def verify_release_deploy(source_sha: str, repo_root: str = "", deployed_link: str = "") -> tuple[str, dict]:
     """RELEASE_DEPLOY authority chain. Returns (verdict, evidence)."""
     source_sha = (source_sha or "")[:12]
     ev: dict = {"source_sha": source_sha, "kind": "RELEASE_DEPLOY"}
@@ -164,9 +189,17 @@ def verify_release_deploy(source_sha: str, repo_root: str = "",
     if not repo_root:
         return "pending", {**ev, "note": "no repo_root — cannot check lineage"}
     try:
-        out = _git(["log", "--format=%H %s", "--reverse",
-                    f"{source_sha}..master", "--grep=^chore\\(release\\)",
-                    "--extended-regexp"], repo_root)
+        out = _git(
+            [
+                "log",
+                "--format=%H %s",
+                "--reverse",
+                f"{source_sha}..master",
+                "--grep=^chore\\(release\\)",
+                "--extended-regexp",
+            ],
+            repo_root,
+        )
     except RuntimeError as e:
         return "pending", {**ev, "note": str(e)}
     successors = [l for l in out.splitlines() if l.strip()]
@@ -206,8 +239,7 @@ def verify_release_deploy(source_sha: str, repo_root: str = "",
                 still = [r for r in _runs_for_sha(succ_sha) if r.get("status") != "completed"]
                 if still:
                     return "pending", {**ev, "note": "deploy propagating (runs in flight)"}
-                return "failure", {**ev,
-                    "note": f"deployed {deployed_id} != successor {succ_sha} (all runs terminal)"}
+                return "failure", {**ev, "note": f"deployed {deployed_id} != successor {succ_sha} (all runs terminal)"}
         else:
             return "failure", {**ev, "note": f"deployed link missing: {deployed_link}"}
 

bridge/daemon/waitstore.py

@@ -16,6 +16,7 @@ Waits persist as waits/<wait_id>.json (atomic tmp+rename).
 Leases persist as leases/<session_id>.lease.json (atomic tmp+rename,
 validated against /proc so a dead or PID-reused owner is stealable).
 """
+
 from __future__ import annotations
 import fcntl, json, os, tempfile, time, uuid
 from pathlib import Path
@@ -27,8 +28,15 @@ LEASES = ROOT / "leases"
 
 SCHEDULABLE = {"READY_TO_RESUME", "READY_TO_RETRY"}
 VALID_STATES = {
-    "RUNNABLE", "WAITING_EXTERNAL", "READY_TO_RESUME", "RESUMING",
-    "COMPLETED", "FAILED", "READY_TO_RETRY", "RESUME_UNVERIFIED", "TIMED_OUT",
+    "RUNNABLE",
+    "WAITING_EXTERNAL",
+    "READY_TO_RESUME",
+    "RESUMING",
+    "COMPLETED",
+    "FAILED",
+    "READY_TO_RETRY",
+    "RESUME_UNVERIFIED",
+    "TIMED_OUT",
 }
 TERMINAL = {"COMPLETED", "FAILED", "RESUME_UNVERIFIED", "TIMED_OUT"}
 
@@ -56,32 +64,53 @@ def now() -> str:
     return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
 
 
-def new_wait(*, session_id: str, provider: str, event_type: str, repo: str,
-             sha: str = "", run_ref: str = "", expected: str = "success",
-             wait_kind: str = "EXACT_WORKFLOW", target_kind: str = "interactive",
-             repo_root: str = "", deployed_link: str = "",
-             objective: str = "", waited_for: str = "", verify_when_woken: str = "",
-             on_success: str = "", on_failure: str = "",
-             resume_cmd: list | None = None) -> dict:
+def new_wait(
+    *,
+    session_id: str,
+    provider: str,
+    event_type: str,
+    repo: str,
+    sha: str = "",
+    run_ref: str = "",
+    expected: str = "success",
+    wait_kind: str = "EXACT_WORKFLOW",
+    target_kind: str = "interactive",
+    repo_root: str = "",
+    deployed_link: str = "",
+    objective: str = "",
+    waited_for: str = "",
+    verify_when_woken: str = "",
+    on_success: str = "",
+    on_failure: str = "",
+    resume_cmd: list | None = None,
+) -> dict:
     wid = f"wr-{int(time.time())}-{uuid.uuid4().hex[:6]}"
     rec = {
-        "wait_id": wid, "session_id": session_id,
+        "wait_id": wid,
+        "session_id": session_id,
         "created_at": now(),
-        "provider": provider, "event_type": event_type, "repo": repo,
-        "sha": sha, "run_ref": run_ref, "expected": expected,
+        "provider": provider,
+        "event_type": event_type,
+        "repo": repo,
+        "sha": sha,
+        "run_ref": run_ref,
+        "expected": expected,
         # T15 §15D: PROCESS_EVENT — a wait whose authoritative verdict is
         # the matching process-exit event itself (exit_code carried on the
         # event), e.g. a builder/controller process completing. The event
         # IS the authority here (the notifier wrapper observed the real
         # process exit); no Gitea lookup applies.
-        "wait_kind": wait_kind if wait_kind in ("EXACT_WORKFLOW", "RELEASE_DEPLOY", "PROCESS_EVENT") else "EXACT_WORKFLOW",
+        "wait_kind": wait_kind
+        if wait_kind in ("EXACT_WORKFLOW", "RELEASE_DEPLOY", "PROCESS_EVENT")
+        else "EXACT_WORKFLOW",
         # CONTENTION POLICY: "interactive" (default, conservative) never
         # auto-dispatches — MiMo exposes NO supported busy/idle session
         # signal, so injecting into a possibly-busy interactive session is
         # NOT YET SAFE. Only explicitly declared headless/disposable
         # targets ("headless") are auto-resumed.
         "target_kind": target_kind if target_kind in ("headless", "interactive") else "interactive",
-        "repo_root": repo_root, "deployed_link": deployed_link,
+        "repo_root": repo_root,
+        "deployed_link": deployed_link,
         # T15 §15D: per-wait resume command override (placeholder template,
         # same substitution as WR_RESUME_CMD: {session_id}/{msg}/{wait_id}).
         # A synthetic controller session (e.g. a Scaffolding stage driver)
@@ -90,12 +119,15 @@ def new_wait(*, session_id: str, provider: str, event_type: str, repo: str,
         # in T12-T14. Per-wait override makes ANY registered continuation
         # target resumable through the same bridge.
         "resume_cmd": list(resume_cmd) if resume_cmd else None,
-        "status": "WAITING_EXTERNAL", "last_verified": "",
+        "status": "WAITING_EXTERNAL",
+        "last_verified": "",
         "resume_attempts": 0,
         "continuation": {
-            "objective": objective, "waited_for": waited_for,
+            "objective": objective,
+            "waited_for": waited_for,
             "verify_when_woken": verify_when_woken,
-            "on_success": on_success, "on_failure": on_failure,
+            "on_success": on_success,
+            "on_failure": on_failure,
         },
     }
     _atomic_write(WAITS / f"{wid}.json", rec)
@@ -124,8 +156,7 @@ def all_waits() -> list[dict]:
 
 
 def pending() -> list[dict]:
-    return [r for r in all_waits()
-            if r.get("status") in {"WAITING_EXTERNAL"} | SCHEDULABLE]
+    return [r for r in all_waits() if r.get("status") in {"WAITING_EXTERNAL"} | SCHEDULABLE]
 
 
 def schedulable() -> list[dict]:
@@ -148,8 +179,10 @@ def requeue(wid: str, note: str) -> dict | None:
 
 # ── events (wake signals, idempotent) ─────────────────────────────────────
 
+
 def record_event(event: dict) -> str:
     import hashlib
+
     body = json.dumps(event, sort_keys=True)
     eid = hashlib.sha256(body.encode()).hexdigest()[:16]
     f = EVENTS / f"{eid}.json"
@@ -216,6 +249,7 @@ def lease_path(session_id: str) -> Path:
     # ids that sanitize identically.
     import hashlib as _hashlib
     import re as _re
+
     safe = _re.sub(r"[^A-Za-z0-9._-]+", "_", session_id)[:80]
     digest = _hashlib.sha256(session_id.encode()).hexdigest()[:10]
     return LEASES / f"{safe}-{digest}.lease.json"
@@ -268,10 +302,13 @@ def acquire_lease(session_id: str, wait_id: str, pid: int, pgid: int) -> dict |
         os.close(fd)
         return None  # live worker owns this session
     lease = {
-        "session_id": session_id, "wait_id": wait_id,
-        "pid": pid, "pgid": pgid,
+        "session_id": session_id,
+        "wait_id": wait_id,
+        "pid": pid,
+        "pgid": pgid,
         "pid_starttime": _pid_starttime(pid),
-        "acquired_at": now(), "state": "RESUMING",
+        "acquired_at": now(),
+        "state": "RESUMING",
     }
     _write_lease_locked(session_id, lease)
     return lease

scripts/install_wr_daemon.py

@@ -11,6 +11,7 @@ way to move code between them:
 
 Never hand-edit the installed copy: edit the repo source, install, restart.
 """
+
 from __future__ import annotations
 
 import argparse
@@ -32,10 +33,8 @@ def sha256(p: Path) -> str:
 
 def main() -> int:
     ap = argparse.ArgumentParser(description=__doc__)
-    ap.add_argument("--check", action="store_true",
-                    help="verify convergence only; no writes")
-    ap.add_argument("--restart", action="store_true",
-                    help="restart the daemon after install")
+    ap.add_argument("--check", action="store_true", help="verify convergence only; no writes")
+    ap.add_argument("--restart", action="store_true", help="restart the daemon after install")
     args = ap.parse_args()
 
     if not CANONICAL.is_dir():
@@ -73,8 +72,7 @@ def main() -> int:
     print("CONVERGED: installed copy == canonical repo source")
 
     if args.restart:
-        rc = subprocess.call([sys.executable,
-                              str(REPO / "scripts" / "start_wr_daemon.py")])
+        rc = subprocess.call([sys.executable, str(REPO / "scripts" / "start_wr_daemon.py")])
         if rc != 0:
             print("restart failed")
             return 1

tests/test_completion_delivery.py

@@ -8,6 +8,7 @@ detection only, not the delivery path). completion_file.py must resolve
 the wr root identically under both env conventions (WR_ROOT / RUNE_WR_ROOT)
 so daemon, waiter, and tests can never disagree about delivery location.
 """
+
 from __future__ import annotations
 
 import importlib
@@ -30,17 +31,16 @@ class CompletionRootResolution(unittest.TestCase):
 
     def _bridge(self, env: dict) -> Path:
         import completion_file
+
         with mock.patch.dict(os.environ, env, clear=False):
             return completion_file._wr_bridge()
 
     def test_rune_wr_root_convention_respected(self):
-        got = self._bridge({"RUNE_WR_ROOT": "/tmp/rune-root-a",
-                            "WR_ROOT": ""})
+        got = self._bridge({"RUNE_WR_ROOT": "/tmp/rune-root-a", "WR_ROOT": ""})
         self.assertEqual(got, Path("/tmp/rune-root-a"))
 
     def test_wr_root_convention_takes_precedence(self):
-        got = self._bridge({"RUNE_WR_ROOT": "/tmp/rune-root-a",
-                            "WR_ROOT": "/tmp/wr-root-b"})
+        got = self._bridge({"RUNE_WR_ROOT": "/tmp/rune-root-a", "WR_ROOT": "/tmp/wr-root-b"})
         self.assertEqual(got, Path("/tmp/wr-root-b"))
 
     def test_default_root_unchanged(self):
@@ -63,6 +63,7 @@ class NotifyWritesDurableCompletion(unittest.TestCase):
                 "RUNE_EVENT_BRIDGE_URL": "http://127.0.0.1:1",  # nothing listens
             }
             import notify_event_bridge as neb
+
             with mock.patch.dict(os.environ, env, clear=False):
                 rc = neb.main()
             self.assertEqual(rc, 0, "notify must never fail CI")
@@ -81,6 +82,7 @@ class NotifyWritesDurableCompletion(unittest.TestCase):
             "RUNE_EVENT_BRIDGE_URL": "http://127.0.0.1:1",
         }
         import notify_event_bridge as neb
+
         with mock.patch.dict(os.environ, env, clear=False):
             rc = neb.main()
         self.assertEqual(rc, 0, "delivery is best-effort; CI must stay green")

tests/test_real_async_integration.py

@@ -13,6 +13,7 @@ Task binding (§9):
   - task-bound resolution scopes to the bound operation; stored state
     wins over model hints
 """
+
 from __future__ import annotations
 
 import importlib
@@ -52,6 +53,7 @@ class _IsolatedRoot(unittest.TestCase):
 
     def _waitstore(self):
         import waitstore
+
         importlib.reload(waitstore)
         return waitstore
 
@@ -59,22 +61,27 @@ class _IsolatedRoot(unittest.TestCase):
         # daemon caches waitstore at import; reload AFTER waitstore so its
         # binding follows this test's isolated root
         import daemon as wrd
+
         importlib.reload(wrd)
         return wrd
 
-    def _new_wait(self, ws, *, session_id, status, task_id="",
-                  resume_cmd=None, stale_secs=0, sha="ab" * 20, **kw):
-        rec = ws.new_wait(session_id=session_id, provider="gitea",
-                          event_type="ci_completed", repo="conductor/rune",
-                          sha=sha, run_ref=f"ci-{sha}",
-                          target_kind="headless", **kw)
+    def _new_wait(self, ws, *, session_id, status, task_id="", resume_cmd=None, stale_secs=0, sha="ab" * 20, **kw):
+        rec = ws.new_wait(
+            session_id=session_id,
+            provider="gitea",
+            event_type="ci_completed",
+            repo="conductor/rune",
+            sha=sha,
+            run_ref=f"ci-{sha}",
+            target_kind="headless",
+            **kw,
+        )
         rec["status"] = status
         rec["ext"] = {"operation_id": rec["run_ref"], "task_id": task_id}
         if resume_cmd is not None:
             rec["resume_cmd"] = resume_cmd
         if stale_secs:
-            rec["last_updated"] = time.strftime(
-                "%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - stale_secs))
+            rec["last_updated"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - stale_secs))
         ws.save(rec)
         return rec
 
@@ -84,22 +91,23 @@ class TestResumeModeSelf(_IsolatedRoot):
 
     def test_self_mode_registers_no_resume_cmd(self):
         import importlib as il
+
         os.environ["RUNE_WR_CODE"] = str(DAEMON_SRC)
         import rune.event_wait as ew
+
         il.reload(ew)
-        rec = ew.register_process_wait("ci-selftest0123", project="rune",
-                                       resume_mode="self")
-        self.assertIsNone(rec.get("resume_cmd"),
-                          "self-mode must not register a delivery worker")
+        rec = ew.register_process_wait("ci-selftest0123", project="rune", resume_mode="self")
+        self.assertIsNone(rec.get("resume_cmd"), "self-mode must not register a delivery worker")
 
     def test_default_mode_preserves_fifo_semantics(self):
         import importlib as il
+
         os.environ["RUNE_WR_CODE"] = str(DAEMON_SRC)
         import rune.event_wait as ew
+
         il.reload(ew)
         rec = ew.register_process_wait("ci-fifotest01234")
-        self.assertIsNotNone(rec.get("resume_cmd"),
-                             "T15 builder semantics must be preserved")
+        self.assertIsNotNone(rec.get("resume_cmd"), "T15 builder semantics must be preserved")
 
 
 class TestTerminalGuard(_IsolatedRoot):
@@ -108,19 +116,16 @@ class TestTerminalGuard(_IsolatedRoot):
     def test_resume_skips_terminal_wait(self):
         ws = self._waitstore()
         wrd = self._daemon()
-        rec = self._new_wait(ws, session_id="s-term", status="COMPLETED",
-                             resume_cmd=["/bin/true"])
+        rec = self._new_wait(ws, session_id="s-term", status="COMPLETED", resume_cmd=["/bin/true"])
         rec["verdict"] = "success"
         ws.save(rec)
         wrd.resume(rec)
         fresh = ws.load(rec["wait_id"])
-        self.assertEqual(fresh["status"], "COMPLETED",
-                         "terminal state was downgraded by delivery path")
+        self.assertEqual(fresh["status"], "COMPLETED", "terminal state was downgraded by delivery path")
         leases = list((self.root / "leases").glob("*.json"))
         self.assertEqual(leases, [], "lease created for terminal wait")
 
 
-
 class TestOrphanSweep(_IsolatedRoot):
     """§4: bounded mid-run reconciliation of orphaned resume flows."""
 
@@ -128,19 +133,22 @@ class TestOrphanSweep(_IsolatedRoot):
         ws = self._waitstore()
         wrd = self._daemon()
         wrd.ORPHAN_GRACE_SECS = 5  # deterministic, small
-        rec = self._new_wait(ws, session_id="s-orphan", status="RESUMING",
-                             resume_cmd=["/bin/true"])
+        rec = self._new_wait(ws, session_id="s-orphan", status="RESUMING", resume_cmd=["/bin/true"])
         rec["resume_attempts"] = 1
         ws.save(rec)
         # backdate AFTER save: save() stamps last_updated=now
-        stale = time.strftime("%Y-%m-%dT%H:%M:%SZ",
-                              time.gmtime(time.time() - 60))
+        stale = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - 60))
         rec["last_updated"] = stale
         (ws.WAITS / f"{rec['wait_id']}.json").write_text(json.dumps(rec))
         # dead-owner lease
-        lease = {"session_id": "s-orphan", "wait_id": rec["wait_id"],
-                 "pid": 2**22, "pgid": 2**22, "pid_starttime": "",
-                 "acquired_at": "2020-01-01T00:00:00Z"}
+        lease = {
+            "session_id": "s-orphan",
+            "wait_id": rec["wait_id"],
+            "pid": 2**22,
+            "pgid": 2**22,
+            "pid_starttime": "",
+            "acquired_at": "2020-01-01T00:00:00Z",
+        }
         ws.lease_path("s-orphan").write_text(json.dumps(lease))
         healed = wrd.orphan_sweep()
         self.assertEqual(healed, 1)
@@ -153,13 +161,13 @@ class TestOrphanSweep(_IsolatedRoot):
         ws = self._waitstore()
         wrd = self._daemon()
         wrd.ORPHAN_GRACE_SECS = 3600
-        rec = self._new_wait(ws, session_id="s-live", status="RESUMING",
-                             resume_cmd=["/bin/true"])
+        rec = self._new_wait(ws, session_id="s-live", status="RESUMING", resume_cmd=["/bin/true"])
         self.assertEqual(wrd.orphan_sweep(), 0)
         self.assertEqual(ws.load(rec["wait_id"])["status"], "RESUMING")
 
     def _waitstore(self):
         import waitstore
+
         importlib.reload(waitstore)
         return waitstore
 
@@ -169,25 +177,25 @@ class TestCliContract(unittest.TestCase):
 
     def test_unknown_flag_rejected_loudly(self):
         import ci_event_wait as cew
-        with mock.patch.object(sys, "argv",
-                               ["ci_event_wait.py", "--sha", "a" * 40,
-                                "--watchdoog", "5"]):
+
+        with mock.patch.object(sys, "argv", ["ci_event_wait.py", "--sha", "a" * 40, "--watchdoog", "5"]):
             rc = cew.main()
         self.assertEqual(rc, 2, "unknown flag must exit 2, not be ignored")
 
     def test_positional_rejected_loudly(self):
         import ci_event_wait as cew
-        with mock.patch.object(sys, "argv",
-                               ["ci_event_wait.py", "extra"]):
+
+        with mock.patch.object(sys, "argv", ["ci_event_wait.py", "extra"]):
             rc = cew.main()
         self.assertEqual(rc, 2)
 
     def test_watchdog_flag_reaches_run(self):
         import ci_event_wait as cew
-        with mock.patch.object(cew, "run", return_value=0) as run_mock, \
-             mock.patch.object(sys, "argv",
-                               ["ci_event_wait.py", "--sha", "a" * 40,
-                                "--watchdog", "123"]):
+
+        with (
+            mock.patch.object(cew, "run", return_value=0) as run_mock,
+            mock.patch.object(sys, "argv", ["ci_event_wait.py", "--sha", "a" * 40, "--watchdog", "123"]),
+        ):
             rc = cew.main()
         self.assertEqual(rc, 0)
         self.assertEqual(run_mock.call_args.kwargs.get("watchdog_secs"), 123)
@@ -198,34 +206,31 @@ class TestTaskBinding(_IsolatedRoot):
 
     def _bind_env(self):
         import await_intent as ai
+
         importlib.reload(ai)
         return ai
 
     def test_task_bound_operation_resolved(self):
         ws = self._waitstore()
         self._new_wait(ws, session_id="s-stale", status="WAITING_EXTERNAL")
-        bound = self._new_wait(ws, session_id="s-bound",
-                               status="WAITING_EXTERNAL", task_id="T9",
-                               sha="cd" * 20)
+        bound = self._new_wait(ws, session_id="s-bound", status="WAITING_EXTERNAL", task_id="T9", sha="cd" * 20)
         ai = self._bind_env()
-        norm = ai.normalize_action({"kind": "await_operation"},
-                                   task_id="T9")
+        norm = ai.normalize_action({"kind": "await_operation"}, task_id="T9")
         self.assertTrue(norm.get("normalized"), norm)
         self.assertEqual(norm.get("operation_id"), bound["run_ref"])
-        self.assertEqual(norm.get("matched_by", norm.get("evidence")),
-                         norm.get("matched_by", norm.get("evidence")))
+        self.assertEqual(norm.get("matched_by", norm.get("evidence")), norm.get("matched_by", norm.get("evidence")))
 
     def test_unbound_task_reports_missing(self):
         ws = self._waitstore()
         self._new_wait(ws, session_id="s-stale2", status="WAITING_EXTERNAL")
         ai = self._bind_env()
-        norm = ai.normalize_action({"kind": "await_operation"},
-                                   task_id="T404")
+        norm = ai.normalize_action({"kind": "await_operation"}, task_id="T404")
         self.assertEqual(norm.get("normalized"), False)
         self.assertIn("T404", norm.get("reason", ""))
 
     def _waitstore(self):
         import waitstore
+
         importlib.reload(waitstore)
         return waitstore