Skip to content

Commit d0665b3

Browse files
solomonneascodex
andcommitted
fix(outcome): preserve colliding decision receipts
Co-Authored-By: Codex <codex@openai.com>
1 parent 0260985 commit d0665b3

3 files changed

Lines changed: 217 additions & 3 deletions

File tree

src/brigade/localio.py

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,26 @@ def write_json(path: Path, payload: dict[str, Any]) -> None:
8585
write_text_atomic(path, json.dumps(payload, indent=2, sort_keys=True) + "\n")
8686

8787

88+
def write_text_exclusive(path: Path, data: str) -> None:
89+
"""Publish complete data atomically without replacing an existing file."""
90+
path.parent.mkdir(parents=True, exist_ok=True)
91+
fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
92+
tmp_path = Path(tmp_name)
93+
try:
94+
with os.fdopen(fd, "w", encoding="utf-8") as handle:
95+
handle.write(data)
96+
handle.flush()
97+
os.fsync(handle.fileno())
98+
os.link(tmp_path, path)
99+
finally:
100+
tmp_path.unlink(missing_ok=True)
101+
102+
103+
def write_json_exclusive(path: Path, payload: dict[str, Any]) -> None:
104+
"""Create path with a JSON payload without replacing an existing file."""
105+
write_text_exclusive(path, json.dumps(payload, indent=2, sort_keys=True) + "\n")
106+
107+
88108
def read_jsonl_dicts(path: Path) -> list[dict[str, Any]]:
89109
"""Read JSONL records from path, keeping only lines that parse to JSON objects."""
90110
if not path.is_file():

src/brigade/outcome_cmd.py

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import json
1616
import os
1717
import platform as platform_mod
18+
import secrets
1819
import sys
1920
from pathlib import Path
2021
from typing import Any
@@ -39,9 +40,29 @@ def _status_path(target: Path) -> Path:
3940

4041

4142
def _decision_path(target: Path, now, artifact_id: str) -> Path:
42-
stamp = now.strftime("%Y%m%d-%H%M%S")
43+
"""Return a collision-resistant path for a new decision receipt."""
44+
stamp = now.strftime("%Y%m%d-%H%M%S-%f")
4345
slug = localio.slugify(artifact_id, fallback="artifact")
44-
return target / "memory" / "outcome" / "decisions" / f"{stamp}-{slug}.json"
46+
token = secrets.token_hex(4)
47+
return target / "memory" / "outcome" / "decisions" / f"{stamp}-{slug}-{token}.json"
48+
49+
50+
def _write_decision_receipt(
51+
target: Path,
52+
now,
53+
artifact_id: str,
54+
receipt: dict[str, Any],
55+
) -> Path:
56+
"""Write a decision receipt exclusively, retrying identity collisions."""
57+
last_error: FileExistsError | None = None
58+
for _ in range(8):
59+
path = _decision_path(target, now, artifact_id)
60+
try:
61+
localio.write_json_exclusive(path, receipt)
62+
return path
63+
except FileExistsError as exc:
64+
last_error = exc
65+
raise FileExistsError(f"could not allocate a unique decision receipt path for {artifact_id}: {last_error}")
4566

4667

