feat(protocol): return the commit revision on write replies by jsam · Pull Request #293 · inputlayer/inputlayer

jsam · 2026-10-05T06:29:39Z

Intent

please continue developing the input layer engine so we make sure that the symbolic engine is ready to go live and completely production ready while we develop the GoTM uh, market.

(Context: the captain approved the neurosymbolic decision engine plan; its milestone "Engine readiness" makes the symbolic engine production-ready for the hosted decision API. This task is item EN-3: commit revision on write replies (E-3): a program whose writes commit returns the revision it committed at (result.revision for committed programs) — why: every API write returns revision; decision records carry revision_commit without a second query)

What Changed

Program results now carry an optional revision field, on QueryResult and on the WebSocket ResultFrame/ResultStartFrame. It holds the knowledge graph revision the program's committed writes become visible at, which is the same revision standing-query pushes carry. It is set for committed fact, schema and rule writes, and for the meta-commands .rel drop, .clear prefix, .index create|drop|rebuild, .kg create and .ontology install|remove|upgrade. When a program makes several such writes, the field holds the last one's revision. Session-only schemas, facts and rules, plain queries, and rolled-back writes leave it unset.
In the storage engine, ProgramCommit now records the revision it published. The storage calls behind those meta-commands now return the revision of the snapshot they published: create_knowledge_graph_at, drop_relation_in, clear_relations_by_prefix_in, and create_index_in/drop_index_in/rebuild_index_in.
Updated the AsyncAPI spec, the WebSocket API guide and the GUI docs bundle to cover the new field. Added tests for programs that commit writes, programs that run meta-commands, programs ending in a pinned proof, and the ordering of standing-query pushes against the returned revision.

Risk Assessment

✅ Low: The change adds an optional revision field to replies. The field is serialized only when a write ran and is read under the KG write lock right after each publish. Every early return in run_program (unpinned proof, pinned proof, messages-only return, failed query) carries it, and the ontology and index paths propagate it consistently, so I found no wrong-value or breaking-change path.

Testing

I built the server from this branch and ran it on a local port with a throwaway data directory and admin password. A Node script logged in as admin over WebSocket with two connections, one writing and one subscribed, and checked 20 things: - Fact inserts, persistent rules, .clear prefix, .rel drop, .index create/rebuild/drop, .kg create, and a write followed by .why all return a revision. - For each write a subscriber was watching, the reply's revision equals the revision on the next subscription_delta. - A write that changed nothing returns the current revision. - Query-only programs, a rolled-back program (no fact persisted and no delta pushed), session rules and schemas, .rel and .kg use carry no revision. The first two attempts had failures caused by bugs in my driver: it counted persistent_update notifications as deltas, and a timed-out wait swallowed the next delta. I fixed the driver, restarted the server on a clean data directory, and all 20 checks passed. I also ran the targeted tests this branch adds or changes; all passed. They cover .ontology install/remove, which I did not drive live, so that scenario is marked untested. I didn't run the full suite. The server and temp directory were removed afterwards, and the worktree is clean. This is an API change with no UI, so there is no screenshot; the evidence is a WebSocket transcript.

Live validation: ✅ go - 8 of 9 scenarios driven live against the product

