fix(outcome): preserve colliding decision receipts (#603)

Co-authored-by: Codex <codex@openai.com>

Author: solomonneas

src/brigade/localio.py

@@ -85,6 +85,26 @@ def write_json(path: Path, payload: dict[str, Any]) -> None:
     write_text_atomic(path, json.dumps(payload, indent=2, sort_keys=True) + "\n")
 
 
+def write_text_exclusive(path: Path, data: str) -> None:
+    """Publish complete data atomically without replacing an existing file."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
+    tmp_path = Path(tmp_name)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.link(tmp_path, path)
+    finally:
+        tmp_path.unlink(missing_ok=True)
+
+
+def write_json_exclusive(path: Path, payload: dict[str, Any]) -> None:
+    """Create path with a JSON payload without replacing an existing file."""
+    write_text_exclusive(path, json.dumps(payload, indent=2, sort_keys=True) + "\n")
+
+
 def read_jsonl_dicts(path: Path) -> list[dict[str, Any]]:
     """Read JSONL records from path, keeping only lines that parse to JSON objects."""
     if not path.is_file():

src/brigade/outcome_cmd.py

@@ -15,6 +15,7 @@
 import json
 import os
 import platform as platform_mod
+import secrets
 import sys
 from pathlib import Path
 from typing import Any
@@ -39,9 +40,29 @@ def _status_path(target: Path) -> Path:
 
 
 def _decision_path(target: Path, now, artifact_id: str) -> Path:
-    stamp = now.strftime("%Y%m%d-%H%M%S")
+    """Return a collision-resistant path for a new decision receipt."""
+    stamp = now.strftime("%Y%m%d-%H%M%S-%f")
     slug = localio.slugify(artifact_id, fallback="artifact")
-    return target / "memory" / "outcome" / "decisions" / f"{stamp}-{slug}.json"
+    token = secrets.token_hex(4)
+    return target / "memory" / "outcome" / "decisions" / f"{stamp}-{slug}-{token}.json"
+
+
+def _write_decision_receipt(
+    target: Path,
+    now,
+    artifact_id: str,
+    receipt: dict[str, Any],
+) -> Path:
+    """Write a decision receipt exclusively, retrying identity collisions."""
+    last_error: FileExistsError | None = None
+    for _ in range(8):
+        path = _decision_path(target, now, artifact_id)
+        try:
+            localio.write_json_exclusive(path, receipt)
+            return path
+        except FileExistsError as exc:
+            last_error = exc
+    raise FileExistsError(f"could not allocate a unique decision receipt path for {artifact_id}: {last_error}")
 
 
 def load_status(target: Path) -> dict[str, dict]:
@@ -1511,7 +1532,7 @@ def reconcile(
                 and decision.action in {"install", "bump"}
             ):
                 receipt["route_policy"] = scorecard_mod.route_policy_marker_for_promotion()
-            localio.write_json(_decision_path(target, now, decision.artifact_id), receipt)
+            _write_decision_receipt(target, now, decision.artifact_id, receipt)
             status_map[decision.artifact_id] = _status_entry_for_transition(
                 new_status=new_status,
                 now=now,

tests/test_outcome_cmd.py

@@ -1,5 +1,9 @@
 import datetime as dt
 import json
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
 
 from brigade import cli, localio, outcome, outcome_cmd, receipts_cmd, scorecard, work_cmd
 
@@ -2128,3 +2132,172 @@ def test_route_breakdown_absent_on_pre_route_ledger(tmp_path, capsys):
     assert outcome_cmd.explain(target=tmp_path, artifact_id="brigade-work", json_output=True) == 0
     payload = json.loads(capsys.readouterr().out)
     assert "route_breakdown" not in payload
+
+
+def _write_legacy_decision_receipt(target, artifact_id, *, stamp="20260620-000000", new_status="promoted"):
+    """Write a receipt under the pre-#564 second-resolution filename scheme.
+
+    The slug-only, second-resolution name is exactly what collide-with-overwrite
+    used to produce. ``load_transitions`` must keep reading these so existing
+    ledgers survive the rollout unchanged.
+    """
+    decisions = target / "memory" / "outcome" / "decisions"
+    decisions.mkdir(parents=True, exist_ok=True)
+    slug = localio.slugify(artifact_id, fallback="artifact")
+    path = decisions / f"{stamp}-{slug}.json"
+    localio.write_json(
+        path,
+        {
+            "artifact_id": artifact_id,
+            "action": "install",
+            "new_status": new_status,
+            "created_at": "2026-06-20T00:00:00+00:00",
+        },
+    )
+    return path
+
+
+def test_decision_path_is_collision_safe_within_the_same_second(tmp_path, monkeypatch):
+    # The pre-#564 scheme returned the same path for the same (now, artifact_id),
+    # so two decisions in one second selected one file and the second write
+    # replaced the first. Use deterministic tokens to prove lossy-equivalent
+    # artifact ids still select distinct paths.
+    tokens = iter(("00000000", "00000001"))
+    monkeypatch.setattr(outcome_cmd.secrets, "token_hex", lambda _n: next(tokens))
+    now = dt.datetime(2026, 6, 20, 0, 0, 0, tzinfo=dt.timezone.utc)
+    a = outcome_cmd._decision_path(tmp_path, now, "Skill X")
+    b = outcome_cmd._decision_path(tmp_path, now, "skill-x")
+    assert a != b
+    assert a.parent == b.parent
+    assert a.name == "20260620-000000-000000-skill-x-00000000.json"
+    assert b.name == "20260620-000000-000000-skill-x-00000001.json"
+
+
+def test_write_json_exclusive_never_replaces_an_existing_receipt(tmp_path):
+    # O_EXCL: the second write to the same path raises and leaves the original
+    # file intact, so an existing receipt can never be overwritten.
+    path = tmp_path / "memory" / "outcome" / "decisions" / "receipt.json"
+    localio.write_json_exclusive(path, {"artifact_id": "first", "new_status": "promoted"})
+    with pytest.raises(FileExistsError):
+        localio.write_json_exclusive(path, {"artifact_id": "second", "new_status": "demoted"})
+    assert json.loads(path.read_text())["artifact_id"] == "first"
+
+
+def test_write_json_exclusive_publishes_only_complete_json(tmp_path, monkeypatch):
+    path = tmp_path / "memory" / "outcome" / "decisions" / "receipt.json"
+    publish_ready = threading.Event()
+    allow_publish = threading.Event()
+    real_link = localio.os.link
+
+    def paused_link(source, destination):
+        publish_ready.set()
+        assert allow_publish.wait(timeout=5)
+        real_link(source, destination)
+
+    monkeypatch.setattr(localio.os, "link", paused_link)
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        future = executor.submit(localio.write_json_exclusive, path, {"artifact_id": "complete"})
+        assert publish_ready.wait(timeout=5)
+        assert not path.exists()
+        allow_publish.set()
+        future.result(timeout=5)
+
+    assert json.loads(path.read_text()) == {"artifact_id": "complete"}
+
+
+def test_write_json_exclusive_allows_exactly_one_concurrent_writer(tmp_path):
+    path = tmp_path / "memory" / "outcome" / "decisions" / "receipt.json"
+    writer_count = 8
+    ready = threading.Barrier(writer_count)
+
+    def write(index):
+        ready.wait()
+        try:
+            localio.write_json_exclusive(path, {"artifact_id": f"writer-{index}"})
+        except FileExistsError:
+            return None
+        return index
+
+    with ThreadPoolExecutor(max_workers=writer_count) as executor:
+        results = list(executor.map(write, range(writer_count)))
+
+    winners = [index for index in results if index is not None]
+    assert len(winners) == 1
+    assert json.loads(path.read_text()) == {"artifact_id": f"writer-{winners[0]}"}
+
+
+def test_concurrent_decision_writers_retry_one_shared_identity(tmp_path, monkeypatch):
+    first_draw = threading.local()
+    first_draw_ready = threading.Barrier(2)
+
+    def token(_n):
+        if not getattr(first_draw, "used", False):
+            first_draw.used = True
+            first_draw_ready.wait()
+            return "deadbeef"
+        return f"{threading.get_ident():x}"
+
+    monkeypatch.setattr(outcome_cmd.secrets, "token_hex", token)
+    now = dt.datetime(2026, 6, 20, 0, 0, 0, tzinfo=dt.timezone.utc)
+
+    def write(artifact_id):
+        return outcome_cmd._write_decision_receipt(
+            tmp_path,
+            now,
+            artifact_id,
+            {"artifact_id": artifact_id, "new_status": "promoted", "created_at": now.isoformat()},
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        paths = list(executor.map(write, ("Skill X", "skill-x")))
+
+    assert paths[0] != paths[1]
+    assert {json.loads(path.read_text())["artifact_id"] for path in paths} == {"Skill X", "skill-x"}
+
+
+def test_write_decision_receipt_writes_two_distinct_files_for_colliding_ids(tmp_path, monkeypatch):
+    # Two artifact ids that slug to the same value ("Skill-X" and "skill-x" both
+    # lower-case to "skill-x") in the same second: the old scheme selected one
+    # path and the second write replaced the first receipt. The new writer draws
+    # a fresh token per call and opens with O_EXCL, so both receipts survive.
+    tokens = iter(("00000000", "00000001"))
+    monkeypatch.setattr(outcome_cmd.secrets, "token_hex", lambda _n: next(tokens))
+    now = dt.datetime(2026, 6, 20, 0, 0, 0, tzinfo=dt.timezone.utc)
+    path_a = outcome_cmd._write_decision_receipt(
+        tmp_path, now, "Skill X", {"artifact_id": "Skill X", "new_status": "promoted", "created_at": now.isoformat()}
+    )
+    path_b = outcome_cmd._write_decision_receipt(
+        tmp_path, now, "skill-x", {"artifact_id": "skill-x", "new_status": "promoted", "created_at": now.isoformat()}
+    )
+    assert path_a != path_b
+    assert path_a.is_file() and path_b.is_file()
+    assert json.loads(path_a.read_text())["artifact_id"] == "Skill X"
+    assert json.loads(path_b.read_text())["artifact_id"] == "skill-x"
+
+
+def test_write_decision_receipt_raises_when_no_unique_path_is_available(tmp_path, monkeypatch):
+    # Force every draw to return the same token, so after the first successful
+    # O_EXCL write every retry collides. The writer must surface FileExistsError
+    # rather than fall back to overwriting the existing receipt.
+    monkeypatch.setattr(outcome_cmd.secrets, "token_hex", lambda _n: "deadbeef")
+    now = dt.datetime(2026, 6, 20, 0, 0, 0, tzinfo=dt.timezone.utc)
+    outcome_cmd._write_decision_receipt(
+        tmp_path, now, "skill-x", {"artifact_id": "skill-x", "new_status": "promoted", "created_at": now.isoformat()}
+    )
+    with pytest.raises(FileExistsError):
+        outcome_cmd._write_decision_receipt(
+            tmp_path,
+            now,
+            "skill-x",
+            {"artifact_id": "skill-x", "new_status": "promoted", "created_at": now.isoformat()},
+        )
+
+
+def test_load_transitions_still_reads_legacy_second_resolution_receipts(tmp_path):
+    # Receipts written before #564 used `{stamp}-{slug}.json` with no microsecond
+    # or token. They must keep loading so an existing ledger survives the rollout.
+    _write_legacy_decision_receipt(tmp_path, "skill-legacy", new_status="promoted")
+    transitions = outcome_cmd.load_transitions(tmp_path)
+    assert len(transitions) == 1
+    assert transitions[0].artifact_id == "skill-legacy"
+    assert transitions[0].new_status == "promoted"