feat(protocol): add expect_revision precondition on execute by jsam · Pull Request #288 · inputlayer/inputlayer

jsam · 2026-10-04T20:57:37Z

Intent

I want you to really hyper focus on making the symbolic engine, you know, crisp and really truly production ready.

(Context: plan ~/firstmate/data/nsdl-plan/report.md §6.1, GitHub milestone 7. Item EN-8: expect_revision precondition on execute — why: exact semantics for the strict guard mode (challenge-2 §2.2))

Substance of the referenced plan item: EN-8 adds an expect_revision precondition to the engine's execute request, needed for the hosted decision API's beta. It gives exact semantics to the decision layer's strict guard mode: a decision made by a model on a state window seen at revision_seen commits only if the window's base relations have not changed since that revision; the check is relation-level, so any write to those relations, by any session, refuses the decision (it suits low-write graphs; the default guard stays candidate_still_valid). Per the plan, execute refuses with precondition_failed when a relation in scope changed after the given revision, and challenge-2 §2.2 asked that specs be able to opt into "state_unchanged" with exact semantics.

What Changed

WebSocket execute frames accept three new optional fields: expect_revision, plus expect_relations and expect_epoch, which both require it. A writing program commits only if nothing in scope changed after that revision. The scope is the named relations and every relation persistent rules derive them from, or the whole knowledge graph when expect_relations is absent; a change to any persistent rule also counts. If the check fails, nothing is applied and the request fails with the new precondition_failed error code. A revision from another engine run (an epoch mismatch) fails the same way.
The storage engine has a new precondition module. Each published snapshot now carries a change log that records the revision at which each base relation and the persistent rules last changed. The commit path checks the precondition against this log before applying any writes.
When a client subscribes to a shared view that hasn't been refreshed, it now joins without re-evaluation if nothing the view depends on has changed. The view's reported revision moves up to the graph's current revision, so a subscriber can pass the revision from its snapshot as expect_revision. Also updated: the AsyncAPI spec, the WebSocket guide's new "Revision Preconditions" section, the errors spec, and the HTML API docs. There are new unit tests and an integration suite (tests/ws_expect_revision_tests.rs).

Risk Assessment

✅ Low: The fix round adds a conservative fast path: a clean shared view joins without re-evaluation only when the subscriber's snapshot ChangeLog shows no rule change and no change to any relation in the view's dependency closure since the view's revision. It falls back to waiting for untracked state (HNSW), for revisions older than since, and for in-flight or due refreshes, and it reuses the existing unchanged-refresh advance publication, so the exact-revision guarantee still holds.

Testing

I built the real server and ran it on a throwaway data directory. Over WebSocket I drove 10 scenarios covering the EN-8 strict-guard semantics and the shared-view join fix, and all 10 passed. Refusals left nothing applied; I confirmed this with follow-up queries. A latency check on an 80k-row view showed that after an unrelated write, a new subscriber joins at the current revision without re-evaluation (0.21s against about 4s). A dependency write still re-evaluates and returns the new row. I also ran the targeted integration test (5 passed) and the unit tests (50 passed). The server and temp files are removed. The worktree is clean.

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