Scenario	Result	Live	Evidence
Client inserts facts over WebSocket and the reply's revision equals the revision on the subscriber's next delta	✅ pass	live	ws-revision-transcript.txt: S1 reply.revision=6, push.revision=6 (subscribed at 5)
A write that changes nothing (inserting a fact already present) returns the current revision	✅ pass	live	ws-revision-transcript.txt: S2 reply.revision=6 expected 6
A query-only program's reply has no revision field	✅ pass	live	ws-revision-transcript.txt: S3 reply keys have no revision
Adversarial: a program whose writes roll back on an arity error carries no revision, persists nothing and pushes no delta	✅ pass	live	ws-revision-transcript.txt: S4 checks (no revision, ?n(3) empty, no push within 1.5s)
A program that writes and then queries returns the query rows plus the commit revision, which matches the delta	✅ pass	live	ws-revision-transcript.txt: S5 reply=7 push=7
Defining a persistent rule returns a newer revision; session rules and session schemas return none	✅ pass	live	ws-revision-transcript.txt: S6 reply=8; S7 revision=undefined for both
Meta commands that change state (.clear prefix, .rel drop, .index create/rebuild/drop, .kg create) each return a newer revision; .clear prefix matches its delta; .rel and .kg use return none	✅ pass	live	ws-revision-transcript.txt: S8 checks (clear 10 = push 10; drop 12; index 15/16/17; kg create 18)
A program that writes then ends in a .why proof still returns its commit revision	✅ pass	live	ws-revision-transcript.txt: S9 reply=19 with proof_trees
.ontology install/remove return their commit revision, and a no-op write afterwards returns that same revision	⏸️ untested	no	The prior payload did not establish a live result: this was checked only by `cargo test --test atomic_program_tests` (failed_pack_installation_applies_nothing), not over WebSocket against the running…

Evidence: Live WebSocket transcript: requests, replies with revision, subscription deltas, and check results

authenticated kg=default stream_epoch=3d315eb95efafbd3
> +n(0)
< {"type":"result","rows":[["Inserted 1 fact(s) into 'n'."]],"revision":5,"errors":[]}
> [subscriber] .subscribe nsub ?n(X)
< {"type":"result","rows":[[0]],"errors":[],"subscribed":{"subscription":"nsub","generation":1,"revision":5}}
> +n(1)\n+other(1)
< {"type":"result","rows":[["Inserted 1 fact(s) into 'n'."],["Inserted 1 fact(s) into 'other'."]],"revision":6,"errors":[]}
< [subscriber push] {"type":"subscription_delta","subscription":"nsub","generation":1,"knowledge_graph":"default","seq":1,"revision":6,"columns":["X"],"inserted":[[1]],"retracted":[]}
PASS S1 committed insert reply carries revision > subscribe revision :: reply.revision=6 subscribed_at=5
PASS S1 push revision equals write reply revision :: push.revision=6
> +n(1) (already present)
< {"type":"result","rows":[["Inserted 0 fact(s) into 'n'."]],"revision":6,"errors":[]}
PASS S2 no-op write names current revision :: reply.revision=6 expected 6
> ?n(X)
< {"type":"result","rows":[[0],[1]],"errors":[]}
PASS S3 query-only reply has no revision field :: keys=type,columns,rows,row_count,total_count,truncated,execution_time_ms,row_provenance,timing_breakdown,errors
> +n(3)\n+n("three", 4)
< {"type":"result","rows":[["Arity mismatch for relation 'n': existing arity is 1, but trying to insert tuples with arity 2 (rolled back: none of the program's 2 write statements 0-1 was applied)"]],"errors":[{"index":1,"code":"validation","message":"Arity mismatch for relation 'n': existing arity is 1, but trying to insert tuples with arity 2 (rolled back: none of the program's 2 write statements 0-1 was applied)"}]}
PASS S4 rolled-back program carries no revision :: errors=1
> ?n(3)
< {"type":"result","rows":[],"errors":[]}
PASS S4 rolled-back write left no fact :: rows=[]
PASS S4 no push for rolled-back program :: push=null
> +n(2)\n?n(X)
< {"type":"result","rows":[[0],[1],[2]],"revision":7,"errors":[]}
< [subscriber push] {"type":"subscription_delta","subscription":"nsub","generation":1,"knowledge_graph":"default","seq":2,"revision":7,"columns":["X"],"inserted":[[2]],"retracted":[]}
PASS S5 write+query names commit revision matching push :: reply=7 push=7
> +big(X) <- n(X), X > 0
< {"type":"result","rows":[["Rule 'big' registered."]],"revision":8,"errors":[]}
PASS S6 persistent rule names revision :: reply=8
> tmp(X) <- n(X) (session rule)
< {"type":"result","rows":[["Session rule added for 'tmp'."]],"errors":[]}
PASS S7 session rule carries no revision :: revision=undefined
> sess(a: int) (session schema)
< {"type":"result","rows":[["Schema for 'sess' registered with 1 columns (session)"]],"errors":[]}
PASS S7 session schema carries no revision :: revision=undefined
> +p_x(1)
< {"type":"result","rows":[["Inserted 1 fact(s) into 'p_x'."]],"revision":9,"errors":[]}
> .clear prefix p_
< {"type":"result","rows":[["Cleared 1 fact(s) from 1 relation(s) with prefix 'p_': p_x (1)"]],"revision":10,"errors":[]}
< [subscriber push] {"type":"subscription_delta","subscription":"psub","generation":2,"knowledge_graph":"default","seq":1,"revision":10,"columns":["X"],"inserted":[],"retracted":[[1]]}
PASS S8 .clear prefix names revision matching push :: reply=10 push=10
> .rel drop gone
< {"type":"result","rows":[["Relation 'gone' dropped."]],"revision":12,"errors":[]}
PASS S8 .rel drop names revision :: reply=12
PASS S8 .rel listing carries no revision :: revision=undefined
> +docs(id: int, emb: vector)
< {"type":"result","rows":[["Schema for 'docs' registered with 2 columns (persistent)"]],"revision":13,"errors":[]}
> +docs[(1,[1,0,0])]
< {"type":"result","rows":[["Inserted 1 fact(s) into 'docs'."]],"revision":14,"errors":[]}
> .index create doc_idx on docs(emb)
< {"type":"result","rows":[["Index 'doc_idx' created on docs.emb (1 vectors)."]],"revision":15,"errors":[]}
PASS S8 .index create doc_idx on docs(emb) names a newer revision :: reply=15 prev=14
> .index rebuild doc_idx
< {"type":"result","rows":[["Index 'doc_idx' rebuilt (1 vectors)."]],"revision":16,"errors":[]}
PASS S8 .index rebuild doc_idx names a newer revision :: reply=16 prev=15
> .index drop doc_idx
< {"type":"result","rows":[["Index 'doc_idx' dropped."]],"revision":17,"errors":[]}
PASS S8 .index drop doc_idx names a newer revision :: reply=17 prev=16
> .kg create fresh_kg
< {"type":"result","rows":[["Knowledge graph 'fresh_kg' created."],["Switched to knowledge graph: fresh_kg"]],"revision":18,"errors":[]}
PASS S8 .kg create names revision :: reply=18
> .kg use fresh_kg
< {"type":"result","rows":[["Switched to knowledge graph: fresh_kg"]],"errors":[]}
PASS S8 .kg use carries no revision :: revision=undefined
> +need("s2","say_eta")\n.why ?need(S, T)
< {"type":"result","revision":19,"errors":[],"keys":["type","columns","rows","row_count","total_count","truncated","execution_time_ms","row_provenance","proof_trees","errors","statements","revision"]}
PASS S9 program ending in .why names its commit revision :: reply=19