4768
def load_status(target: Path) -> dict[str, dict]:
@@ -1511,7 +1532,7 @@ def reconcile(
15111532
and decision.action in {"install", "bump"}
15121533
):
15131534
receipt["route_policy"] = scorecard_mod.route_policy_marker_for_promotion()
1514-
localio.write_json(_decision_path(target, now, decision.artifact_id), receipt)
1535+
_write_decision_receipt(target, now, decision.artifact_id, receipt)
15151536
status_map[decision.artifact_id] = _status_entry_for_transition(
15161537
new_status=new_status,
15171538
now=now,

tests/test_outcome_cmd.py

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
import datetime as dt
22
import json
3+
import threading
4+
from concurrent.futures import ThreadPoolExecutor
5+
6+
import pytest
37

48
from brigade import cli, localio, outcome, outcome_cmd, receipts_cmd, scorecard, work_cmd
59

@@ -2128,3 +2132,172 @@ def test_route_breakdown_absent_on_pre_route_ledger(tmp_path, capsys):
21282132
assert outcome_cmd.explain(target=tmp_path, artifact_id="brigade-work", json_output=True) == 0
21292133
payload = json.loads(capsys.readouterr().out)
21302134
assert "route_breakdown" not in payload
2135+
2136+
2137+
def _write_legacy_decision_receipt(target, artifact_id, *, stamp="20260620-000000", new_status="promoted"):
2138+
"""Write a receipt under the pre-#564 second-resolution filename scheme.
2139+
2140+
The slug-only, second-resolution name is exactly what collide-with-overwrite
2141+
used to produce. ``load_transitions`` must keep reading these so existing
2142+
ledgers survive the rollout unchanged.
2143+
"""
2144+
decisions = target / "memory" / "outcome" / "decisions"
2145+
decisions.mkdir(parents=True, exist_ok=True)
2146+
slug = localio.slugify(artifact_id, fallback="artifact")
2147+
path = decisions / f"{stamp}-{slug}.json"
2148+
localio.write_json(
2149+
path,
2150+
{
2151+
"artifact_id": artifact_id,
2152+
"action": "install",
2153+
"new_status": new_status,
2154+
"created_at": "2026-06-20T00:00:00+00:00",
2155+
},
2156+
)
2157+
return path
2158+
2159+
2160+
def test_decision_path_is_collision_safe_within_the_same_second(tmp_path, monkeypatch):
2161+
# The pre-#564 scheme returned the same path for the same (now, artifact_id),
2162+
# so two decisions in one second selected one file and the second write
2163+
# replaced the first. Use deterministic tokens to prove lossy-equivalent
2164+
# artifact ids still select distinct paths.
2165+
tokens = iter(("00000000", "00000001"))
2166+
monkeypatch.setattr(outcome_cmd.secrets, "token_hex", lambda _n: next(tokens))
2167+
now = dt.datetime(2026, 6, 20, 0, 0, 0, tzinfo=dt.timezone.utc)
2168+
a = outcome_cmd._decision_path(tmp_path, now, "Skill X")
2169+
b = outcome_cmd._decision_path(tmp_path, now, "skill-x")
2170+
assert a != b
2171+
assert a.parent == b.parent
2172+
assert a.name == "20260620-000000-000000-skill-x-00000000.json"
2173+
assert b.name == "20260620-000000-000000-skill-x-00000001.json"
2174+
2175+
2176+
def test_write_json_exclusive_never_replaces_an_existing_receipt(tmp_path):
2177+
# O_EXCL: the second write to the same path raises and leaves the original
2178+
# file intact, so an existing receipt can never be overwritten.
2179+
path = tmp_path / "memory" / "outcome" / "decisions" / "receipt.json"
2180+
localio.write_json_exclusive(path, {"artifact_id": "first", "new_status": "promoted"})
2181+
with pytest.raises(FileExistsError):
2182+
localio.write_json_exclusive(path, {"artifact_id": "second", "new_status": "demoted"})
2183+
assert json.loads(path.read_text())["artifact_id"] == "first"
2184+
2185+
2186+
def test_write_json_exclusive_publishes_only_complete_json(tmp_path, monkeypatch):
2187+
path = tmp_path / "memory" / "outcome" / "decisions" / "receipt.json"
2188+
publish_ready = threading.Event()
2189+
allow_publish = threading.Event()
2190+
real_link = localio.os.link
2191+
2192+
def paused_link(source, destination):
2193+
publish_ready.set()
2194+
assert allow_publish.wait(timeout=5)
2195+
real_link(source, destination)
2196+
2197+
monkeypatch.setattr(localio.os, "link", paused_link)
2198+
with ThreadPoolExecutor(max_workers=1) as executor:
2199+
future = executor.submit(localio.write_json_exclusive, path, {"artifact_id": "complete"})
2200+
assert publish_ready.wait(timeout=5)
2201+
assert not path.exists()
2202+
allow_publish.set()
2203+
future.result(timeout=5)
2204+
2205+
assert json.loads(path.read_text()) == {"artifact_id": "complete"}
2206+
2207+
2208+
def test_write_json_exclusive_allows_exactly_one_concurrent_writer(tmp_path):
2209+
path = tmp_path / "memory" / "outcome" / "decisions" / "receipt.json"
2210+
writer_count = 8
2211+
ready = threading.Barrier(writer_count)
2212+
2213+
def write(index):
2214+
ready.wait()
2215+
try:
2216+
localio.write_json_exclusive(path, {"artifact_id": f"writer-{index}"})
2217+
except FileExistsError:
2218+
return None
2219+
return index
2220+
2221+
with ThreadPoolExecutor(max_workers=writer_count) as executor:
2222+
results = list(executor.map(write, range(writer_count)))
2223+
2224+
winners = [index for index in results if index is not None]
2225+
assert len(winners) == 1
2226+
assert json.loads(path.read_text()) == {"artifact_id": f"writer-{winners[0]}"}
2227+
2228+
2229+
def test_concurrent_decision_writers_retry_one_shared_identity(tmp_path, monkeypatch):
2230+
first_draw = threading.local()
2231+
first_draw_ready = threading.Barrier(2)
2232+
2233+
def token(_n):
2234+
if not getattr(first_draw, "used", False):
2235+
first_draw.used = True
2236+
first_draw_ready.wait()
2237+
return "deadbeef"
2238+
return f"{threading.get_ident():x}"
2239+
2240+
monkeypatch.setattr(outcome_cmd.secrets, "token_hex", token)
2241+
now = dt.datetime(2026, 6, 20, 0, 0, 0, tzinfo=dt.timezone.utc)
2242+
2243+
def write(artifact_id):
2244+
return outcome_cmd._write_decision_receipt(
2245+
tmp_path,
2246+
now,
2247+
artifact_id,
2248+
{"artifact_id": artifact_id, "new_status": "promoted", "created_at": now.isoformat()},
2249+
)
2250+
2251+
with ThreadPoolExecutor(max_workers=2) as executor:
2252+
paths = list(executor.map(write, ("Skill X", "skill-x")))
2253+
2254+
assert paths[0] != paths[1]
2255+
assert {json.loads(path.read_text())["artifact_id"] for path in paths} == {"Skill X", "skill-x"}
2256+
2257+
2258+
def test_write_decision_receipt_writes_two_distinct_files_for_colliding_ids(tmp_path, monkeypatch):
2259+
# Two artifact ids that slug to the same value ("Skill-X" and "skill-x" both
2260+
# lower-case to "skill-x") in the same second: the old scheme selected one
2261+
# path and the second write replaced the first receipt. The new writer draws
2262+
# a fresh token per call and opens with O_EXCL, so both receipts survive.
2263+
tokens = iter(("00000000", "00000001"))
2264+
monkeypatch.setattr(outcome_cmd.secrets, "token_hex", lambda _n: next(tokens))
2265+
now = dt.datetime(2026, 6, 20, 0, 0, 0, tzinfo=dt.timezone.utc)
2266+
path_a = outcome_cmd._write_decision_receipt(
2267+
tmp_path, now, "Skill X", {"artifact_id": "Skill X", "new_status": "promoted", "created_at": now.isoformat()}
2268+
)
2269+
path_b = outcome_cmd._write_decision_receipt(
2270+
tmp_path, now, "skill-x", {"artifact_id": "skill-x", "new_status": "promoted", "created_at": now.isoformat()}
2271+
)
2272+
assert path_a != path_b
2273+
assert path_a.is_file() and path_b.is_file()
2274+
assert json.loads(path_a.read_text())["artifact_id"] == "Skill X"
2275+
assert json.loads(path_b.read_text())["artifact_id"] == "skill-x"
2276+
2277+
2278+
def test_write_decision_receipt_raises_when_no_unique_path_is_available(tmp_path, monkeypatch):
2279+
# Force every draw to return the same token, so after the first successful
2280+
# O_EXCL write every retry collides. The writer must surface FileExistsError
2281+
# rather than fall back to overwriting the existing receipt.
2282+
monkeypatch.setattr(outcome_cmd.secrets, "token_hex", lambda _n: "deadbeef")
2283+
now = dt.datetime(2026, 6, 20, 0, 0, 0, tzinfo=dt.timezone.utc)
2284+
outcome_cmd._write_decision_receipt(
2285+
tmp_path, now, "skill-x", {"artifact_id": "skill-x", "new_status": "promoted", "created_at": now.isoformat()}
2286+
)
2287+
with pytest.raises(FileExistsError):
2288+
outcome_cmd._write_decision_receipt(
2289+
tmp_path,
2290+
now,
2291+
"skill-x",
2292+
{"artifact_id": "skill-x", "new_status": "promoted", "created_at": now.isoformat()},
2293+
)
2294+
2295+
2296+
def test_load_transitions_still_reads_legacy_second_resolution_receipts(tmp_path):
2297+
# Receipts written before #564 used `{stamp}-{slug}.json` with no microsecond
2298+
# or token. They must keep loading so an existing ledger survives the rollout.
2299+
_write_legacy_decision_receipt(tmp_path, "skill-legacy", new_status="promoted")
2300+
transitions = outcome_cmd.load_transitions(tmp_path)
2301+
assert len(transitions) == 1
2302+
assert transitions[0].artifact_id == "skill-legacy"
2303+
assert transitions[0].new_status == "promoted"

0 commit comments

Comments
 (0)