Scenario	Result	Live	Evidence
S1: A decider subscribes to ?eta, another session writes an unrelated relation, and the strict decision (expect_revision=seen, expect_relations=[eta], expect_epoch) commits	✅ pass	live	ws-expect-revision-transcript.txt S1
S2: Another session writes eta, so the decision at the old revision is refused with precondition_failed ('Nothing was applied') and the claim count stays 1	✅ pass	live	ws-expect-revision-transcript.txt S2
S3: Without expect_relations (whole-graph scope), any later write refuses the decision	✅ pass	live	ws-expect-revision-transcript.txt S3
S4: An expect_epoch from another engine run is refused with precondition_failed	✅ pass	live	ws-expect-revision-transcript.txt S4
S5: Malformed expectations (relations or epoch without a revision, empty scope, empty relation name, a query, .subscribe) return invalid_request and the connection stays usable	✅ pass	live	ws-expect-revision-transcript.txt S5
S6: 6 clients race to write slot at the same revision with scope [slot]: exactly 1 commits, 5 get precondition_failed, slot has 2 rows	✅ pass	live	ws-expect-revision-transcript.txt S6
S7 (review fix): After an unrelated write, a second subscriber to a shared view is reported the current revision (14, not 13) and can commit with it	✅ pass	live	ws-expect-revision-transcript.txt S7; shared-view-join-latency.txt (joins in 0.21s without re-evaluation, against about 4s for an evaluation)
S8 (adversarial): A write to the view's own relation is not skipped, and a late subscriber sees the new row at the new revision	✅ pass	live	ws-expect-revision-transcript.txt S8; shared-view-join-latency.txt (dependency write re-evaluates, 80201 rows)
S9 (adversarial): A new rule feeding a shared derived view forces re-evaluation, and the late subscriber sees the derived row from the new rule	✅ pass	live	ws-expect-revision-transcript.txt S9
S10: A revision that is stale after a dependency write is refused, and a re-read at the fresh revision commits	✅ pass	live	ws-expect-revision-transcript.txt S10

Evidence: Live /ws transcript of the EN-8 scenarios S1–S10

## S1 strict decision commits when its window is unchanged (unrelated write in between)
  writer >> {"program": "+eta(\"s1\", 3)"}
  writer << {"type": "result", "revision": 6, "rows": [["Inserted 1 fact(s) into 'eta'."]]}
  decider >> {"program": ".subscribe window ?eta(S, D)"}
  decider << {"type": "result", "subscribed": {"subscription": "window", "generation": 1, "revision": 6}, "rows": [["s1", 3]]}
  writer >> {"program": "+noise(1)"}
  writer << {"type": "result", "revision": 7, "rows": [["Inserted 1 fact(s) into 'noise'."]]}
  decider >> {"program": "+claim(\"s1\", \"call_carrier\")", "expect_revision": 6, "expect_relations": ["eta"], "expect_epoch": "d129d344f1b08ac6"}
  decider << {"type": "result", "revision": 8, "rows": [["Inserted 1 fact(s) into 'claim'."]]}
==> S1: PASS seen=6

## S2 another session writes the window -> precondition_failed, nothing applied
  writer >> {"program": "+eta(\"s2\", 5)"}
  writer << {"type": "result", "revision": 9, "rows": [["Inserted 1 fact(s) into 'eta'."]]}
  decider >> {"program": "+claim(\"s1\", \"say_eta\")", "expect_revision": 6, "expect_relations": ["eta"]}
  decider << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: relation 'eta' changed at revision 9, after revision 6. Nothing was applied."}
==> S2: PASS claim rows=1

## S3 whole-graph scope (no expect_relations): any write refuses
  decider >> {"program": "+claim(\"s9\", \"x\")", "expect_revision": 6}
  decider << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: relation 'claim' changed at revision 8, after revision 6. Nothing was applied."}
==> S3: PASS 

## S4 expect_epoch of another engine run is refused
  decider >> {"program": ".subscribe w4 ?eta(S, D)"}
  decider << {"type": "result", "subscribed": {"subscription": "w4", "generation": 2, "revision": 9}, "rows": [["s1", 3], ["s2", 5]]}
  decider >> {"program": "+claim(\"s4\", \"x\")", "expect_revision": 9, "expect_epoch": "not-this-run"}
  decider << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: expect_epoch is not this engine run's stream epoch (d129d344f1b08ac6); revisions restart with the engine. Nothing was applied."}
==> S4: PASS 