Evidence: Node WebSocket driver used for the live run

const URL_ = 'ws://127.0.0.1:18791/ws';
function client() {
  return new Promise((res, rej) => {
    const ws = new WebSocket(URL_);
    const q = []; const waiters = []; const pushes = []; const pushWaiters = [];
    ws.onmessage = (ev) => {
      const m = JSON.parse(ev.data);
      if (m.type === 'result' || m.type === 'error' || m.type === 'authenticated' || m.type === 'auth_error' || m.type==='result_start' || m.type==='result_end') {
        const w = waiters.shift(); if (w) w(m); else q.push(m);
      } else if (m.type === 'subscription_delta' || m.type === 'subscription_error') { const w = pushWaiters.shift(); if (w) w(m); else pushes.push(m); }
    };
    ws.onerror = rej;
    const next = () => q.length ? Promise.resolve(q.shift()) : new Promise(r => waiters.push(r));
    const nextPush = (ms=5000) => pushes.length ? Promise.resolve(pushes.shift()) : new Promise((r) => { pushWaiters.push(r); setTimeout(()=>{ const i=pushWaiters.indexOf(r); if(i>=0){pushWaiters.splice(i,1); r(null);} }, ms); });
    ws.onopen = async () => {
      ws.send(JSON.stringify({type:'login', username:'admin', password:'testpass123'}));
      const a = await next(); if (a.type !== 'authenticated') rej(a);
      res({ ws, auth: a, exec: async (p) => { ws.send(JSON.stringify({type:'execute', program:p})); return next(); }, nextPush, close: () => ws.close() });
    };
  });
}
const out = [];
const results = [];
function log(s){ out.push(s); console.log(s); }
function check(name, cond, detail){ results.push({name, pass: !!cond}); log(`${cond?'PASS':'FAIL'} ${name} :: ${detail}`); }
const brief = (r) => JSON.stringify({type:r.type, rows:r.rows, revision:r.revision, errors:r.errors, subscribed:r.subscribed, message:r.message});

