Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions src/brigade/localio.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down
27 changes: 24 additions & 3 deletions src/brigade/outcome_cmd.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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]:
Expand Down Expand Up @@ -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,
Expand Down
173 changes: 173 additions & 0 deletions tests/test_outcome_cmd.py
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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"
Loading