## S5 malformed expectations are invalid_request; connection stays usable
  decider >> {"program": "+a(1)", "expect_relations": ["eta"]}
  decider << {"type": "error", "code": "invalid_request", "message": "expect_relations and expect_epoch require expect_revision"}
  decider >> {"program": "+a(1)", "expect_epoch": "e"}
  decider << {"type": "error", "code": "invalid_request", "message": "expect_relations and expect_epoch require expect_revision"}
  decider >> {"program": "+a(1)", "expect_revision": 1, "expect_relations": []}
  decider << {"type": "error", "code": "invalid_request", "message": "expect_relations names no relation; omit it to expect the whole knowledge graph unchanged"}
  decider >> {"program": "+a(1)", "expect_revision": 1, "expect_relations": [""]}
  decider << {"type": "error", "code": "invalid_request", "message": "expect_relations names an empty relation"}
  decider >> {"program": "?eta(S, D)", "expect_revision": 1}
  decider << {"type": "error", "code": "invalid_request", "message": "expect_revision needs a program that writes persistent state (facts, schemas or rules): it is checked when those writes commit. Nothing ran."}
  decider >> {"program": ".subscribe s ?eta(S, D)", "expect_revision": 1}
  decider << {"type": "error", "code": "invalid_request", "message": "expect_revision applies to a program that writes, not to .subscribe or .unsubscribe"}
  decider >> {"program": "+a(1)"}
  decider << {"type": "result", "revision": 10, "rows": [["Inserted 1 fact(s) into 'a'."]]}
==> S5: PASS codes=['invalid_request', 'invalid_request', 'invalid_request', 'invalid_request', 'invalid_request', 'invalid_request']

## S6 six clients race on one revision of one scope -> exactly one commits
  writer >> {"program": "+slot(0)"}
  writer << {"type": "result", "revision": 11, "rows": [["Inserted 1 fact(s) into 'slot'."]]}
  writer >> {"program": ".subscribe slot ?slot(X)"}
  writer << {"type": "result", "subscribed": {"subscription": "slot", "generation": 1, "revision": 11}, "rows": [[0]]}
  racer0 >> {"program": "+slot(1)", "expect_revision": 11, "expect_relations": ["slot"]}
  racer0 << {"type": "result", "revision": 12, "rows": [["Inserted 1 fact(s) into 'slot'."]]}
  racer5 >> {"program": "+slot(6)", "expect_revision": 11, "expect_relations": ["slot"]}
  racer5 << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: relation 'slot' changed at revision 12, after revision 11. Nothing was applied."}
  racer2 >> {"program": "+slot(3)", "expect_revision": 11, "expect_relations": ["slot"]}
  racer2 << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: relation 'slot' changed at revision 12, after revision 11. Nothing was applied."}
  racer1 >> {"program": "+slot(2)", "expect_revision": 11, "expect_relations": ["slot"]}
  racer1 << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: relation 'slot' changed at revision 12, after revision 11. Nothing was applied."}
  racer3 >> {"program": "+slot(4)", "expect_revision": 11, "expect_relations": ["slot"]}
  racer3 << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: relation 'slot' changed at revision 12, after revision 11. Nothing was applied."}
  racer4 >> {"program": "+slot(5)", "expect_revision": 11, "expect_relations": ["slot"]}
  racer4 << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: relation 'slot' changed at revision 12, after revision 11. Nothing was applied."}
==> S6: PASS committed=1 refused=5 slot rows=2

## S7 (review fix) unrelated write: second subscriber joins shared view at the CURRENT revision, and can commit with it
  writer >> {"program": "+av(1)"}
  writer << {"type": "result", "revision": 13, "rows": [["Inserted 1 fact(s) into 'av'."]]}
  subA >> {"program": ".subscribe va ?av(X)"}
  subA << {"type": "result", "subscribed": {"subscription": "va", "generation": 1, "revision": 13}, "rows": [[1]]}
  writer >> {"program": "+bv(1)"}
  writer << {"type": "result", "revision": 14, "rows": [["Inserted 1 fact(s) into 'bv'."]]}
  subB >> {"program": ".subscribe vb ?av(X)"}
  subB << {"type": "result", "subscribed": {"subscription": "vb", "generation": 1, "revision": 14}, "rows": [[1]]}
  subB >> {"program": "+claimv(1)", "expect_revision": 14}
  subB << {"type": "result", "revision": 15, "rows": [["Inserted 1 fact(s) into 'claimv'."]]}
==> S7: PASS A saw 13, unrelated write rev 14, B subscribed at 14