const w = await client(); const s = await client();
log(`authenticated kg=${w.auth.knowledge_graph} stream_epoch=${w.auth.stream_epoch}`);
let r;
r = await w.exec('+n(0)'); log('> +n(0)\n< '+brief(r));
r = await s.exec('.subscribe nsub ?n(X)'); log('> [subscriber] .subscribe nsub ?n(X)\n< '+brief(r));
const subAt = r.subscribed.revision;

// S1 insert -> revision equals push
r = await w.exec('+n(1)\n+other(1)'); log('> +n(1)\\n+other(1)\n< '+brief(r));
const c1 = r.revision;
let p = await s.nextPush(); log('< [subscriber push] '+JSON.stringify(p));
check('S1 committed insert reply carries revision > subscribe revision', typeof c1==='number' && c1 > subAt, `reply.revision=${c1} subscribed_at=${subAt}`);
check('S1 push revision equals write reply revision', p && p.revision === c1, `push.revision=${p&&p.revision}`);

// S2 noop write -> current revision
r = await w.exec('+n(1)'); log('> +n(1) (already present)\n< '+brief(r));
check('S2 no-op write names current revision', r.revision === c1, `reply.revision=${r.revision} expected ${c1}`);

// S3 query only -> none
r = await w.exec('?n(X)'); log('> ?n(X)\n< '+brief(r));
check('S3 query-only reply has no revision field', !('revision' in r), `keys=${Object.keys(r)}`);

// S4 rolled back
r = await w.exec('+n(3)\n+n("three", 4)'); log('> +n(3)\\n+n("three", 4)\n< '+brief(r));
check('S4 rolled-back program carries no revision', r.errors?.length>0 && !('revision' in r), `errors=${r.errors?.length}`);
r = await w.exec('?n(3)'); log('> ?n(3)\n< '+brief(r));
check('S4 rolled-back write left no fact', (r.rows||[]).length===0, `rows=${JSON.stringify(r.rows)}`);
p = await s.nextPush(1500); check('S4 no push for rolled-back program', p===null, `push=${JSON.stringify(p)}`);

// S5 write + query
r = await w.exec('+n(2)\n?n(X)'); log('> +n(2)\\n?n(X)\n< '+brief(r));
const c5 = r.revision; p = await s.nextPush(); log('< [subscriber push] '+JSON.stringify(p));
check('S5 write+query names commit revision matching push', typeof c5==='number' && c5 > c1 && p?.revision===c5, `reply=${c5} push=${p?.revision}`);

// S6 persistent rule
r = await w.exec('+big(X) <- n(X), X > 0'); log('> +big(X) <- n(X), X > 0\n< '+brief(r));
check('S6 persistent rule names revision', typeof r.revision==='number' && r.revision > c5, `reply=${r.revision}`);
const c6 = r.revision;

// S7 session-scoped
r = await w.exec('tmp(X) <- n(X)'); log('> tmp(X) <- n(X) (session rule)\n< '+brief(r));
check('S7 session rule carries no revision', !('revision' in r), `revision=${r.revision}`);
r = await w.exec('sess(a: int)'); log('> sess(a: int) (session schema)\n< '+brief(r));
check('S7 session schema carries no revision', !('revision' in r), `revision=${r.revision}`);

