Drive Append-Only Audit from your own code
The page reads the event-sourcing code behind one Python service in the browser for free — the
Postgres schema (.sql), the event store written with asyncpg (or psycopg, or SQLAlchemy
text()), the services that append to it and the projectors that build read models from
it, one or more files joined into one paste, each introduced by a # file: <path>
line (-- file: <path> for SQL) — with its line numbers. It lexes Python
strings, f-strings, triple-quoted strings and comments out first (and SQL comments, quoted text and
dollar-quoted bodies), rebuilds the indentation structure (classes, defs, loops, with-blocks,
try/except), pulls every SQL statement out of the .sql files and out of the Python strings
(following module constants to the execute and fetch calls that run them),
reads the schema (tables, columns, primary keys, unique constraints and indexes), works out which table
holds the events, which the checkpoints and which are read models, and finds every projector loop.
Then it runs 21 deterministic checks drawn from event-store and projection practice: store ids
ES-01..ES-10 (no UNIQUE (stream, version), appends with no expected version,
UPDATE, DELETE or TRUNCATE on the events table, multi-event
appends outside a transaction, global reads by a sequence position with no guard against transactions
still in flight, event reads with no ORDER BY or no LIMIT, no unique event
id, no metadata column, no schema version), projection ids PJ-01..PJ-10 (a checkpoint saved outside the
read model’s transaction or before the event is applied, writes that are not safe to replay,
handler failures swallowed while the checkpoint moves on, events applied concurrently, rebuilds that
reset the checkpoint without clearing the read model, handlers with several writes outside one
transaction, polling loops that never wait, projectors with no checkpoint, handler lookups with no
default), and EV-01 (a credential literal), plus EV-00 for context. Nothing is run: every flag is a
reading of the text. Its three paid steps are available over HTTP. You send one JSON object with
task set to "store", "projection" or "fix", the
paste as code and the page’s check as a facts string, and the model
replies with one JSON object: an audit that rates the lane’s areas, answers every flag of the
lane and gives exact before/after edits per finding, or the whole fixed file(s) with a list of every
change tied to the page’s flags.
Base URL: https://api.skillsafe.ai/v1/app-api. Every call sends
Content-Type: application/json and, once you have a token,
Authorization: Bearer <token>. The paths below (/guest,
/me, /estimate, /run, /jobs/{job_id},
/run-stream) are relative to it, exactly as sdk.js calls them. A
plain-text summary of the app for agents is at /llms.txt. All three lanes run
on the gpt-terra model with a 10% publisher markup. A run needs a signed-in (user) token
and is metered: the platform places a hold on your credits when the run starts (the
estimate’s hold_credits, a reservation) and then charges only what
the run used (the job’s charged_credits). /estimate is free.
The response envelope
Responses are JSON of the form {"ok": ..., "data": ..., "error": {...}}. The page’s
SDK (sdk.js) decides success from the HTTP status, not from ok: on a
2xx it returns data; otherwise it raises an error whose message is
error.message (or the HTTP status text when the body carries none), with
status, code (error.code) and details
(error.details) attached. A body that is not JSON is treated as {}.
{ "ok": true, "data": { ... } }
{ "ok": false, "error": { "code": "...", "message": "...", "details": { ... } } }
Error codes
Branch on the HTTP status and error.code, show error.message. The page
handles 401 and 402 by name; the rest reach it as a generic failure.
| HTTP | error.code | What it means / what to do |
|---|---|---|
400 | VALIDATION_ERROR | The body is not a JSON object, a required field (task, code, facts) is missing, or a declared field has the wrong type. Every field is a string; sending facts or audit_handoff as an object is the usual cause. |
401 | UNAUTHORIZED | No token, or one that expired or was revoked. The page forgets the signed-in user and says “Your session expired - sign in again.” Get a new token. |
402 | INSUFFICIENT_CREDITS | The balance cannot cover the run. The page says “Not enough credits for this run.” with a top-up link, and tries never to submit into a 402: it disables the run button when the /me balance is below the estimate’s min_credits. |
404 | NOT_FOUND | A wrong path, or a job id that does not belong to this token. |
409 | CONFLICT | An Idempotency-Key replayed with a different body. Send the original body, or use a new key (the page bumps its :a<attempt> suffix for a retry). |
429 | RATE_LIMITED | Too many requests. Back off with a delay; do not tight-loop, and never resend a metered run without its original Idempotency-Key. |
Failures that do not arrive as an HTTP status:
| What | What happens |
|---|---|
event: error on /run-stream | The SDK raises an error carrying the event’s message, code and job_id. |
Job status: "failed" | Polling stops; the job’s error field says why. |
| Polling past 180 s | The SDK’s own client-side limit (waitForJob, 1 s interval, 180 000 ms default): it throws “job timed out”. The job itself is not cancelled. |
The task field: three lanes
task routes the run, and it takes exactly three values: "store",
"projection" and "fix". The two audits share one reply shape (with
different areas and catalogue ids), the fix lane has its own, and a reply never mixes two contracts.
The skill’s instructions tell the model to do the store lane when task is missing or
anything else (and the page’s own Eskit.buildInput turns any lane it does not know
into "store"), so a typo silently gets you a store audit: always send one of the three
values.
task | What it does | verdict | Use it when |
|---|---|---|---|
"store" | Reviews the event store the way a senior engineer reviews it before it holds production data: can a stream fork, can an append land half-way, can an event change after it was written, does every reader see every event in order. Rates three areas, in this order: schema (the events table’s constraints, event id, metadata, schema version, positions), appends (optimistic concurrency, one transaction per append, immutability) and reads (ordering, batching, and gaps a global reader can skip), each sound, fragile or broken. Confirms or dismisses every store flag (ES-01..ES-10, EV-01), adds what text rules cannot see (a version check outside the transaction that inserts, a unique violation that surfaces as a raw database error, a command that decides on a stream it read and appends without that version, SQL built with an f-string), and gives the verbatim evidence and exact before/after edits per finding, with a plan that ends in a pytest run against a real Postgres. | blocked | needs_work | ready | Before the store holds real data: constraints, versions, transactions and reads. |
"projection" | Reviews the projectors the way a senior engineer reviews them before a read model is trusted: does every event reach the read model exactly once in effect, in order, across crashes, restarts, a second instance and a rebuild. Rates four areas, in this order: delivery (checkpoint order and atomicity with the writes, a checkpoint at all, what a handler failure does), idempotency (what a replay does to every write), ordering (one event at a time, in position order; polling) and rebuild (reset and clear together, unknown event types). Confirms or dismisses every projection flag (PJ-01..PJ-10, EV-01), with the same findings, edits and plan as the store audit. | blocked | needs_work | ready | When read models drift, double count or lose events after a deploy, a crash or a rebuild. |
"fix" | Writes the fixed version of the pasted code: the whole of every file it changes (at most one new file next to them), with a unique (stream, version) and an expected version, appends in one transaction, ordered reads, checkpoints committed with the read model, and writes guarded against replay — plus a changes list tied to the page flags, the verify steps (pytest, and the tests to run against a real Postgres) and what it left open. It closes every high and medium flag of both catalogues unless the note asks to leave something as it is. Every public function keeps its name, its parameters (a new one only at the end, with a default), its return type and async def or def; a schema change goes at the end of the pasted .sql file as a new statement. | fixed | partial | When you want the code changed, not a list of findings: a store or projector you already know is broken, or what is left after an audit. |
The page chains the lanes: after either audit, “Apply every edit and fix the rest” puts
every audit edit into the paste, switches to the fix lane on that edited paste and adds an
audit_handoff string (below) with the audit’s lane, verdict, headline and findings;
after a store audit, “Apply every edit and audit the projections” puts the store edits
into the paste and selects the projection audit instead (no handoff string: it is a plain projection run on the edited paste). Over the API a
handoff is a second run with task: "fix", the edited paste as code,
facts recomputed from the edited paste for the fix lane, and that string. The page drops
the handoff as soon as the paste is edited again.
Guards the page applies before a run and the API does not: it refuses to price or run a paste it cannot read (“Nothing was pasted.”, “No Python or SQL file was found. Paste .py and .sql files, each with a # file: <path> line (-- file: for SQL) above it.”, or “This does not read as Python or SQL: no def, class, import or SQL statement was found.”), and it warns before a run on a paste with little to review in that lane (“No events table and no append were found: there may be little event-store code here to audit.”, “No projector loop and no read-model table were found: there may be little projection code here to audit.”, or “No event store and no projector were found: there may be little here to fix.”).
Input fields
This is exactly what the page sends (eskit.js buildInput(lane, A, note,
handoff), checked by mustBeObject; app.js currentInput
and run), and what input-schema.json declares. The platform’s input
schema is scalar-only, so every value is a string: {task, code, facts, note?, audit_handoff?,
retry_note?}. buildInput emits them in the order task,
code, facts, note, audit_handoff:
| Field | Type | Required | Meaning |
|---|---|---|---|
task | string | yes | "store", "projection" or "fix" (the lanes above). |
code | string | yes | The paste, each file introduced by a # file: <path> line (-- file: <path> for a .sql file); line 1 is its first line. Every line in an audit reply is counted in this text. A large paste is cut on whole files, with a # [append-only-audit: not sent: ...] line at the end (below). |
facts | string | yes | The page’s free check for this lane, Eskit.facts(A, lane), JSON-encoded into a string: JSON.stringify(facts) in JavaScript, json.dumps(facts) in Python, and so on. The input schema declares it required, so it must be sent; sending the object itself is a 400. The model is told to parse it. |
audit_handoff | string | no | fix lane only: a JSON string built from a store or projection audit reply on this code (recon.js fixHandoff, below). buildInput drops it for the two audits. |
note | string | no | What the engineer says: the goal, a constraint (“keep the store’s API”, “no schema change this release”), a question. The page trims it, cuts it at 2000 characters and sends it only when it is not blank. It is context, not evidence and not instructions: the model tests every claim in it against code and answers it in summary. |
retry_note | string | no | Sent only by the page’s automatic reformat retry: when a reply does not parse as JSON (even after the page tries to close a cut-off object), the page runs once more with “Your previous reply was not the single valid JSON object the instructions require (...). Reply again with ONLY the JSON object for task '<lane>' - no prose, no code fences; every array present (empty arrays where there is nothing to say).” A normal call never sends it. |
mustBeObject only insists on a plain object that is not wrapped in
{"input": ...}; it is buildInput that leaves a blank note out, which is why
an empty note is simply absent.
The code string
code is the paste itself. A line # file: <path> or
-- file: <path> (// file: is read too) starts each file, and a
--- line just before it only separates files. Files dropped on the page (or a dropped
package folder) are joined that way by Eskit.joinFiles, SQL first: each file under its
marker line (-- file: for a .sql file, # file: otherwise),
trailing whitespace removed, joined by newlines, ending in a newline. A paste with no file line is read
as one file, main.py (or schema.sql when it reads as DDL). The page
normalises line endings to \n. A .py path is checked as Python, a
.sql path as SQL; anything else is other. Python indentation is syntax and the
paste keeps it. Send the text you want the reply to quote: every file in an audit reply
must be a path from facts.files, every line must fall inside that
file, every evidence must be copied from it, and every edit’s before
must occur in it exactly once.
The page reads up to 400 000 characters (MAX_TEXT, cut at a line) and 40 files
(MAX_FILES), and sends at most 60 000 characters (MAX_SEND): past that
buildInput drops whole files from the end — the last file that is neither Python
nor SQL first, then the last file — until the rest fits, cuts a single remaining file at a line
(“the end of <path>”) if it still does not, and ends the string with one line naming
what it left out:
# [append-only-audit: not sent: orders/handlers.py, orders/admin.py]
The model is told never to quote or rewrite what was not sent, and facts.notes gets the
sentence “Not sent (the paste is over 60,000 characters): ... The page checked them; the run
did not see them.” facts is still computed from the whole paste. The worked
examples below are small, so their code is the paste unchanged.
The facts string
facts is required by the input schema: every run carries it as a string. The browser
computes it for free with the page’s check: Eskit.analyze splits the paste into
files, lexes them, rebuilds their structure, extracts the SQL, reads the schema, gives every table a
role, inventories the statements and projector loops and runs the rules, and
Eskit.facts(A, lane) turns that analysis into the object below. It is cut per
lane: the store audit gets the store flags and the shared ones (EV-01, EV-00), the projection
audit the projection flags and the shared ones, the fix lane all of them, and lane and
verdict_candidate follow the lane. The flag uids (F1,
F2, ...) are numbered over the whole paste, so a lane’s list can skip numbers. It is
how the model knows which flags to answer, which file paths and line ranges to name, and which
functions must survive. Its top-level keys, in the order built:
| Key | Contents |
|---|---|
lane | The lane these facts were cut for: store, projection or fix. Send it with the same task. |
summary | python_files, sql_files, functions, statements, schema_pasted (a CREATE TABLE of the events table is in the paste) and projector_loops. |
files | Every file: path (use it verbatim), kind (python, sql, other), line and end_line (the paste lines of the file’s content, without its marker line). |
functions | Every def (up to 200): name (Class.method for methods), file, line, async, private (the name starts with _), roles (append, reads, projection, checkpoint, checkpoint_reset, clears, rebuild). |
tables | Every table the paste creates or writes (up to 60): name, role (events, checkpoint, read_model, snapshot, store, tracking, other), created, file, line, columns, and for an events table stream_column, version_column, position_column, position_is_sequence, unique_stream_version, event_id_unique, metadata, schema_version. |
statements | Every SQL statement outside the DDL (up to 160): file, line, func, kind, table, conn, conn_from (own, param, shared, unknown), in_tx, in_loop, run_at, and where they apply order_by, limit, on_conflict, increments, guarded, formatted. |
projectors | Every loop that applies events (up to 30): file, line, func, handler_calls, checkpoint_saves (lines), checkpoint (after the apply, before the apply or none), atomic (the checkpoint and the handler’s writes commit in one transaction). |
checkpoint_saves | Every call or statement that saves a checkpoint (up to 40): file, line, call. |
flags | Every flag of the lane: uid, id (ES-01..ES-10, PJ-01..PJ-10, EV-01, EV-00), sev (high | medium | low | info), file, line, text, detail, and lines (every line an aggregated flag covers, only when it covers more than one). An audit must answer every non-info flag by its uid; a fix must close every high and medium flag or name it in open. |
verdict_candidate | The verdict the lane’s flags alone point at: blocked with any high flag, needs_work with any medium, otherwise ready. |
notes | Plain sentences about what was not read, not sent, or cut to fit the run (empty for a paste that fits). The facts are kept under about 40 000 characters: past that, long entries are clipped, the longest lists shortened and the flags kept most severe first, and a note names every cut. An audit_handoff over 30 000 characters is cut the same way. |
The simplest source is eskit.js itself: the page loads it from
/eskit.js, and it also runs in Node (it sets module.exports).
This script prints what the page would send as task, code and
facts for one lane (with the 60 000-character cut applied), with the facts decoded so you
can read them; the step 4 samples compact them back into the page’s string byte for byte. Save
the paste as ledger-paste.txt and redirect the output into
ledger-check.json:
// build-check.js -- run: node build-check.js store ledger-paste.txt > ledger-check.json (eskit.js saved next to it)
const fs = require("fs");
const Eskit = require("./eskit.js");
const lane = process.argv[2]; // store | projection | fix: facts are cut per lane
if (Eskit.LANES.indexOf(lane) === -1) { console.error("lane must be store, projection or fix"); process.exit(1); }
const A = Eskit.analyze(fs.readFileSync(process.argv[3], "utf8"));
if (!A.ok) { console.error(A.errors.join("\n")); process.exit(1); }
const inp = Eskit.buildInput(lane, A, "", null);
console.log(JSON.stringify({ task: lane, code: inp.code, facts: JSON.parse(inp.facts) }, null, 2));
const c = A.laneCounts[lane], real = A.flags.filter((f) => Eskit.inLane(f, lane) && f.sev !== "info");
console.error(lane + ": " + A.files.length + " file(s), " + A.functions.length + " function(s), " + A.statements.length + " SQL statement(s), " + A.runners.length + " projector loop(s); " +
real.length + " flags (" + c.high + " high, " + c.medium + " medium, " + c.low + " low); verdict_candidate " + A.candidates[lane]);
For the three worked examples it reports (node build-check.js store ledger-paste.txt,
node build-check.js projection orders-paste.txt, node build-check.js fix
loyalty-paste.txt):
store: 3 file(s), 9 function(s), 5 SQL statement(s), 0 projector loop(s); 7 flags (3 high, 3 medium, 1 low); verdict_candidate blocked
projection: 2 file(s), 12 function(s), 7 SQL statement(s), 1 projector loop(s); 6 flags (0 high, 6 medium, 0 low); verdict_candidate needs_work
fix: 2 file(s), 6 function(s), 6 SQL statement(s), 1 projector loop(s); 4 flags (1 high, 3 medium, 0 low); verdict_candidate blocked
For all three its code and facts are exactly what the page sends. The same
paste gives different facts per lane, so build the file for the lane you will run (a
store file sent with task: "projection" carries no projection flags to
answer).
Without the page’s check. A programmatic caller that does not run
eskit.js must build facts itself: it is required, and the platform only
checks that it is a string. A minimal object works — "{}", or a hand-built one such
as {"lane": "store", "flags": [], "notes": ["No page check was run; read the code
directly."]} — and the model still reads code. But send the fields the model
needs, or you lose what the check adds: with no flags there is nothing to confirm, dismiss
or close, with no files there are no paths and line ranges to name, with no
tables the model does not know which table holds the events or whether the schema was
pasted, with no projectors it does not know where each checkpoint is saved, and with no
functions nothing says what must survive a fix and stay async def or
def. Build it with eskit.js as above when you can; it costs nothing.
The audit_handoff string
Only for task: "fix", and only for the code the audit was about with every audit edit
applied. The page builds both from the normalised audit reply res
(recon.js normalize, below) and the analysis A of the audited
paste, in recon.js fixHandoff:
const run = Recon.applyFindings(A, res.findings, Eskit); // every finding's edits, in order, each on top of the last
JSON.stringify({ lane: res.task, verdict: res.verdict, headline: res.headline,
findings: res.findings.map(f => ({ id: f.id, severity: f.severity, file: f.file, line: f.line, title: f.title, why: f.why, edits: f.edits.length })),
applied: /* edits that applied */, failed: /* edits that did not */ })
// run.A.text is the edited paste: it becomes code, and facts is recomputed from it for the fix lane
That is {"lane", "verdict", "headline", "findings": [{id, severity, file, line, title, why,
edits}], "applied", "failed"}, where lane is store or
projection and each finding’s edits is the number of edits it
carried. applyFindings applies every finding’s edits with
Eskit.applyEdits, in the order listed: each before must occur exactly once
in its file as the earlier edits left it, or that edit is reported as failed and skipped — it is
never guessed. The handoff is a previous reply, not evidence: the model checks it against
code and fixes what is left. This script does the whole step — the edited paste,
its recomputed fix-lane facts and the handoff string, in the same shape as build-check.js
prints, so the samples read it the same way:
// handoff.js -- run: node handoff.js ledger-paste.txt ledger-reply.txt store > ledger-fix-check.json (eskit.js, recon.js next to it)
const fs = require("fs");
const Eskit = require("./eskit.js");
const Recon = require("./recon.js");
const A = Eskit.analyze(fs.readFileSync(process.argv[2], "utf8")); // the audited paste
const lane = process.argv[4] || "store"; // the audit's lane: store or projection
const res = Recon.normalize(Recon.parseResult(fs.readFileSync(process.argv[3], "utf8")), lane);
const ho = Recon.fixHandoff(A, res); // every edit, in order
const B = Eskit.analyze(ho.text); // facts recomputed from the edited paste
const inp = Eskit.buildInput("fix", B, "", ho.handoff);
console.log(JSON.stringify({ task: "fix", code: inp.code, facts: JSON.parse(inp.facts), audit_handoff: inp.audit_handoff }, null, 2));
console.error(ho.applied + " edit(s) applied, " + ho.failed + " did not apply");
Save the audit’s output.output as ledger-reply.txt; for the ledger
store audit below it reports 7 edit(s) applied, 0 did not apply. The handoff string it
builds, decoded, with each why left out:
{
"lane": "store",
"verdict": "blocked",
"headline": "The wallet store can fork a stream, land an append half-finished, and lets redact_iban rewrite already-committed events, so it is not ready to hold real money.",
"findings": [
{
"id": "ES-03",
"severity": "high",
"file": "wallet/store.py",
"line": 92,
"title": "redact_iban rewrites already-committed events",
"why": ...,
"edits": 0
},
{
"id": "ES-01",
"severity": "high",
"file": "migrations/001_wallet_events.sql",
"line": 13,
"title": "wallet_events has no unique constraint on (stream_id, version)",
"why": ...,
"edits": 1
},
{
"id": "ES-04",
"severity": "high",
"file": "wallet/store.py",
"line": 54,
"title": "append() inserts outside a transaction and never checks the version it expects",
"why": ...,
"edits": 2
},
{
"id": "ES-05",
"severity": "medium",
"file": "wallet/store.py",
"line": 78,
"title": "global reads by global_position have no guard against a gap from an in-flight transaction",
"why": ...,
"edits": 1
},
{
"id": "ES-02",
"severity": "medium",
"file": "wallet/service.py",
"line": 133,
"title": "withdraw decides on a balance it read, then appends without passing that version",
"why": ...,
"edits": 1
},
{
"id": "ES-06",
"severity": "medium",
"file": "wallet/store.py",
"line": 69,
"title": "read_stream returns events with no ORDER BY",
"why": ...,
"edits": 1
},
{
"id": "ES-10",
"severity": "low",
"file": "migrations/001_wallet_events.sql",
"line": 2,
"title": "wallet_events stores no schema version",
"why": ...,
"edits": 1
}
],
"applied": 7,
"failed": 0
}
The page sends that fix run with the audit’s note and the key
append-only-audit:fix:dcda6d34-3f23:a1. Its recomputed facts still raise one high flag
(ES-03, the redact_iban UPDATE the audit left without an edit), which the fix run must
close or name in open.
The output contracts
The model is instructed (SKILL.md) to reply with one JSON object and
nothing else — no prose, no code fences — with every array of its lane
present ([] when empty). The run’s output text is that object. In every lane it
must ground everything in code: a caller, a migration, an index, a trigger, a Postgres
version, an isolation level, a deployment (how many projector instances run) or a library’s
behaviour that is not in the paste is never claimed to exist — it goes into
questions (“if store.read_all orders by global_position ...”, never “it
does”). It must never write a secret value: a line that holds a literal credential is never
copied into the reply (not in evidence, not in an edit, not in a file), and no credential, hostname or
endpoint is invented. It must never change what the code offers its callers: public classes, functions
and methods keep their names, their parameters (a new one only at the end, with a default), their
return types and whether they are async def or def, unless the note allows
more. A schema change goes at the end of the pasted .sql file as a new statement (never an
edit of a CREATE TABLE that has already run), or in one new migration file when no .sql
file was pasted, and must work on a table that already has rows. It uses only asyncpg APIs that exist
(create_pool, Pool.acquire, Connection.transaction,
execute, executemany, fetch, fetchrow,
fetchval, asyncpg.exceptions.UniqueViolationError), the standard library
and what the paste already imports. Every flag of the lane whose sev is not
info is answered exactly once (audit lanes: in flag_answers), and an info
flag is never reported as a finding.
task: "store" and task: "projection"
task:"store"or"projection", the lane that was asked for.verdict:blocked(any finding is high) |needs_work(the worst is medium) |ready(nothing above low; low findings and questions do not stop a release).headline: one sentence.system: one or two sentences — what the code stores or projects, and what runs concurrently in it.areas[]: exactly one entry per area of the lane, in the lane’s order — store:schema,appends,reads; projection:delivery,idempotency,ordering,rebuild— as{area, status, note}withstatussound|fragile|broken(brokenwhen a high finding sits in that area,fragilefor a medium one).findings[]:{id, flags[], file, line, severity, area, title, evidence, why, edits[]}.idis a catalogue id of the lane (store: ES-01..ES-10 or EV-01; projection: PJ-01..PJ-10 or EV-01) orother;flagslists the pageuids it confirms ([]for something the page did not flag);fileis apathfromfacts.files, copied exactly;lineis the line ofcodewith the offending text, inside that file’sline..end_line;severityishigh(events are lost, forked, rewritten or silently skipped: ES-01, ES-03, ES-04, PJ-03) |medium(it goes wrong under concurrency, a crash, a replay or a restart: ES-02, ES-05, ES-06, PJ-01, PJ-04..PJ-09, EV-01, and PJ-02 unless the checkpoint is atomic) |low(bounded risk or hygiene: ES-07..ES-10, PJ-10, and PJ-02 when the checkpoint commits with the writes);evidenceis one to three whole lines copied verbatim fromcodeas pasted (indentation may be trimmed). A problem of the other audit is named in one question (“run the projection audit”), not reported as a finding.edits[]:{before, after}on that one file. Eachbeforeis a contiguous, exact slice of the file — whole lines with their indentation, as pasted — that occurs exactly once in the file as it stands when the edit is applied;afteris its replacement, keeping the indentation of the code around it (spaces, never tabs), including SQL inside a triple-quoted string. Findings are applied in the order listed and their edits in order, each on top of the ones before, so a later finding that touches the same lines writes itsbeforeas the earlier one left it. A schema change is an edit whosebeforeis the last statement of the.sqlfile and whoseafteris that statement followed by the new one. Every line an edit adds follows the house rules itself, so the page’s re-check raises nothing new.editsmay be[]only for a low finding or a change outside the paste.flag_answers[]:{flag, status, reason}, exactly one per non-info flag of the lane,statusconfirmed(a finding lists that uid) |dismissed(with the reason).plan[]: the findings in the order to apply them — what loses, forks or rewrites events first, then concurrency and atomicity, then replay safety and ordering, then hygiene — ending with a test run with pytest against a real Postgres (two concurrent appends with the same expected version, one of which must fail; a batch applied twice that must leave the read model unchanged; a projector stopped between the write and the checkpoint).questions[](only what the paste cannot answer) andsummary(two to four sentences, answeringnoteif one was sent).
task: "fix"
task:"fix".verdict:fixed(openis empty) |partial.headline: one sentence.files[]:{path, content}— every pasted file it changes, whole (every line, without the# file:or-- file:line, indented exactly as the file already is: a top-leveldeforclassstarts in column one), under itspathfromfacts.files. At most one new file next to them (a migration when no.sqlfile was pasted, a small helper module).changes[]:{file, function, flags[], change, why}, one per change.functionis the function it is in, asfacts.functionsnames it (Class.method), or(module)for a change outside any function (an import, a constant, a class attribute) or(schema)for SQL in a.sqlfile;flagsare the page flags the change closes. Every high and medium flag appears in theflagsof a change or of anopenitem.verify[]: always the test suite (pytest), plus the tests that would have caught the worst problems, run against a real Postgres: two concurrent appends with the same expected version (one raises the concurrency error); a batch applied twice (the read model is unchanged the second time); the projector stopped after the writes and before the checkpoint (a restart applies nothing twice); a rebuild that ends with the same read model as the live projection.open[]:{item, flags[], reason}for what it left unfixed — a decision for the engineer, or code outside the paste — with each item’s risk as it stands in the returned files.questions[]andsummary, as in the audits; with anaudit_handoff, it fixes what the audit left.
How the page reads a reply (recon.js parseResult and
normalize): it strips a code fence and parses from the first { to the
last }. A reply is read as the lane that was asked for; one naming the fix lane when an
audit was asked (or the other way round) is read as that lane only when it carries that lane’s
body (a files array for fix, a findings array for an audit), and one naming
the other audit is read as the audit asked for, with the mismatch counted as a disagreement. An unknown
verdict shows as needs_work or partial and counts as a
disagreement; an unknown severity becomes medium; ids like ES1
or pj 2 become ES-01 and PJ-02; flags keeps only
F<n> values; findings with no file, evidence and title are dropped;
system may also arrive as feature or service, file
as path, a change’s function as func and its
change as what, and an open item as path;
questions is capped at 10, plan at 30 and verify at 20. An audit
reply carries 9 sections (verdict, headline, system, areas, findings, flag_answers, plan, questions,
summary) and a fix reply 8 (verdict, headline, files, changes, verify, open, questions, summary),
which is what the page counts when it recovers a partial reply.
Worked examples
One per lane, all three examples the page ships (the example buttons on the app
page; the full pastes are in /example.js, ids ledger,
orders and loyalty), built exactly as the page builds them:
Eskit.buildInput(lane, Eskit.analyze(text), note, null), with the real model reply the
page replays for each (/example-result.js). The pastes are shown
whole; the request bodies cut the long code and facts strings with a visible
..., and the replies are pretty-printed excerpts, with ... where entries
were left out. Every value shown is copied from the input or the reply. No paste holds a credential,
and the services are invented.
Store audit: a wallet ledger whose events table has no unique (stream_id, version), whose append inserts one event at a time outside a transaction with no expected version, and which rewrites stored events to erase an IBAN
Three files: migrations/001_wallet_events.sql (the wallet_events table, with
a BIGSERIAL global_position, a unique event_id UUID, a
metadata JSONB column, no schema version, and only a plain index on
(stream_id, version)), wallet/store.py and wallet/service.py: a
wallet’s deposits and withdrawals. WalletEventStore.append takes its own pool
connection, reads COALESCE(MAX(version), 0) and inserts each event in a loop, with no
transaction and no expected version; read_stream selects a stream with no
ORDER BY; read_all reads WHERE global_position > $1 ORDER BY
global_position LIMIT $2; redact_iban runs UPDATE wallet_events SET data =
data - 'iban'. WalletService.withdraw reads the stream, folds it into a balance,
checks it and appends FundsWithdrawn. The page’s free check raises 7 store flags
(3 high, 3 medium, 1 low) plus the EV-00 context flag; its own reading of the verdict is
blocked. The engineer’s note: “We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request.”
The code string, whole: 4,472 characters, 136 lines ending in a newline.
Save it as ledger-paste.txt and run
node build-check.js store ledger-paste.txt > ledger-check.json:
-- file: migrations/001_wallet_events.sql
CREATE TABLE wallet_events (
global_position BIGSERIAL PRIMARY KEY,
event_id UUID NOT NULL UNIQUE,
stream_id TEXT NOT NULL,
version INTEGER NOT NULL,
event_type TEXT NOT NULL,
data JSONB NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}',
recorded_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_wallet_events_stream ON wallet_events (stream_id, version);
---
# file: wallet/store.py
import json
from dataclasses import dataclass, field
from typing import List, Optional
from uuid import UUID, uuid4
import asyncpg
@dataclass
class NewEvent:
event_type: str
data: dict
metadata: dict = field(default_factory=dict)
event_id: UUID = field(default_factory=uuid4)
@dataclass
class StoredEvent:
stream_id: str
version: int
event_type: str
data: dict
global_position: int
class WalletEventStore:
def __init__(self, pool: asyncpg.Pool):
self.pool = pool
async def append(self, stream_id: str, events: List[NewEvent]) -> int:
async with self.pool.acquire() as conn:
current = await conn.fetchval(
"SELECT COALESCE(MAX(version), 0) FROM wallet_events WHERE stream_id = $1",
stream_id,
)
for i, event in enumerate(events, start=1):
await conn.execute(
"""
INSERT INTO wallet_events (event_id, stream_id, version, event_type, data, metadata)
VALUES ($1, $2, $3, $4, $5, $6)
""",
event.event_id,
stream_id,
current + i,
event.event_type,
json.dumps(event.data),
json.dumps(event.metadata),
)
return current + len(events)
async def read_stream(self, stream_id: str) -> List[StoredEvent]:
async with self.pool.acquire() as conn:
rows = await conn.fetch(
"SELECT stream_id, version, event_type, data, global_position FROM wallet_events WHERE stream_id = $1",
stream_id,
)
return [self._row(r) for r in rows]
async def read_all(self, after_position: int, limit: int = 500) -> List[StoredEvent]:
async with self.pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT stream_id, version, event_type, data, global_position
FROM wallet_events
WHERE global_position > $1
ORDER BY global_position
LIMIT $2
""",
after_position,
limit,
)
return [self._row(r) for r in rows]
async def redact_iban(self, stream_id: str) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"UPDATE wallet_events SET data = data - 'iban' WHERE stream_id = $1",
stream_id,
)
@staticmethod
def _row(r: asyncpg.Record) -> StoredEvent:
return StoredEvent(
stream_id=r["stream_id"],
version=r["version"],
event_type=r["event_type"],
data=json.loads(r["data"]),
global_position=r["global_position"],
)
---
# file: wallet/service.py
from typing import Optional
from .store import NewEvent, WalletEventStore
class WalletService:
def __init__(self, store: WalletEventStore):
self.store = store
async def deposit(self, wallet_id: str, amount_cents: int, reference: str) -> int:
return await self.store.append(
f"Wallet-{wallet_id}",
[NewEvent("FundsDeposited", {"amount_cents": amount_cents, "reference": reference})],
)
async def withdraw(self, wallet_id: str, amount_cents: int) -> int:
stream = f"Wallet-{wallet_id}"
history = await self.store.read_stream(stream)
balance = 0
for e in history:
if e.event_type == "FundsDeposited":
balance += e.data["amount_cents"]
elif e.event_type == "FundsWithdrawn":
balance -= e.data["amount_cents"]
if balance < amount_cents:
raise ValueError("insufficient funds")
return await self.store.append(
stream,
[NewEvent("FundsWithdrawn", {"amount_cents": amount_cents})],
)
Its store-lane facts, decoded (the request carries this object as one compact JSON string):
{
"lane": "store",
"summary": {
"python_files": 2,
"sql_files": 1,
"functions": 9,
"statements": 5,
"schema_pasted": true,
"projector_loops": 0
},
"files": [
{
"path": "migrations/001_wallet_events.sql",
"kind": "sql",
"line": 2,
"end_line": 13
},
{
"path": "wallet/store.py",
"kind": "python",
"line": 16,
"end_line": 104
},
{
"path": "wallet/service.py",
"kind": "python",
"line": 107,
"end_line": 137
}
],
"functions": [
{
"name": "WalletEventStore.__init__",
"file": "wallet/store.py",
"line": 42,
"async": false,
"private": false,
"roles": []
},
{
"name": "WalletEventStore.append",
"file": "wallet/store.py",
"line": 45,
"async": true,
"private": false,
"roles": [
"append",
"reads"
]
},
{
"name": "WalletEventStore.read_stream",
"file": "wallet/store.py",
"line": 66,
"async": true,
"private": false,
"roles": [
"reads"
]
},
{
"name": "WalletEventStore.read_all",
"file": "wallet/store.py",
"line": 74,
"async": true,
"private": false,
"roles": [
"reads"
]
},
{
"name": "WalletEventStore.redact_iban",
"file": "wallet/store.py",
"line": 89,
"async": true,
"private": false,
"roles": []
},
{
"name": "WalletEventStore._row",
"file": "wallet/store.py",
"line": 97,
"async": false,
"private": true,
"roles": []
},
{
"name": "WalletService.__init__",
"file": "wallet/service.py",
"line": 113,
"async": false,
"private": false,
"roles": []
},
{
"name": "WalletService.deposit",
"file": "wallet/service.py",
"line": 116,
"async": true,
"private": false,
"roles": []
},
{
"name": "WalletService.withdraw",
"file": "wallet/service.py",
"line": 122,
"async": true,
"private": false,
"roles": []
}
],
"tables": [
{
"name": "wallet_events",
"role": "events",
"created": true,
"file": "migrations/001_wallet_events.sql",
"line": 2,
"columns": [
"global_position",
"event_id",
"stream_id",
"version",
"event_type",
"data",
"metadata",
"recorded_at"
],
"stream_column": "stream_id",
"version_column": "version",
"position_column": "global_position",
"position_is_sequence": true,
"unique_stream_version": false,
"event_id_unique": true,
"metadata": true,
"schema_version": false
}
],
"statements": [
{
"file": "wallet/store.py",
"line": 48,
"func": "WalletEventStore.append",
"kind": "SELECT",
"table": "wallet_events",
"conn": "conn",
"conn_from": "own",
"in_tx": false,
"in_loop": false,
"run_at": "wallet/store.py:47",
"order_by": null,
"limit": false
},
{
"file": "wallet/store.py",
"line": 54,
"func": "WalletEventStore.append",
"kind": "INSERT",
"table": "wallet_events",
"conn": "conn",
"conn_from": "own",
"in_tx": false,
"in_loop": true,
"run_at": "wallet/store.py:52",
"on_conflict": null
},
{
"file": "wallet/store.py",
"line": 69,
"func": "WalletEventStore.read_stream",
"kind": "SELECT",
"table": "wallet_events",
"conn": "conn",
"conn_from": "own",
"in_tx": false,
"in_loop": false,
"run_at": "wallet/store.py:68",
"order_by": null,
"limit": false
},
{
"file": "wallet/store.py",
"line": 78,
"func": "WalletEventStore.read_all",
"kind": "SELECT",
"table": "wallet_events",
"conn": "conn",
"conn_from": "own",
"in_tx": false,
"in_loop": false,
"run_at": "wallet/store.py:76",
"order_by": "global_position",
"limit": true
},
{
"file": "wallet/store.py",
"line": 92,
"func": "WalletEventStore.redact_iban",
"kind": "UPDATE",
"table": "wallet_events",
"conn": "conn",
"conn_from": "own",
"in_tx": false,
"in_loop": false,
"run_at": "wallet/store.py:91",
"increments": [
"data"
]
}
],
"projectors": [],
"checkpoint_saves": [],
"flags": [
{
"uid": "F1",
"id": "ES-01",
"sev": "high",
"file": "migrations/001_wallet_events.sql",
"line": 2,
"text": "events table wallet_events has no unique constraint on (stream_id, version)",
"detail": "Two appends that both read the same current version both insert version n+1 and the stream forks silently. A PRIMARY KEY or UNIQUE (stream_id, version) - or a unique index on exactly those columns - turns the race into an error the append can report as a concurrency conflict."
},
{
"uid": "F2",
"id": "ES-04",
"sev": "high",
"file": "wallet/store.py",
"line": 54,
"text": "WalletEventStore.append inserts several events outside a transaction",
"detail": "Each INSERT commits on its own: a failure half way leaves the first events of the batch in the stream without the rest, and a subscriber can read the first half before the second exists. Wrap the version check and every insert in one transaction (async with conn.transaction():)."
},
{
"uid": "F3",
"id": "ES-03",
"sev": "high",
"file": "wallet/store.py",
"line": 92,
"text": "UPDATE on the events table wallet_events in WalletEventStore.redact_iban",
"detail": "Events are immutable facts: changing or deleting one rewrites history under every projection and subscriber that already read it, and a rebuild no longer reproduces what users saw. Append a compensating event instead (a correction, a reversal, a redaction event with crypto-shredding for personal data)."
},
{
"uid": "F4",
"id": "ES-02",
"sev": "medium",
"file": "wallet/store.py",
"line": 54,
"text": "WalletEventStore.append appends without an expected version",
"detail": "Nothing in the append path takes the version the caller last read, so a command decided on stale state is written anyway (lost update). Take expected_version, compare it with the stream's current version in the same transaction and raise a concurrency error on a mismatch."
},
{
"uid": "F5",
"id": "ES-06",
"sev": "medium",
"file": "wallet/store.py",
"line": 69,
"text": "events of wallet_events read with no ORDER BY in WalletEventStore.read_stream",
"detail": "Without ORDER BY, Postgres returns rows in whatever order the plan produces - usually insertion order, until a parallel scan, an index choice or a vacuum changes it. Replaying a stream out of order rebuilds the wrong state. ORDER BY the version (stream reads) or the global position (subscriptions)."
},
{
"uid": "F6",
"id": "ES-05",
"sev": "medium",
"file": "wallet/store.py",
"line": 78,
"text": "global reads of wallet_events by global_position (a sequence) with no guard against transactions still in flight",
"detail": "A sequence value is taken when a row is inserted, not when it commits: position 41 can commit after 42, and a subscriber that has already moved its checkpoint to 42 never sees 41. Serialise appends (SELECT pg_advisory_xact_lock(...) in the append transaction) or only read below the oldest running transaction (a transaction-id column compared with pg_snapshot_xmin(pg_current_snapshot()))."
},
{
"uid": "F7",
"id": "ES-10",
"sev": "low",
"file": "migrations/001_wallet_events.sql",
"line": 2,
"text": "events table wallet_events stores no schema version",
"detail": "When an event's payload shape changes, old and new events of the same type cannot be told apart by an upcaster. Record a schema version per event (a column, or in metadata) from the first event on."
},
{
"uid": "F8",
"id": "EV-00",
"sev": "info",
"file": "migrations/001_wallet_events.sql",
"line": 2,
"text": "context: 2 Python files, 1 SQL file, 5 SQL statement(s) in 9 function(s); events table wallet_events; 0 read-model table(s), 0 checkpoint table(s), 0 projector loop(s)",
"detail": "Only what is pasted is checked: the schema checks need the CREATE TABLE (and its indexes); transactions are read from async with ....transaction()/begin() blocks and explicit start()/commit(); what a function in another file does, the isolation level and how many projector instances run are outside the check."
}
],
"verdict_candidate": "blocked",
"notes": []
}
The request body (POST /estimate, then POST /run or
/run-stream with Idempotency-Key: append-only-audit:store:7c46ba71-30c5:a1 as
the page sends it):
{
"task": "store",
"code": "-- file: migrations/001_wallet_events.sql\nCREATE TABLE wallet_events (\n global_position BIGSERIAL PRIMARY KEY,\n event_id UUID NOT NULL UNIQUE,\n...",
"facts": "{\"lane\":\"store\",\"summary\":{\"python_files\":2,\"sql_files\":1,\"functions\":9,\"statements\":5,\"schema_pasted\":true,\"projector_loops\":0},\"files\":...",
"note": "We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request."
}
The reply (output.output), an excerpt: the verdict, the headline, the system, the three
areas, and two of the findings with their edits:
{
"task": "store",
"verdict": "blocked",
"headline": "The wallet store can fork a stream, land an append half-finished, and lets redact_iban rewrite already-committed events, so it is not ready to hold real money.",
"system": "This is a Postgres/asyncpg event store for one wallet's deposits and withdrawals (wallet_events), with a WalletService that reads a wallet's history to check its balance before appending a withdrawal event. The paste says nothing about how many instances of the service or store run at once, so concurrent appends to the same stream must be assumed possible.",
"areas": [
{
"area": "schema",
"status": "broken",
"note": "wallet_events has no unique constraint on (stream_id, version) and stores no schema version."
},
{
"area": "appends",
"status": "broken",
"note": "redact_iban rewrites events in place, and append() inserts several events outside a transaction with no expected-version check."
},
{
"area": "reads",
"status": "fragile",
"note": "read_stream has no ORDER BY and read_all's global-position read has no guard against a still-committing transaction."
}
],
"findings": [
...,
{
"id": "ES-01",
"flags": [
"F1"
],
"file": "migrations/001_wallet_events.sql",
"line": 13,
"severity": "high",
"area": "schema",
"title": "wallet_events has no unique constraint on (stream_id, version)",
"evidence": "CREATE INDEX idx_wallet_events_stream ON wallet_events (stream_id, version);",
"why": "The index on (stream_id, version) is not unique, so two appends that both read the same current version (store.py's append, lines 47-50) both insert the same next version and the stream forks silently instead of one failing. A UNIQUE constraint turns that race into a database error the append can report as a concurrency conflict. If wallet_events already has duplicate (stream_id, version) rows, this ALTER TABLE fails until they are deduplicated first.",
"edits": [
{
"before": "CREATE INDEX idx_wallet_events_stream ON wallet_events (stream_id, version);",
"after": "CREATE INDEX idx_wallet_events_stream ON wallet_events (stream_id, version);\n\nALTER TABLE wallet_events ADD CONSTRAINT wallet_events_stream_version_key UNIQUE (stream_id, version);"
}
]
},
...,
{
"id": "ES-06",
"flags": [
"F5"
],
"file": "wallet/store.py",
"line": 69,
"severity": "medium",
"area": "reads",
"title": "read_stream returns events with no ORDER BY",
"evidence": " \"SELECT stream_id, version, event_type, data, global_position FROM wallet_events WHERE stream_id = $1\",",
"why": "Without ORDER BY, Postgres returns wallet_events rows in whatever order its query plan produces, not necessarily the order they were appended; withdraw folds this list into a running balance, so an out-of-order read can silently produce a wrong balance. ORDER BY version returns them in the order they happened.",
"edits": [
{
"before": " \"SELECT stream_id, version, event_type, data, global_position FROM wallet_events WHERE stream_id = $1\",",
"after": " \"SELECT stream_id, version, event_type, data, global_position FROM wallet_events WHERE stream_id = $1 ORDER BY version\","
}
]
},
...
],
"flag_answers": [
{
"flag": "F1",
"status": "confirmed",
"reason": ""
},
...
],
"plan": [
"Add a UNIQUE constraint on (stream_id, version) to wallet_events so a race between two appends errors instead of forking the stream (ES-01).",
...,
"Write a pytest suite against a real Postgres: two concurrent appends to the same stream with the same expected_version where one must raise the concurrency error, and a withdraw retried after a crash that must not double-spend or skip the checkpoint of a downstream projector."
],
"questions": [
"Are there existing rows in wallet_events today, and could any already share a (stream_id, version) pair? The new UNIQUE constraint (ES-01) fails to apply if duplicates exist.",
...
],
"summary": "Not yet safe for real deposits: wallet_events has no unique constraint on (stream_id, version) so a race can fork a stream, append() inserts a batch outside a transaction with no expected-version check so a crash can half-land an append or a stale balance check can overspend, and redact_iban rewrites events in place. The schema and append fixes below are mechanical, but redact_iban needs a design decision (keep the IBAN out of the event, or encrypt it) that this paste can't resolve alone. Two medium ordering/replay gaps (no ORDER BY on stream reads, no guard on the global read) and a missing schema-version column round out the list; none of the low or open items should block the mechanical fixes above."
}
The reply has 7 findings and confirms all 7 flags; they carry 7 edits in all. In order: ES-03
(redact_iban, F3) with no edit, because keeping the IBAN out of the event or
encrypting it with a per-subject key is a design decision the paste cannot settle; ES-01 (F1) adds
UNIQUE (stream_id, version) as a new statement at the end of the migration; ES-04 carries
F2 and F4 together: a ConcurrencyError class, and append wrapped in
async with conn.transaction(): with a new trailing parameter
expected_version: Optional[int] = None and the unique violation turned into that error;
ES-05 (F6) takes pg_advisory_xact_lock as the first statement of the append transaction;
a finding the page did not flag (ES-02, flags: []) makes withdraw pass the
version it read; ES-06 (F5) adds ORDER BY version; and ES-10 (F7) adds a
schema_version column with a DEFAULT, written on top of the ES-01 statement.
The public append keeps its name, its first two parameters and async def.
Projection audit: an order-summary projector whose handlers each take their own pool connection and add to totals with no guard, which logs and skips failed events, saves its checkpoint in a separate transaction and rebuilds without clearing the table
Two files: db/schema.sql (an order_events table that is sound — an
identity global_position, a unique event_id, UNIQUE (stream_id,
version), schema_version and metadata — the
order_summaries read model and projection_checkpoints) and
orders/projections.py. OrderSummaryProjection looks its handlers up with
self.handlers.get(event.event_type), and every handler takes its own pool connection:
_on_placed is a plain INSERT, _on_item_added and
_on_item_removed add to item_count and total_cents.
Projector.run_once loads the checkpoint, reads a batch with
self.store.read_all, applies each event inside except Exception: that only
logs, and after the loop upserts the checkpoint on another connection; run_forever
sleeps a second when a batch is empty; rebuild deletes the checkpoint row and replays. The
page’s free check raises 6 projection flags (all medium) plus EV-00; its own reading of the
verdict is needs_work. The engineer’s note: “The order totals drift upwards after every deploy, and last week's rebuild made them worse. What do we fix first? Nothing outside this file calls OrderSummaryProjection or Projector except our worker's run_forever() and an admin command's rebuild(), so their methods may change.”
The code string, whole: 5,360 characters, 159 lines ending in a newline.
Save it as orders-paste.txt and run
node build-check.js projection orders-paste.txt > orders-check.json:
-- file: db/schema.sql
CREATE TABLE order_events (
global_position BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
event_id UUID NOT NULL UNIQUE,
stream_id TEXT NOT NULL,
version INTEGER NOT NULL,
event_type TEXT NOT NULL,
schema_version SMALLINT NOT NULL DEFAULT 1,
data JSONB NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}',
recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (stream_id, version)
);
CREATE TABLE order_summaries (
order_id UUID PRIMARY KEY,
customer_id UUID NOT NULL,
status TEXT NOT NULL,
item_count INTEGER NOT NULL DEFAULT 0,
total_cents BIGINT NOT NULL DEFAULT 0,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE projection_checkpoints (
projection TEXT PRIMARY KEY,
last_position BIGINT NOT NULL DEFAULT 0,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
---
# file: orders/projections.py
import asyncio
import logging
from typing import Awaitable, Callable, Dict
import asyncpg
log = logging.getLogger(__name__)
Handler = Callable[[dict], Awaitable[None]]
class OrderSummaryProjection:
name = "order_summary"
def __init__(self, pool: asyncpg.Pool):
self.pool = pool
self.handlers: Dict[str, Handler] = {
"OrderPlaced": self._on_placed,
"ItemAdded": self._on_item_added,
"ItemRemoved": self._on_item_removed,
"OrderShipped": self._on_shipped,
}
async def apply(self, event) -> None:
handler = self.handlers.get(event.event_type)
if handler is not None:
await handler(event.data)
async def _on_placed(self, data: dict) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"INSERT INTO order_summaries (order_id, customer_id, status) VALUES ($1, $2, 'placed')",
data["order_id"],
data["customer_id"],
)
async def _on_item_added(self, data: dict) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"""
UPDATE order_summaries
SET item_count = item_count + $2,
total_cents = total_cents + $3,
updated_at = now()
WHERE order_id = $1
""",
data["order_id"],
data["quantity"],
data["quantity"] * data["unit_price_cents"],
)
async def _on_item_removed(self, data: dict) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"""
UPDATE order_summaries
SET item_count = item_count - $2,
total_cents = total_cents - $3,
updated_at = now()
WHERE order_id = $1
""",
data["order_id"],
data["quantity"],
data["quantity"] * data["unit_price_cents"],
)
async def _on_shipped(self, data: dict) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"UPDATE order_summaries SET status = 'shipped', updated_at = now() WHERE order_id = $1",
data["order_id"],
)
class Projector:
def __init__(self, store, pool: asyncpg.Pool, projection: OrderSummaryProjection, batch_size: int = 200):
self.store = store
self.pool = pool
self.projection = projection
self.batch_size = batch_size
async def _load_checkpoint(self) -> int:
async with self.pool.acquire() as conn:
position = await conn.fetchval(
"SELECT last_position FROM projection_checkpoints WHERE projection = $1",
self.projection.name,
)
return position or 0
async def _save_checkpoint(self, position: int) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"""
INSERT INTO projection_checkpoints (projection, last_position)
VALUES ($1, $2)
ON CONFLICT (projection) DO UPDATE
SET last_position = EXCLUDED.last_position, updated_at = now()
""",
self.projection.name,
position,
)
async def run_once(self) -> int:
position = await self._load_checkpoint()
events = await self.store.read_all(position, self.batch_size)
for event in events:
try:
await self.projection.apply(event)
except Exception:
log.exception("order_summary failed on event %s", event.global_position)
position = event.global_position
if events:
await self._save_checkpoint(position)
return len(events)
async def run_forever(self) -> None:
while True:
applied = await self.run_once()
if applied == 0:
await asyncio.sleep(1.0)
async def rebuild(self) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"DELETE FROM projection_checkpoints WHERE projection = $1",
self.projection.name,
)
while await self.run_once():
pass
Its projection-lane facts, decoded, with functions, tables and
statements left out:
{
"lane": "projection",
"summary": {
"python_files": 1,
"sql_files": 1,
"functions": 12,
"statements": 7,
"schema_pasted": true,
"projector_loops": 1
},
"files": [
{
"path": "db/schema.sql",
"kind": "sql",
"line": 2,
"end_line": 28
},
{
"path": "orders/projections.py",
"kind": "python",
"line": 31,
"end_line": 160
}
],
"functions": ...,
"tables": ...,
"statements": ...,
"projectors": [
{
"file": "orders/projections.py",
"line": 136,
"func": "Projector.run_once",
"handler_calls": [
"self.projection.apply @138"
],
"checkpoint_saves": [
143
],
"checkpoint": "after the apply",
"atomic": false
}
],
"checkpoint_saves": [
{
"file": "orders/projections.py",
"line": 143,
"call": "self._save_checkpoint"
},
{
"file": "orders/projections.py",
"line": 122,
"call": "INSERT projection_checkpoints"
}
],
"flags": [
{
"uid": "F1",
"id": "PJ-02",
"sev": "medium",
"file": "orders/projections.py",
"line": 62,
"text": "OrderSummaryProjection._on_placed inserts into order_summaries with no ON CONFLICT - not safe to replay",
"detail": "Applied twice (a redelivery after a crash, a second projector instance, a rebuild that did not clear the table) the insert fails on the primary key and stalls the projection, or adds a duplicate row. Guard the write with the event's position (store last_position per row and only apply newer events), or record processed event ids, or recompute the value from state."
},
{
"uid": "F2",
"id": "PJ-02",
"sev": "medium",
"file": "orders/projections.py",
"line": 71,
"text": "OrderSummaryProjection._on_item_added adds to item_count, total_cents - not safe to replay",
"detail": "Applied twice (a redelivery after a crash, a second projector instance, a rebuild that did not clear the table) the increment counts the event twice. Guard the write with the event's position (store last_position per row and only apply newer events), or record processed event ids, or recompute the value from state."
},
{
"uid": "F3",
"id": "PJ-02",
"sev": "medium",
"file": "orders/projections.py",
"line": 86,
"text": "OrderSummaryProjection._on_item_removed adds to item_count, total_cents - not safe to replay",
"detail": "Applied twice (a redelivery after a crash, a second projector instance, a rebuild that did not clear the table) the increment counts the event twice. Guard the write with the event's position (store last_position per row and only apply newer events), or record processed event ids, or recompute the value from state."
},
{
"uid": "F4",
"id": "PJ-04",
"sev": "medium",
"file": "orders/projections.py",
"line": 139,
"text": "Projector.run_once swallows a handler failure and moves on",
"detail": "The except catches the failure, the loop continues and the checkpoint moves past the event, so the read model silently misses it - and every later event of that entity is applied to the wrong state. Let it raise (the projector stops and retries from the checkpoint), or park the event in a dead-letter table and alert."
},
{
"uid": "F5",
"id": "PJ-01",
"sev": "medium",
"file": "orders/projections.py",
"line": 143,
"text": "Projector.run_once saves the checkpoint outside the transaction that writes the read model",
"detail": "The handlers take their own connection from the pool, so each write commits on its own; a crash between the two either replays events that were already applied (double counts unless every write is idempotent) or skips events whose writes never committed. Write the read model and the checkpoint on one connection inside one transaction."
},
{
"uid": "F6",
"id": "PJ-06",
"sev": "medium",
"file": "orders/projections.py",
"line": 154,
"text": "Projector.rebuild resets the checkpoint but never clears the read model",
"detail": "The rebuild replays every event on top of the rows that are already there: increments double and inserts collide. Clear (or rebuild into a fresh table and swap) in the same step as the checkpoint reset."
},
{
"uid": "F7",
"id": "EV-00",
"sev": "info",
"file": "db/schema.sql",
"line": 2,
"text": "context: 1 Python file, 1 SQL file, 7 SQL statement(s) in 12 function(s); events table order_events; 1 read-model table(s), 1 checkpoint table(s), 1 projector loop(s)",
"detail": "Only what is pasted is checked: the schema checks need the CREATE TABLE (and its indexes); transactions are read from async with ....transaction()/begin() blocks and explicit start()/commit(); what a function in another file does, the isolation level and how many projector instances run are outside the check."
}
],
"verdict_candidate": "needs_work",
"notes": []
}
The request body (Idempotency-Key: append-only-audit:projection:c444b495-3b17:a1 as the page
sends it):
{
"task": "projection",
"code": "-- file: db/schema.sql\nCREATE TABLE order_events (\n global_position BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,\n event_id UUID NOT NULL UNIQUE,\n...",
"facts": "{\"lane\":\"projection\",\"summary\":{\"python_files\":1,\"sql_files\":1,\"functions\":12,\"statements\":7,\"schema_pasted\":true,\"projector_loops\":1},\"files\":...",
"note": "The order totals drift upwards after every deploy, and last week's rebuild made them worse. What do we fix first? Nothing outside this file calls OrderSummaryProjection or Projector except our worker's run_forever() and an admin command's rebuild(), so their methods may change."
}
The reply, an excerpt: the verdict, the headline, the system, the four areas, the rebuild finding with its edit, every flag answer, and the summary that answers the note:
{
"task": "projection",
"verdict": "needs_work",
"headline": "The checkpoint isn't atomic with the read-model write, three handlers aren't safe to replay, and rebuild never clears order_summaries - together that is exactly the deploy-time drift and the worse rebuild the note describes.",
"system": "The paste projects OrderPlaced, ItemAdded, ItemRemoved and OrderShipped from order_events into one order_summaries read model through Projector.run_once, backed by a projection_checkpoints row; run_forever drives the loop continuously and an admin rebuild() resets it, with each handler currently taking its own pooled connection.",
"areas": [
{
"area": "delivery",
"status": "fragile",
"note": "The checkpoint is saved on its own connection after the write commits, and a handler failure is logged and skipped instead of stopping the batch."
},
{
"area": "idempotency",
"status": "fragile",
"note": "order_summaries has no per-row guard, so the insert and the two increments are not safe to replay."
},
{
"area": "ordering",
"status": "sound",
"note": "run_once reads one batch and applies events one at a time in a plain for loop, so per-stream order is preserved."
},
{
"area": "rebuild",
"status": "fragile",
"note": "rebuild() resets the checkpoint but leaves order_summaries's existing totals in place."
}
],
"findings": [
...,
{
"id": "PJ-06",
"flags": [
"F6"
],
"file": "orders/projections.py",
"line": 154,
"severity": "medium",
"area": "rebuild",
"title": "Rebuild resets the checkpoint but never clears order_summaries",
"evidence": " async with self.pool.acquire() as conn:\n await conn.execute(\n \"DELETE FROM projection_checkpoints WHERE projection = $1\",",
"why": "rebuild() only deletes the checkpoint row, so run_once replays every event from position 0 on top of the totals already sitting in order_summaries - which is why last week's rebuild made the drift worse instead of fixing it.",
"edits": [
{
"before": " async def rebuild(self) -> None:\n async with self.pool.acquire() as conn:\n await conn.execute(\n \"DELETE FROM projection_checkpoints WHERE projection = $1\",\n self.projection.name,\n )\n while await self.run_once():\n pass",
"after": " async def rebuild(self) -> None:\n async with self.pool.acquire() as conn:\n async with conn.transaction():\n await conn.execute(\"DELETE FROM order_summaries\")\n await conn.execute(\n \"DELETE FROM projection_checkpoints WHERE projection = $1\",\n self.projection.name,\n )\n while await self.run_once():\n pass"
}
]
}
],
"flag_answers": [
{
"flag": "F1",
"status": "confirmed",
"reason": "The insert has no ON CONFLICT; fixed with a last_position column and ON CONFLICT (order_id) DO NOTHING."
},
{
"flag": "F2",
"status": "confirmed",
"reason": "item_count/total_cents are incremented with no guard; fixed with a per-row last_position guard on the UPDATE."
},
{
"flag": "F3",
"status": "confirmed",
"reason": "Same increment problem as F2 on the removal path; fixed the same way."
},
{
"flag": "F4",
"status": "confirmed",
"reason": "The except logs and continues while the checkpoint still advances past the failed event; fixed by letting it raise so the batch stops and retries."
},
{
"flag": "F5",
"status": "confirmed",
"reason": "The checkpoint commits on a separate connection/transaction from the write it follows; fixed by putting both on one connection in one transaction."
},
{
"flag": "F6",
"status": "confirmed",
"reason": "rebuild() deletes only the checkpoint row; fixed by clearing order_summaries in the same transaction."
}
],
"plan": [
...,
"Add a pytest suite run against a real Postgres: the same batch of events applied twice leaves order_summaries unchanged; the projector stopped after a write and before its checkpoint save replays nothing extra on restart; rebuild() ends with the same totals as the live projection."
],
"questions": [
"How many instances of this projector run at once? _load_checkpoint reads the row on its own connection with no FOR UPDATE or advisory lock, which only matters if more than one instance runs against the same checkpoint.",
...
],
"summary": "The upward drift after every deploy is F5 and F2/F3 together: the checkpoint commits on a separate connection after the write, so a deploy that stops the projector between an applied event and its checkpoint replays that event, and the unguarded increments in _on_item_added/_on_item_removed count it twice; F4's swallowed exception can advance the checkpoint past a partly-applied event the same way. The rebuild made it worse because it resets the checkpoint (F6) without clearing order_summaries, replaying the whole history on top of the existing totals instead of from zero. Fix the transaction/failure handling and the replay guards first, then rebuild's clear step, in that order. Since nothing outside this file calls anything but Projector.run_forever() and Projector.rebuild(), the fixes thread a connection and the event's position through apply and the private handlers without changing any signature actually used by a caller."
}
The reply has 4 findings carrying 15 edits and confirms all 6 flags. In order: a
last_position BIGINT NOT NULL DEFAULT 0 column on order_summaries, added as a
new statement at the end of the schema (PJ-02, flags: []: the guard needs it first);
PJ-01 with F4 and F5, which passes one connection through apply, every handler and
_save_checkpoint, commits each event’s writes and its checkpoint in one
transaction, lets a handler failure raise and moves the retry into run_forever (8 edits,
which the note allows: only run_forever() and rebuild() are called from
outside); PJ-02 with F1, F2 and F3, which passes the event’s position to each handler and guards
the writes with ON CONFLICT (order_id) DO NOTHING and last_position < $n
(5 edits); and PJ-06 (F6), shown above. The ordering area is sound: the loop
applies one event at a time.
Fix: a loyalty projector that saves its checkpoint before it applies each event, outside any transaction, and adds points with no guard - with rebuild() left as it is on purpose
Two files and no events table: db/loyalty.sql (loyalty_balances and
projection_checkpoints) and loyalty/projector.py.
LoyaltyProjector.run_once takes one connection, reads the checkpoint and a batch, and for
each event first runs UPDATE projection_checkpoints SET last_position = $2 and only then
_apply, with no transaction around either; _apply inserts
MemberJoined with ON CONFLICT (member_id) DO NOTHING and adds or subtracts
points with points = points + $2; rebuild sets the checkpoint to 0 and
replays without clearing loyalty_balances. The page’s free check raises 4 flags for
the fix lane (1 high, 3 medium) plus EV-00; the verdict its flags point at is blocked. The
engineer’s note: “Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine.”
The code string, whole: 2,812 characters, 81 lines ending in a newline.
Save it as loyalty-paste.txt and run
node build-check.js fix loyalty-paste.txt > loyalty-check.json:
-- file: db/loyalty.sql
CREATE TABLE loyalty_balances (
member_id UUID PRIMARY KEY,
points INTEGER NOT NULL DEFAULT 0,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE projection_checkpoints (
projection TEXT PRIMARY KEY,
last_position BIGINT NOT NULL DEFAULT 0
);
---
# file: loyalty/projector.py
import asyncio
import logging
import asyncpg
log = logging.getLogger(__name__)
class LoyaltyProjector:
name = "loyalty_balances"
def __init__(self, store, pool: asyncpg.Pool, batch_size: int = 500):
self.store = store
self.pool = pool
self.batch_size = batch_size
async def _position(self, conn: asyncpg.Connection) -> int:
value = await conn.fetchval(
"SELECT last_position FROM projection_checkpoints WHERE projection = $1",
self.name,
)
return value or 0
async def _apply(self, conn: asyncpg.Connection, event) -> None:
if event.event_type == "MemberJoined":
await conn.execute(
"INSERT INTO loyalty_balances (member_id, points) VALUES ($1, 0) ON CONFLICT (member_id) DO NOTHING",
event.data["member_id"],
)
elif event.event_type == "PointsEarned":
await conn.execute(
"UPDATE loyalty_balances SET points = points + $2, updated_at = now() WHERE member_id = $1",
event.data["member_id"],
event.data["points"],
)
elif event.event_type == "PointsRedeemed":
await conn.execute(
"UPDATE loyalty_balances SET points = points - $2, updated_at = now() WHERE member_id = $1",
event.data["member_id"],
event.data["points"],
)
async def run_once(self) -> int:
async with self.pool.acquire() as conn:
position = await self._position(conn)
events = await self.store.read_all(position, self.batch_size)
for event in events:
await conn.execute(
"UPDATE projection_checkpoints SET last_position = $2 WHERE projection = $1",
self.name,
event.global_position,
)
await self._apply(conn, event)
return len(events)
async def run_forever(self) -> None:
while True:
if await self.run_once() == 0:
await asyncio.sleep(2.0)
async def rebuild(self) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"UPDATE projection_checkpoints SET last_position = 0 WHERE projection = $1",
self.name,
)
while await self.run_once():
log.info("loyalty_balances rebuild: batch applied")
Its fix-lane facts, decoded, with functions, tables and
statements left out:
{
"lane": "fix",
"summary": {
"python_files": 1,
"sql_files": 1,
"functions": 6,
"statements": 6,
"schema_pasted": false,
"projector_loops": 1
},
"files": [
{
"path": "db/loyalty.sql",
"kind": "sql",
"line": 2,
"end_line": 11
},
{
"path": "loyalty/projector.py",
"kind": "python",
"line": 14,
"end_line": 82
}
],
"functions": ...,
"tables": ...,
"statements": ...,
"projectors": [
{
"file": "loyalty/projector.py",
"line": 60,
"func": "LoyaltyProjector.run_once",
"handler_calls": [
"self._apply @66"
],
"checkpoint_saves": [
61
],
"checkpoint": "before the apply",
"atomic": false
}
],
"checkpoint_saves": [
{
"file": "loyalty/projector.py",
"line": 61,
"call": "UPDATE projection_checkpoints"
}
],
"flags": [
{
"uid": "F1",
"id": "PJ-03",
"sev": "high",
"file": "loyalty/projector.py",
"line": 61,
"text": "LoyaltyProjector.run_once saves the checkpoint before it applies the event",
"detail": "A crash, a deploy or an exception between the save and the apply skips that event for good: on restart the projector resumes after it and the read model silently misses it. Save the checkpoint after the event is applied - in the same transaction as the read-model write."
},
{
"uid": "F2",
"id": "PJ-02",
"sev": "medium",
"file": "loyalty/projector.py",
"line": 45,
"text": "LoyaltyProjector._apply adds to points - not safe to replay",
"detail": "Applied twice (a redelivery after a crash, a second projector instance, a rebuild that did not clear the table) the increment counts the event twice. Guard the write with the event's position (store last_position per row and only apply newer events), or record processed event ids, or recompute the value from state.",
"lines": [
45,
51
]
},
{
"uid": "F3",
"id": "PJ-01",
"sev": "medium",
"file": "loyalty/projector.py",
"line": 61,
"text": "LoyaltyProjector.run_once saves the checkpoint outside the transaction that writes the read model",
"detail": "The checkpoint and the handler's writes do not share one transaction; a crash between the two either replays events that were already applied (double counts unless every write is idempotent) or skips events whose writes never committed. Write the read model and the checkpoint on one connection inside one transaction."
},
{
"uid": "F4",
"id": "PJ-06",
"sev": "medium",
"file": "loyalty/projector.py",
"line": 76,
"text": "LoyaltyProjector.rebuild resets the checkpoint but never clears the read model",
"detail": "The rebuild replays every event on top of the rows that are already there: increments double and inserts collide. Clear (or rebuild into a fresh table and swap) in the same step as the checkpoint reset."
},
{
"uid": "F5",
"id": "EV-00",
"sev": "info",
"file": "db/loyalty.sql",
"line": 2,
"text": "context: 1 Python file, 1 SQL file, 6 SQL statement(s) in 6 function(s); no events table found; 1 read-model table(s), 1 checkpoint table(s), 1 projector loop(s)",
"detail": "Only what is pasted is checked: the schema checks need the CREATE TABLE (and its indexes); transactions are read from async with ....transaction()/begin() blocks and explicit start()/commit(); what a function in another file does, the isolation level and how many projector instances run are outside the check."
}
],
"verdict_candidate": "blocked",
"notes": []
}
The request body (Idempotency-Key: append-only-audit:fix:5002a346-24b6:a1 as the page sends
it; no audit_handoff, since this run fixes the paste directly):
{
"task": "fix",
"code": "-- file: db/loyalty.sql\nCREATE TABLE loyalty_balances (\n member_id UUID PRIMARY KEY,\n points INTEGER NOT NULL DEFAULT 0,\n...",
"facts": "{\"lane\":\"fix\",\"summary\":{\"python_files\":1,\"sql_files\":1,\"functions\":6,\"statements\":6,\"schema_pasted\":false,\"projector_loops\":1},\"files\":...",
"note": "Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine."
}
The reply, an excerpt: everything except the content of loyalty/projector.py, which
follows it decoded:
{
"task": "fix",
"verdict": "partial",
"headline": "The checkpoint now commits atomically after each event (via upsert) and PointsEarned/PointsRedeemed are guarded by a new last_position column, so a crash or replay can no longer skip or double-count points; rebuild() is left exactly as it was, per the note.",
"files": [
{
"path": "db/loyalty.sql",
"content": "CREATE TABLE loyalty_balances (\n member_id UUID PRIMARY KEY,\n points INTEGER NOT NULL DEFAULT 0,\n updated_at TIMESTAMPTZ NOT NULL DEFAULT now()\n);\n\nCREATE TABLE projection_checkpoints (\n projection TEXT PRIMARY KEY,\n last_position BIGINT NOT NULL DEFAULT 0\n);\n\nALTER TABLE loyalty_balances ADD COLUMN last_position BIGINT NOT NULL DEFAULT 0;\n"
},
{
"path": "loyalty/projector.py",
"content": ...
}
],
"changes": [
{
"file": "db/loyalty.sql",
"function": "(schema)",
"flags": [
"F2"
],
"change": "Added a last_position BIGINT NOT NULL DEFAULT 0 column to loyalty_balances as a new statement at the end of the file.",
"why": "gives each row a place to record the position of the last event applied to it, so a handler can refuse to re-apply an older or repeated event."
},
{
"file": "loyalty/projector.py",
"function": "LoyaltyProjector._apply",
"flags": [
"F2"
],
"change": "PointsEarned and PointsRedeemed now UPDATE points and last_position together WHERE member_id = $1 AND last_position < $3, and MemberJoined also records last_position on insert.",
"why": "a redelivered or replayed event can no longer add or subtract points a second time."
},
{
"file": "loyalty/projector.py",
"function": "LoyaltyProjector.run_once",
"flags": [
"F1",
"F3"
],
"change": "Moved the checkpoint save to after self._apply and wrapped both in one async with conn.transaction(); the checkpoint write is now an INSERT ... ON CONFLICT (projection) DO UPDATE instead of a bare UPDATE.",
"why": "the read-model write and the checkpoint now commit or roll back together, so a crash between them can no longer skip an event, and the checkpoint is saved even before its row has ever been seeded."
}
],
"verify": [
"Run the existing pytest suite to confirm nothing else broke.",
"pytest against a real Postgres: apply the same batch of events through run_once twice and assert loyalty_balances is unchanged the second time (catches PJ-02 regressions in _apply).",
"pytest against a real Postgres: kill the connection inside the async with conn.transaction() block after _apply has run but before the checkpoint UPDATE would commit, restart, and assert the event is applied exactly once (catches PJ-01/PJ-03 regressions).",
"pytest against a real Postgres: run rebuild() after a manual TRUNCATE of loyalty_balances and assert the resulting table matches a fresh run_once from position 0 (checks the manual-truncate runbook the note describes still produces a correct rebuild).",
"Store-side, when that code is available: two concurrent appends with the same expected version against a real Postgres, asserting exactly one raises a concurrency error - the append/store code is not in this paste, but the projector's exactly-once guarantee depends on it."
],
"open": [
{
"item": "rebuild() still resets the checkpoint with a bare UPDATE and never clears loyalty_balances itself.",
"flags": [
"F4"
],
"reason": "left exactly as the note asked - the DBA truncates loyalty_balances by hand before every rebuild today. When that runbook changes next quarter, rebuild() will need to clear the table (e.g. TRUNCATE) and upsert its own checkpoint reset, in one transaction."
},
{
"item": "After this migration, every existing row's last_position defaults to 0, so a PointsEarned/PointsRedeemed event already applied before the deploy could be re-applied once if redelivered before a newer event for that member is seen.",
"flags": [],
"reason": "a one-time cutover risk that code in this paste cannot close by itself - decide whether to backfill last_position (e.g. from the projector's checkpoint position at deploy time) before rollout."
}
],
"questions": [
"store.read_all(position, batch_size) is outside this paste - does it return events strictly in ascending global_position order with no gaps from transactions still in flight? The atomic checkpoint fix here assumes so; without that guarantee a projector instance can still advance its checkpoint past a position whose event has not committed yet."
],
"summary": "F1 (checkpoint saved before the apply) and F3 (checkpoint not atomic with the write) are fixed together in run_once: the checkpoint now upserts after _apply, both inside one transaction. F2 (non-idempotent increments) is fixed by a new last_position guard column, populated on every write. Per the note, rebuild() is left byte-for-byte unchanged and F4 is carried in open, since the DBA's manual truncate covers it today but not once that runbook changes. Adding the last_position column was explicitly allowed by the note; no public signature changed."
}
loyalty/projector.py as the reply returns it, whole:
import asyncio
import logging
import asyncpg
log = logging.getLogger(__name__)
class LoyaltyProjector:
name = "loyalty_balances"
def __init__(self, store, pool: asyncpg.Pool, batch_size: int = 500):
self.store = store
self.pool = pool
self.batch_size = batch_size
async def _position(self, conn: asyncpg.Connection) -> int:
value = await conn.fetchval(
"SELECT last_position FROM projection_checkpoints WHERE projection = $1",
self.name,
)
return value or 0
async def _apply(self, conn: asyncpg.Connection, event) -> None:
if event.event_type == "MemberJoined":
await conn.execute(
"INSERT INTO loyalty_balances (member_id, points, last_position) "
"VALUES ($1, 0, $2) ON CONFLICT (member_id) DO NOTHING",
event.data["member_id"],
event.global_position,
)
elif event.event_type == "PointsEarned":
await conn.execute(
"UPDATE loyalty_balances SET points = points + $2, updated_at = now(), last_position = $3 "
"WHERE member_id = $1 AND last_position < $3",
event.data["member_id"],
event.data["points"],
event.global_position,
)
elif event.event_type == "PointsRedeemed":
await conn.execute(
"UPDATE loyalty_balances SET points = points - $2, updated_at = now(), last_position = $3 "
"WHERE member_id = $1 AND last_position < $3",
event.data["member_id"],
event.data["points"],
event.global_position,
)
async def run_once(self) -> int:
async with self.pool.acquire() as conn:
position = await self._position(conn)
events = await self.store.read_all(position, self.batch_size)
for event in events:
async with conn.transaction():
await self._apply(conn, event)
await conn.execute(
"INSERT INTO projection_checkpoints (projection, last_position) "
"VALUES ($1, $2) ON CONFLICT (projection) DO UPDATE "
"SET last_position = EXCLUDED.last_position",
self.name,
event.global_position,
)
return len(events)
async def run_forever(self) -> None:
while True:
if await self.run_once() == 0:
await asyncio.sleep(2.0)
async def rebuild(self) -> None:
async with self.pool.acquire() as conn:
await conn.execute(
"UPDATE projection_checkpoints SET last_position = 0 WHERE projection = $1",
self.name,
)
while await self.run_once():
log.info("loyalty_balances rebuild: batch applied")
The verdict is partial because the note asked to leave rebuild() exactly as
it is, so the medium F4 is named in open rather than fixed; rebuild comes
back byte for byte, every function keeps its name and its async def, and the new column
is one the note allowed. With every write now guarded, the page’s re-check still raises the
rebuild flag, at low.
The page re-checks the reply against the code; the API does not
What comes back from the API is the model’s raw text in output.output. The page
holds it to account in the browser before showing it (recon.js
reconcile, with the facts, the code and the handoff it sent): it applies every edit or
returned file to the paste and re-runs its own free check on the result. None of that happens in the
API. A direct API caller gets the reply unchecked and must do the applying and re-checking itself.
- All lanes: the reply is the lane you asked for (one written for another task is a disagreement), it carries a valid verdict for that lane, and no credential from the paste is repeated anywhere in it (every credential value the free check found in the paste).
- Audits: the areas table. One row per area of the lane (schema, appends, reads; or delivery, idempotency, ordering, rebuild), each with a valid status.
- Audits: every finding is anchored. Its
idis in the lane’s catalogue (ES-01..ES-10 or PJ-01..PJ-10, EV-01, orother; EV-00 is context and never a finding); itsareabelongs to the lane; itsfileis a pasted file; itslineis inside that file; itsevidenceis found line by line in that file ofcode(indentation trimmed) and on or nearline; every citeduidis a non-info flag the page raised for this lane. A finding that carries a high flag at a lower severity, or a medium or high finding with no edit, is pointed out. - Audits: every flag is answered once, consistently. Every non-info flag of the lane answered exactly once and no other; a
confirmedanswer needs a finding carrying that uid; adismissedone must not be carried by a finding. The verdict must fit the worst severity (blockedwith a high finding,needs_workwith a medium, otherwiseready). - Audits: every edit is put in place and re-checked. The findings’ edits are applied in the listed order, each on top of the ones before (
Eskit.applyEdits): abeforethat is not found or is found more than once is a disagreement. After each finding the free check runs again: the code must still read and no new high flag may appear. With every edit applied, every confirmed flag a finding with edits carries must be gone, every function of the paste must still exist, and a new medium flag is pointed out. - Fix: the files. At least one file, at most one new one; none empty, none with a
# file:or-- file:line, every path a.pyor.sqlfile; a file indented with tabs is pointed out, and a Python file sent back with every line indented is a disagreement. - Fix: put into the paste and re-checked. The files replace the pasted files of the same path (or are added) (
Eskit.replaceFiles) and the free check runs again: the code must read, no high or medium flag of the paste may remain unless anopenitem names it, no new high or medium flag may appear, every function must still exist, and none may change betweendefandasync def. - Fix: the changes list matches the code. Every change names a returned file, a function in the fixed code (or
(module),(schema)or(top level)) and real page flags; a high or medium flag no change and no open item names is pointed out. The verdict must fit theopenlist, and an emptyverify, or one with no test in it, is pointed out.
recon.js also runs in Node, so you can hold a reply to exactly the page’s standard.
Save /eskit.js and /recon.js next to this script,
pass the file you built the body from (build-check.js or handoff.js output,
which carries the lane as task) and the reply text; it prints every statement, the count,
and the paste with every edit (or file) put back:
// recheck.js -- run: node recheck.js ledger-check.json ledger-reply.txt > ledger-edited.txt (eskit.js, recon.js next to it)
const fs = require("fs");
const Recon = require("./recon.js"); // loads ./eskit.js itself
const sent = JSON.parse(fs.readFileSync(process.argv[2], "utf8")); // task, code, facts (and audit_handoff) you sent
const res = Recon.normalize(Recon.parseResult(fs.readFileSync(process.argv[3], "utf8")), sent.task); // output.output
const r = Recon.reconcile(res, { facts: sent.facts, code: sent.code, handoff: sent.audit_handoff || null });
r.notes.forEach((n) => console.error(n.kind + ": " + n.text));
console.error(r.disagreements + " disagreement(s) in " + r.statements + " statements checked");
if (r.after) (r.after.flags || []).forEach((f) => console.error("still raised: " + f.id + " " + f.sev + " " + f.file + ":" + f.line + " " + f.text));
if (r.after && r.after.text) console.log(r.after.text.replace(/\n$/, "")); // the paste with every edit (or file) put back
For the ledger store audit reply it reports:
ok: The areas table rates schema, appends, reads once each.
warn: Finding 1 (ES-03) is high and gives no edit; the page cannot check a fix for it.
ok: Every one of the 7 findings names a pasted file, a line inside it and evidence copied from it.
ok: The verdict (blocked) fits the worst finding.
ok: Every one of the page's 7 store flags is answered once.
ok: All 7 edits applied in order, each before text found exactly once.
ok: Every confirmed flag a finding carries is gone once all edits are applied.
ok: Every function of the paste is still there after the edits.
ok: The reply repeats no credential from the paste.
0 disagreement(s) in 9 statements checked
still raised: ES-03 high wallet/store.py:116 UPDATE on the events table wallet_events in WalletEventStore.redact_iban
for the orders projection audit reply (node recheck.js orders-check.json orders-reply.txt):
ok: The areas table rates delivery, idempotency, ordering, rebuild once each.
ok: Every one of the 4 findings names a pasted file, a line inside it and evidence copied from it.
ok: The verdict (needs_work) fits the worst finding.
ok: Every one of the page's 6 projection flags is answered once.
ok: All 15 edits applied in order, each before text found exactly once.
ok: Every confirmed flag a finding carries is gone once all edits are applied.
ok: Every function of the paste is still there after the edits.
ok: The reply repeats no credential from the paste.
0 disagreement(s) in 8 statements checked
and for the loyalty fix reply (node recheck.js loyalty-check.json loyalty-reply.txt):
ok: 2 files returned whole.
ok: With the fixed files in place the free check raises no high or medium flag beyond the ones left open (1 low).
info: F4 left open by the reply.
ok: Every one of the 6 functions is still there.
ok: Every one of the 3 changes names a returned file, a function in the fixed code and real page flags.
ok: The verdict (partial) fits the open list.
ok: The reply repeats no credential from the paste.
0 disagreement(s) in 6 statements checked
still raised: PJ-06 low loyalty/projector.py:87 LoyaltyProjector.rebuild resets the checkpoint but never clears the read model
The page does not run the code. After that, run the test suite (pytest) and the tests the
reply’s plan or verify names against a real Postgres (a CI service or
a disposable container): two concurrent appends with the same expected version, one of which must
fail; a batch applied twice that must leave the read model unchanged; a projector stopped between the
write and the checkpoint; and a rebuild that ends with the same read model as the live projection,
before you ship anything.
Step 1 — a tiny client helper
One function per language that does what sdk.js does for every call: JSON in,
bearer token if you have one (a personal token from /tokens.html, step 2), data out on a 2xx, an error carrying
status, error.code and error.message otherwise. The later
steps assume this helper is in the same file (for Go, Java and C#, the later snippets go inside
main / the top-level statements; Java’s makeInput from step 4 goes
in the class).
BASE="https://api.skillsafe.ai/v1/app-api"
TOKEN="${SKILLSAFE_TOKEN:-YOUR_TOKEN}" # step 2
# api METHOD PATH [extra curl args...] -- prints the whole envelope; pipe to jq
api() {
method="$1"; path="$2"; shift 2
curl -sS -X "$method" "$BASE$path" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" "$@"
}
# success: jq '.data' failure: jq '.error | {code, message}'
import hashlib, json, os, time, urllib.parse
import requests # pip install requests
BASE = "https://api.skillsafe.ai/v1/app-api"
TOKEN = os.environ.get("SKILLSAFE_TOKEN", "YOUR_TOKEN") # step 2
class ApiError(Exception):
def __init__(self, status, code, message, details=None):
super().__init__(f"{status} {code}: {message}")
self.status, self.code, self.details = status, code, details
def api(method, path, body=None, headers=None):
h = {"Content-Type": "application/json"}
if TOKEN:
h["Authorization"] = "Bearer " + TOKEN
h.update(headers or {})
r = requests.request(method, BASE + path, headers=h,
data=None if body is None else json.dumps(body))
try:
env = r.json()
except ValueError:
env = {}
if not r.ok:
err = env.get("error") or {}
raise ApiError(r.status_code, err.get("code"), err.get("message") or r.reason, err.get("details"))
return env.get("data")
// append-only-audit.mjs -- Node 18+ (global fetch), run with: node append-only-audit.mjs
import { readFileSync } from "node:fs";
import { createHash } from "node:crypto";
const BASE = "https://api.skillsafe.ai/v1/app-api";
let TOKEN = "YOUR_TOKEN"; // step 2
async function api(method, path, body, headers = {}) {
const h = { "Content-Type": "application/json", ...headers };
if (TOKEN) h.Authorization = "Bearer " + TOKEN;
const res = await fetch(BASE + path, {
method,
headers: h,
body: body === undefined ? undefined : JSON.stringify(body),
});
const json = await res.json().catch(() => ({}));
if (!res.ok) {
const err = new Error((json.error && json.error.message) || res.statusText);
err.status = res.status;
err.code = json.error && json.error.code;
err.details = json.error && json.error.details;
throw err;
}
return json.data;
}
package main
import (
"bufio"
"bytes"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"net/url"
"os"
"strings"
"time"
)
const base = "https://api.skillsafe.ai/v1/app-api"
var token = os.Getenv("SKILLSAFE_TOKEN") // step 2
type apiError struct {
Status int
Code string
Message string
}
func (e *apiError) Error() string { return fmt.Sprintf("%d %s: %s", e.Status, e.Code, e.Message) }
type envelope struct {
OK bool `json:"ok"`
Data json.RawMessage `json:"data"`
Error *struct {
Code string `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
func newRequest(method, path string, body any, headers map[string]string) (*http.Request, error) {
var buf []byte
if body != nil {
b, err := json.Marshal(body)
if err != nil {
return nil, err
}
buf = b
}
req, err := http.NewRequest(method, base+path, bytes.NewReader(buf))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
if token != "" {
req.Header.Set("Authorization", "Bearer "+token)
}
for k, v := range headers {
req.Header.Set(k, v)
}
return req, nil
}
func errorFrom(res *http.Response, env envelope) error {
e := &apiError{Status: res.StatusCode, Message: http.StatusText(res.StatusCode)}
if env.Error != nil {
e.Code = env.Error.Code
if env.Error.Message != "" {
e.Message = env.Error.Message
}
}
return e
}
// api sends body as JSON and decodes the envelope's data into out (out may be nil).
func api(method, path string, body any, headers map[string]string, out any) error {
req, err := newRequest(method, path, body, headers)
if err != nil {
return err
}
res, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer res.Body.Close()
var env envelope
_ = json.NewDecoder(res.Body).Decode(&env)
if res.StatusCode < 200 || res.StatusCode > 299 {
return errorFrom(res, env)
}
if out != nil && len(env.Data) > 0 {
return json.Unmarshal(env.Data, out)
}
return nil
}
func main() {
// steps 2-6 go here; remove any import a step you skip no longer uses
_, _, _, _, _ = bufio.NewScanner, sha256.Sum256, hex.EncodeToString, url.PathEscape, strings.Index
_ = time.Second
}
// AppendOnlyAudit.java -- Java 17+, needs com.fasterxml.jackson.core:jackson-databind on the classpath
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.net.URI;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class AppendOnlyAudit {
static final String BASE = "https://api.skillsafe.ai/v1/app-api";
static String token = System.getenv().getOrDefault("SKILLSAFE_TOKEN", "YOUR_TOKEN"); // step 2
static final HttpClient HTTP = HttpClient.newHttpClient();
static final ObjectMapper JSON = new ObjectMapper();
static class ApiException extends Exception {
final int status;
final String code;
ApiException(int status, String code, String message) {
super(status + " " + code + ": " + message);
this.status = status;
this.code = code;
}
}
static HttpRequest.Builder request(String method, String path, Object body, Map<String, String> headers) throws Exception {
HttpRequest.Builder b = HttpRequest.newBuilder(URI.create(BASE + path))
.header("Content-Type", "application/json")
.method(method, body == null ? HttpRequest.BodyPublishers.noBody()
: HttpRequest.BodyPublishers.ofString(JSON.writeValueAsString(body)));
if (token != null && !token.isEmpty()) b.header("Authorization", "Bearer " + token);
if (headers != null) headers.forEach(b::header);
return b;
}
static ApiException errorFrom(int status, JsonNode env) {
JsonNode err = env.path("error");
return new ApiException(status, err.path("code").asText(null), err.path("message").asText("HTTP " + status));
}
static JsonNode api(String method, String path, Object body, Map<String, String> headers) throws Exception {
HttpResponse<String> res = HTTP.send(request(method, path, body, headers).build(), HttpResponse.BodyHandlers.ofString());
JsonNode env;
try { env = JSON.readTree(res.body()); } catch (Exception e) { env = JSON.createObjectNode(); }
if (env == null) env = JSON.createObjectNode();
if (res.statusCode() / 100 != 2) throw errorFrom(res.statusCode(), env);
return env.path("data");
}
public static void main(String[] args) throws Exception {
// steps 2-6 go here (makeInput from step 4 goes next to api())
}
}
require "digest"
require "json"
require "net/http"
require "uri"
BASE = "https://api.skillsafe.ai/v1/app-api"
$token = ENV.fetch("SKILLSAFE_TOKEN", "YOUR_TOKEN") # step 2
class ApiError < StandardError
attr_reader :status, :code
def initialize(status, code, message)
super("#{status} #{code}: #{message}")
@status = status
@code = code
end
end
def build_request(method, path, body = nil, headers = {})
uri = URI(BASE + path)
req = Net::HTTPGenericRequest.new(method, !body.nil?, true, uri.request_uri)
req["Content-Type"] = "application/json"
req["Authorization"] = "Bearer #{$token}" if $token && !$token.empty?
headers.each { |k, v| req[k] = v }
req.body = JSON.generate(body) unless body.nil?
[uri, req]
end
def api(method, path, body = nil, headers = {})
uri, req = build_request(method, path, body, headers)
res = Net::HTTP.start(uri.host, uri.port, use_ssl: true) { |h| h.request(req) }
env = (JSON.parse(res.body) rescue {})
env = {} unless env.is_a?(Hash)
unless res.is_a?(Net::HTTPSuccess)
err = env["error"] || {}
raise ApiError.new(res.code.to_i, err["code"], err["message"] || res.message)
end
env["data"]
end
<?php
// needs the curl extension
const BASE = "https://api.skillsafe.ai/v1/app-api";
$TOKEN = getenv("SKILLSAFE_TOKEN") ?: "YOUR_TOKEN"; // step 2
class ApiError extends Exception {
public $status;
public $apiCode;
public function __construct($status, $code, $message) {
parent::__construct("$status $code: $message");
$this->status = $status;
$this->apiCode = $code;
}
}
function headers_for(array $extra = []): array {
global $TOKEN;
$h = ["Content-Type: application/json"];
if ($TOKEN) $h[] = "Authorization: Bearer $TOKEN";
foreach ($extra as $k => $v) $h[] = "$k: $v";
return $h;
}
function api(string $method, string $path, $body = null, array $headers = []) {
$ch = curl_init(BASE . $path);
curl_setopt_array($ch, [
CURLOPT_CUSTOMREQUEST => $method,
CURLOPT_HTTPHEADER => headers_for($headers),
CURLOPT_RETURNTRANSFER => true,
]);
if ($body !== null) curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($body, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE));
$raw = curl_exec($ch);
if ($raw === false) throw new RuntimeException(curl_error($ch));
$status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
curl_close($ch);
$env = json_decode($raw);
if (!is_object($env)) $env = new stdClass();
if ($status < 200 || $status > 299) {
$e = $env->error ?? new stdClass();
throw new ApiError($status, $e->code ?? null, $e->message ?? "HTTP $status");
}
return $env->data ?? null;
}
// Program.cs -- .NET 6+ console app (implicit usings on) with top-level statements; steps 2-6 (with the MakeInput local function) go ABOVE this class
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
static class Api
{
public const string Base = "https://api.skillsafe.ai/v1/app-api";
public static string Token = Environment.GetEnvironmentVariable("SKILLSAFE_TOKEN") ?? "YOUR_TOKEN"; // step 2
public static readonly HttpClient Http = new HttpClient();
public class ApiException : Exception
{
public int Status;
public string? Code;
public ApiException(int status, string? code, string message) : base($"{status} {code}: {message}")
{
Status = status;
Code = code;
}
}
public static HttpRequestMessage Request(string method, string path, object? body = null, IDictionary<string, string>? headers = null)
{
var req = new HttpRequestMessage(new HttpMethod(method), Base + path);
if (body != null) req.Content = new StringContent(JsonSerializer.Serialize(body), Encoding.UTF8, "application/json");
if (!string.IsNullOrEmpty(Token)) req.Headers.TryAddWithoutValidation("Authorization", "Bearer " + Token);
if (headers != null) foreach (var kv in headers) req.Headers.TryAddWithoutValidation(kv.Key, kv.Value);
return req;
}
public static ApiException ErrorFrom(HttpResponseMessage res, JsonNode? env)
{
var err = env?["error"];
return new ApiException((int)res.StatusCode, err?["code"]?.GetValue<string>(),
err?["message"]?.GetValue<string>() ?? res.ReasonPhrase ?? "error");
}
public static async Task<JsonNode?> Call(string method, string path, object? body = null, IDictionary<string, string>? headers = null)
{
using var req = Request(method, path, body, headers);
using var res = await Http.SendAsync(req);
var text = await res.Content.ReadAsStringAsync();
JsonNode? env = null;
try { env = JsonNode.Parse(text); } catch (JsonException) { }
if (!res.IsSuccessStatusCode) throw ErrorFrom(res, env);
return env?["data"];
}
}
Step 2 — a token
Two kinds work:
-
Personal — needed for a run. All three lanes are metered in credits on
gpt-terra, so a run needs a signed-in (user) token (the page says “The store audit run is metered, so it needs an account.”, or the same of the projection audit and the fix run) unless the estimate sayssponsor_enabled. Sign in and copy your token from /tokens.html, then export it asSKILLSAFE_TOKEN(the samples fall back to the placeholderYOUR_TOKEN). -
Guest — enough for
/meand/estimate.POST /guestwith{"slug": "append-only-audit"}and noAuthorizationheader, exactly assdk.jsguest()does. The response’sdatacarriestokenandguest_id. Reuse a token you already hold rather than minting a new one each time.
# personal token (needed for a run): copy it from /tokens.html, then
# export SKILLSAFE_TOKEN=... and skip the rest of this block
# guest token (enough for /me and /estimate): no Authorization header on this one call
TOKEN=$(curl -sS -X POST "$BASE/guest" \
-H "Content-Type: application/json" \
-d '{"slug":"append-only-audit"}' | jq -r '.data.token')
echo "guest token: ${TOKEN:0:12}..."
# personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if TOKEN in ("", "YOUR_TOKEN"):
TOKEN = "" # no Authorization header on /guest
guest = api("POST", "/guest", {"slug": "append-only-audit"})
TOKEN = guest["token"]
print("guest", guest.get("guest_id"))
// personal token (needed for a run): paste it from /tokens.html into TOKEN and skip this
if (!TOKEN || TOKEN === "YOUR_TOKEN") {
TOKEN = ""; // no Authorization header on /guest
const guest = await api("POST", "/guest", { slug: "append-only-audit" });
TOKEN = guest.token;
console.log("guest", guest.guest_id);
}
// inside main -- personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if token == "" {
var guest struct {
Token string `json:"token"`
GuestID string `json:"guest_id"`
}
if err := api("POST", "/guest", map[string]string{"slug": "append-only-audit"}, nil, &guest); err != nil {
panic(err)
}
token = guest.Token
fmt.Println("guest", guest.GuestID)
}
// inside main -- personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if (token.isEmpty() || token.equals("YOUR_TOKEN")) {
token = ""; // no Authorization header on /guest
JsonNode guest = api("POST", "/guest", Map.of("slug", "append-only-audit"), null);
token = guest.path("token").asText();
System.out.println("guest " + guest.path("guest_id").asText());
}
# personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if $token.empty? || $token == "YOUR_TOKEN"
$token = "" # no Authorization header on /guest
guest = api("POST", "/guest", { slug: "append-only-audit" })
$token = guest["token"]
puts "guest #{guest["guest_id"]}"
end
// personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if ($TOKEN === "" || $TOKEN === "YOUR_TOKEN") {
$TOKEN = ""; // no Authorization header on /guest
$guest = api("POST", "/guest", ["slug" => "append-only-audit"]);
$TOKEN = $guest->token;
echo "guest {$guest->guest_id}\n";
}
// top-level statements (above the Api class) -- personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if (Api.Token == "" || Api.Token == "YOUR_TOKEN")
{
Api.Token = ""; // no Authorization header on /guest
var guest = await Api.Call("POST", "/guest", new { slug = "append-only-audit" });
Api.Token = guest!["token"]!.GetValue<string>();
Console.WriteLine($"guest {guest["guest_id"]}");
}
Step 3 — GET /me
Who the token belongs to and what it can spend. sdk.js documents the shape as
{subject_type, subject_id, credits, profile?}. The page treats the caller as signed
in only when subject_type is "user", shows credits as the
balance (1 credit = $0.0001, so it divides by 10 000 for the dollar figure), and compares it
with the estimate in step 4 before it lets you run.
api GET /me | jq '.data // .error'
me = api("GET", "/me")
print(me["subject_type"], me.get("subject_id"), me.get("credits"))
signed_in = me["subject_type"] == "user"
const me = await api("GET", "/me");
console.log(me.subject_type, me.subject_id, me.credits);
const signedIn = me.subject_type === "user";
var me struct {
SubjectType string `json:"subject_type"`
SubjectID any `json:"subject_id"`
Credits any `json:"credits"`
}
if err := api("GET", "/me", nil, nil, &me); err != nil {
panic(err)
}
fmt.Println(me.SubjectType, me.SubjectID, me.Credits)
signedIn := me.SubjectType == "user"
_ = signedIn
JsonNode me = api("GET", "/me", null, null);
System.out.println(me.path("subject_type").asText() + " " + me.path("subject_id").asText() + " " + me.path("credits"));
boolean signedIn = me.path("subject_type").asText().equals("user");
me = api("GET", "/me")
puts [me["subject_type"], me["subject_id"], me["credits"]].join(" ")
signed_in = me["subject_type"] == "user"
$me = api("GET", "/me");
echo $me->subject_type, " ", $me->subject_id ?? "", " ", $me->credits ?? "", "\n";
$signedIn = $me->subject_type === "user";
var me = await Api.Call("GET", "/me");
Console.WriteLine($"{me?["subject_type"]} {me?["subject_id"]} {me?["credits"]}");
var signedIn = me?["subject_type"]?.GetValue<string>() == "user";
Step 4 — build the input, then POST /estimate
First a small makeInput that does what buildInput does: task,
then code as the paste the page would send, facts as a compact JSON string
(required), the note when it is not blank, and — for the fix lane only — the
audit_handoff string when the file carries one. It reads the file
build-check.js (or handoff.js) printed; that file’s
facts.lane must be the task you pass, since the facts are cut per lane. The
samples build the store input of the ledger example from
ledger-check.json; step 5 runs it, and step 6 reuses makeInput for the
loyalty (fix) example. Then price it. The body of /estimate is the same input you will
send to /run; no job is created and nothing is charged — /estimate is
free. From the response the page reads:
| Field | Meaning |
|---|---|
hold_credits | A reservation, not a price: the worst case the platform holds while the run is in flight. The page shows it as “reserves N credits ($...) for the store audit run” and adds that you are charged only for what the run actually uses, usually far less. What you pay is the job’s charged_credits after settlement (step 5). A balance below the hold is allowed, but the page warns that the reply may be cut short (truncated: true). |
min_credits | The smallest balance that can start a run. Below it the page disables the run button and names the shortfall, because the call would be a 402. |
sponsor_enabled | When true, a caller who is not signed in may run; otherwise the page asks them to sign in. |
warnings | Input warnings. They do not refuse the call; the page shows them next to the price (“input warnings: ...”), so check them before you run. |
The page re-prices whenever the paste, the note or the lane change. The amounts grow with the size
of code and facts, so read them from the call rather than hard-coding them.
# ledger-check.json: printed by `node build-check.js store ledger-paste.txt` (see "The facts string").
# (.facts | tojson) turns the facts into ONE compact string, as JSON.stringify does; build the file for this lane.
BODY=$(jq -c --arg note "We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request." \
'{task: "store", code, facts: (.facts | tojson), note: $note}' ledger-check.json)
# facts is required. No page check at hand? "{}" is accepted as a string, but the reply then has no page flags to answer.
# Fix lane after an audit: read handoff.js's output instead and add its string:
# '{task: "fix", code, facts: (.facts | tojson), note: $note} + (if .audit_handoff then {audit_handoff} else {} end)'
api POST /estimate -d "$BODY" \
| jq '.error // (.data | {hold_credits, min_credits, sponsor_enabled, warnings})'
def make_input(task, check_path, note=None):
"""The body the page sends: task, code, facts as a JSON STRING, an optional note, and
(fix lane only) the audit_handoff string -- every value a string."""
with open(check_path, encoding="utf-8") as f:
check = json.load(f) # build-check.js (or handoff.js) output
inp = {
"task": task,
"code": check["code"], # the paste as the page sends it
"facts": json.dumps(check["facts"], separators=(",", ":"), ensure_ascii=False), # a STRING, like JSON.stringify
}
n = (note or "").strip()
if n:
inp["note"] = n[:2000] # optional; the page caps it at 2000 chars
if task == "fix" and check.get("audit_handoff"):
inp["audit_handoff"] = check["audit_handoff"] # optional; already a JSON string
return inp
inp = make_input("store", "ledger-check.json",
"We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request.")
est = api("POST", "/estimate", inp)
print("hold", est.get("hold_credits"), "min", est.get("min_credits"),
"sponsored", est.get("sponsor_enabled"), "warnings", est.get("warnings") or [])
// The body the page sends: task, code, facts as a JSON STRING, an optional note,
// and (fix lane only) the audit_handoff string -- every value a string.
function makeInput(task, checkPath, note) {
const check = JSON.parse(readFileSync(checkPath, "utf8")); // build-check.js (or handoff.js) output
const input = {
task,
code: check.code, // the paste as the page sends it
facts: JSON.stringify(check.facts), // a STRING: the page's free check
};
const n = (note || "").trim();
if (n) input.note = n.length > 2000 ? n.slice(0, 2000) : n; // optional; max 2000 chars
if (task === "fix" && check.audit_handoff) input.audit_handoff = check.audit_handoff; // optional
return input;
}
const input = makeInput("store", "ledger-check.json",
"We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request.");
const est = await api("POST", "/estimate", input);
console.log("hold", est.hold_credits, "min", est.min_credits,
"sponsored", est.sponsor_enabled, "warnings", est.warnings || []);
// makeInput builds the body the page sends: task, code, facts as a JSON STRING,
// an optional note and (fix lane only) the audit_handoff string.
makeInput := func(task, checkPath, note string) map[string]any {
raw, err := os.ReadFile(checkPath) // build-check.js (or handoff.js) output
if err != nil {
panic(err)
}
var check struct {
Code string `json:"code"`
Facts json.RawMessage `json:"facts"`
AuditHandoff string `json:"audit_handoff"`
}
if err := json.Unmarshal(raw, &check); err != nil {
panic(err)
}
var facts bytes.Buffer
if err := json.Compact(&facts, check.Facts); err != nil { // keeps key order; travels as a STRING
panic(err)
}
in := map[string]any{"task": task, "code": check.Code, "facts": facts.String()}
if n := []rune(strings.TrimSpace(note)); len(n) > 0 {
if len(n) > 2000 {
n = n[:2000] // optional; the page caps it at 2000 chars
}
in["note"] = string(n)
}
if task == "fix" && check.AuditHandoff != "" {
in["audit_handoff"] = check.AuditHandoff // optional; already a JSON string
}
return in
}
input := makeInput("store", "ledger-check.json",
"We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request.")
var est struct {
HoldCredits any `json:"hold_credits"`
MinCredits any `json:"min_credits"`
SponsorEnabled bool `json:"sponsor_enabled"`
Warnings []any `json:"warnings"`
}
if err := api("POST", "/estimate", input, nil, &est); err != nil {
panic(err)
}
fmt.Printf("hold %v min %v sponsored %v warnings %v\n", est.HoldCredits, est.MinCredits, est.SponsorEnabled, est.Warnings)
// ---- add to the AppendOnlyAudit class, next to api(): the body the page sends
static Map<String, Object> makeInput(String task, String checkPath, String note) throws Exception {
JsonNode check = JSON.readTree(Files.readString(Path.of(checkPath))); // build-check.js (or handoff.js) output
Map<String, Object> in = new LinkedHashMap<>();
in.put("task", task);
in.put("code", check.path("code").asText()); // the paste as the page sends it
in.put("facts", JSON.writeValueAsString(check.path("facts"))); // a STRING, compact
String n = note == null ? "" : note.strip();
if (!n.isEmpty()) in.put("note", n.length() > 2000 ? n.substring(0, 2000) : n); // optional; max 2000 chars
if (task.equals("fix") && check.hasNonNull("audit_handoff")) in.put("audit_handoff", check.get("audit_handoff").asText());
return in;
}
// ---- inside main
Map<String, Object> input = makeInput("store", "ledger-check.json",
"We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request.");
JsonNode est = api("POST", "/estimate", input, null);
System.out.println("hold " + est.path("hold_credits") + " min " + est.path("min_credits")
+ " sponsored " + est.path("sponsor_enabled").asBoolean(false) + " warnings " + est.path("warnings"));
# The body the page sends: task, code, facts as a JSON STRING, an optional note,
# and (fix lane only) the audit_handoff string -- every value a string.
def make_input(task, check_path, note = nil)
check = JSON.parse(File.read(check_path, encoding: "UTF-8")) # build-check.js (or handoff.js) output
input = {
task: task,
code: check["code"], # the paste as the page sends it
facts: JSON.generate(check["facts"]) # a STRING, compact
}
n = note.to_s.strip
input[:note] = n[0, 2000] unless n.empty? # optional; max 2000 chars
input[:audit_handoff] = check["audit_handoff"] if task == "fix" && !check["audit_handoff"].to_s.empty?
input
end
input = make_input("store", "ledger-check.json",
"We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request.")
est = api("POST", "/estimate", input)
puts "hold #{est["hold_credits"]} min #{est["min_credits"]} " \
"sponsored #{est["sponsor_enabled"]} warnings #{(est["warnings"] || []).inspect}"
// The body the page sends: task, code, facts as a JSON STRING, an optional note,
// and (fix lane only) the audit_handoff string -- every value a string.
function make_input(string $task, string $checkPath, string $note = ""): array {
// decode to objects (not arrays) so an empty {} stays {} when re-encoded
$check = json_decode(file_get_contents($checkPath)); // build-check.js (or handoff.js) output
$input = [
"task" => $task,
"code" => $check->code, // the paste as the page sends it
"facts" => json_encode($check->facts, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE), // a STRING
];
$n = trim($note);
if ($n !== "") $input["note"] = mb_substr($n, 0, 2000); // optional; max 2000 chars
if ($task === "fix" && !empty($check->audit_handoff)) $input["audit_handoff"] = $check->audit_handoff;
return $input;
}
$input = make_input("store", "ledger-check.json",
"We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request.");
$est = api("POST", "/estimate", $input);
echo "hold ", $est->hold_credits ?? "-", " min ", $est->min_credits ?? "-",
" sponsored ", json_encode($est->sponsor_enabled ?? false),
" warnings ", json_encode($est->warnings ?? []), "\n";
// The body the page sends: task, code, facts as a JSON STRING, an optional note,
// and (fix lane only) the audit_handoff string -- every value a string.
static Dictionary<string, string> MakeInput(string task, string checkPath, string? note = null)
{
var compact = new JsonSerializerOptions { Encoder = System.Text.Encodings.Web.JavaScriptEncoder.UnsafeRelaxedJsonEscaping };
var check = JsonNode.Parse(File.ReadAllText(checkPath))!; // build-check.js (or handoff.js) output
var input = new Dictionary<string, string>
{
["task"] = task,
["code"] = check["code"]!.GetValue<string>(), // the paste as the page sends it
["facts"] = check["facts"]!.ToJsonString(compact), // a STRING, compact
};
var n = (note ?? "").Trim();
if (n.Length > 0) input["note"] = n.Length > 2000 ? n[..2000] : n; // optional; max 2000 chars
var handoff = check["audit_handoff"]?.GetValue<string>();
if (task == "fix" && !string.IsNullOrEmpty(handoff)) input["audit_handoff"] = handoff;
return input;
}
var input = MakeInput("store", "ledger-check.json",
"We start taking real deposits on this next week. Is the store safe to put money on? redact_iban exists because compliance needs an IBAN erased on request.");
var est = await Api.Call("POST", "/estimate", input);
Console.WriteLine($"hold {est?["hold_credits"]} min {est?["min_credits"]} " +
$"sponsored {est?["sponsor_enabled"]} warnings {est?["warnings"]?.ToJsonString() ?? "[]"}");
Step 5 — POST /run, then poll the job (store audit)
Send the same input to /run with an Idempotency-Key header, using a
personal token. Its data carries a job_id. Poll
GET /jobs/{job_id} the way sdk.js waitForJob does: every
second until status is succeeded or failed, giving up on
the client side after 180 seconds. On the finished job, read:
output.output— the model’s reply as a string: the JSON object from the output contract of the lane you asked for. Parse it yourself (from the first{to the last}). The samples print the verdict and headline, each area’s status, each finding’sid,file,line,severityand number of edits, then each flag answer. Keep the text:recheck.jsreads it.truncated—truewhen the reply was cut short because the balance ran out. The page then closes the cut-off JSON, shows what arrived as a partial result (“N of 9 sections recovered” for an audit, of 8 for a fix run), and says it was cut short by the available balance.charged_credits— the real cost of the run, as opposed to the estimate’shold_creditsreservation. The page re-reads/meafterwards for the new balance.error— on afailedjob.
Idempotency key. Resending a request with the same key must not start a second
paid run, so derive the key from the input, not from the clock — and include the lane, so a
store audit, a projection audit and a fix run of the same paste never share a key. The page uses
append-only-audit:<lane>:<inputHash>:a<attempt> (app.js
run), where <lane> is store, projection or
fix, <inputHash> is a 32-bit djb2 hash of
JSON.stringify(input) in hex, a dash, and the string’s length in hex
(app.js inputHash, over the input without
retry_note), and <attempt> is 1 for a normal run and
2 for the automatic reformat retry (which also adds retry_note). For the
three worked examples that is append-only-audit:store:7c46ba71-30c5:a1 (ledger store
audit), append-only-audit:projection:c444b495-3b17:a1 (orders projection audit) and
append-only-audit:fix:5002a346-24b6:a1 (loyalty fix); the fix run of the ledger audit handed
on (above) is append-only-audit:fix:dcda6d34-3f23:a1. The format is the page’s
convention, not a requirement: any stable, unique string does. The samples below use the lane and a
SHA-256 prefix.
The page itself uses /run-stream (step 6), not /run; both take the
same body and header. For the projection audit, run the same code with
node build-check.js projection orders-paste.txt > orders-check.json,
makeInput("projection", "orders-check.json", note) and a key that starts
append-only-audit:projection:; the reply has the same shape, with the four projection
areas.
KEY="append-only-audit:store:$(printf '%s' "$BODY" | shasum -a 256 | cut -c1-16):a1"
JOB=$(api POST /run -H "Idempotency-Key: $KEY" -d "$BODY" | jq -r '.data.job_id')
for i in $(seq 180); do # 1 s interval, 180 s client-side limit
J=$(api GET "/jobs/$JOB")
S=$(printf '%s' "$J" | jq -r '.data.status')
if [ "$S" = "succeeded" ] || [ "$S" = "failed" ]; then break; fi
sleep 1
done
printf '%s' "$J" | jq '.data | {status, truncated, charged_credits, error}'
printf '%s' "$J" | jq -r '.data.output.output' \
| jq '{verdict, headline, areas: [(.areas // [])[] | {area, status}],
findings: [(.findings // [])[] | {id, file, line, severity, edits: ((.edits // []) | length)}],
flag_answers: [(.flag_answers // [])[] | {flag, status}]}'
body = json.dumps(inp)
key = "append-only-audit:store:" + hashlib.sha256(body.encode()).hexdigest()[:16] + ":a1"
job_id = api("POST", "/run", inp, {"Idempotency-Key": key})["job_id"]
deadline = time.time() + 180 # 1 s interval, 180 s client-side limit
while True:
job = api("GET", "/jobs/" + urllib.parse.quote(job_id, safe=""))
if job["status"] in ("succeeded", "failed"):
break
if time.time() > deadline:
raise TimeoutError("job timed out")
time.sleep(1)
if job["status"] == "failed":
raise SystemExit(f"run failed: {job.get('error')}")
print("truncated:", job.get("truncated") is True, "charged:", job.get("charged_credits"))
text = (job.get("output") or {}).get("output") or ""
reply = json.loads(text[text.index("{"): text.rindex("}") + 1])
print(reply["verdict"], "-", reply["headline"])
for s in reply.get("areas", []):
print("area", s["area"], s["status"])
for f in reply.get("findings", []):
print(f["id"], f["file"], f.get("line"), f["severity"], len(f.get("edits") or []), "edit(s)")
for a in reply.get("flag_answers", []):
print("answer", a["flag"], a["status"])
const key = "append-only-audit:store:" +
createHash("sha256").update(JSON.stringify(input)).digest("hex").slice(0, 16) + ":a1";
const { job_id } = await api("POST", "/run", input, { "Idempotency-Key": key });
let job;
const deadline = Date.now() + 180000; // 1 s interval, 180 s client-side limit
for (;;) {
job = await api("GET", "/jobs/" + encodeURIComponent(job_id));
if (job.status === "succeeded" || job.status === "failed") break;
if (Date.now() > deadline) throw new Error("job timed out");
await new Promise((r) => setTimeout(r, 1000));
}
if (job.status === "failed") throw new Error("run failed: " + JSON.stringify(job.error));
console.log("truncated:", job.truncated === true, "charged:", job.charged_credits);
const text = (job.output && job.output.output) || "";
const reply = JSON.parse(text.slice(text.indexOf("{"), text.lastIndexOf("}") + 1));
console.log(reply.verdict, "-", reply.headline);
for (const s of reply.areas || []) console.log("area", s.area, s.status);
for (const f of reply.findings || []) console.log(f.id, f.file, f.line, f.severity, (f.edits || []).length, "edit(s)");
for (const a of reply.flag_answers || []) console.log("answer", a.flag, a.status);
inputJSON, _ := json.Marshal(input)
sum := sha256.Sum256(inputJSON)
key := "append-only-audit:store:" + hex.EncodeToString(sum[:8]) + ":a1"
var started struct {
JobID string `json:"job_id"`
}
if err := api("POST", "/run", input, map[string]string{"Idempotency-Key": key}, &started); err != nil {
panic(err)
}
type jobT struct {
Status string `json:"status"`
Output *struct {
Output string `json:"output"`
} `json:"output"`
Truncated bool `json:"truncated"`
ChargedCredits any `json:"charged_credits"`
Error any `json:"error"`
}
var job jobT
deadline := time.Now().Add(180 * time.Second) // 1 s interval, 180 s client-side limit
for {
job = jobT{}
if err := api("GET", "/jobs/"+url.PathEscape(started.JobID), nil, nil, &job); err != nil {
panic(err)
}
if job.Status == "succeeded" || job.Status == "failed" {
break
}
if time.Now().After(deadline) {
panic("job timed out")
}
time.Sleep(time.Second)
}
if job.Status == "failed" {
panic(fmt.Sprintf("run failed: %v", job.Error))
}
fmt.Println("truncated:", job.Truncated, "charged:", job.ChargedCredits)
text := ""
if job.Output != nil {
text = job.Output.Output
}
var reply map[string]any
if a, b := strings.Index(text, "{"), strings.LastIndex(text, "}"); a >= 0 && b > a {
if err := json.Unmarshal([]byte(text[a:b+1]), &reply); err != nil {
panic(err)
}
}
fmt.Println(reply["verdict"], "-", reply["headline"])
areas, _ := reply["areas"].([]any)
for _, e := range areas {
s, _ := e.(map[string]any)
fmt.Println("area", s["area"], s["status"])
}
findings, _ := reply["findings"].([]any)
for _, e := range findings {
f, _ := e.(map[string]any)
edits, _ := f["edits"].([]any)
fmt.Println(f["id"], f["file"], f["line"], f["severity"], len(edits), "edit(s)")
}
answers, _ := reply["flag_answers"].([]any)
for _, e := range answers {
a, _ := e.(map[string]any)
fmt.Println("answer", a["flag"], a["status"])
}
byte[] digest = MessageDigest.getInstance("SHA-256").digest(JSON.writeValueAsBytes(input));
String key = "append-only-audit:store:" + HexFormat.of().formatHex(digest, 0, 8) + ":a1";
String jobId = api("POST", "/run", input, Map.of("Idempotency-Key", key)).path("job_id").asText();
JsonNode job;
long deadline = System.currentTimeMillis() + 180_000; // 1 s interval, 180 s client-side limit
while (true) {
job = api("GET", "/jobs/" + URLEncoder.encode(jobId, StandardCharsets.UTF_8), null, null);
String s = job.path("status").asText();
if (s.equals("succeeded") || s.equals("failed")) break;
if (System.currentTimeMillis() > deadline) throw new RuntimeException("job timed out");
Thread.sleep(1000);
}
if (job.path("status").asText().equals("failed")) throw new RuntimeException("run failed: " + job.path("error"));
System.out.println("truncated: " + job.path("truncated").asBoolean(false) + ", charged: " + job.path("charged_credits"));
String text = job.path("output").path("output").asText("");
JsonNode reply = JSON.readTree(text.substring(text.indexOf('{'), text.lastIndexOf('}') + 1));
System.out.println(reply.path("verdict").asText() + " - " + reply.path("headline").asText());
for (JsonNode s : reply.path("areas")) {
System.out.println("area " + s.path("area").asText() + " " + s.path("status").asText());
}
for (JsonNode f : reply.path("findings")) {
System.out.println(f.path("id").asText() + " " + f.path("file").asText() + " " + f.path("line") + " "
+ f.path("severity").asText() + " " + f.path("edits").size() + " edit(s)");
}
for (JsonNode a : reply.path("flag_answers")) {
System.out.println("answer " + a.path("flag").asText() + " " + a.path("status").asText());
}
key = "append-only-audit:store:#{Digest::SHA256.hexdigest(JSON.generate(input))[0, 16]}:a1"
job_id = api("POST", "/run", input, { "Idempotency-Key" => key })["job_id"]
deadline = Time.now + 180 # 1 s interval, 180 s client-side limit
job = nil
loop do
job = api("GET", "/jobs/#{URI.encode_www_form_component(job_id)}")
break if %w[succeeded failed].include?(job["status"])
raise "job timed out" if Time.now > deadline
sleep 1
end
raise "run failed: #{job["error"].inspect}" if job["status"] == "failed"
puts "truncated: #{job["truncated"] == true}, charged: #{job["charged_credits"]}"
text = (job["output"] || {})["output"].to_s
reply = JSON.parse(text[text.index("{")..text.rindex("}")])
puts "#{reply["verdict"]} - #{reply["headline"]}"
(reply["areas"] || []).each { |s| puts "area #{s["area"]} #{s["status"]}" }
(reply["findings"] || []).each { |f| puts "#{f["id"]} #{f["file"]} #{f["line"].inspect} #{f["severity"]} #{(f["edits"] || []).size} edit(s)" }
(reply["flag_answers"] || []).each { |a| puts "answer #{a["flag"]} #{a["status"]}" }
$key = "append-only-audit:store:" . substr(hash("sha256", json_encode($input)), 0, 16) . ":a1";
$jobId = api("POST", "/run", $input, ["Idempotency-Key" => $key])->job_id;
$deadline = time() + 180; // 1 s interval, 180 s client-side limit
while (true) {
$job = api("GET", "/jobs/" . rawurlencode($jobId));
if (in_array($job->status, ["succeeded", "failed"], true)) break;
if (time() > $deadline) throw new RuntimeException("job timed out");
sleep(1);
}
if ($job->status === "failed") throw new RuntimeException("run failed: " . json_encode($job->error ?? null));
echo "truncated: ", json_encode(($job->truncated ?? false) === true), ", charged: ", $job->charged_credits ?? "-", "\n";
$text = $job->output->output ?? "";
$reply = json_decode(substr($text, strpos($text, "{"), strrpos($text, "}") - strpos($text, "{") + 1));
echo $reply->verdict, " - ", $reply->headline, "\n";
foreach ($reply->areas ?? [] as $s) echo "area ", $s->area, " ", $s->status, "\n";
foreach ($reply->findings ?? [] as $f) echo $f->id, " ", $f->file, " ", json_encode($f->line ?? null), " ", $f->severity, " ", count($f->edits ?? []), " edit(s)\n";
foreach ($reply->flag_answers ?? [] as $a) echo "answer ", $a->flag, " ", $a->status, "\n";
var digest = System.Security.Cryptography.SHA256.HashData(Encoding.UTF8.GetBytes(JsonSerializer.Serialize(input)));
var key = "append-only-audit:store:" + Convert.ToHexString(digest, 0, 8).ToLowerInvariant() + ":a1";
var started = await Api.Call("POST", "/run", input, new Dictionary<string, string> { ["Idempotency-Key"] = key });
var jobId = started!["job_id"]!.GetValue<string>();
JsonNode job;
var deadline = DateTime.UtcNow.AddSeconds(180); // 1 s interval, 180 s client-side limit
while (true)
{
job = (await Api.Call("GET", "/jobs/" + Uri.EscapeDataString(jobId)))!;
var s = job["status"]?.GetValue<string>();
if (s == "succeeded" || s == "failed") break;
if (DateTime.UtcNow > deadline) throw new TimeoutException("job timed out");
await Task.Delay(1000);
}
if (job["status"]?.GetValue<string>() == "failed") throw new Exception($"run failed: {job["error"]?.ToJsonString()}");
var truncated = job["truncated"] is JsonValue t && t.TryGetValue(out bool tb) && tb;
Console.WriteLine($"truncated: {truncated}, charged: {job["charged_credits"]}");
var text = job["output"]?["output"]?.GetValue<string>() ?? "";
var reply = JsonNode.Parse(text[text.IndexOf('{')..(text.LastIndexOf('}') + 1)])!;
Console.WriteLine($"{reply["verdict"]} - {reply["headline"]}");
foreach (var s in reply["areas"]?.AsArray() ?? new JsonArray())
Console.WriteLine($"area {s?["area"]} {s?["status"]}");
foreach (var f in reply["findings"]?.AsArray() ?? new JsonArray())
Console.WriteLine($"{f?["id"]} {f?["file"]} {f?["line"]?.ToJsonString() ?? "null"} {f?["severity"]} {f?["edits"]?.AsArray().Count ?? 0} edit(s)");
foreach (var a in reply["flag_answers"]?.AsArray() ?? new JsonArray())
Console.WriteLine($"answer {a?["flag"]} {a?["status"]}");
Step 6 — POST /run-stream (Server-Sent Events, fix)
The alternative to step 5, and what the page uses: same kind of body, same
Idempotency-Key header, one request. Run one or the other for a given input, not
both. To show another lane, the samples build the fix input of the
loyalty example (with its own key) and stream it; to stream the store (or projection) input instead,
pass the step 4 input and key. The response is text/event-stream. sdk.js splits
it into frames on a blank line, takes the event name from an event: line (default
message) and concatenates the data: lines, skips frames with no data or
with data that is not JSON, and handles these events:
| Event | Data | What the SDK does |
|---|---|---|
delta | {text} | Passes text to onDelta. |
job | job info | Passes it to onJob (the page uses it to advance its progress display). |
done | the final payload: {job_id, status, charged_credits, output} | Keeps it as the result. |
pending | a payload | Treated exactly like done. If it carries a job_id and no output, poll that job as in step 5. |
error | {message, code, job_id} | After the stream ends, raises an error with those fields. |
If the response is not text/event-stream, it is an ordinary JSON envelope:
the SDK returns its data (an idempotent replay returns this way) or raises the
error on a non-2xx status. Handle both shapes.
From the final payload the page reads output.output (falling back to the
concatenated delta text), truncated and charged_credits.
Browsers receive progress ticks rather than deltas from this endpoint, so the page relies on the
final payload; if the stream fails part-way, it tries to recover a partial reply from whatever
delta text arrived. The samples print the verdict and headline, each returned file’s path and
size, each change (function, flags and change), each
verification step, each open item and each question. To put the files into the paste and re-check
them the way the page’s “Put every file into the paste” does, keep the reply text
and run recheck.js with the lane fix. To fix code you audited, read handoff.js’s output instead of
loyalty-check.json: makeInput then carries its audit_handoff.
# the fix worked example: loyalty-check.json from `node build-check.js fix loyalty-paste.txt`
# (for audited code, read handoff.js's output instead; the + (...) adds its audit_handoff)
FBODY=$(jq -c --arg note "Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine." \
'{task: "fix", code, facts: (.facts | tojson), note: $note} + (if .audit_handoff then {audit_handoff} else {} end)' \
loyalty-check.json)
FKEY="append-only-audit:fix:$(printf '%s' "$FBODY" | shasum -a 256 | cut -c1-16):a1"
# -N turns off buffering; the whole stream lands in $STREAM
STREAM=$(api POST /run-stream -N -H "Idempotency-Key: $FKEY" -d "$FBODY")
# frames look like:
# event: job
# data: {...}
#
# event: done
# data: {"job_id":"...","status":"...","charged_credits":...,"output":{"output":"{\"task\":\"fix\",...}"}}
# the final reply text, from the done (or pending) frame:
REPLY=$(printf '%s\n' "$STREAM" \
| awk '/^event:/{ev=$2} /^data:/ && (ev=="done" || ev=="pending"){sub(/^data: */,""); print}' \
| jq -r '.output.output')
# an idempotent replay answers with a plain JSON envelope instead: REPLY=$(printf '%s' "$STREAM" | jq -r '.data.output.output')
printf '%s' "$REPLY" | jq '{verdict, headline,
files: [(.files // [])[] | {path, characters: (.content | length)}],
changes: [(.changes // [])[] | {function, flags, change}], verify, open, questions}'
# the fix worked example: loyalty-check.json from build-check.js
# (for audited code, pass handoff.js's output instead: make_input adds its audit_handoff)
fix_inp = make_input("fix", "loyalty-check.json",
"Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine.")
fix_key = "append-only-audit:fix:" + hashlib.sha256(json.dumps(fix_inp).encode()).hexdigest()[:16] + ":a1"
h = {"Content-Type": "application/json", "Authorization": "Bearer " + TOKEN, "Idempotency-Key": fix_key}
done, failure, streamed = None, None, ""
with requests.post(BASE + "/run-stream", headers=h, data=json.dumps(fix_inp), stream=True) as r:
if "text/event-stream" not in r.headers.get("content-type", ""):
try:
env = r.json()
except ValueError:
env = {}
if not r.ok:
err = env.get("error") or {}
raise ApiError(r.status_code, err.get("code"), err.get("message") or r.reason)
done = env.get("data")
else:
event, data = "message", ""
for line in r.iter_lines(decode_unicode=True):
if line == "": # a blank line ends a frame
if data:
try:
payload = json.loads(data)
except ValueError:
payload = None
if event == "delta" and payload is not None:
streamed += payload.get("text") or ""
elif event == "job":
print("job", payload)
elif event in ("done", "pending"):
done = payload
elif event == "error":
failure = payload
event, data = "message", ""
elif line.startswith("event:"):
event = line[6:].strip()
elif line.startswith("data:"):
data += line[5:].strip()
if failure:
raise SystemExit(f"{failure.get('code')}: {failure.get('message')} (job {failure.get('job_id')})")
done = done or {}
full = (done.get("output") or {}).get("output") or streamed
print("truncated:", done.get("truncated") is True, "charged:", done.get("charged_credits"))
result = json.loads(full[full.index("{"): full.rindex("}") + 1])
print(result["verdict"], "-", result["headline"])
for f in result.get("files", []):
print("file", f["path"], len(f.get("content") or ""), "characters")
for c in result.get("changes", []):
print("change", c.get("function"), ",".join(c.get("flags") or []) or "-", "-", c.get("change"))
for step in result.get("verify", []):
print("verify", step)
for o in result.get("open", []):
print("open", o["item"], "-", o.get("reason"))
for q in result.get("questions", []):
print("question", q)
// the fix worked example: loyalty-check.json from `node build-check.js fix loyalty-paste.txt`
// (for audited code, pass handoff.js's output instead: makeInput adds its audit_handoff)
const fixInput = makeInput("fix", "loyalty-check.json",
"Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine.");
const fixKey = "append-only-audit:fix:" +
createHash("sha256").update(JSON.stringify(fixInput)).digest("hex").slice(0, 16) + ":a1";
const res = await fetch(BASE + "/run-stream", {
method: "POST",
headers: { "Content-Type": "application/json", Authorization: "Bearer " + TOKEN, "Idempotency-Key": fixKey },
body: JSON.stringify(fixInput),
});
let done = null, failure = null, streamed = "";
if (!(res.headers.get("content-type") || "").includes("text/event-stream")) {
const json = await res.json().catch(() => ({}));
if (!res.ok) throw new Error((json.error && json.error.message) || res.statusText);
done = json.data;
} else {
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
for (;;) {
const chunk = await reader.read();
if (chunk.done) break;
buffer += decoder.decode(chunk.value, { stream: true });
let idx;
while ((idx = buffer.indexOf("\n\n")) >= 0) {
const frame = buffer.slice(0, idx);
buffer = buffer.slice(idx + 2);
let event = "message", dataStr = "";
for (const line of frame.split("\n")) {
if (line.startsWith("event:")) event = line.slice(6).trim();
else if (line.startsWith("data:")) dataStr += line.slice(5).trim();
}
if (!dataStr) continue;
let data;
try { data = JSON.parse(dataStr); } catch { continue; }
if (event === "delta") streamed += data.text || "";
else if (event === "job") console.log("job", data);
else if (event === "done" || event === "pending") done = data;
else if (event === "error") failure = data;
}
}
}
if (failure) throw new Error(`${failure.code}: ${failure.message} (job ${failure.job_id})`);
const full = (done && done.output && done.output.output) || streamed;
console.log("truncated:", !!(done && done.truncated), "charged:", done && done.charged_credits);
const result = JSON.parse(full.slice(full.indexOf("{"), full.lastIndexOf("}") + 1));
console.log(result.verdict, "-", result.headline);
for (const f of result.files || []) console.log("file", f.path, (f.content || "").length, "characters");
for (const c of result.changes || []) console.log("change", c.function, (c.flags || []).join(",") || "-", "-", c.change);
for (const step of result.verify || []) console.log("verify", step);
for (const o of result.open || []) console.log("open", o.item, "-", o.reason);
for (const q of result.questions || []) console.log("question", q);
// the fix worked example: loyalty-check.json from `node build-check.js fix loyalty-paste.txt`
// (for audited code, pass handoff.js's output instead: makeInput adds its audit_handoff)
fixInput := makeInput("fix", "loyalty-check.json",
"Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine.")
fixJSON, _ := json.Marshal(fixInput)
fixSum := sha256.Sum256(fixJSON)
fixKey := "append-only-audit:fix:" + hex.EncodeToString(fixSum[:8]) + ":a1"
sreq, err := newRequest("POST", "/run-stream", fixInput, map[string]string{"Idempotency-Key": fixKey})
if err != nil {
panic(err)
}
sres, err := http.DefaultClient.Do(sreq)
if err != nil {
panic(err)
}
defer sres.Body.Close()
var done, failure map[string]any
var streamed strings.Builder
if !strings.Contains(sres.Header.Get("Content-Type"), "text/event-stream") {
var env envelope
_ = json.NewDecoder(sres.Body).Decode(&env)
if sres.StatusCode < 200 || sres.StatusCode > 299 {
panic(errorFrom(sres, env))
}
_ = json.Unmarshal(env.Data, &done)
} else {
sc := bufio.NewScanner(sres.Body)
sc.Buffer(make([]byte, 64*1024), 16*1024*1024) // the done frame can be long
event, data := "message", ""
for sc.Scan() {
line := sc.Text()
switch {
case line == "": // a blank line ends a frame
if data != "" {
var p map[string]any
if json.Unmarshal([]byte(data), &p) == nil {
switch event {
case "delta":
if t, ok := p["text"].(string); ok {
streamed.WriteString(t)
}
case "job":
fmt.Println("job", p)
case "done", "pending":
done = p
case "error":
failure = p
}
}
}
event, data = "message", ""
case strings.HasPrefix(line, "event:"):
event = strings.TrimSpace(line[6:])
case strings.HasPrefix(line, "data:"):
data += strings.TrimSpace(line[5:])
}
}
}
if failure != nil {
panic(fmt.Sprintf("%v: %v (job %v)", failure["code"], failure["message"], failure["job_id"]))
}
full := streamed.String()
if out, ok := done["output"].(map[string]any); ok {
if s, ok := out["output"].(string); ok && s != "" {
full = s
}
}
fmt.Println("truncated:", done["truncated"] == true, "charged:", done["charged_credits"])
var result map[string]any
if a, b := strings.Index(full, "{"), strings.LastIndex(full, "}"); a >= 0 && b > a {
if err := json.Unmarshal([]byte(full[a:b+1]), &result); err != nil {
panic(err)
}
}
fmt.Println(result["verdict"], "-", result["headline"])
files, _ := result["files"].([]any)
for _, e := range files {
f, _ := e.(map[string]any)
c, _ := f["content"].(string)
fmt.Println("file", f["path"], len([]rune(c)), "characters")
}
changes, _ := result["changes"].([]any)
for _, e := range changes {
c, _ := e.(map[string]any)
var flags []string
if fs, ok := c["flags"].([]any); ok {
for _, fl := range fs {
flags = append(flags, fmt.Sprint(fl))
}
}
fl := strings.Join(flags, ",")
if fl == "" {
fl = "-"
}
fmt.Println("change", c["function"], fl, "-", c["change"])
}
verify, _ := result["verify"].([]any)
for _, st := range verify {
fmt.Println("verify", st)
}
open, _ := result["open"].([]any)
for _, e := range open {
o, _ := e.(map[string]any)
fmt.Println("open", o["item"], "-", o["reason"])
}
questions, _ := result["questions"].([]any)
for _, q := range questions {
fmt.Println("question", q)
}
// the fix worked example: loyalty-check.json from `node build-check.js fix loyalty-paste.txt`
// (for audited code, pass handoff.js's output instead: makeInput adds its audit_handoff)
Map<String, Object> fixInput = makeInput("fix", "loyalty-check.json",
"Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine.");
byte[] fixDigest = MessageDigest.getInstance("SHA-256").digest(JSON.writeValueAsBytes(fixInput));
String fixKey = "append-only-audit:fix:" + HexFormat.of().formatHex(fixDigest, 0, 8) + ":a1";
HttpRequest sreq = request("POST", "/run-stream", fixInput, Map.of("Idempotency-Key", fixKey)).build();
HttpResponse<Stream<String>> sres = HTTP.send(sreq, HttpResponse.BodyHandlers.ofLines());
JsonNode done = null, failure = null;
StringBuilder streamed = new StringBuilder();
if (!sres.headers().firstValue("content-type").orElse("").contains("text/event-stream")) {
JsonNode env = JSON.readTree(sres.body().collect(Collectors.joining("\n")));
if (env == null) env = JSON.createObjectNode();
if (sres.statusCode() / 100 != 2) throw errorFrom(sres.statusCode(), env);
done = env.path("data");
} else {
String event = "message";
StringBuilder data = new StringBuilder();
Iterator<String> it = sres.body().iterator();
while (it.hasNext()) {
String line = it.next();
if (line.isEmpty()) { // a blank line ends a frame
if (data.length() > 0) {
JsonNode p = null;
try { p = JSON.readTree(data.toString()); } catch (Exception ignored) { }
if (p != null) {
switch (event) {
case "delta": streamed.append(p.path("text").asText("")); break;
case "job": System.out.println("job " + p); break;
case "done": case "pending": done = p; break;
case "error": failure = p; break;
default: break;
}
}
}
event = "message";
data.setLength(0);
} else if (line.startsWith("event:")) event = line.substring(6).trim();
else if (line.startsWith("data:")) data.append(line.substring(5).trim());
}
}
if (failure != null) {
throw new RuntimeException(failure.path("code").asText() + ": " + failure.path("message").asText()
+ " (job " + failure.path("job_id").asText() + ")");
}
String full = done == null ? "" : done.path("output").path("output").asText("");
if (full.isEmpty()) full = streamed.toString();
System.out.println("truncated: " + (done != null && done.path("truncated").asBoolean(false))
+ ", charged: " + (done == null ? "-" : done.path("charged_credits").toString()));
JsonNode result = JSON.readTree(full.substring(full.indexOf('{'), full.lastIndexOf('}') + 1));
System.out.println(result.path("verdict").asText() + " - " + result.path("headline").asText());
for (JsonNode f : result.path("files")) {
System.out.println("file " + f.path("path").asText() + " " + f.path("content").asText("").codePointCount(0, f.path("content").asText("").length()) + " characters");
}
for (JsonNode c : result.path("changes")) {
List<String> flags = new ArrayList<>();
c.path("flags").forEach(fl -> flags.add(fl.asText()));
System.out.println("change " + c.path("function").asText() + " " + (flags.isEmpty() ? "-" : String.join(",", flags))
+ " - " + c.path("change").asText());
}
for (JsonNode step : result.path("verify")) System.out.println("verify " + step.asText());
for (JsonNode o : result.path("open")) System.out.println("open " + o.path("item").asText() + " - " + o.path("reason").asText());
for (JsonNode q : result.path("questions")) System.out.println("question " + q.asText());
# the fix worked example: loyalty-check.json from build-check.js
# (for audited code, pass handoff.js's output instead: make_input adds its audit_handoff)
fix_input = make_input("fix", "loyalty-check.json",
"Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine.")
fix_key = "append-only-audit:fix:#{Digest::SHA256.hexdigest(JSON.generate(fix_input))[0, 16]}:a1"
uri, req = build_request("POST", "/run-stream", fix_input, { "Idempotency-Key" => fix_key })
done = failure = nil
streamed = +""
Net::HTTP.start(uri.host, uri.port, use_ssl: true, read_timeout: 300) do |http|
http.request(req) do |res|
unless res["content-type"].to_s.include?("text/event-stream")
env = (JSON.parse(res.body) rescue {})
unless res.is_a?(Net::HTTPSuccess)
raise ApiError.new(res.code.to_i, env.dig("error", "code"), env.dig("error", "message") || res.message)
end
done = env["data"]
next
end
buf = +""
res.read_body do |chunk|
buf << chunk
while (i = buf.index("\n\n"))
frame = buf.slice!(0, i + 2)
event = "message"
data = +""
frame.split("\n").each do |line|
if line.start_with?("event:") then event = line[6..].strip
elsif line.start_with?("data:") then data << line[5..].strip
end
end
next if data.empty?
payload = (JSON.parse(data) rescue nil)
next if payload.nil?
case event
when "delta" then streamed << payload["text"].to_s
when "job" then puts "job #{payload.inspect}"
when "done", "pending" then done = payload
when "error" then failure = payload
end
end
end
end
end
raise "#{failure["code"]}: #{failure["message"]} (job #{failure["job_id"]})" if failure
done ||= {}
full = (done["output"] || {})["output"].to_s
full = streamed if full.empty?
puts "truncated: #{done["truncated"] == true}, charged: #{done["charged_credits"]}"
result = JSON.parse(full[full.index("{")..full.rindex("}")])
puts "#{result["verdict"]} - #{result["headline"]}"
(result["files"] || []).each { |f| puts "file #{f["path"]} #{f["content"].to_s.length} characters" }
(result["changes"] || []).each do |c|
flags = (c["flags"] || []).join(",")
puts "change #{c["function"]} #{flags.empty? ? "-" : flags} - #{c["change"]}"
end
(result["verify"] || []).each { |step| puts "verify #{step}" }
(result["open"] || []).each { |o| puts "open #{o["item"]} - #{o["reason"]}" }
(result["questions"] || []).each { |q| puts "question #{q}" }
// the fix worked example: loyalty-check.json from `node build-check.js fix loyalty-paste.txt`
// (for audited code, pass handoff.js's output instead: make_input adds its audit_handoff)
$fixInput = make_input("fix", "loyalty-check.json",
"Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine.");
$fixKey = "append-only-audit:fix:" . substr(hash("sha256", json_encode($fixInput)), 0, 16) . ":a1";
$done = null; $failure = null; $streamed = ""; $buf = ""; $ctype = "";
$handle = function (string $event, string $data) use (&$done, &$failure, &$streamed) {
$p = json_decode($data);
if ($p === null) return; // not JSON: skipped, as sdk.js does
switch ($event) {
case "delta": $streamed .= $p->text ?? ""; break;
case "job": echo "job ", json_encode($p), "\n"; break;
case "done": case "pending": $done = $p; break;
case "error": $failure = $p; break;
}
};
$ch = curl_init(BASE . "/run-stream");
curl_setopt_array($ch, [
CURLOPT_POST => true,
CURLOPT_HTTPHEADER => headers_for(["Idempotency-Key" => $fixKey]),
CURLOPT_POSTFIELDS => json_encode($fixInput, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE),
CURLOPT_HEADERFUNCTION => function ($ch, $line) use (&$ctype) {
if (stripos($line, "content-type:") === 0) $ctype = strtolower(trim(substr($line, 13)));
return strlen($line);
},
CURLOPT_WRITEFUNCTION => function ($ch, $chunk) use (&$buf, &$ctype, $handle) {
$buf .= $chunk; // collects the response in memory
if (strpos($ctype, "text/event-stream") === false) return strlen($chunk); // plain JSON: parse at the end
while (($i = strpos($buf, "\n\n")) !== false) {
$frame = substr($buf, 0, $i);
$buf = substr($buf, $i + 2);
$event = "message"; $data = "";
foreach (explode("\n", $frame) as $line) {
if (strpos($line, "event:") === 0) $event = trim(substr($line, 6));
elseif (strpos($line, "data:") === 0) $data .= trim(substr($line, 5));
}
if ($data !== "") $handle($event, $data);
}
return strlen($chunk);
},
]);
if (curl_exec($ch) === false) throw new RuntimeException(curl_error($ch));
$status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
curl_close($ch);
if (strpos($ctype, "text/event-stream") === false) {
$env = json_decode($buf) ?? new stdClass();
if ($status < 200 || $status > 299) {
throw new ApiError($status, $env->error->code ?? null, $env->error->message ?? "HTTP $status");
}
$done = $env->data ?? null;
}
if ($failure) throw new RuntimeException(($failure->code ?? "") . ": " . ($failure->message ?? "") . " (job " . ($failure->job_id ?? "") . ")");
$full = $done->output->output ?? "";
if ($full === "") $full = $streamed;
echo "truncated: ", json_encode(($done->truncated ?? false) === true), ", charged: ", $done->charged_credits ?? "-", "\n";
$result = json_decode(substr($full, strpos($full, "{"), strrpos($full, "}") - strpos($full, "{") + 1));
echo $result->verdict, " - ", $result->headline, "\n";
foreach ($result->files ?? [] as $f) echo "file ", $f->path, " ", mb_strlen($f->content ?? ""), " characters\n";
foreach ($result->changes ?? [] as $c) {
$flags = implode(",", $c->flags ?? []);
echo "change ", $c->function ?? "", " ", $flags === "" ? "-" : $flags, " - ", $c->change ?? "", "\n";
}
foreach ($result->verify ?? [] as $step) echo "verify ", $step, "\n";
foreach ($result->open ?? [] as $o) echo "open ", $o->item, " - ", $o->reason ?? "", "\n";
foreach ($result->questions ?? [] as $q) echo "question ", $q, "\n";
// the fix worked example: loyalty-check.json from `node build-check.js fix loyalty-paste.txt`
// (for audited code, pass handoff.js's output instead: MakeInput adds its audit_handoff)
var fixInput = MakeInput("fix", "loyalty-check.json",
"Members lose points after a crash and gain them twice after a restart. Fix the checkpoint and the double counting. Leave rebuild() exactly as it is - the DBA truncates loyalty_balances by hand before every rebuild, and that runbook changes next quarter. Adding a column to loyalty_balances is fine.");
var fixDigest = System.Security.Cryptography.SHA256.HashData(Encoding.UTF8.GetBytes(JsonSerializer.Serialize(fixInput)));
var fixKey = "append-only-audit:fix:" + Convert.ToHexString(fixDigest, 0, 8).ToLowerInvariant() + ":a1";
using var sreq = Api.Request("POST", "/run-stream", fixInput, new Dictionary<string, string> { ["Idempotency-Key"] = fixKey });
using var sres = await Api.Http.SendAsync(sreq, HttpCompletionOption.ResponseHeadersRead);
JsonNode? done = null, failure = null;
var streamed = new StringBuilder();
if (sres.Content.Headers.ContentType?.MediaType != "text/event-stream")
{
JsonNode? env = null;
try { env = JsonNode.Parse(await sres.Content.ReadAsStringAsync()); } catch (JsonException) { }
if (!sres.IsSuccessStatusCode) throw Api.ErrorFrom(sres, env);
done = env?["data"];
}
else
{
using var reader = new StreamReader(await sres.Content.ReadAsStreamAsync());
var ev = "message";
var data = new StringBuilder();
string? line;
while ((line = await reader.ReadLineAsync()) != null)
{
if (line.Length == 0) // a blank line ends a frame
{
if (data.Length > 0)
{
JsonNode? p = null;
try { p = JsonNode.Parse(data.ToString()); } catch (JsonException) { }
if (p != null)
{
switch (ev)
{
case "delta": streamed.Append(p["text"]?.GetValue<string>() ?? ""); break;
case "job": Console.WriteLine($"job {p.ToJsonString()}"); break;
case "done": case "pending": done = p; break;
case "error": failure = p; break;
}
}
}
ev = "message";
data.Clear();
}
else if (line.StartsWith("event:")) ev = line[6..].Trim();
else if (line.StartsWith("data:")) data.Append(line[5..].Trim());
}
}
if (failure != null) throw new Exception($"{failure["code"]}: {failure["message"]} (job {failure["job_id"]})");
var full = done?["output"]?["output"]?.GetValue<string>();
if (string.IsNullOrEmpty(full)) full = streamed.ToString();
var cut = done?["truncated"] is JsonValue tv && tv.TryGetValue(out bool cutShort) && cutShort;
Console.WriteLine($"truncated: {cut}, charged: {done?["charged_credits"]}");
var result = JsonNode.Parse(full[full.IndexOf('{')..(full.LastIndexOf('}') + 1)])!;
Console.WriteLine($"{result["verdict"]} - {result["headline"]}");
foreach (var f in result["files"]?.AsArray() ?? new JsonArray())
Console.WriteLine($"file {f?["path"]} {(f?["content"]?.GetValue<string>() ?? "").Length} characters");
foreach (var c in result["changes"]?.AsArray() ?? new JsonArray())
{
var flags = string.Join(",", (c?["flags"]?.AsArray() ?? new JsonArray()).Select(fl => fl?.GetValue<string>()));
Console.WriteLine($"change {c?["function"]} {(flags == "" ? "-" : flags)} - {c?["change"]}");
}
foreach (var step in result["verify"]?.AsArray() ?? new JsonArray()) Console.WriteLine($"verify {step}");
foreach (var o in result["open"]?.AsArray() ?? new JsonArray()) Console.WriteLine($"open {o?["item"]} - {o?["reason"]}");
foreach (var q in result["questions"]?.AsArray() ?? new JsonArray()) Console.WriteLine($"question {q}");
What the page does that you now do
- Reads the paste in the browser and builds
codeand a lane-cutfactsfrom it for free (eskit.js), joining dropped files under# file:lines (-- file:for SQL, SQL first) and cutting a large paste on whole files with a# [append-only-audit: not sent: ...]line; you send the same strings, andfactsis required. - Prices every input with
/estimate(thehold_creditsreservation, themin_creditsfloor) and refuses to run belowmin_credits, or on a paste with no Python or SQL it can read. - Sends each attempt with an input- and lane-derived
Idempotency-Key, and retries once withretry_notewhen a reply does not parse; after a run it re-reads/mefor the balance left oncecharged_creditssettled. - Closes a cut-off JSON reply and shows what arrived when
truncatedis true or the stream ends early. - Hands a store audit to the projection audit with its edits applied, and either audit to the fix lane as
audit_handoff, with every edit applied to the paste first, and only while that edited paste is unchanged. - Re-checks the reply against the code (
recon.js), in the browser: files, lines, evidence, catalogue ids of the lane, every edit applied in order and the free check re-run after each finding, flag answers and verdict; the fixed files put back, re-checked and matched against the changes list and the open items, with every function still there and stillasync defordef; and repeated credentials.recheck.jsabove runs the same checks for you.
Or skip all of this and use Append-Only Audit in the browser: the check, the flags, the table, statement and projector inventories, putting an edit or the fixed files back into the paste, and the Markdown and CSV exports are free.