## S8 adversarial: a write to the view's dependency is NOT skipped -> late subscriber sees new rows
  writer >> {"program": "+av(2)"}
  writer << {"type": "result", "revision": 16, "rows": [["Inserted 1 fact(s) into 'av'."]]}
  subC >> {"program": ".subscribe vc ?av(X)"}
  subC << {"type": "result", "subscribed": {"subscription": "vc", "generation": 1, "revision": 16}, "rows": [[1], [2]]}
   subscribed rows: [[1], [2]]
==> S8: PASS write rev 16, C at 16

## S9 adversarial: a new rule feeding a shared derived view forces re-evaluation
  writer >> {"program": "+der(X) <- av(X)"}
  writer << {"type": "result", "revision": 17, "rows": [["Rule 'der' registered."]]}
  subA >> {"program": ".subscribe d1 ?der(X)"}
  subA << {"type": "result", "subscribed": {"subscription": "d1", "generation": 2, "revision": 17}, "rows": [[1], [2]]}
  writer >> {"program": "+der(X) <- extra(X), X > 0"}
  writer << {"type": "result", "revision": 18, "rows": [["Rule 'der' registered."]]}
  writer >> {"program": "+extra(7)"}
  writer << {"type": "result", "revision": 19, "rows": [["Inserted 1 fact(s) into 'extra'."]]}
  subD >> {"program": ".subscribe d2 ?der(X)"}
  subD << {"type": "result", "subscribed": {"subscription": "d2", "generation": 1, "revision": 19}, "rows": [[1], [2], [7]]}
   subscribed rows: [[1], [2], [7]]
==> S9: PASS first 17, later 19

## S10 stale-view revision after a dependency write cannot commit; re-read revision can
  decider >> {"program": ".subscribe s10 ?av(X)"}
  decider << {"type": "result", "subscribed": {"subscription": "s10", "generation": 3, "revision": 19}, "rows": [[1], [2]]}
  writer >> {"program": "+av(3)"}
  writer << {"type": "result", "revision": 20, "rows": [["Inserted 1 fact(s) into 'av'."]]}
  decider >> {"program": "+claimv(2)", "expect_revision": 19, "expect_relations": ["av"]}
  decider << {"type": "error", "code": "precondition_failed", "message": "Precondition failed: relation 'av' changed at revision 20, after revision 19. Nothing was applied."}
  subE >> {"program": ".subscribe s10b ?av(X)"}
  subE << {"type": "result", "subscribed": {"subscription": "s10b", "generation": 1, "revision": 20}, "rows": [[1], [2], [3]]}
  decider >> {"program": "+claimv(2)", "expect_revision": 20, "expect_relations": ["av"]}
  decider << {"type": "result", "revision": 21, "rows": [["Inserted 1 fact(s) into 'claimv'."]]}
==> S10: PASS 19->20

SUMMARY {"S1": "PASS", "S2": "PASS", "S3": "PASS", "S4": "PASS", "S5": "PASS", "S6": "PASS", "S7": "PASS", "S8": "PASS", "S9": "PASS", "S10": "PASS"}

Evidence: Shared-view join latency: unrelated vs dependency write

first subscriber (cold evaluation): revision=24 rows=80200 took=4.308s second subscriber, no write: revision=24 took=0.225s unrelated write at revision 25 third subscriber after UNRELATED write: revision=25 rows=80200 took=0.212s dependency write (edge) at revision 26 fourth subscriber after DEPENDENCY write: revision=26 rows=80201 took=3.952s

first subscriber (cold evaluation): revision=24 rows=80200 took=4.308s
second subscriber, no write: revision=24 rows=80200 took=0.225s
unrelated write at revision 25
third subscriber after UNRELATED write: revision=25 rows=80200 took=0.212s
dependency write (edge) at revision 26
fourth subscriber after DEPENDENCY write: revision=26 rows=80201 took=3.952s
RESULT unrelated_join_rev_is_current=True dep_join_rev_is_current=True dep_rows_grew=True unrelated_fast=True

Evidence: Targeted test output

    Finished `test` profile [unoptimized + debuginfo] target(s) in 13.26s
     Running tests/ws_expect_revision_tests.rs (target/debug/deps/ws_expect_revision_tests-e5a480dfac6845e8)