// S8 meta commands
r = await w.exec('+p_x(1)'); const cp = r.revision; log('> +p_x(1)\n< '+brief(r));
await s.exec('.subscribe psub ?p_x(X)');
r = await w.exec('.clear prefix p_'); log('> .clear prefix p_\n< '+brief(r));
const cc = r.revision; p = await s.nextPush(); log('< [subscriber push] '+JSON.stringify(p));
check('S8 .clear prefix names revision matching push', typeof cc==='number' && cc>cp && p?.revision===cc, `reply=${cc} push=${p?.revision}`);
r = await w.exec('+gone(1)'); r = await w.exec('.rel drop gone'); log('> .rel drop gone\n< '+brief(r));
check('S8 .rel drop names revision', typeof r.revision==='number' && r.revision>cc, `reply=${r.revision}`);
r = await w.exec('.rel'); check('S8 .rel listing carries no revision', !('revision' in r), `revision=${r.revision}`);
r = await w.exec('+docs(id: int, emb: vector)'); log('> +docs(id: int, emb: vector)\n< '+brief(r));
r = await w.exec('+docs[(1, [1.0, 0.0, 0.0])]'); log('> +docs[(1,[1,0,0])]\n< '+brief(r)); let prev = r.revision;
for (const cmd of ['.index create doc_idx on docs(emb)', '.index rebuild doc_idx', '.index drop doc_idx']) {
  r = await w.exec(cmd); log(`> ${cmd}\n< `+brief(r));
  check(`S8 ${cmd} names a newer revision`, typeof r.revision==='number' && r.revision>prev, `reply=${r.revision} prev=${prev}`);
  prev = r.revision;
}
r = await w.exec('.kg create fresh_kg'); log('> .kg create fresh_kg\n< '+brief(r));
check('S8 .kg create names revision', typeof r.revision==='number', `reply=${r.revision}`);
r = await w.exec('.kg use fresh_kg'); log('> .kg use fresh_kg\n< '+brief(r));
check('S8 .kg use carries no revision', !('revision' in r), `revision=${r.revision}`);

// S9 pinned proof
r = await w.exec('+need("s2", "say_eta")\n.why ?need(S, T)'); log('> +need("s2","say_eta")\\n.why ?need(S, T)\n< '+JSON.stringify({type:r.type, revision:r.revision, errors:r.errors, keys:Object.keys(r)}));
check('S9 program ending in .why names its commit revision', typeof r.revision==='number', `reply=${r.revision}`);

w.close(); s.close();
const fs = await import('fs');
fs.writeFileSync(process.argv[2], out.join('\n')+'\n');
const failed = results.filter(x=>!x.pass).length; console.log(`\n${results.length-failed}/${results.length} checks passed`);
process.exit(failed?1:0);

Evidence: Key excerpt

> +n(1)\n+other(1)
< {"type":"result",...,"revision":6,"errors":[]}
< [subscriber push] {"type":"subscription_delta","subscription":"nsub",...,"revision":6,"inserted":[[1]]}
> +n(1) (already present)
< {..."revision":6}
> ?n(X)
< {"type":"result","rows":[[0],[1]],"errors":[]} (no revision)
> +n(3)\n+n("three", 4)
< rolled back, errors=[...], no revision
20/20 checks passed

Pipeline

Updates from git push no-mistakes

✅ **intent** - passed

✅ No issues found.

🔧 **Rebase** - 1 issue found → auto-fixed ✅

⚠️ src/protocol/handler.rs - merge conflict rebasing onto origin/main

🔧 Fix applied.
✅ Re-checked - no issues remain.

✅ **Review** - passed

✅ No issues found.

✅ **Test** - passed

✅ No issues found.

Live validation: ✅ go - 8 of 9 scenarios driven live against the product

