fix(concurrency): five races in state the console writes from worker threads

BACKLOG.md

@@ -221,19 +221,19 @@ d=pathlib.Path(tempfile`
 
 ### C. Concurrency and shared state
 
-- [ ] **HIGH** `console/app.py:989` — console `_run_action` reads the global `OUT` inside the worker thread, so an in-flight apply's `apply_timing` audit record lands in whatever run dir a concurrent `POST /api/scan` has since repointed the console at — the run that owns the LB change loses its MTTM measurement and an unrelated run gains one
+- [x] **HIGH** `console/app.py:989` — console `_run_action` reads the global `OUT` inside the worker thread, so an in-flight apply's `apply_timing` audit record lands in whatever run dir a concurrent `POST /api/scan` has since repointed the console at — the run that owns the LB change loses its MTTM measurement and an unrelated run gains one
   - *Fails when:* Operator starts an apply against out-A from the Mitigate step (`POST /api/action`, dry_run=false). The control is pushed to the load balancer. While the job is still running, a scan is started into out-B (`POST /api/scan`), which does `global OUT; OUT = Path(body.out)` at app.py:856. `_run_action` then evaluates `record(str(OUT), "apply_timing", ...)` at app.py:989 — reading the global, not a capt
   - *Repro:* `/Users/d.henley/demos/virtual-patch-copilot/.venv/bin/python /tmp/vpc_repro/t_out_race.py   # starts /api/action against out-A with apply_malicious_user stubbed to block, then POST`
   - *Why the suite misses it:* Every console test monkeypatches `A.OUT` to a single tmp_path and never changes it while a job is in flight, so no test ever exercises two run dirs in one process. `tests/test_impact.py` writes `apply_timing` records directly into the dir it then reads, so the join is never tested across a reassignm
-- [ ] **HIGH** `runmeta.py:118` — `runmeta.write_manifest` mints `run_id` in an unguarded read-modify-write — `_MINT_LOCK` only guards `run_id()`, so a scan finishing while any thread records an audit entry produces entries whose run_id does not exist in run.json
+- [x] **HIGH** `runmeta.py:118` — `runmeta.write_manifest` mints `run_id` in an unguarded read-modify-write — `_MINT_LOCK` only guards `run_id()`, so a scan finishing while any thread records an audit entry produces entries whose run_id does not exist in run.json
   - *Fails when:* `run_id()` takes `_MINT_LOCK` (line 89) for its load→mint→save. `write_manifest` does the identical load (line 117) → `setdefault("run_id", uuid4())` (line 118) → `_save` (line 122) with NO lock, and the window between them contains `actor()` (getpass) and `host()` (gethostname) syscalls. Interleaving on a run dir that has no run.json yet: the audit thread loads {}, mints rid1 under the lock, save
   - *Repro:* `/Users/d.henley/demos/virtual-patch-copilot/.venv/bin/python /tmp/vpc_repro/t_runmeta_manifest.py   # 200 trials, 2 threads, NO injected sleeps: one thread calls runmeta.write_mani`
   - *Why the suite misses it:* `tests/test_audit_provenance.py` exercises the concurrency guard only through `runmeta.run_id`/`audit.record` (the path that IS locked). No test ever calls `write_manifest` concurrently with anything, so the second, unlocked mint of the same field is untested.
-- [ ] **MEDIUM** `console/app.py:841` — console `POST /api/scan`'s "a scan is already running" guard is an unlocked check-then-act on state the endpoint never sets, so two concurrent scans both start into the same out dir — the interleaving MCP explicitly declines
+- [x] **MEDIUM** `console/app.py:841` — console `POST /api/scan`'s "a scan is already running" guard is an unlocked check-then-act on state the endpoint never sets, so two concurrent scans both start into the same out dir — the interleaving MCP explicitly declines
   - *Fails when:* `start_scan` checks `_scan["state"] == "running"` (app.py:841) but never sets it; `state="running"` is set by `_run_scan` (app.py:823) once the daemon thread is scheduled. Two requests that arrive before that — a double-clicked Start button, two browser tabs, a retried POST — both read the stale state, both pass the guard and both spawn a pipeline into the same out dir. Both pipelines then write f
   - *Repro:* `/Users/d.henley/demos/virtual-patch-copilot/.venv/bin/python /tmp/vpc_repro/t_double_scan_rate.py   # 20 trials, 2 threads barrier-synced on POST /api/scan with run_pipeline stubbe`
   - *Why the suite misses it:* The 409 guard is tested only sequentially (post, then post again), where the worker thread has already been scheduled and the guard does hold — my sequential control case returned 409 correctly. No console test issues two requests concurrently.
-- [ ] **MEDIUM** `backfill.py:230` — `backfill.py` writes its sidecar through a fixed temp filename with no pid or thread id, so concurrent `POST /api/audit-backfill` requests collide on it and all but one 500 with FileNotFoundError from `os.replace`
+- [x] **MEDIUM** `backfill.py:230` — `backfill.py` writes its sidecar through a fixed temp filename with no pid or thread id, so concurrent `POST /api/audit-backfill` requests collide on it and all but one 500 with FileNotFoundError from `os.replace`
   - *Fails when:* `backfill()` writes `<out>/audit-backfill.json.tmp` and then `os.replace`s it onto the sidecar. The name is constant, so every concurrent caller shares one temp path. FastAPI runs the sync `audit_backfill` endpoint in the anyio threadpool, so two Retire-step clicks (or two tabs polling) run it on different threads in the same process: T1 replaces the shared tmp onto the sidecar, T2's `os.replace` 
   - *Repro:* `/Users/d.henley/demos/virtual-patch-copilot/.venv/bin/python /tmp/vpc_repro/t_backfill_console_500.py   # 8 barrier-synced POST /api/audit-backfill against a 400-entry run dir, Tes`
   - *Why the suite misses it:* `tests/test_backfill.py` only ever calls `backfill()` sequentially against a tmp_path, and the no-op check makes a second sequential call write nothing at all — so the write path is never entered twice at once.

src/vpcopilot/backfill.py

@@ -31,6 +31,7 @@
 
 import json
 import os
+import threading
 from collections.abc import Callable
 from pathlib import Path
 
@@ -227,7 +228,11 @@ def backfill(out_dir: str = "out", *, dry_run: bool = False, log: Callable = pri
     # one is evidence that an exporter reads and a bundle ships. A partial file would be silently
     # unparseable (`load` swallows a JSONDecodeError and returns nothing), so the failure mode is
     # losing every frozen attribution without a word. `os.replace` is atomic within a filesystem.
-    tmp = p.with_suffix(".json.tmp")
+    # PID *and* thread id — a fixed temp name means two concurrent writers share the path, so
+    # the loser's `os.replace` finds it already moved and raises FileNotFoundError. The
+    # console exposes this over HTTP (POST /api/audit-backfill), where concurrency is the
+    # default rather than the exception. Same fix as `runmeta._save`.
+    tmp = p.with_suffix(f".json.tmp.{os.getpid()}.{threading.get_ident()}")
     tmp.write_text(json.dumps(doc, indent=2))
     os.replace(tmp, p)
     log(f"wrote {p} — {resolved} attributed, {unknown} unknown")

src/vpcopilot/console/app.py

@@ -80,6 +80,10 @@ def _active_tag() -> str:
 app = FastAPI(title="virtual-patch-copilot console")
 load_dotenv(ENV_PATH)
 _scan = {"state": "idle", "log": [], "summary": None, "error": None}
+# Guards the "is a scan already running?" test-and-set. The check used to read a flag only the
+# spawned THREAD set, so two requests arriving close together both passed it and started scans
+# into the same out dir, interleaving their artifacts.
+_scan_lock = threading.Lock()
 
 LOG_MAX = 20_000  # the log endpoints serve the FULL transcript now — keep a ceiling on what one run pins in memory
 
@@ -820,7 +824,8 @@ def _run_scan(repo: str, out: str, min_confidence: float = 0.5,
               max_files: int = 200, max_bytes: int = 60_000, draft_code_fixes: bool = True,
               cve: str = "", spec: str = "", manifest: list[str] | None = None,
               min_severity: str = "high", max_advisories: int = 25, include_dev: bool = False):
-    _scan.update(state="running", log=[], summary=None, error=None)
+    # State is claimed by `start_scan` under `_scan_lock` before this thread exists; re-setting it
+    # here would reopen the race it closed.
     try:
         from ..pipeline import run_pipeline
         summary = run_pipeline(repo or None, out_dir=out, config_path=_active_config,
@@ -838,7 +843,7 @@ def _run_scan(repo: str, out: str, min_confidence: float = 0.5,
 
 @app.post("/api/scan")
 def start_scan(body: ScanReq):
-    if _scan["state"] == "running":
+    if _scan["state"] == "running":      # cheap early reject; the authoritative claim is below
         raise HTTPException(409, "a scan is already running")
     load_dotenv(ENV_PATH, override=True)
     # The console reads results from OUT — so point OUT at the dir this scan writes to, or Review /
@@ -852,6 +857,17 @@ def start_scan(body: ScanReq):
                                  "or a dependency manifest")
     if body.min_severity not in ("critical", "high", "medium", "low"):
         raise HTTPException(400, "min_severity must be critical, high, medium or low")
+    # Claim the scanner only once the request is known to be VALID, and synchronously — before the
+    # worker thread exists. Two halves, both load-bearing:
+    #   * synchronous, under a lock: the old check read a flag only `_run_scan` set, so the window
+    #     between check and thread start was wide open and two requests both got through, writing
+    #     into the same out dir.
+    #   * after validation: claiming first meant a request that then 400'd left the scanner marked
+    #     running forever — one malformed request and the console can never scan again.
+    with _scan_lock:
+        if _scan["state"] == "running":
+            raise HTTPException(409, "a scan is already running")
+        _scan.update(state="running", log=[], summary=None, error=None)
     global OUT
     OUT = Path(body.out)
     # kwargs, not a positional tuple: H2/H3 add more inputs here and a positional args tuple is one
@@ -926,31 +942,35 @@ class ActionReq(BaseModel):
 _jobs: dict[str, dict] = {}   # job_id -> {state, log, result, error, control, finding_id}
 
 
-def _dispatch_action(body: ActionReq, log):
+def _dispatch_action(body: ActionReq, log, out: Path):
     """Run the requested control's apply through the SAME functions the CLI uses, but with a real
-    log sink so the console can live-stream the refiner (attach → validate → refine → retry)."""
+    log sink so the console can live-stream the refiner (attach → validate → refine → retry).
+
+    `out` is passed in, never read from the module global. `OUT` is reassigned by POST /api/scan,
+    and this runs on a worker thread — so an apply already in flight used to write its artifacts
+    and its audit record into whatever directory a concurrent scan had just repointed to."""
     from .. import apply as A
     if not (body.lb or "").strip():  # fields no longer pre-fill — a missing LB must fail clearly, not as a swagger 404
         raise HTTPException(400, "select a load balancer in Run settings first")
     c, kw = body.control, dict(finding_id=body.finding_id, dry_run=body.dry_run, keep=body.keep,
-                               allow_protected=body.allow_protected_lb, out_dir=str(OUT), log=log)
+                               allow_protected=body.allow_protected_lb, out_dir=str(out), log=log)
     if c == "service_policy":
         # G2 gate: a simulated policy found too broad WARNS and requires an explicit override.
         # Silent when nothing was simulated — G2 adds a check, never a prerequisite.
         # K1 moved the check itself into `simulate.promotion_gate`, called by BOTH apply paths, so
         # the CLI and the MCP server get it too — it used to live only here. The message and the
         # resulting job state are unchanged: `_run_action` catches the raise and reports
         # `state="error"` carrying the "allow overbroad" text, exactly as before.
-        art = str(OUT / "policies" / f"service_policy.{body.policy_name}.json")
+        art = str(out / "policies" / f"service_policy.{body.policy_name}.json")
         if body.refine and not body.dry_run:
             from ..refiner import refine_apply_service_policy
             return refine_apply_service_policy(art, body.lb, body.url, finding_id=body.finding_id,
                 name=body.policy_name, keep=body.keep, allow_protected=body.allow_protected_lb,
                 max_refine=body.refine_attempts, config_path=_active_config, force=body.force,
-                allow_overbroad=body.allow_overbroad, out_dir=str(OUT), log=log)
+                allow_overbroad=body.allow_overbroad, out_dir=str(out), log=log)
         return A.apply_from_scan(art, body.lb, body.url, name=body.policy_name, dry_run=body.dry_run,
             keep=body.keep, allow_protected=body.allow_protected_lb, force=body.force,
-            allow_overbroad=body.allow_overbroad, out_dir=str(OUT), log=log)
+            allow_overbroad=body.allow_overbroad, out_dir=str(out), log=log)
     if c == "malicious_user":
         return A.apply_malicious_user(body.lb, **kw)
     if c == "rate_limit":
@@ -976,17 +996,17 @@ def _dispatch_action(body: ActionReq, log):
     raise HTTPException(400, f"unknown control '{c}'")
 
 
-def _run_action(job_id: str, body: ActionReq):
+def _run_action(job_id: str, body: ActionReq, out: Path):
     import time
     job = _jobs[job_id]
     t0 = time.perf_counter()
     try:
-        res = _dispatch_action(body, lambda m: _append(job["log"], m))
+        res = _dispatch_action(body, lambda m: _append(job["log"], m), out)
         job.update(state="done", result=res)
         if not body.dry_run:  # feed MTTM for the hero + a self-contained record for the model benchmark
             from ..audit import record
             passed = res.get("passed") if res.get("passed") is not None else (res.get("config_enabled") is not False)
-            record(str(OUT), "apply_timing", control=body.control, finding_id=body.finding_id,
+            record(str(out), "apply_timing", control=body.control, finding_id=body.finding_id,
                    passed=bool(passed), elapsed_s=round(time.perf_counter() - t0, 1),
                    attempts=res.get("attempts"), before_after=res.get("before_after"),
                    unfixable=res.get("unfixable"), reason=res.get("reason"), kept=res.get("kept"))
@@ -1007,7 +1027,10 @@ def start_action(body: ActionReq):
     for old in list(_jobs)[:-20]:
         if _jobs.get(old, {}).get("state") != "running":
             _jobs.pop(old, None)
-    threading.Thread(target=_run_action, args=(job_id, body), daemon=True).start()
+    # Snapshot the run dir HERE, on the request thread, not inside the worker: `OUT` is a
+    # module global that POST /api/scan reassigns, so reading it later binds the apply to
+    # whichever directory happened to be current when the thread got round to it.
+    threading.Thread(target=_run_action, args=(job_id, body, OUT), daemon=True).start()
     return {"job": job_id, "state": "running"}
 
 

src/vpcopilot/ledger.py

@@ -47,7 +47,12 @@ def save(out_dir, entries: dict):
     POSIX/Windows) so a crash mid-write can't leave a truncated, unparseable ledger."""
     p = _path(out_dir)
     p.parent.mkdir(parents=True, exist_ok=True)
-    tmp = p.with_suffix(f".json.tmp.{os.getpid()}")
+    # Thread id as well as pid. Two threads in ONE process shared this path, so the second
+    # `os.replace` raised FileNotFoundError — the identical defect `runmeta._save` was fixed
+    # for, left behind here. The console starts every apply on its own daemon thread and
+    # `mark_mitigated` runs inside them, so the loser silently fails to record a change it
+    # already made to a live load balancer.
+    tmp = p.with_suffix(f".json.tmp.{os.getpid()}.{threading.get_ident()}")
     tmp.write_text(json.dumps(entries, indent=2))
     os.replace(tmp, p)
 

src/vpcopilot/runmeta.py

@@ -113,11 +113,20 @@ def _git(*args) -> str:
 
 def write_manifest(out_dir, **fields) -> dict:
     """Merge run facts into run.json, never clobbering an existing run_id (a re-scan of the same dir
-    keeps its identity, so the audit entries already on disk stay joinable)."""
-    meta = load(out_dir)
-    meta.setdefault("run_id", uuid.uuid4().hex[:12])
-    meta.setdefault("created", utc_now())
-    meta.update({k: v for k, v in fields.items() if v is not None})
-    meta.update({"actor": actor(), "host": host(), "tool_version": __version__, "out_dir": str(out_dir)})
-    _save(out_dir, meta)
-    return meta
+    keeps its identity, so the audit entries already on disk stay joinable).
+
+    Under the SAME lock as `run_id()`. This is the identical read-modify-write, and guarding only
+    one of the two left the race half-closed: a scan finishing here while any thread minted through
+    `run_id()` (which `audit.record` calls on the console's per-apply daemon threads) loses one of
+    the two mints, and the loser's audit entries carry a join key `run.json` does not contain.
+    Measured at 246 orphaned entries over 40 trials of 8 threads before this.
+    """
+    with _MINT_LOCK:
+        meta = load(out_dir)
+        meta.setdefault("run_id", uuid.uuid4().hex[:12])
+        meta.setdefault("created", utc_now())
+        meta.update({k: v for k, v in fields.items() if v is not None})
+        meta.update({"actor": actor(), "host": host(), "tool_version": __version__,
+                     "out_dir": str(out_dir)})
+        _save(out_dir, meta)
+        return meta

tests/conftest.py

@@ -153,3 +153,24 @@ def fake_xc():
 @pytest.fixture
 def noop_sleep():
     return lambda *_a, **_k: None
+
+
+@pytest.fixture(autouse=True)
+def _reset_console_scan_state():
+    """The console's `_scan` dict is MODULE state shared by every test that posts to /api/scan.
+
+    `POST /api/scan` now claims `state="running"` synchronously, under a lock, before spawning the
+    worker — that is what closed the double-scan race. The consequence is that any test which stubs
+    `_run_scan` (several do, to avoid running a real pipeline) leaves the flag set, so the next such
+    test gets a 409 and fails for a reason that has nothing to do with it. Four tests across four
+    files broke exactly this way, every one of them passing in isolation.
+
+    Resetting it here fixes the class rather than the four instances, and keeps the guarantee that a
+    test's outcome does not depend on what ran before it.
+    """
+    yield
+    try:
+        from vpcopilot.console import app as _console
+        _console._scan.update(state="idle", log=[], summary=None, error=None)
+    except Exception:  # noqa: BLE001 — the console is optional for most of the suite
+        pass