running 5 tests
test a_revision_of_another_engine_run_is_refused ... ok
test malformed_expectations_are_invalid_requests ... ok
test a_strict_commit_holds_until_its_window_changes ... ok
test a_refused_decider_that_reads_again_can_commit ... ok
test of_clients_racing_on_one_revision_exactly_one_commits ... ok

test result: ok. 5 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 4.21s

test protocol::handler::expect_revision_tests::a_change_in_scope_refuses_the_program_with_precondition_failed ... ok
test storage_engine::precondition::tests::a_program_that_changes_nothing_is_still_refused ... ok
test storage_engine::precondition::tests::a_rule_change_refuses_every_scope ... ok
test storage_engine::precondition::tests::the_change_log_stamps_only_what_a_publish_changed ... ok
test storage_engine::precondition::tests::a_change_in_scope_after_the_revision_refuses_and_writes_nothing ... ok
test protocol::subscription::connection::tests::an_unrelated_write_lets_a_new_subscriber_join_at_the_current_revision ... ok
test storage_engine::precondition::tests::dropping_a_relation_in_scope_is_a_change ... ok
test storage_engine::precondition::tests::changes_outside_the_scope_do_not_refuse ... ok
test storage_engine::precondition::tests::misspelled_scope_relations_are_rejected_and_declared_ones_accepted ... ok
test storage_engine::precondition::tests::a_derived_scope_covers_the_relations_it_is_derived_from ... ok
test storage_engine::precondition::tests::revisions_not_issued_or_from_before_the_graph_existed_are_refused ... ok
test storage_engine::precondition::tests::a_restarted_engine_refuses_revisions_of_the_previous_run ... ok
test storage_engine::precondition::tests::of_concurrent_writers_expecting_one_revision_of_one_scope_exactly_one_commits ... ok
test storage_engine::precondition::tests::writers_outside_the_scope_never_refuse_a_scoped_commit ... ok
test result: ok. 50 passed; 0 failed; 0 ignored; 0 measured; 3090 filtered out; finished in 0.56s

Evidence: Driver scripts

import asyncio, json, sys, websockets
URL="ws://127.0.0.1:18731/ws?kg=expect"
PW="nm-en8-test-password"
LOG=[]
def log(*a):
    s=" ".join(str(x) for x in a); print(s); LOG.append(s)

class C:
    n=0
    def __init__(s,name): s.name=name
    async def open(s, url=URL):
        s.ws=await websockets.connect(url)
        await s.ws.send(json.dumps({"type":"login","username":"admin","password":PW}))
        a=json.loads(await s.ws.recv()); s.epoch=a["stream_epoch"]; return s
    async def reply(s, id):
        while True:
            f=json.loads(await asyncio.wait_for(s.ws.recv(),60))
            if f.get("id")==id: return f
    async def req(s, frame, quiet=False):
        C.n+=1; id=f"{s.name}-{C.n}"
        frame=dict(frame, type="execute", id=id)
        await s.ws.send(json.dumps(frame))
        r=await s.reply(id)
        if not quiet:
            short={k:r.get(k) for k in ("type","code","message","revision","subscribed","rows") if k in r}
            log(f"  {s.name} >> {json.dumps({k:v for k,v in frame.items() if k not in ('type','id')})}")
            log(f"  {s.name} << {json.dumps(short)}")
        return r
    async def sub(s, name, q):
        r=await s.req({"program":f".subscribe {name} {q}"}); return r["subscribed"]["revision"], r
    async def count(s, rel):
        r=await s.req({"program":f"?{rel}(X, Y)"} if rel in("eta","claim") else {"program":f"?{rel}(X)"}, quiet=True)
        return r.get("row_count")

results={}
def check(name, cond, detail=""):
    results[name]= "PASS" if cond else "FAIL"
    log(f"==> {name}: {'PASS' if cond else 'FAIL'} {detail}\n")