Scenario	Result	Live	Evidence
Client inserts facts over WebSocket and the reply's revision equals the revision on the subscriber's next delta	✅ pass	live	ws-revision-transcript.txt: S1 reply.revision=6, push.revision=6 (subscribed at 5)
A write that changes nothing (inserting a fact already present) returns the current revision	✅ pass	live	ws-revision-transcript.txt: S2 reply.revision=6 expected 6
A query-only program's reply has no revision field	✅ pass	live	ws-revision-transcript.txt: S3 reply keys have no revision
Adversarial: a program whose writes roll back on an arity error carries no revision, persists nothing and pushes no delta	✅ pass	live	ws-revision-transcript.txt: S4 checks (no revision, ?n(3) empty, no push within 1.5s)
A program that writes and then queries returns the query rows plus the commit revision, which matches the delta	✅ pass	live	ws-revision-transcript.txt: S5 reply=7 push=7
Defining a persistent rule returns a newer revision; session rules and session schemas return none	✅ pass	live	ws-revision-transcript.txt: S6 reply=8; S7 revision=undefined for both
Meta commands that change state (.clear prefix, .rel drop, .index create/rebuild/drop, .kg create) each return a newer revision; .clear prefix matches its delta; .rel and .kg use return none	✅ pass	live	ws-revision-transcript.txt: S8 checks (clear 10 = push 10; drop 12; index 15/16/17; kg create 18)
A program that writes then ends in a .why proof still returns its commit revision	✅ pass	live	ws-revision-transcript.txt: S9 reply=19 with proof_trees
.ontology install/remove return their commit revision, and a no-op write afterwards returns that same revision	⏸️ untested	no	The prior payload did not establish a live result: this was checked only by `cargo test --test atomic_program_tests` (failed_pack_installation_applies_nothing), not over WebSocket against the running…

cargo build --bin inputlayer-server, then started it with --port 18791 --data-dir <tmp>/data and INPUTLAYER_ADMIN_PASSWORD in a temp dir
node drive.mjs: a Node WebSocket driver with a writer and a subscriber connection that ran 20 live checks against the server
cargo test --test standing_query_tests test_write_reply_names_the_revision_its_push_carries
cargo test --test standing_query_tests test_relation_commands_name_the_revision_their_push_carries
cargo test --test atomic_program_tests (includes ontology install/remove revision assertions in failed_pack_installation_applies_nothing)
cargo test --test hnsw_index_tests revision
cargo test --lib pinned_proof_tests::the_reply_names_the_revision_the_writes_committed_at
cargo test -p inputlayer-ws-protocol revision

✅ **Document** - passed

✅ No issues found.

✅ **Lint** - passed

✅ No issues found.

✅ **Push** - passed

✅ No issues found.

A program whose writes commit now replies with `revision`: the revision of the snapshot its commit published, the one standing-query pushes carry. A write that changes nothing names the current revision. Queries, rolled back programs and session-scoped writes carry none.

…e for session schemas

…s nothing durable

…evision

… docs

jsam added 6 commits October 5, 2026 06:16

no-mistakes(review): Name revision on direct meta-command writes; non…

4a6ff28

…e for session schemas

no-mistakes(review): Keep meta-command revision when write run commit…

b47947a

…s nothing durable

test(protocol): a program ending in a pinned proof names its commit r…

4692ee6

…evision

no-mistakes(document): Document result revision field in AsyncAPI spec

402f881

no-mistakes(document): Regenerate GUI docs bundle for result revision…

27bc8b0

… docs

jsam mentioned this pull request Oct 5, 2026

feat(protocol): return the committed revision on write replies #277

Closed

jsam merged commit 0a0fdaf into main Oct 5, 2026
26 of 27 checks passed

jsam deleted the fm/en-en3-sync branch October 5, 2026 06:49

This was referenced Oct 5, 2026

Engine: commit revision on write replies (EN-3) #256

Closed

fix(protocol): share the probe's round so a bound family evaluates once per revision #303

Merged

Engine release tracker: checkpoints 1-2 live status #199

Closed

Navigation Menu

Repository navigation

feat(protocol): return the commit revision on write replies - #293

feat(protocol): return the commit revision on write replies#293
jsam merged 6 commits into
mainfrom
fm/en-en3-sync

jsam commented Oct 5, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

jsam commented Oct 5, 2026

Intent

What Changed

Risk Assessment

Testing

Pipeline

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant