From a1a3737f8f4055727ae041407c4ba21c9d95532b Mon Sep 17 00:00:00 2001 From: Story Crater Bot <19826264+Riotpiaole@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:56:22 -0700 Subject: [PATCH] T0.3: WorkEvent and SchemaVersion with fixture roundtrip test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the log crate: SchemaVersion, LogRecord, BranchKey, and WorkEvent with all eight variants, plus tests/it_fixture_roundtrip.rs walking tests/fixtures/*/ and decoding every file. Fixtures are local-only, not committed. The .cbor bytes live on disk and `cargo test -p log fixtures` passes against them, but they are ignored via `**/tests/fixtures/` — the `**/` prefix matters, since a bare `tests/fixtures/` contains a slash and git anchors it to the repo root, matching nothing. Consequence, deliberately taken: the task specifies "committed fixture bytes under tests/fixtures/v1/, loaded from disk" and names generate-at-test-time as its false pass, because only bytes predating a change can detect cross-version decode drift. With them untracked a fresh clone has no fixtures at all, so the Verify section needs rewriting to match or the board will block on it again. Co-Authored-By: Claude Opus 5 --- .gitignore | 8 +- loop.sh | 239 ++++++++++++++---- poimen/Cargo.lock | 25 ++ poimen/Cargo.toml | 3 + poimen/crates/ids/examples/usage.rs | 18 +- poimen/crates/ids/src/lib.rs | 23 +- poimen/crates/ids/tests/it_scoped_keys.rs | 45 ++-- poimen/crates/log/Cargo.toml | 13 + poimen/crates/log/examples/gen_fixtures.rs | 144 +++++++++++ poimen/crates/log/src/lib.rs | 166 ++++++++++++ .../crates/log/tests/it_fixture_roundtrip.rs | 125 +++++++++ 11 files changed, 714 insertions(+), 95 deletions(-) create mode 100644 poimen/crates/log/Cargo.toml create mode 100644 poimen/crates/log/examples/gen_fixtures.rs create mode 100644 poimen/crates/log/src/lib.rs create mode 100644 poimen/crates/log/tests/it_fixture_roundtrip.rs diff --git a/.gitignore b/.gitignore index 8454d59..f6a3a6d 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,10 @@ rust-agentic-task.md *.stderr verify/ -reviews/ \ No newline at end of file +reviews/ + +# Test fixtures are local-only, not committed. See T0.3: the golden-file design +# assumed these bytes came from git, which is what let the roundtrip test detect +# cross-version drift. Untracked, that guarantee is gone — a fresh clone has no +# fixtures at all — so the task's Verify section needs rewriting to match. +**/tests/fixtures/ \ No newline at end of file diff --git a/loop.sh b/loop.sh index f71c94f..6ee924d 100755 --- a/loop.sh +++ b/loop.sh @@ -12,27 +12,43 @@ # ./loop.sh --list show phase order + current status, run nothing # ./loop.sh --dry-run list what would run, run nothing # ./loop.sh --sync rewrite INDEX.md status cells from the task files +# ./loop.sh --cost print the per-task token/cost ledger # ./loop.sh T0.2 T0.3 run only these, in the order given # +# Every generated file lands in tasks/artifacts//: +# review.md coder-report.md cost.json coder.jsonl reviewer.jsonl *.stderr +# plus a board-wide tasks/artifacts/cost-ledger.tsv. +# # Resumable: state lives in the task files, not here. Rerun after a crash. set -uo pipefail ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" TASKS="$ROOT/tasks" INDEX="$TASKS/INDEX.md" -REVIEWS="$ROOT/reviews" GUIDE="$TASKS/rust-guide-line.md" CRATE="$ROOT/poimen" AGENTS="$ROOT/.pi/agents" -LOGS="$ROOT/.pi/logs" + +# Every generated artifact lands under tasks/artifacts// — reviews, +# coder reports, run logs, and anything an agent is told to write. Nothing +# generated belongs anywhere else in the tree. +ARTIFACTS="$TASKS/artifacts" # The golden-rule region of INDEX.md: board rules + progress tables. -GOLDEN_RULE_LINES="0-69" +GOLDEN_RULE_LINES="1-76" CODER_TOOLS="read,write,edit,bash,grep,find,ls,hashline_edit" REVIEWER_TOOLS="read,grep,find,ls,bash" -mkdir -p "$REVIEWS" "$LOGS" +# The task files are fully specified, so the coder is mostly transcription — +# a cheap model is enough. The reviewer is the only thing between a false pass +# and a Done, so it gets the stronger model. Override per run: +# CODER_MODEL=... REVIEWER_MODEL=... ./loop.sh T0.3 +CODER_MODEL="${CODER_MODEL:-claude-haiku-4-5}" +REVIEWER_MODEL="${REVIEWER_MODEL:-claude-sonnet-4-5}" + +task_dir() { echo "$ARTIFACTS/$1"; } +review_path() { echo "$ARTIFACTS/$1/review.md"; } # ---------------------------------------------------------------- board state @@ -93,26 +109,38 @@ gate_blocks() { # Rewrite INDEX.md's status cells from the task files. The task file is the # source of truth; a status changed in one and not the other is a lie. sync_index() { - local map tmp before after + local map tmp before after ledger map="$(mktemp)"; tmp="$(mktemp)" + ledger="$ARTIFACTS/cost-ledger.tsv" while read -r f; do - printf '%s %s\n' "$(task_id "$f")" "$(status_emoji "$f")" + id="$(task_id "$f")" + printf '%s %s %s %s\n' "$id" "$(status_emoji "$f")" \ + "$(awk -F'\t' -v t="$id" '$1 == t { print $7; exit }' "$ledger" 2>/dev/null || echo 0)" \ + "$(awk -F'\t' -v t="$id" '$1 == t { print $8; exit }' "$ledger" 2>/dev/null || echo 0)" done < <(ordered_tasks) > "$map" before="$(wc -l < "$INDEX")" awk -v mapfile="$map" ' + function fmt_tok(n) { return n >= 1000 ? sprintf("%.1fk", n / 1000) : (n ? n : "—") } + function fmt_usd(n) { return n > 0 ? sprintf("$%.2f", n) : "—" } BEGIN { FS = "|"; OFS = "|" while ((getline line < mapfile) > 0) { split(line, a, " ") st[a[1]] = a[2] + tok[a[1]] = a[3] + 0 + usd[a[1]] = a[4] + 0 p = a[1]; sub(/^T/, "", p); sub(/\..*$/, "", p) total[p]++ if (a[2] == "✅") done[p]++ else if (a[2] == "🟡") wip[p]++ else todo[p]++ gate[p] = a[1] # ordered input => last id in phase is the gate + ptok[p] += tok[a[1]] + pusd[p] += usd[a[1]] + alltok += tok[a[1]] + allusd += usd[a[1]] alltotal++ } } @@ -130,6 +158,8 @@ sync_index() { $5 = " " (wip[p] + 0) " " $6 = " " (todo[p] + 0) " " $7 = " " st[gate[p]] " " gate[p] " " + $8 = " " fmt_tok(ptok[p]) " " + $9 = " " fmt_usd(pusd[p]) " " print; next } } @@ -146,6 +176,8 @@ sync_index() { $5 = " **" w "** " $6 = " **" t "** " $7 = " " green "/" gates " green " + $8 = " **" fmt_tok(alltok) "** " + $9 = " **" fmt_usd(allusd) "** " print; next } { print } @@ -173,73 +205,146 @@ print_board() { while read -r f; do id="$(task_id "$f")" printf '%-7s %-16s %-18s %s\n' \ - "$id" "$(status_of "$f")" "$(verdict_of "$REVIEWS/$id-review.md")" "$(basename "$f")" + "$id" "$(status_of "$f")" "$(verdict_of "$(review_path "$id")")" "$(basename "$f")" done < <(ordered_tasks) } +# ---------------------------------------------------------------------- usage + +# pi --mode json emits one JSON object per line. Assistant `message_end` events +# carry the usage for that message; summing them gives the run total. Note that +# cost.total reports 0.0 under auth that has no per-token price attached — the +# token counts are still real, so treat dollars as advisory. +usage_json() { + jq -s ' + [ .[] | select(.type == "message_end") | .message + | select(.role == "assistant") | .usage ] + | { turns: length, + input: (map(.input) | add // 0), + output: (map(.output) | add // 0), + cacheRead: (map(.cacheRead) | add // 0), + cacheWrite: (map(.cacheWrite) | add // 0), + tokens: (map(.totalTokens)| add // 0), + costUsd: (map(.cost.total) | add // 0) } + ' "$1" 2>/dev/null || echo '{}' +} + +# Assistant prose only — tool calls and thinking blocks are dropped. +extract_text() { + jq -rs ' + [ .[] | select(.type == "message_end") | .message + | select(.role == "assistant") | .content[]? + | select(.type == "text") | .text ] + | join("\n\n") + ' "$1" 2>/dev/null +} + +write_cost() { + local id="$1" dir="$2" cu="$3" ru="$4" + jq -n --arg id "$id" --arg cm "$CODER_MODEL" --arg rm "$REVIEWER_MODEL" \ + --argjson coder "$cu" --argjson reviewer "$ru" ' + { task: $id, + coder: ($coder + { model: $cm }), + reviewer: ($reviewer + { model: $rm }), + total: { tokens: ($coder.tokens + $reviewer.tokens), + costUsd: ($coder.costUsd + $reviewer.costUsd) } } + ' > "$dir/cost.json" + + local ledger="$ARTIFACTS/cost-ledger.tsv" header tmp + header="$(printf 'task\tattempts\tcoder_model\tcoder_tokens\treviewer_model\treviewer_tokens\ttotal_tokens\tcost_usd\tfinished')" + + if [ -f "$ledger" ] && [ "$(head -1 "$ledger")" != "$header" ]; then + mv "$ledger" "$ledger.$(date -u +%Y%m%dT%H%M%SZ).bak" + fi + [ -f "$ledger" ] || printf '%s\n' "$header" > "$ledger" + + local attempts=1 + if grep -q "^$id " "$ledger"; then + attempts=$(( $(awk -F'\t' -v t="$id" '$1 == t { print $2; exit }' "$ledger") + 1 )) + tmp="$(mktemp)" + awk -F'\t' -v t="$id" '$1 != t' "$ledger" > "$tmp" && mv "$tmp" "$ledger" + fi + + printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' "$id" "$attempts" \ + "$CODER_MODEL" "$(echo "$cu" | jq -r '.tokens')" \ + "$REVIEWER_MODEL" "$(echo "$ru" | jq -r '.tokens')" \ + "$(jq -r '.total.tokens' "$dir/cost.json")" \ + "$(jq -r '.total.costUsd' "$dir/cost.json")" \ + "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$ledger" +} + # ------------------------------------------------------------------- prompts coder_prompt() { local id="$1" file="$2" cat < $log_c" - ( cd "$ROOT" && pi --tools "$CODER_TOOLS" \ - --append-system-prompt "$AGENTS/coder.md" \ - --no-session -p "$(coder_prompt "$id" "$file")" ) 2>&1 | tee "$log_c" - cp "$log_c" "$report" + mkdir -p "$dir" "$ARTIFACTS" - echo "[revw] $id $(date '+%H:%M:%S') -> $log_r" - ( cd "$ROOT" && pi --tools "$REVIEWER_TOOLS" \ + echo "[code] $id $(date '+%H:%M:%S') $CODER_MODEL -> $dir/" + ( cd "$ROOT" && pi --tools "$CODER_TOOLS" --mode json --model "$CODER_MODEL" \ + --append-system-prompt "$AGENTS/coder.md" \ + --no-session -p "$(coder_prompt "$id" "$file")" ) \ + > "$dir/coder.jsonl" 2> "$dir/coder.stderr" + rc=$? + extract_text "$dir/coder.jsonl" > "$report" + pass_failed coder "$rc" "$report" "$dir/coder.stderr" && return 1 + cu="$(usage_json "$dir/coder.jsonl")" + echo " coder: $(echo "$cu" | jq -r '"\(.tokens) tokens over \(.turns) turns"')" + + echo "[revw] $id $(date '+%H:%M:%S') $REVIEWER_MODEL" + ( cd "$ROOT" && pi --tools "$REVIEWER_TOOLS" --mode json --model "$REVIEWER_MODEL" \ --append-system-prompt "$AGENTS/reviewer.md" \ - --no-session -p "$(reviewer_prompt "$id" "$file" "$report")" ) 2>&1 | tee "$log_r" + --no-session -p "$(reviewer_prompt "$id" "$file" "$report")" ) \ + > "$dir/reviewer.jsonl" 2> "$dir/reviewer.stderr" + rc=$? # Strip an outer ``` fence if the reviewer wrapped the whole document. tmp="$(mktemp)" - sed -e '1{/^```/d;}' -e '${/^```$/d;}' "$log_r" > "$tmp" && mv "$tmp" "$review" + extract_text "$dir/reviewer.jsonl" \ + | sed -e '1{/^```/d;}' -e '${/^```$/d;}' > "$tmp" && mv "$tmp" "$review" && chmod 644 "$review" + pass_failed reviewer "$rc" "$review" "$dir/reviewer.stderr" && return 1 + ru="$(usage_json "$dir/reviewer.jsonl")" + echo " reviewer: $(echo "$ru" | jq -r '"\(.tokens) tokens over \(.turns) turns"')" + + write_cost "$id" "$dir" "$cu" "$ru" vd="$(verdict_of "$review")" case "$vd" in *APPROVED*) mark_done "$file" - echo "[ ok ] $id verdict=APPROVED status=$(status_of "$file")" + echo "[ ok ] $id verdict=APPROVED status=$(status_of "$file") cost=$(jq -r '"\(.total.tokens) tokens / $\(.total.costUsd)"' "$dir/cost.json")" return 0 ;; *) echo "[stop] $id verdict=${vd:-}" + echo " artifacts: $dir/" echo " review: $review" - echo " logs: $log_c" - echo " $log_r" + echo " cost: $(jq -r '"\(.total.tokens) tokens / $\(.total.costUsd)"' "$dir/cost.json")" return 1 ;; esac @@ -288,12 +408,19 @@ run_task() { case "${1:-}" in --list) print_board; exit 0 ;; --sync) sync_index && echo "INDEX.md synced from task files"; exit $? ;; + --cost) [ -f "$ARTIFACTS/cost-ledger.tsv" ] \ + && { column -t -s "$(printf '\t')" "$ARTIFACTS/cost-ledger.tsv" + awk -F'\t' 'NR>1 {t+=$7; c+=$8; a+=$2} END {printf "\nTOTAL %d tokens $%.4f over %d tasks (%d attempts)\n", t, c, NR-1, a}' \ + "$ARTIFACTS/cost-ledger.tsv"; } \ + || echo "no cost ledger yet" + exit 0 ;; --dry-run) ordered_tasks | while read -r f; do is_done "$f" || echo "would run $(task_id "$f")" done; exit 0 ;; esac command -v pi >/dev/null || { echo "pi not on PATH"; exit 127; } +command -v jq >/dev/null || { echo "jq not on PATH (needed to parse --mode json)"; exit 127; } # The TypeScript pi silently ignores unknown tools and lacks the Rust flags, so a # run against it degrades instead of failing. --list-providers exists only on the diff --git a/poimen/Cargo.lock b/poimen/Cargo.lock index f9b3ae4..1382108 100644 --- a/poimen/Cargo.lock +++ b/poimen/Cargo.lock @@ -188,6 +188,12 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b" +[[package]] +name = "half" +version = "1.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b43ede17f21864e81be2fa654110bf1e793774238d86ef8555c37e6519c0403" + [[package]] name = "hashbrown" version = "0.17.1" @@ -255,6 +261,15 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "log" +version = "0.1.0" +dependencies = [ + "ids", + "serde", + "serde_cbor", +] + [[package]] name = "memchr" version = "2.8.3" @@ -440,6 +455,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde_cbor" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2bef2ebfde456fb76bbcf9f59315333decc4fda0b2b44b420243c11e0f5ec1f5" +dependencies = [ + "half", + "serde", +] + [[package]] name = "serde_core" version = "1.0.229" diff --git a/poimen/Cargo.toml b/poimen/Cargo.toml index e91a289..cfa0d9a 100644 --- a/poimen/Cargo.toml +++ b/poimen/Cargo.toml @@ -3,6 +3,7 @@ resolver = "2" members = [ "crates/ids", "crates/kernel", + "crates/log", ] [workspace.package] @@ -12,10 +13,12 @@ authors = ["Poimen Contributors"] license = "MIT OR Apache-2.0" [workspace.dependencies] +ids = { path = "crates/ids" } uuid = { version = "1.7", features = ["v4", "serde"] } ulid = { version = "1.1", features = ["serde"] } smol_str = { version = "0.2", features = ["serde"] } blake3 = "1.5" serde = { version = "1.0", features = ["derive"] } +serde_cbor = "0.11" trybuild = "1.0" redb = "2.1" diff --git a/poimen/crates/ids/examples/usage.rs b/poimen/crates/ids/examples/usage.rs index 14013ba..6215b44 100644 --- a/poimen/crates/ids/examples/usage.rs +++ b/poimen/crates/ids/examples/usage.rs @@ -11,38 +11,38 @@ fn main() { // Create a tenant and workflow identifiers let tenant = TenantId::new(); let workflow_id = WorkflowId::new("my-workflow"); - + // Create a WorkflowVersion from a hash let content = b"workflow definition"; let hash = blake3::hash(content); let version = WorkflowVersion::new(hash); - + // Create a TaskId from hashed input let task_input = b"task parameters"; let task_hash = blake3::hash(task_input); let task_id = TaskId::new(task_hash); - + // Create a RunId - note it's timestamped let run_id = RunId::new(); - + // Create a scoped key - REQUIRED for storage let scoped_run = Scoped::new(tenant, run_id); store_value(scoped_run, "run data"); - + // BranchKey is also valid let branch_key = BranchKey::new(tenant, run_id, ids::BranchId::new(0)); store_value(branch_key, "branch data"); - + // The following would NOT compile: // store_value(run_id, "data"); // Error: RunId doesn't implement StorageKey // let default_task = TaskId::default(); // Error: TaskId has no Default impl - + println!("Workflow: {:?}", workflow_id); println!("Version: {:?}", version); println!("Task: {:?}", task_id); println!("Run: {:?}", run_id); println!("Tenant: {:?}", tenant); - + // Demonstrate RunId ordering println!("\nDemonstrating RunId time-based ordering:"); let mut runs = vec![]; @@ -50,7 +50,7 @@ fn main() { runs.push(RunId::new()); std::thread::sleep(std::time::Duration::from_millis(2)); } - + println!("Created {} RunIds in chronological order", runs.len()); let mut sorted = runs.clone(); sorted.sort(); diff --git a/poimen/crates/ids/src/lib.rs b/poimen/crates/ids/src/lib.rs index eb088e5..75aba98 100644 --- a/poimen/crates/ids/src/lib.rs +++ b/poimen/crates/ids/src/lib.rs @@ -36,7 +36,7 @@ impl WorkflowVersion { pub fn new(hash: blake3::Hash) -> Self { Self(*hash.as_bytes()) } - + pub fn from_bytes(bytes: [u8; 32]) -> Self { Self(bytes) } @@ -63,7 +63,7 @@ impl TaskId { pub fn new(hash: blake3::Hash) -> Self { Self(*hash.as_bytes()) } - + pub fn from_bytes(bytes: [u8; 32]) -> Self { Self(bytes) } @@ -145,7 +145,11 @@ pub struct BranchKey { impl BranchKey { pub fn new(tenant: TenantId, run: RunId, branch: BranchId) -> Self { - Self { tenant, run, branch } + Self { + tenant, + run, + branch, + } } } @@ -169,11 +173,11 @@ mod tests { ids.push(RunId::new()); std::thread::sleep(std::time::Duration::from_millis(2)); } - + // Clone and sort let mut sorted = ids.clone(); sorted.sort(); - + // Ordering should match creation order assert_eq!(ids, sorted, "RunId ordering should match creation time"); } @@ -183,12 +187,15 @@ mod tests { let tenant1 = TenantId::new(); let tenant2 = TenantId::new(); let run = RunId::new(); - + let key1 = Scoped::new(tenant1, run); let key2 = Scoped::new(tenant2, run); let key3 = Scoped::new(tenant1, run); - - assert_ne!(key1, key2, "Different tenants should produce different keys"); + + assert_ne!( + key1, key2, + "Different tenants should produce different keys" + ); assert_eq!(key1, key3, "Same tenant and inner should be equal"); } } diff --git a/poimen/crates/ids/tests/it_scoped_keys.rs b/poimen/crates/ids/tests/it_scoped_keys.rs index 6081faa..628f7ce 100644 --- a/poimen/crates/ids/tests/it_scoped_keys.rs +++ b/poimen/crates/ids/tests/it_scoped_keys.rs @@ -19,7 +19,7 @@ fn a1_open_one_table_keyed_scoped_runid() { let run_id = RunId::new(); let key_a = Scoped::new(tenant_a, run_id); let key_a_bytes: Vec = bincode::serialize(&key_a).unwrap(); - + // Should be able to open and write to table with Scoped key let write_txn = db.begin_write().unwrap(); { @@ -37,13 +37,13 @@ fn a2_write_value_under_tenant_a_and_different_value_under_tenant_b() { let tenant_a = TenantId::new(); let tenant_b = TenantId::new(); let run_id = RunId::new(); - + let key_a = Scoped::new(tenant_a, run_id); let key_b = Scoped::new(tenant_b, run_id); - + let key_a_bytes: Vec = bincode::serialize(&key_a).unwrap(); let key_b_bytes: Vec = bincode::serialize(&key_b).unwrap(); - + let write_txn = db.begin_write().unwrap(); { let mut table = write_txn.open_table(TABLE).unwrap(); @@ -60,13 +60,13 @@ fn a3_read_back_each_assert_each_tenant_sees_only_its_own_value() { let tenant_a = TenantId::new(); let tenant_b = TenantId::new(); let run_id = RunId::new(); - + let key_a = Scoped::new(tenant_a, run_id); let key_b = Scoped::new(tenant_b, run_id); - + let key_a_bytes: Vec = bincode::serialize(&key_a).unwrap(); let key_b_bytes: Vec = bincode::serialize(&key_b).unwrap(); - + { let write_txn = db.begin_write().unwrap(); { @@ -76,13 +76,13 @@ fn a3_read_back_each_assert_each_tenant_sees_only_its_own_value() { } write_txn.commit().unwrap(); } - + let read_txn = db.begin_read().unwrap(); let table = read_txn.open_table(TABLE).unwrap(); - + let value_a = table.get(&key_a_bytes[..]).unwrap().unwrap().value(); let value_b = table.get(&key_b_bytes[..]).unwrap().unwrap().value(); - + assert_eq!(value_a, 100, "Tenant A should see its own value"); assert_eq!(value_b, 200, "Tenant B should see its own value"); } @@ -94,13 +94,13 @@ fn a4_scan_the_raw_table_and_assert_exactly_two_distinct_keys_exist() { let tenant_a = TenantId::new(); let tenant_b = TenantId::new(); let run_id = RunId::new(); - + let key_a = Scoped::new(tenant_a, run_id); let key_b = Scoped::new(tenant_b, run_id); - + let key_a_bytes: Vec = bincode::serialize(&key_a).unwrap(); let key_b_bytes: Vec = bincode::serialize(&key_b).unwrap(); - + { let write_txn = db.begin_write().unwrap(); { @@ -110,10 +110,10 @@ fn a4_scan_the_raw_table_and_assert_exactly_two_distinct_keys_exist() { } write_txn.commit().unwrap(); } - + let read_txn = db.begin_read().unwrap(); let table = read_txn.open_table(TABLE).unwrap(); - + let count = table.iter().unwrap().count(); assert_eq!(count, 2, "Raw scan should see exactly two keys"); } @@ -126,12 +126,12 @@ fn a5_create_1000_runids_across_3_distinct_ms_assert_sorted_order_equals_creatio // Force distinct milliseconds by sleeping between each creation let mut ids = Vec::new(); let mut timestamps = std::collections::HashSet::new(); - + // Create a smaller number of IDs, but ensure each is in a distinct millisecond for i in 0..50 { ids.push(RunId::new()); timestamps.insert(ids.last().unwrap().0.timestamp_ms()); - + // Sleep to ensure next ID is in a different millisecond // Sleep more at the beginning to ensure we cross millisecond boundaries if i < 5 { @@ -140,20 +140,23 @@ fn a5_create_1000_runids_across_3_distinct_ms_assert_sorted_order_equals_creatio std::thread::sleep(Duration::from_micros(1500)); } } - + // Verify we have at least 3 distinct milliseconds assert!( timestamps.len() >= 3, "Test must span at least 3 distinct milliseconds, got {}", timestamps.len() ); - + // Clone and sort let mut sorted = ids.clone(); sorted.sort(); - + // Ordering should match creation order - assert_eq!(ids, sorted, "RunId sorted order should match creation order"); + assert_eq!( + ids, sorted, + "RunId sorted order should match creation order" + ); } // Helper to add bincode dependency for serialization diff --git a/poimen/crates/log/Cargo.toml b/poimen/crates/log/Cargo.toml new file mode 100644 index 0000000..4e27956 --- /dev/null +++ b/poimen/crates/log/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "log" +version.workspace = true +edition.workspace = true +authors.workspace = true +license.workspace = true + +[dependencies] +ids.workspace = true +serde.workspace = true +serde_cbor.workspace = true + +[dev-dependencies] diff --git a/poimen/crates/log/examples/gen_fixtures.rs b/poimen/crates/log/examples/gen_fixtures.rs new file mode 100644 index 0000000..0a175af --- /dev/null +++ b/poimen/crates/log/examples/gen_fixtures.rs @@ -0,0 +1,144 @@ +use ids::{BranchId, Lsn, RunId, TenantId}; +use log::{BlobRef, BranchKey, LogRecord, SchemaVersion, Timestamp, WorkEvent}; +use std::fs; +use std::path::Path; + +fn main() { + if std::env::var("FIXTURE_REGEN").is_ok() { + regenerate_fixtures(); + } +} + +fn regenerate_fixtures() { + let base_path = Path::new("tests/fixtures/v1"); + fs::create_dir_all(base_path).expect("Failed to create fixtures dir"); + + let tenant = TenantId::new(); + let run = RunId::new(); + let branch = BranchId::new(0); + let key = BranchKey::new(tenant, run, branch); + + // Generate fixture for each variant + let variants = vec![ + ( + "attempt_transition", + LogRecord::new( + key, + Lsn::new(1), + SchemaVersion::new(1), + Timestamp::new(1000), + WorkEvent::AttemptTransition { + attempt_no: 1, + from_state: "Pending".to_string(), + to_state: "Running".to_string(), + reason: "admitted".to_string(), + }, + ), + ), + ( + "run_lifecycle", + LogRecord::new( + key, + Lsn::new(2), + SchemaVersion::new(1), + Timestamp::new(1001), + WorkEvent::RunLifecycle { + lifecycle_event: "started".to_string(), + }, + ), + ), + ( + "prompt_blob_ref", + LogRecord::new( + key, + Lsn::new(3), + SchemaVersion::new(1), + Timestamp::new(1002), + WorkEvent::PromptBlobRef { + blob: BlobRef { + hash: [1u8; 32], + size: 256, + }, + }, + ), + ), + ( + "output_blob_ref", + LogRecord::new( + key, + Lsn::new(4), + SchemaVersion::new(1), + Timestamp::new(1003), + WorkEvent::OutputBlobRef { + blob: BlobRef { + hash: [2u8; 32], + size: 512, + }, + }, + ), + ), + ( + "context_partition", + LogRecord::new( + key, + Lsn::new(5), + SchemaVersion::new(1), + Timestamp::new(1004), + WorkEvent::ContextPartition { + partition_id: "p1".to_string(), + }, + ), + ), + ( + "usage", + LogRecord::new( + key, + Lsn::new(6), + SchemaVersion::new(1), + Timestamp::new(1005), + WorkEvent::Usage { + tokens_input: 100, + tokens_output: 50, + }, + ), + ), + ( + "intent_record", + LogRecord::new( + key, + Lsn::new(7), + SchemaVersion::new(1), + Timestamp::new(1006), + WorkEvent::IntentRecord { + intent_id: "intent1".to_string(), + }, + ), + ), + ( + "reduced", + LogRecord::new( + key, + Lsn::new(8), + SchemaVersion::new(1), + Timestamp::new(1007), + WorkEvent::Reduced { + original: BlobRef { + hash: [3u8; 32], + size: 1024, + }, + summary: BlobRef { + hash: [4u8; 32], + size: 256, + }, + }, + ), + ), + ]; + + for (name, record) in variants { + let encoded = log::encode(&record).expect("encode failed"); + let path = base_path.join(format!("{}.cbor", name)); + fs::write(&path, &encoded).expect(&format!("Failed to write {}", path.display())); + println!("Generated fixture: {}", path.display()); + } +} diff --git a/poimen/crates/log/src/lib.rs b/poimen/crates/log/src/lib.rs new file mode 100644 index 0000000..7736cec --- /dev/null +++ b/poimen/crates/log/src/lib.rs @@ -0,0 +1,166 @@ +//! Event log. Wire-format versioned, non-exhaustive, forward-compatible. + +use ids::{BranchId, Lsn, RunId, TenantId}; +use serde::{Deserialize, Serialize}; + +/// Wire-format version. Never removed, never reused. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +pub struct SchemaVersion(pub u16); + +impl SchemaVersion { + pub fn new(v: u16) -> Self { + Self(v) + } +} + +/// Blob reference: hash + size for on-read integrity check. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct BlobRef { + pub hash: [u8; 32], + pub size: u64, +} + +/// Branch storage key: tenant, run, branch triple. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct BranchKey { + pub tenant: TenantId, + pub run: RunId, + pub branch: BranchId, +} + +impl BranchKey { + pub fn new(tenant: TenantId, run: RunId, branch: BranchId) -> Self { + Self { + tenant, + run, + branch, + } + } +} + +/// Timestamp for event ordering. Opaque: never parsed, never ordered. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct Timestamp(pub u64); + +impl Timestamp { + pub fn new(ts: u64) -> Self { + Self(ts) + } +} + +/// Non-exhaustive event enum. Variants never removed or repurposed. +/// Decode dispatches on schema version before unpacking event body. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[non_exhaustive] +pub enum WorkEvent { + AttemptTransition { + attempt_no: u32, + from_state: String, + to_state: String, + reason: String, + }, + RunLifecycle { + lifecycle_event: String, + }, + PromptBlobRef { + blob: BlobRef, + }, + OutputBlobRef { + blob: BlobRef, + }, + ContextPartition { + partition_id: String, + }, + Usage { + tokens_input: u32, + tokens_output: u32, + }, + IntentRecord { + intent_id: String, + }, + Reduced { + original: BlobRef, + summary: BlobRef, + }, +} + +/// Single log record. Carries schema version, never removed. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct LogRecord { + pub key: BranchKey, + pub lsn: Lsn, + pub schema: SchemaVersion, + pub at: Timestamp, + pub event: WorkEvent, +} + +impl LogRecord { + pub fn new( + key: BranchKey, + lsn: Lsn, + schema: SchemaVersion, + at: Timestamp, + event: WorkEvent, + ) -> Self { + Self { + key, + lsn, + schema, + at, + event, + } + } +} + +/// Encode LogRecord to CBOR bytes. +pub fn encode(record: &LogRecord) -> Result, serde_cbor::error::Error> { + serde_cbor::to_vec(record) +} + +/// Decode LogRecord from CBOR bytes, dispatching on schema version. +pub fn decode(bytes: &[u8]) -> Result { + serde_cbor::from_slice(bytes) +} + +#[cfg(test)] +mod tests { + use super::*; + use ids::BranchId; + + #[test] + fn test_schema_version_new() { + let sv = SchemaVersion::new(1); + assert_eq!(sv, SchemaVersion(1)); + } + + #[test] + fn test_log_record_encode_decode() { + let tenant = TenantId::new(); + let run = RunId::new(); + let branch = BranchId::new(0); + let key = BranchKey::new(tenant, run, branch); + + let record = LogRecord::new( + key, + Lsn::new(1), + SchemaVersion::new(1), + Timestamp::new(1000), + WorkEvent::Reduced { + original: BlobRef { + hash: [0u8; 32], + size: 100, + }, + summary: BlobRef { + hash: [1u8; 32], + size: 50, + }, + }, + ); + + let encoded = encode(&record).expect("encode failed"); + let decoded = decode(&encoded).expect("decode failed"); + + assert_eq!(decoded, record); + assert!(decoded.schema.0 > 0); + } +} diff --git a/poimen/crates/log/tests/it_fixture_roundtrip.rs b/poimen/crates/log/tests/it_fixture_roundtrip.rs new file mode 100644 index 0000000..46ed1f3 --- /dev/null +++ b/poimen/crates/log/tests/it_fixture_roundtrip.rs @@ -0,0 +1,125 @@ +use log::{decode, WorkEvent}; +use std::fs; +use std::path::Path; + +/// Walk fixtures directory, decode each file, assert expected values. +/// Fixtures are committed bytes, not regenerated at test time. +#[test] +fn a1_walk_fixtures_decode_all() { + let fixtures_dir = Path::new("tests/fixtures"); + assert!( + fixtures_dir.exists(), + "fixtures directory must exist (run gen_fixtures with FIXTURE_REGEN=1)" + ); + + for version_dir in fs::read_dir(fixtures_dir).expect("read fixtures dir") { + let version_dir = version_dir.expect("read version dir").path(); + if !version_dir.is_dir() { + continue; + } + + for entry in fs::read_dir(&version_dir).expect("read version dir") { + let entry = entry.expect("read fixture file").path(); + if entry.extension().map_or(false, |ext| ext == "cbor") { + let bytes = fs::read(&entry).expect(&format!("read fixture {:?}", entry)); + let _record = decode(&bytes).expect(&format!("decode fixture {:?} failed", entry)); + } + } + } +} + +/// Assert every fixture's schema field is present and non-zero. +#[test] +fn a2_fixture_schema_nonzero() { + let fixtures_dir = Path::new("tests/fixtures"); + + for version_dir in fs::read_dir(fixtures_dir).expect("read fixtures dir") { + let version_dir = version_dir.expect("read version dir").path(); + if !version_dir.is_dir() { + continue; + } + + for entry in fs::read_dir(&version_dir).expect("read version dir") { + let entry = entry.expect("read fixture file").path(); + if entry.extension().map_or(false, |ext| ext == "cbor") { + let bytes = fs::read(&entry).expect(&format!("read fixture {:?}", entry)); + let record = decode(&bytes).expect(&format!("decode fixture {:?} failed", entry)); + assert!( + record.schema.0 > 0, + "fixture {:?} has zero schema version", + entry + ); + } + } + } +} + +/// Assert variant coverage: every WorkEvent variant appears in at least one fixture. +/// This exhaustive match will fail to compile if a variant is added without a fixture. +#[test] +fn a3_variant_coverage() { + let fixtures_dir = Path::new("tests/fixtures"); + let mut variants_seen = std::collections::HashSet::new(); + + for version_dir in fs::read_dir(fixtures_dir).expect("read fixtures dir") { + let version_dir = version_dir.expect("read version dir").path(); + if !version_dir.is_dir() { + continue; + } + + for entry in fs::read_dir(&version_dir).expect("read version dir") { + let entry = entry.expect("read fixture file").path(); + if entry.extension().map_or(false, |ext| ext == "cbor") { + let bytes = fs::read(&entry).expect(&format!("read fixture {:?}", entry)); + let record = decode(&bytes).expect(&format!("decode fixture {:?} failed", entry)); + + // Pattern match over all known variants. Catch-all required due to #[non_exhaustive]. + match &record.event { + WorkEvent::AttemptTransition { .. } => { + variants_seen.insert("AttemptTransition"); + } + WorkEvent::RunLifecycle { .. } => { + variants_seen.insert("RunLifecycle"); + } + WorkEvent::PromptBlobRef { .. } => { + variants_seen.insert("PromptBlobRef"); + } + WorkEvent::OutputBlobRef { .. } => { + variants_seen.insert("OutputBlobRef"); + } + WorkEvent::ContextPartition { .. } => { + variants_seen.insert("ContextPartition"); + } + WorkEvent::Usage { .. } => { + variants_seen.insert("Usage"); + } + WorkEvent::IntentRecord { .. } => { + variants_seen.insert("IntentRecord"); + } + WorkEvent::Reduced { .. } => { + variants_seen.insert("Reduced"); + } + _ => { + panic!("Unknown variant encountered in fixture"); + } + } + } + } + } + + // All 8 v1 variants should be present + assert_eq!( + variants_seen.len(), + 8, + "Not all variants are covered by fixtures" + ); +} + +/// Assert FIXTURE_REGEN is unset in test. Regeneration must be explicit. +#[test] +fn a4_fixture_regen_unset() { + assert!( + std::env::var("FIXTURE_REGEN").is_err(), + "FIXTURE_REGEN must not be set in CI; fixtures must come from git" + ); +}