async def main():
    admin=await C("setup").open("ws://127.0.0.1:18731/ws")
    await admin.req({"program":".kg create expect"}, quiet=True)
    decider=await C("decider").open(); writer=await C("writer").open()

    log("## S1 strict decision commits when its window is unchanged (unrelated write in between)")
    await writer.req({"program":'+eta("s1", 3)'})
    seen,_=await decider.sub("window","?eta(S, D)")
    await writer.req({"program":"+noise(1)"})
    r=await decider.req({"program":'+claim("s1", "call_carrier")',"expect_revision":seen,"expect_relations":["eta"],"expect_epoch":decider.epoch})
    check("S1", r["type"]=="result" and await decider.count("claim")==1, f"seen={seen}")

    log("## S2 another session writes the window -> precondition_failed, nothing applied")
    await writer.req({"program":'+eta("s2", 5)'})
    r=await decider.req({"program":'+claim("s1", "say_eta")',"expect_revision":seen,"expect_relations":["eta"]})
    c=await decider.count("claim")
    check("S2", r["type"]=="error" and r.get("code")=="precondition_failed" and c==1, f"claim rows={c}")

    log("## S3 whole-graph scope (no expect_relations): any write refuses")
    r=await decider.req({"program":'+claim("s9", "x")',"expect_revision":seen})
    check("S3", r.get("code")=="precondition_failed" and await decider.count("claim")==1)

    log("## S4 expect_epoch of another engine run is refused")
    s4,_=await decider.sub("w4","?eta(S, D)")
    r=await decider.req({"program":'+claim("s4", "x")',"expect_revision":s4,"expect_epoch":"not-this-run"})
    check("S4", r.get("code")=="precondition_failed" and await decider.count("claim")==1)

    log("## S5 malformed expectations are invalid_request; connection stays usable")
    bad=[{"program":"+a(1)","expect_relations":["eta"]},
         {"program":"+a(1)","expect_epoch":"e"},
         {"program":"+a(1)","expect_revision":1,"expect_relations":[]},
         {"program":"+a(1)","expect_revision":1,"expect_relations":[""]},
         {"program":"?eta(S, D)","expect_revision":1},
         {"program":".subscribe s ?eta(S, D)","expect_revision":1}]
    codes=[(await decider.req(b)).get("code") for b in bad]
    ok=(await decider.req({"program":"+a(1)"}))["type"]=="result"
    check("S5", all(c=="invalid_request" for c in codes) and ok, f"codes={codes}")

    log("## S6 six clients race on one revision of one scope -> exactly one commits")
    await writer.req({"program":"+slot(0)"})
    s6,_=await writer.sub("slot","?slot(X)")
    racers=[await C(f"racer{i}").open() for i in range(6)]
    rs=await asyncio.gather(*[c.req({"program":f"+slot({i+1})","expect_revision":s6,"expect_relations":["slot"]}) for i,c in enumerate(racers)])
    committed=sum(r["type"]=="result" for r in rs); refused=sum(r.get("code")=="precondition_failed" for r in rs)
    n=await writer.count("slot")
    check("S6", committed==1 and refused==5 and n==2, f"committed={committed} refused={refused} slot rows={n}")

    log("## S7 (review fix) unrelated write: second subscriber joins shared view at the CURRENT revision, and can commit with it")
    a=await C("subA").open(); b=await C("subB").open()
    await writer.req({"program":"+av(1)"})
    ra,_=await a.sub("va","?av(X)")
    w=await writer.req({"program":"+bv(1)"})
    rb,rep=await b.sub("vb","?av(X)")
    r=await b.req({"program":"+claimv(1)","expect_revision":rb})
    check("S7", rb==w["revision"] and rb>ra and r["type"]=="result", f"A saw {ra}, unrelated write rev {w['revision']}, B subscribed at {rb}")

    log("## S8 adversarial: a write to the view's dependency is NOT skipped -> late subscriber sees new rows")
    w=await writer.req({"program":"+av(2)"})
    rc,rep=await (await C("subC").open()).sub("vc","?av(X)")
    rows=sorted(x[0] for x in rep["subscribed"].get("rows",[])) if isinstance(rep["subscribed"],dict) and "rows" in rep["subscribed"] else rep["subscribed"]
    log("   subscribed rows:", rep["rows"])
    check("S8", rc>=w["revision"] and [2] in rep["rows"], f"write rev {w['revision']}, C at {rc}")

    log("## S9 adversarial: a new rule feeding a shared derived view forces re-evaluation")
    await writer.req({"program":"+der(X) <- av(X)"})
    d1,_=await a.sub("d1","?der(X)")
    w=await writer.req({"program":"+der(X) <- extra(X), X > 0"})
    await writer.req({"program":"+extra(7)"})
    d2,rep=await (await C("subD").open()).sub("d2","?der(X)")
    log("   subscribed rows:", rep["rows"])
    check("S9", d2>d1 and [7] in rep["rows"], f"first {d1}, later {d2}")

    log("## S10 stale-view revision after a dependency write cannot commit; re-read revision can")
    s10,_=await decider.sub("s10","?av(X)")
    await writer.req({"program":"+av(3)"})
    r1=await decider.req({"program":"+claimv(2)","expect_revision":s10,"expect_relations":["av"]})
    s10b,_=await (await C("subE").open()).sub("s10b","?av(X)")
    r2=await decider.req({"program":"+claimv(2)","expect_revision":s10b,"expect_relations":["av"]})
    check("S10", r1.get("code")=="precondition_failed" and r2["type"]=="result", f"{s10}->{s10b}")

    log("SUMMARY", json.dumps(results))
    open(sys.argv[1],"w").write("\n".join(LOG)+"\n")
asyncio.run(main())

Pipeline

Updates from git push no-mistakes

✅ **intent** - passed

✅ No issues found.

✅ **Rebase** - passed

✅ No issues found.

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

ℹ️ src/protocol/subscription/views.rs:213 - Subscribing to a shared view can now cost a full re-evaluation. attach joins a live view at once only when live.latest.revision >= revision, where revision is the graph's current snapshot revision (connection.rs:92). A view's revision only moves forward when the view itself refreshes, and that happens only when its dependencies change. So after any write to an unrelated relation of the same knowledge graph, every new subscriber to a clean shared view takes the Waiting path, which starts an evaluation (it is coalesced while one is in flight). Example: subscribe to ?a(X), then +b(1) in the same graph, then a second client subscribes to ?a(X). The second client now waits for ?a(X) to be re-evaluated, where before it joined the cached result. This is correct and conservative: it makes a subscription's revision usable as expect_revision without spurious refusals. On graphs with many writes and many shared subscribers it adds evaluation load. This is a deliberate tradeoff (commit b3c3300), noted for awareness only.

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

✅ **Test** - passed

✅ No issues found.

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

Scenario	Result	Live	Evidence
S1: A decider subscribes to ?eta, another session writes an unrelated relation, and the strict decision (expect_revision=seen, expect_relations=[eta], expect_epoch) commits	✅ pass	live	ws-expect-revision-transcript.txt S1
S2: Another session writes eta, so the decision at the old revision is refused with precondition_failed ('Nothing was applied') and the claim count stays 1	✅ pass	live	ws-expect-revision-transcript.txt S2
S3: Without expect_relations (whole-graph scope), any later write refuses the decision	✅ pass	live	ws-expect-revision-transcript.txt S3
S4: An expect_epoch from another engine run is refused with precondition_failed	✅ pass	live	ws-expect-revision-transcript.txt S4
S5: Malformed expectations (relations or epoch without a revision, empty scope, empty relation name, a query, .subscribe) return invalid_request and the connection stays usable	✅ pass	live	ws-expect-revision-transcript.txt S5
S6: 6 clients race to write slot at the same revision with scope [slot]: exactly 1 commits, 5 get precondition_failed, slot has 2 rows	✅ pass	live	ws-expect-revision-transcript.txt S6
S7 (review fix): After an unrelated write, a second subscriber to a shared view is reported the current revision (14, not 13) and can commit with it	✅ pass	live	ws-expect-revision-transcript.txt S7; shared-view-join-latency.txt (joins in 0.21s without re-evaluation, against about 4s for an evaluation)
S8 (adversarial): A write to the view's own relation is not skipped, and a late subscriber sees the new row at the new revision	✅ pass	live	ws-expect-revision-transcript.txt S8; shared-view-join-latency.txt (dependency write re-evaluates, 80201 rows)
S9 (adversarial): A new rule feeding a shared derived view forces re-evaluation, and the late subscriber sees the derived row from the new rule	✅ pass	live	ws-expect-revision-transcript.txt S9
S10: A revision that is stale after a dependency write is refused, and a re-read at the fresh revision commits	✅ pass	live	ws-expect-revision-transcript.txt S10

cargo build --bin inputlayer-server, then started it on a throwaway data dir (/tmp, port 18731) with INPUTLAYER_ADMIN_PASSWORD; drove it over real WebSocket connections using a Python websockets client in a disposable venv
drive.py: scenarios S1–S10 (strict commit, in-scope write refusal, whole-graph scope, foreign epoch, malformed expectations, 6-client race, shared-view join at the current revision, dependency-write and rule-change adversarial checks, stale-then-fresh revision)
latency.py 400: built an 80,200-row transitive-closure shared view and timed subscribe after no write, after an unrelated write and after a dependency write
cargo test --test ws_expect_revision_tests (5 passed)
cargo test --lib -- expect_revision precondition subscription::connection subscription::views (50 passed, including an_unrelated_write_lets_a_new_subscriber_join_at_the_current_revision)

✅ **Document** - passed

✅ No issues found.

✅ **Lint** - passed

✅ No issues found.

✅ **Push** - passed

✅ No issues found.

An execute that writes may carry expect_revision (with optional expect_relations and expect_epoch): its writes commit only if no relation in scope (the named relations closed over the persistent rules that derive them, or the whole knowledge graph) and no persistent rule changed after that revision. The check runs under the knowledge graph's write lock with the commit, so of concurrent writers expecting one revision of one scope at most one commits. A refused program applies nothing and fails with the new precondition_failed code. Every published snapshot carries a change log stamping, per base relation and for the rules, the revision of the last snapshot that changed them, computed at publish by comparing with the previous snapshot.

…o in-scope writers

…ed view

…stfmt

# Conflicts: # src/protocol/handler.rs # src/storage_engine/program_commit.rs

…endencies are unchanged

jsam added 5 commits October 4, 2026 20:25

no-mistakes(review): Qualify expect_revision mutual-exclusion claim t…

b132e51

…o in-scope writers

no-mistakes(test): Report current revision when subscribing to a shar…

b3c3300

…ed view

no-mistakes(document): Add resource_exhausted to HTML error codes; ru…

a99a061

…stfmt

no-mistakes(lint): Fix differential_oracle attach call arity

7281e66

jsam marked this pull request as draft October 4, 2026 21:04

jsam added 2 commits October 5, 2026 06:54

Merge remote-tracking branch 'origin/main' into fm/en-en8-sync

2a6f8fe

# Conflicts: # src/protocol/handler.rs # src/storage_engine/program_commit.rs

Merge remote-tracking branch 'origin/main' into fm/en-en8-sync

b111b28

jsam marked this pull request as ready for review October 5, 2026 08:09

jsam added 2 commits October 5, 2026 08:18

no-mistakes(review): Join shared views without re-evaluation when dep…

f7006db

…endencies are unchanged

no-mistakes(document): Fix private intra-doc link breaking doc-check

aac0eca

jsam merged commit 09e05b3 into main Oct 5, 2026
21 checks passed

jsam deleted the fm/en-en8 branch October 5, 2026 08:34

This was referenced Oct 5, 2026

Engine: expect_revision precondition on execute (EN-8) #261

Closed

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

Closed

Navigation Menu

Repository navigation

feat(protocol): add expect_revision precondition on execute - #288

feat(protocol): add expect_revision precondition on execute#288
jsam merged 9 commits into
mainfrom
fm/en-en8

jsam commented Oct 4, 2026 •

edited

Loading

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

jsam commented Oct 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Intent

What Changed

Risk Assessment

Testing

Pipeline

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

jsam commented Oct 4, 2026 •

edited

Loading