diff --git a/.claude/board/EPIPHANIES.md b/.claude/board/EPIPHANIES.md index 3468b802..000bc37e 100644 --- a/.claude/board/EPIPHANIES.md +++ b/.claude/board/EPIPHANIES.md @@ -1,3 +1,41 @@ +## 2026-08-02 — E-THE-DURABLE-UNIT-IS-THE-CYCLE-NOT-THE-CAST-ONE-WAL-WRITE-PER-SWEEP-1 — the persistence seam is reshaped to the WAL-amortized generation; per-cast durability retired + +**Status:** FINDING (operator-ruled 2026-08-02 — the decision on the fork Correction 2(c) of `E-THE-PAIRED-MOVE-MUST-BE-DURABLE-CO-LOCATED-NOT-IN-MEMORY-ONLY-1` surfaced). **Confidence:** High for the **storage/race CONTRACT** (`lance-graph-planner` 344 lib tests, 13 in `persist_sink`, clippy+fmt clean); crash-durability remains contract-probed over an in-process fake, NOT storage-proven (`compile+test green ≠ storage proven`, the Ladybug lesson). Builds NO concrete sink. + +**The ruling.** The durable unit is the **cycle/sweep** — one globally-aligned 64k-row frame — NOT the per-cast `DurableWitness`. `persist_cast`-per-thought would pay WAL header + durability-wait + version-metadata 64k times; the reshape **amortizes the WAL**: 64k concurrent thoughts stage into an owned cast vector, one `persist_cycle` folds+freezes them, and `WalSink::commit_cycle` does exactly **one atomic append → one `DatasetVersion`**. cast ⊂ chunk ⊂ cycle. + +**The layer's ONLY concerns are storage + that 64k concurrent thoughts hit no physical or epistemic race.** It models no semantics: it mints NO witness / ancestry / awareness / branch / rung / temporal-policy type. `CycleFrame` is storage-only `{cycle, base_version}`; `SweepSlot` is a boring landing (`where`, never `why` — no `basis`). + +**Physical race → WRITE-SIDE order, never read-time repair.** Thoughts finish in arbitrary CPU order; `DetachedCycleBatch::freeze` stable-orders casts by their EXISTING `stream_position` (`order_cycle_stably`, generic over the key) BEFORE the single append, so the durable image is already canonical. `scan_sealed` returns stored order and **never sorts** (a read-time sort would lose the whole amortization). This write-side ordering lives in `persist_sink.rs`, **NOT `temporal.rs`** — that module owns query-time reader-horizon/version reading; the write-side deinterlace additions made there this session were reverted. + +**Epistemic race → the sealed read horizon.** Every thought reads only the sealed predecessor `Vn` (`frame.base_version`); the commit publishes `Vn+1` once, all-or-nothing. `scan_sealed` excludes any uncommitted cycle, so no thought reads a concurrent sibling's in-flight output as though it were sealed history (`read Vn / write Vn+1`). + +**Coalescing is a real per-row fold, not last-chunk-wins.** The final image is `row -> last payload in stream order`; chunks are disjoint/mergeable slices of that image. + +**Vocabulary reused, not re-minted.** `DatasetVersion` (scheduler), `KanbanMove`/`KanbanColumn`/`RubiconTransitionError` (kanban), `MailboxId`/`MailboxSoaOwner`, `stream_position` (the existing canonical key). `recover_and_apply` + the durable **watermark** idempotence (not phase equality — the lifecycle is cyclic) + the `StalePhase`/`OwnerMismatch` corruption guards survive from the prior model. + +**Retired (this reshape).** `DurableWitness`, `DurableReceipt`, `DurableCoordinate`, `DurableWrite`, `persist_cast`, `apply_durable_step`, `scan_witnesses`, `LandedWitness` — the per-cast durability surface. Their intent is preserved in the cycle machinery (`CycleFrame`, `SweepSlot`, `LandedSlot`, `DetachedCycleBatch`, `WalSink::commit_cycle`, `persist_cycle`, `order_cycle_stably`). Supersedes the seam-shape half of `E-THE-PAIRED-MOVE-MUST-BE-DURABLE-CO-LOCATED-NOT-IN-MEMORY-ONLY-1` (its ordering/recovery/watermark CONTRACT stands). Still deferred: the concrete Lance sink, gated on crash falsifiers. PR #878 (reshaped in place). + +## 2026-08-01 — E-THE-PAIRED-MOVE-MUST-BE-DURABLE-CO-LOCATED-NOT-IN-MEMORY-ONLY-1 — the persistence sink's crash gap, and temporal.rs's missing layer-1 + +**Status:** FINDING (operator-ruled 2026-08-01; review-hardened twice — see the two Corrections below). **Confidence:** High for the **ordering/recovery CONTRACT** (349 planner lib tests, clippy+fmt clean); the crash-durability is **contract-probed over an in-process fake sink, NOT storage-proven** — real MemWAL/restart/atomic co-location durability is demonstrated only by the deferred concrete Lance sink (`compile+test green ≠ storage proven`, the Ladybug lesson). Builds NO concrete sink. + +**The gap the split-phase left open.** At the prior two-clock-domain shape, `persist_cast` appended only the PAYLOAD; the paired `KanbanMove` lived solely in the in-memory `DurableReceipt`. Sequence: WAL append lands (durable) → process dies before `apply_durable_step` → the receipt evaporates → the paired move is lost → the KanbanStep never fires — **even though the durable SoA state moved**. Silent: storage advanced, lifecycle did not. An in-memory receipt is not a durable record; a step that lives only in it is a step that a crash can delete. + +**The rule (operator).** The paired transition witness must be **durably co-located with the SoA state in the same persistence generation**. The in-memory receipt may REFERENCE that durable material (via its `DurableCoordinate`) but must never be its only copy. NOT via a separate ack/confirmation ledger (`E-ACK-ELIMINATED-1` stands) — the witness rides IN the generation and is read back with the state. So `DurableWrite::append(&DurableWitness, &payload)` lands both atomically; `scan_witnesses` reads them back; `recover_and_apply` replays the pending tail. + +**temporal.rs is TWO layers, and layer-1 was missing.** Layer-2 (epistemic projection — `classify`/`deinterlace`: contemporary/anachronistic/spoiler/unknowable) was well-developed. Layer-1 (**causal** deinterlacing) did not exist: the global durable log interleaves every owner's writes (`A@s0, C@s0, B@s0, A@s1`), and reconstructing owner A's OWN local chain `[A@s0, A@s1]` — with the interleaved owners *removed* from A's timeline — is what makes crash-replay in cast order possible. New `LocalCausalRow` + `local_trajectories`/`local_trajectory_of`. The two layers compose: layer-1 rebuilds the chain, layer-2 filters it by the reader's horizon. `DurableWitness` implements `LocalCausalRow`. + +**Durability proof is the coordinate, not a `LanceVersion`.** A MemWAL write is queryable/durable BEFORE any base manifest version attaches, so recovery identifies the latest LOCAL state from the co-located witness's cast order — not from a dataset version that does not yet exist (falsifier 5). This is why `DurableCoordinate` (shard/epoch/wal-position), never `DatasetVersion`, is what `append` returns. + +**Five integration falsifiers (all green).** (1) layer-1 deinterlaces the interleaved global log into an owner-local chain, with a vacuity guard that the other owners' rows are *dropped* not reordered; (2) one owner's batch replays in cast order over a deliberately seq-descending log; (3) WAL succeeds + crash (receipt dropped) → recovery reconstructs+applies the move from the durable witness, idempotent on re-run; (4) WAL fails → no witness lands → no move, no step; (5) WAL-visible-before-manifest → recovery uses cast order, no dataset version exists. Plus a synchronous-path `StalePhase` guard (a stale/out-of-order apply is surfaced loudly on `apply_durable_step`, while recovery silently skips a stale move as already-reflected). + +**Still deferred (operator ruling — do NOT build yet).** The concrete `LanceShardSink` (`ShardWriter::put`, `enable_memtable + durable_write`) comes AFTER, gated on these falsifiers; the durable local causal chain lands first. No confirmation ledger, replay queue, per-thought ack, custom WAL, or ractor hot-path callback. Extends `E-KANBANMOVE-IS-THE-PARCEL-ADDRESS-STEP-IS-THE-DELIVERY-SCAN-1`. + +**Correction (PR #878 review, same PR — Bugbot High + Codex/CodeRabbit Critical).** The first cut ordered replay by `cast_id` and inferred "already applied" from phase equality. Both are unsound and were fixed within the PR: (a) **`CastId` resets to 0 on every writer restart** (`BatchWriter::next_id`), so cross-lifetime witnesses collide — replay now orders by the durable **`DurableCoordinate::log_order`** (WAL entry position, monotonic across restarts, never reset); `cast_id` survives only as provenance. `scan_witnesses` returns `LandedWitness{coordinate, witness}` and is the `LocalCausalRow` implementor. (b) **Phase equality is not a sound idempotence key on a CYCLIC lifecycle** (`Planning→CognitiveWork→Evaluation→Plan→Planning`): after a lap the owner is back at `Planning`, so a phase-only check replays the whole lap. `recover_and_apply` now takes a durable **watermark** (`applied_through`) — skip ≤ watermark, and above it a non-matching `from` is genuine corruption (`StalePhase`), not a benign skip; it returns the new watermark to persist WITH the SoA phase. Also: `paired_move.mailbox == owner` is validated at persist time (no cross-owner move becomes durable); `scan_witnesses(from)` is bounded; `WriteFailed`/`PersistError` impl `Display`+`Error`. New falsifiers: cyclic-idempotence with a negative control proving the watermark is load-bearing, and cast_id-collision ordering-by-durable-position. The layer-1 `cast_seq` doc now states the monotonic-across-restarts precondition. + +**Correction 2 (PR #878 review — ChatGPT, taken with a grain of salt).** Three further points held up and were fixed: (a) **reordered receipts must stay retryable, NOT dropped** — a stale receipt dropped on the *happy path* would only be re-applied on a crash+restart (recovery is not the happy-path backstop for concurrent-completion reordering), so `apply_durable_step` now **borrows** `&DurableReceipt`; on `StalePhase` the caller keeps it and retries once the prefix lands (falsifier `a_reordered_receipt_is_retryable_not_lost`, no crash needed). (b) **API-honesty on the durable coordinate** — `ShardWriter::put` returns a *batch position*, not `WalAppendResult::entry_position`, so the field `wal_entry_position` was renamed to an opaque, backend-defined `seq` (the sink maps whichever its API yields); no field claims a WAL offset the chosen API does not return. (c) **honest test/claims framing** — the `FakeSink` tests are labelled **ordering/recovery CONTRACT probes**, not crash/WAL integration tests (no real WAL, restart, atomic RecordBatch co-location, or fencing); the fake now STORES the payload (was ignored) and asserts it landed alongside the witness, while the module doc states plainly that atomic co-location is unproven until the concrete sink. Points taken-with-salt and NOT actioned as blocking: the per-cast-vs-generation persistence *seam shape* (finding 5) is a real architectural fork but not a correctness bug (the operator's own `bb5fcdd` used per-cast) — surfaced to the operator for decision rather than unilaterally reshaped; a `DurableWitness::try_from_cast` constructor was deemed redundant given the persist-time guard. + ## 2026-08-01 — E-KANBANMOVE-IS-THE-PARCEL-ADDRESS-STEP-IS-THE-DELIVERY-SCAN-1 — the D-MBX-A6 move is pre-write intent; the KanbanStep is post-write; the persistence sink wires Lance 7's existing MemWAL and invents nothing **Status:** FINDING (operator-ruled 2026-08-01). **Confidence:** High — the pre-write half is shipped with 5 falsifiable probes (`lance_graph_planner::owner_adapter`); the persistence half's API is verified against real `lance-7.0.0` source. diff --git a/.claude/board/INTEGRATION_PLANS.md b/.claude/board/INTEGRATION_PLANS.md index 291a4279..0ba26e5a 100644 --- a/.claude/board/INTEGRATION_PLANS.md +++ b/.claude/board/INTEGRATION_PLANS.md @@ -38,6 +38,32 @@ vectors → four Ψ ripple fields; cloned-lens control as the can-it-stay-silent falsifier), LLM tail last. deepnsm-v2 crate untouched by ruling. A withdrawn probe verdict (tesseract-side, retention-axis category error) is recorded in §8's provenance + tesseract-rs CLAUDE.md. +## 2026-08-02 — persistence-cycle-wal-bootstrap v1 — ACTIVE (bootstrap SHIPPED #878; temporal/revision upgrade PLANNED) — main thread + +**Plan:** `.claude/plans/persistence-cycle-wal-bootstrap-v1.md` +Documentation-only architectural ruling recording the *role* of the #878 +persistence seam and the larger two-dimensional temporal architecture it +bootstraps toward. #878 is a **primitive cycle/WAL bootstrap** (concurrent +results → primitive slot collection/order → freeze one cycle → one amortized +WAL write → one `DatasetVersion`), with six hard guarantees (single sealed `Vn` +read horizon; open-cycle results invisible as `Vn` input; freeze-before-I/O; one +seal = one WAL write = one version; no live mutable SoA borrow to the WAL). The +scalar slot/order model is **explicitly provisional** — sufficient for the +execution/durability plumbing, not the final temporal representation. Two +orthogonal dimensions: **horizontal** = `temporal.rs` coherence within a frame +(book→chapter→verse→span for deterministic streams; partially-ordered +neighbourhoods for medical/higher-order thought — hence a scalar key is +insufficient); **vertical** = `DatasetVersion` succession + cheap +Stockfish-style hindsight lookup. Planned upgrade: a wait-free-emit shadow +`temporal.rs` coherence pass feeding `revision.rs` forward-correction into a +later cycle — a sealed version is never silently rewritten. Cross-refs (does not +re-specify): horizontal detail → `temporal-markov-and-style-classes-v1.md`; +D-MBX-A6 tracking → STATUS_BOARD; per-row write gate → +`mailbox-cycle-aware-write-contract-v1.md`; reshape ruling → EPIPHANIES +`E-THE-DURABLE-UNIT-IS-THE-CYCLE-...-1`. Scope exclusions: no cohort internals / +slot count / actor-neighbour waiting; no new semantic/temporal/rung/witness/ +branch/ancestry types; no reviving ThoughtWitness/basis/awareness_seq/per-cast +WAL. ## 2026-07-26 — rosetta-codebook-convergence v1 — PROPOSED (D-RCC-1 calibrator runnable today) — main thread diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md index d632f6de..d8cfcc1a 100644 --- a/.claude/board/LATEST_STATE.md +++ b/.claude/board/LATEST_STATE.md @@ -1,3 +1,17 @@ +## 2026-08-02 — branch `claude/medcare-rs-continue-ufsazd` — D-MBX-A6-P3e persistence-sink reshaped to WAL-amortized cycle (one write per sweep) + +- `lance_graph_planner::persist_sink` — **reshaped in place** (PR #878) from the per-cast durable-witness model to the **cycle/WAL** model (operator ruling on the seam-shape fork surfaced 2026-08-01). The durable unit is the **cycle/sweep**, NOT the cast: 64k concurrent thoughts stage into an owned `Vec`, `persist_cycle(sink, frame, casts)` folds+freezes them, and `WalSink::commit_cycle(base, DetachedCycleBatch)` does **exactly one atomic append → one `DatasetVersion`** (WAL amortization — 64k thoughts, one write). Types: `CycleId`, `CycleFrame{cycle, base_version}` (storage-only — NO rung/branch/semantic tags), `SweepSlot{cycle, stream_position, owner, row, paired_move, payload}` (boring landing, no `basis`), `LandedSlot{version, slot}`, `DetachedCycleBatch{frame, landings, image}` (frozen, deinterlaced, coalesced). **Write-side ordering** = `order_cycle_stably(rows, key)` (generic over the caller's canonical key; stable-orders casts by the EXISTING `stream_position` in `freeze` BEFORE the append) — completion order never becomes storage order (physical race); `scan_sealed` returns stored order and NEVER sorts. **Epistemic horizon** = sealed read (`read Vn / write Vn+1`); an uncommitted cycle is invisible to `scan_sealed`. **Coalescing** = real per-row fold (`row -> last payload in stream order`), not last-chunk-wins. `recover_and_apply(owner, sealed, applied_through)` + the durable **watermark** idempotence + `StalePhase`/`OwnerMismatch` guards survive unchanged. `versions()` = the cheap coarse timeline a downstream time-series consumer (stockfish-rs, another session) looks up, no landing replay. +- **Write-side ordering lives HERE, not in `temporal.rs`** — that module owns query-time reader-horizon/version reading; only the write-side `deinterlace_write` additions made there earlier this session were **reverted**. `temporal.rs`'s existing query-time causal helpers (`LocalCausalRow`, `local_trajectories`/`local_trajectory_of`, layer-1 causal deinterlace) landed in the PR's earlier commits and **remain** — the reshape commit itself leaves `temporal.rs` untouched (this commit's `git diff` on it is empty), NOT the whole PR. The rung a cycle aligns at is decided by whoever schedules it, never modelled in `CycleFrame`. +- **Retired:** `DurableWitness`, `DurableReceipt`, `DurableCoordinate`, `DurableWrite`, `persist_cast`, `apply_durable_step`, `scan_witnesses`, `LandedWitness` (the per-cast surface). Vocabulary reused, not re-minted: `DatasetVersion`, `KanbanMove`/`KanbanColumn`, `MailboxId`/`MailboxSoaOwner`, `stream_position`. **No concrete Lance sink built** (deferred per operator, gated on crash falsifiers). Tests are storage/race CONTRACT probes over an in-process `FakeWalSink`, honestly labelled — real MemWAL/restart/atomic-append durability UNPROVEN. See `EPIPHANIES.md` E-THE-DURABLE-UNIT-IS-THE-CYCLE-NOT-THE-CAST-ONE-WAL-WRITE-PER-SWEEP-1. + +## 2026-08-01 — branch `claude/medcare-rs-continue-ufsazd` — D-MBX-A6-P3d persistence-sink durable-witness reshape + temporal layer-1 + +### Current Contract Inventory — new/changed modules (lance-graph-planner) +- `lance_graph_planner::persist_sink` — the POST-write half, two clock domains, **crash-durable**. `DurableWitness{owner, cast_id, cycle, paired_move}` is CO-LOCATED with the SoA payload in one persistence generation via `DurableWrite::append(&witness, &payload)`; the in-memory `DurableReceipt` merely REFERENCES it (via `DurableCoordinate`), never the only copy of the move. **Replay order = the durable `DurableCoordinate::log_order` (WAL position, monotonic across restarts), NOT `cast_id`** (`BatchWriter`'s counter resets on restart → cross-lifetime collisions; `cast_id` is provenance only). `DurableWrite::scan_witnesses(from)` returns `LandedWitness{coordinate, witness}` (bounded tail read; `LandedWitness` is the `LocalCausalRow` implementor). `recover_and_apply(owner, landed, applied_through)` = crash recovery: temporal layer-1 → per-owner tail in durable order; **idempotence is a durable WATERMARK** (`applied_through`), NOT phase equality (unsound on the cyclic lifecycle — a completed lap returns to `Planning`); above the watermark a non-matching `from` is corruption (`StalePhase`); returns `Recovered{applied, watermark}` to persist with the SoA phase. `persist_cast` validates `paired_move.mailbox == owner` (no cross-owner move becomes durable); `apply_durable_step` keeps the sync `from==phase` guard (a stale drop is safe — the durable witness replays). `WriteFailed`/`PersistError` impl `Display`+`Error`. Async `persist_cast` (no owner borrow) / sync `apply_durable_step` (no await) split preserved. Durability proof = `DurableCoordinate`, NEVER `LanceVersion`. **No concrete `LanceShardSink` built** (deferred per operator, gated on the crash falsifiers). +- `lance_graph_planner::temporal` — **layer-1 CAUSAL deinterlacing added** (the missing half): `LocalCausalRow{owner, cast_seq}` + `local_trajectories`/`local_trajectory_of` split a globally-interleaved durable log into per-owner LOCAL chains ordered by `cast_seq` (`A@s0,C@s0,B@s0,A@s1` → A's `[A@s0,A@s1]`; interleaved owners removed). `cast_seq` doc states the monotonic-across-restarts precondition (a resettable counter is NOT valid; use a durable-log position). Composes with the pre-existing layer-2 epistemic projection (`classify`/`deinterlace`). `LandedWitness` implements `LocalCausalRow`. +- Round-2 review (ChatGPT, grain of salt): `apply_durable_step` now **borrows** `&DurableReceipt` so a reordered/stale receipt stays RETRYABLE on the happy path (not dropped-until-crash); `DurableCoordinate.wal_entry_position` renamed to opaque `seq` (API-honest — `ShardWriter::put` returns a batch position, not a WAL offset); the `FakeSink` tests are labelled **CONTRACT probes** (the fake now stores + asserts the payload) — real MemWAL/restart/atomic-co-location durability is UNPROVEN until the concrete sink (`compile+test green ≠ storage proven`). +- **Honest status:** the ordering/recovery CONTRACT is green (349 planner lib tests, clippy + fmt clean); crash-durability is contract-probed over an in-process fake, NOT storage-proven. Deferred per operator: the concrete `LanceShardSink`; and the per-cast-vs-**generation/batch** persistence seam shape (finding 5 — surfaced for operator decision, not a correctness bug). `E-THE-PAIRED-MOVE-MUST-BE-DURABLE-CO-LOCATED-NOT-IN-MEMORY-ONLY-1`. + ## 2026-08-01 — branch `claude/medcare-rs-continue-ufsazd` — D-MBX-A6-P3c owner-consume adapter (rebased onto main dcd9cc9) ### Current Contract Inventory — new module (lance-graph-planner) diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 823de65e..0270d578 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -963,6 +963,7 @@ Plan path: `.claude/plans/unified-soa-convergence-v1.md`. Handover `.claude/hand | D-MBX-A6-P3a | StyleStrategy: thinking-style -> cluster -> mechanism -> recipe_kernels Tactic selection (planning substrate; carries tau JIT addr) | lance-graph-planner | 130 | LOW | **In PR** | #439; first cut of A6-P3 consumer wiring; planner now consumes contract recipes/styles; deferred: i4-32D decode, Outcome->Candidate, tau->JIT, membrane commit | | D-MBX-A6-P3b | output overhaul: `StrategyOutcome{reliability, intended_move: Option}` carrier on `PlanInput.outcome`; StyleStrategy retires the dead-store `_reliability`, SURFACES reliability + a bootstrap intended move (Planning→CognitiveWork, owner 0, warden-BOOTSTRAP-OK) — plan still pure | lance-graph-planner | 130 | LOW | **In progress** | additive Option field (6 in-crate literals); UNBLOCKED (no mint, not OQ-11.7); deferred: compose thread-out + contract-promote + owner-consume; E-STRATEGY-OUTCOME-CARRIER-1 | | D-MBX-A6-P3c | owner-consume: `lance_graph_planner::owner_adapter` = the `Outcome → KanbanMove` bootstrap-rebind + ahead-cast adapter. `rebind_bootstrap` (mailbox 0/cycle 0 sentinel → live owner; refuses an already-owned move = no ownership theft) + `emit_bootstrap_intent` → `BatchWriter::cast(on_behalf = owner)`. Fire-and-forget (no ack/ledger/WAL/arbitration/callback); the move is the pre-write "parcel address", the lifecycle STEP stays post-write. Completes P3b's deferred `owner-consume`. | lance-graph-planner | 90 | LOW | **In PR** | 5 falsifiable probes (rebind 0→live anti-vacuity + no-theft + on-behalf cast + non-vacuous no-op silence); lance-free, builds without protoc. Persistence sink (drain→Lance 7 `mem_wal::WalAppender::append`) verified-but-gated (protoc missing + disk); knowledge doc `.claude/v3/knowledge/d-mbx-a6-owner-consume-and-persistence.md`; `E-KANBANMOVE-IS-THE-PARCEL-ADDRESS-STEP-IS-THE-DELIVERY-SCAN-1` | +| D-MBX-A6-P3d | persistence sink ORDERING CORE + durable-witness reshape + temporal layer-1 (the POST-write half). `lance_graph_planner::persist_sink`: two clock domains (async `persist_cast` no-owner-borrow → `DurableReceipt`; sync `apply_durable_step` no-await → `try_advance_phase`). Crash-durability: `DurableWitness{owner,cast_id,cycle,paired_move}` CO-LOCATED with the SoA payload in one generation via `DurableWrite::append(&witness,&payload)`; `scan_witnesses(from)` bounded replay seam returning `LandedWitness{coordinate,witness}`; `recover_and_apply(owner,landed,applied_through)` replays the pending tail in **durable-log order** (`DurableCoordinate::log_order`, NOT the resettable `cast_id`) with a durable **watermark** for cyclic-safe idempotence, returning `Recovered{applied,watermark}`; `StalePhase` = corruption above the watermark (sync path: safe-to-drop stale). `temporal::{LocalCausalRow, local_trajectories, local_trajectory_of}` = layer-1 CAUSAL deinterlacing (global interleaved log → per-owner local chain), composing with the existing layer-2 epistemic projection. Durability proof = `DurableCoordinate` (opaque `seq`, API-honest), never `LanceVersion`. | lance-graph-planner | 175 | LOW | **In PR** | #878; ordering/recovery CONTRACT probed (349 planner lib tests, clippy+fmt clean) — crash-durability NOT storage-proven (in-process fake, no real MemWAL/restart); review-hardened ×2 (Bugbot: cast_id-resets→durable-position order; Codex/CodeRabbit Critical: cyclic idempotence via durable watermark + negative control; cross-owner reject; concurrent-drain retryable receipt; bounded scan; contract-probe honesty); builds NO concrete `LanceShardSink`; generation-vs-per-cast seam (finding 5) surfaced for operator decision; `E-THE-PAIRED-MOVE-MUST-BE-DURABLE-CO-LOCATED-NOT-IN-MEMORY-ONLY-1` | | D-MBX-A6-P3-M1 | `Tactic::requires() -> ThoughtMask` + `ThoughtField`/`ThoughtMask` (checklist-as-data keystone): 34 tactics declare their ThoughtCtx field-reads; `covered_by` = reliability-coverage gate | lance-graph-contract | 120 | LOW | **In PR** | #439; the panel-recalibrated keystone (extraction not construction); makes P1/P7/P11 derived; teeth-test asserts masks varied not stub | | D-CLS-FM | `class_view`: FieldMask(u64 presence) + ClassView meta-DTO resolver trait + ClassProjection (the class flies ABOVE the SoA; labels resolved late from OGIT cache, zero in the bytes) — extends ObjectView, reuses class_id | lance-graph-contract | 270 | LOW | **Shipped** | #441 D-CLS contract foundation; OD-gates ratified; presence!=semantics (C2); N3 stable positions; 3 teeth-tests | | D-CLS-RES | `class_resolver`: `RegistryClassView` impls `ClassView` over the live OntologyRegistry — the ontology-side 'parser' (class_id -> shape, DOLCE resolved LATE via classify_odoo from the cache URI, memoized over the O(n) registry scan) | lance-graph-ontology | 200 | LOW | **Shipped** | #441 D-CLS; makes the contract trait live; field-set supplied (D-CLS audit deferred); 4 teeth-tests | diff --git a/.claude/plans/persistence-cycle-wal-bootstrap-v1.md b/.claude/plans/persistence-cycle-wal-bootstrap-v1.md new file mode 100644 index 00000000..100eb94c --- /dev/null +++ b/.claude/plans/persistence-cycle-wal-bootstrap-v1.md @@ -0,0 +1,224 @@ +# persistence-cycle-wal-bootstrap-v1 — the primitive cycle/WAL seam and its temporal/revision upgrade path + +> **Status:** ACTIVE (bootstrap SHIPPED in PR #878; upgrade phases PLANNED). +> **Date:** 2026-08-02. +> **Scope:** documentation-only architectural ruling. Records the *role* of the +> #878 persistence seam and the intended larger two-dimensional temporal +> architecture it bootstraps toward. Changes **no** Rust code, tests, public +> APIs, or `temporal.rs`. +> **Owns (narrowly):** "what #878 is and is NOT", the horizontal/vertical +> temporal split as it touches persistence, and the later +> `temporal.rs`-shadow + `revision.rs` correction path. +> **Does NOT own (cross-refs, never re-specifies):** +> - Horizontal temporal-stream detail → `temporal-markov-and-style-classes-v1.md` +> (the ratified 2026-07-10 arc; `E-MARKOV-TEMPORAL-STREAM-1`). +> - D-MBX-A6 deliverable tracking → `.claude/board/STATUS_BOARD.md` +> (D-MBX-A6-P1…P3e rows). +> - Per-row `write_row` cycle-gate → `mailbox-cycle-aware-write-contract-v1.md` +> (a *different* deliverable — the SoA setter gate, not the WAL seam). +> - The reshape ruling itself → `.claude/board/EPIPHANIES.md` +> `E-THE-DURABLE-UNIT-IS-THE-CYCLE-NOT-THE-CAST-ONE-WAL-WRITE-PER-SWEEP-1` +> and `E-THE-PAIRED-MOVE-MUST-BE-DURABLE-CO-LOCATED-NOT-IN-MEMORY-ONLY-1`. +> +> This is the bootstrap-and-upgrade companion, not a competing architecture. + +--- + +## 1. Current #878 role — a PRIMITIVE cycle/WAL bootstrap (deliberate) + +#878 deliberately establishes a **primitive** persistence seam so that thinking +can become *wired and runnable* without first solving the complete cognitive +topology. It is scaffolding that carries load, chosen so the execution and +durability plumbing exists and is exercised **before** the final temporal or +cognitive representation is settled. + +Its accepted responsibilities, end to end: + +``` +concurrent thought results + → primitive slot collection and ordering + → freeze one complete cycle + → one amortized WAL write + → one Lance DatasetVersion +``` + +The **hard guarantees** #878 makes (and only these): + +1. All work in a cycle reads exactly **one sealed predecessor version `Vn`**. +2. **Open-cycle results are not visible as `Vn` input** (an unsealed cycle is + excluded from the sealed read horizon). +3. The cycle is **frozen before durable I/O** (order + coalesce happen on a + detached snapshot, then the append runs). +4. One successful cycle seal produces **exactly one WAL write**. +5. One successful cycle seal produces **exactly one `DatasetVersion`**. +6. The WAL **never receives a live mutable SoA borrow** (the persistence path + holds no owner across I/O). + +### The scalar slot/order model is explicitly PROVISIONAL + +The current scalar slot + single-key ordering model is **sufficient** to +establish the execution and durability plumbing, and **nothing more**. It is +**not** claimed to be the final representation of temporal or cognitive order. +This document does **not** redesign or fix that limitation — recording the +boundary is the whole point. The upgrade path is §2–§3; the accepted debts are +§4. + +--- + +## 2. The intended larger architecture — two orthogonal dimensions + +The final architecture is two **orthogonal** dimensions. #878 supplies the +vertical axis primitively and leaves the horizontal axis to a later phase. + +### 2.1 Horizontal dimension — coherence within a frame (`temporal.rs`) + +`temporal.rs` preserves and reconstructs **horizontal causal and contextual +coherence** — the relationships *among* the results that make up one frame. + +For **deterministic streamed material** (e.g. the Bible), the initial +horizontal position is strongly determined by the source structure: + +``` +book → chapter → verse → token/span +``` + +Preserving this horizontal structure is required because it is what the local +awareness reads: + +``` +signed i4 neighbourhood pointers + → relative-pronoun and anaphora relations + → local grammatical and causal relationships + → coherent standing-wave awareness +``` + +For **medical knowledge and higher-order thought**, horizontal relationships +may **not** be monotonically ordered — they may form **concurrent or partially +ordered neighbourhoods**. The final temporal architecture must therefore be able +to resolve **more than a scalar total order**. (This is precisely why the #878 +scalar key is labelled provisional in §1.) + +> The horizontal-stream mechanism itself — the version-range read +> (`QueryReference::at(v, rung)` + deinterlace), the ±5-generalizing window, the +> `LocalCausalRow`/`local_trajectories` reconstruction — is owned by +> `temporal-markov-and-style-classes-v1.md` and is **not** re-specified here. +> This document only records *that* the horizontal dimension is `temporal.rs`'s +> responsibility and *why* a scalar key is insufficient for it. + +### 2.2 Vertical dimension — successive durable frames (`DatasetVersion`) + +Lance `DatasetVersion` is the **vertical** axis. Each successful cycle seal +advances the durable frame: + +``` +Vn → Vn+1 → Vn+2 +``` + +The version table therefore provides **both**: + +- **vertical cognitive/rung progression** (each sealed frame is one discrete + advance), and +- **cheap historical time-series lookup** (a version is selected without + replaying any landing). + +Consumers such as **Stockfish-style hindsight tests** can select versions from +the Lance version table because the frames have already been sealed as discrete +durable states — the lookup is over already-coherent frames. + +> The version table performs **vertical** frame succession and lookup **only**. +> It does **not** perform horizontal causal ordering — that is the horizontal +> dimension's job (§2.1). Conflating the two is the error this section forecloses. + +--- + +## 3. Planned temporal error-correction phase (later; not #878) + +The primitive #878 slot model may initially produce frames whose **horizontal +coherence is incomplete** — the scalar order captures *that* results happened, +not the full partial-order structure among them. + +A later phase may run `temporal.rs` as a **metacognitive / shadow correction +layer** over the produced frame. This layer **must not** require ractor actors +to wait synchronously for neighbours during thought execution — the emit path +stays wait-free. + +Intended direction: + +``` +actors emit without neighbour waits + → primitive cycle is collected + → temporal coherence is evaluated (shadow pass) + → inconsistencies become correction / revision input + → a LATER cycle carries the corrected interpretation +``` + +`revision.rs` is the **planned complementary mechanism** for reinterpretation, +circular reasoning, and correction from the reflective side. (Distinct from the +NARS *belief-revision* of policy atoms in `triangle-tenants-gestalt-separation-v1.md` +— that revises `LearnedStyle`; this `revision.rs` feeds justified *reinterpretation* +into subsequent cognition.) + +Conceptually, the three responsibilities stay separate: + +``` +temporal.rs detects or reconstructs horizontal coherence +revision.rs feeds justified correction into subsequent cognition +Lance versions preserve the successive durable vertical frames +``` + +**Iron constraint on the correction path:** a sealed historical version **must +not be silently rewritten** merely because a later metacognitive pass revised +its interpretation. Correction flows *forward* — into a later cycle / later +version — never *backward* over a sealed frame. (This is the vertical-axis +immutability that keeps the hindsight-lookup in §2.2 honest.) + +--- + +## 4. Known limitations accepted for #878 + +Recorded explicitly as **upgrade points**, not as claims that #878 already +solves the final temporal model: + +- The scalar slot model is **provisional**. +- A scalar key **may not capture** future partial-order cognition (§2.1). +- Cross-owner and equal-position **conflict semantics are not finalized**. +- Cycle **retry** and **production WAL idempotence** still require + concrete-sink hardening. +- The **fake WAL sink proves the contract shape, not real crash durability** + (`compile+test green ≠ storage proven`, the Ladybug lesson). +- The **complete horizontal temporal projection remains future work** (§2.1, + cross-ref `temporal-markov-and-style-classes-v1.md`). + +--- + +## 5. Scope exclusions (this document and the #878 bootstrap) + +- Do **not** introduce or document detailed **cohort internals**. +- Do **not** mention a **fixed number of cohort slots**. +- Do **not** design **actor-neighbour waiting or firing dependencies** (the + emit path is wait-free by §3). +- Do **not** invent new **semantic, temporal, rung, witness, branch, or + ancestry** types. +- Do **not** revive `ThoughtWitness`, `basis`, `awareness_seq`, or **per-cast + WAL persistence** (all retired by the reshape ruling; the durable unit is the + cycle, not the cast). +- The **cohort execution model** is out of scope — it belongs to the separate + cohort architecture work, not here. + +--- + +## 6. Status snapshot + +| Aspect | State | +|---|---| +| Cycle/WAL bootstrap seam (`persist_sink`) | **SHIPPED** in PR #878 (bootstrap; contract-probed, not storage-proven) | +| Vertical axis (`DatasetVersion` succession + lookup) | Established primitively by the seam | +| Horizontal axis (`temporal.rs` coherence over a frame) | **PLANNED** — owned by `temporal-markov-and-style-classes-v1.md` | +| Shadow temporal-coherence correction pass | **PLANNED** (§3) | +| `revision.rs` forward-correction mechanism | **PLANNED** (§3) | +| Concrete Lance sink (real crash durability) | **DEFERRED** — gated on crash falsifiers (§4) | + +The bootstrap exists so the rest can be built on a running, durable seam. The +scalar order is a load-bearing placeholder, and this document is the record that +it is a placeholder — nothing here claims the primitive ordering is the final +temporal architecture. diff --git a/crates/lance-graph-planner/src/lib.rs b/crates/lance-graph-planner/src/lib.rs index 2605836a..77e51672 100644 --- a/crates/lance-graph-planner/src/lib.rs +++ b/crates/lance-graph-planner/src/lib.rs @@ -78,6 +78,9 @@ pub mod batch_writer; // === D-MBX-A6 Outcome→KanbanMove emit adapter (bootstrap rebind + ahead-cast) === pub mod owner_adapter; +// === D-MBX-A6 persistence-sink ordering core (post-write kanbanstep; DurableWrite seam) === +pub mod persist_sink; + // === Internal API (same-binary, zero-serde) === pub mod api; diff --git a/crates/lance-graph-planner/src/persist_sink.rs b/crates/lance-graph-planner/src/persist_sink.rs new file mode 100644 index 00000000..4b7805c5 --- /dev/null +++ b/crates/lance-graph-planner/src/persist_sink.rs @@ -0,0 +1,1231 @@ +//! The D-MBX-A6 persistence sink — **one WAL write per cycle** (the amortization). +//! +//! This is the POST-write half of the fire-and-forget flow (pre-write half = +//! [`crate::owner_adapter`]). Its ONLY concerns are **storage** and that 64k +//! concurrent thoughts hit **no physical or epistemic race condition** (operator +//! ruling). It does not model semantics: semantic causality already lives in the +//! witness substrate (`causal_witness`, AriGraph SPO, NARS inference direction) — +//! this layer never invents a second causality graph. +//! +//! > The sweep globally aligns **durability**; `temporal.rs` locally aligns +//! > **order** before durability. One semantic generation = one globally-aligned +//! > 64k-row **cycle** that seals into exactly one [`DatasetVersion`]. +//! +//! ## Three levels — cast ⊂ chunk ⊂ cycle +//! +//! - **cast** — one logical thought staged into the open cycle (no WAL write). +//! - **chunk** — a disjoint / row-indexed slice of the cycle's final image. +//! - **cycle** — the 64k-row frame. Its **seal is ONE WAL write** → one version. +//! +//! Modelling the durable unit as one *cast* (a WAL write per thought) is the +//! anti-pattern this file corrects: it would pay WAL header + durability-wait + +//! version-metadata 64k times. Here the WAL is amortized — many logical thoughts +//! → one durable boundary → one published version. +//! +//! ## Physical race safety — WRITE-SIDE order, not read-time repair +//! +//! 64k thoughts run concurrently and finish in ARBITRARY CPU order. In +//! [`DetachedCycleBatch::freeze`], the casts are stable-ordered by their existing +//! `stream_position` via [`order_cycle_stably`] BEFORE the single WAL append — so +//! the durable image is already canonical. [`WalSink::scan_sealed`] returns rows in +//! that stored order and **never sorts**: completion order is fixed to canonical +//! order at WRITE time, not repaired on every read (which would lose the whole +//! amortization and hand every downstream reader the wrong weave first). The +//! ordering is write-path WAL preparation, so it lives here — NOT in `temporal.rs`, +//! which owns query-time reading. +//! +//! ## Epistemic race safety — the sealed read horizon +//! +//! Every thought in an open cycle reads only the sealed predecessor +//! (`frame.base_version` = `Vn`); the seal publishes `Vn+1` once. An open cycle's +//! output is **never visible as `Vn` input** — [`WalSink::scan_sealed`] excludes +//! any unsealed cycle. So no thought can read a concurrent sibling's in-flight +//! result as though it were sealed history (`read Vn / write Vn+1`). +//! +//! ## Coalescing — fold ordered updates into owned rows, then chunk +//! +//! The final cycle image is the ordered per-row fold of the staged updates (a +//! later stream position overwrites the same row); chunks are disjoint / mergeable +//! slices of that final image, NOT successive whole-image replacements. +//! +//! ## Scope boundary — storage only, no semantic / rung types minted here +//! +//! This layer owns storage + race-freedom, nothing else. It does NOT mint (or +//! carry) semantic-witness, ancestry, awareness, branch, rung, or temporal-policy +//! types — those live upstream (the semantic witness substrate) and downstream +//! (`temporal.rs` owns *query-time* reader-horizon / version reading). The +//! write-side ordering here is generic over a caller-supplied canonical key +//! ([`order_cycle_stably`]); the rung a cycle is aligned at is decided by whoever +//! schedules it, not modelled in [`CycleFrame`]. +//! +//! ## What the tests prove — CONTRACT probes, NOT crash/WAL integration (honest) +//! +//! The `#[cfg(test)]` `FakeWalSink` is in-process maps, not a WAL/Lance table. The +//! tests are **storage/race CONTRACT probes**: one WAL write per cycle, write-side +//! order under scrambled completion, per-row coalescing, no unsealed visibility, +//! sealed read horizon, recovery idempotence. They do **NOT** prove real +//! durability (no MemWAL, restart, atomic `RecordBatch`, or manifest). +//! **`compile+test green ≠ storage proven`** (the Ladybug lesson). **This module +//! builds NO concrete Lance sink.** + +use lance_graph_contract::kanban::{KanbanColumn, KanbanMove, RubiconTransitionError}; +use lance_graph_contract::scheduler::DatasetVersion; +use lance_graph_contract::soa_view::MailboxSoaOwner; + +pub use lance_graph_contract::collapse_gate::MailboxId; + +/// **Write-side stable ordering** — the loom before the WAL. Stable-order +/// concurrently-produced rows by a caller-supplied canonical key BEFORE the single +/// WAL append, so completion order never becomes storage order (physical race) and +/// no read-time repair is ever needed. Generic over the key until the repository +/// proves a canonical key type; equal keys keep arrival order (stable). +/// +/// **Note (the dense-key nuance):** when the key is already a dense slot index into +/// the cycle image, a stable scatter into predetermined positions is cheaper than a +/// general `O(n log n)` sort on the 64k path — that choice belongs beside the cycle +/// batch (here), not in `temporal.rs`. This general sort is the fallback form. +/// +/// This lives with the persistence cycle, NOT in `temporal.rs`: that module owns +/// query-time reader-horizon / version reading; this is write-path WAL preparation. +pub fn order_cycle_stably(rows: &mut [T], key: impl FnMut(&T) -> K) { + rows.sort_by_key(key); +} + +/// A cycle — one globally-aligned 64k-row sweep whose commit is ONE WAL append and +/// ONE [`DatasetVersion`]. A boring scheduling identity (monotonic), not a +/// semantic object. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct CycleId(pub u64); + +/// A cycle's storage identity: which cycle, and the sealed predecessor its thoughts +/// read (`Vn` — the epistemic read horizon; the commit publishes its successor). +/// Storage-only: it carries NO rung / projection / branch / semantic tags (those +/// are decided upstream, never minted in this layer). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CycleFrame { + pub cycle: CycleId, + /// The sealed predecessor every thought in this cycle reads (`Vn`). + pub base_version: DatasetVersion, +} + +impl CycleFrame { + /// A cycle reading `base_version` (`Vn`) as its sealed predecessor. + #[must_use] + pub fn new(cycle: CycleId, base_version: DatasetVersion) -> Self { + Self { + cycle, + base_version, + } + } +} + +/// A persistence **landing** record — WHERE a staged thought lands in the cycle. +/// Boring by design: it says where, never why. Semantic causality (ancestry, +/// evidence, revision) stays in the witness substrate — this record has no +/// `basis`. `stream_position` is the caller's EXISTING canonical order key, the +/// sole write-side ordering input. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SweepSlot { + pub cycle: CycleId, + /// The existing canonical (textual/stream) order key — the write-side + /// deinterlace input. NOT a new coordinate. + /// + /// **Contract: monotonically increasing per owner ACROSS cycles, not + /// per cycle.** [`recover_and_apply`] uses it as the durable watermark over a + /// multi-cycle [`WalSink::scan_sealed`] stream (`stream_position <= watermark` + /// ⇒ already applied), so a per-cycle restart to 0 would make recovery skip + /// every later-cycle landing at or below an earlier cycle's watermark. The + /// caller owns this monotonicity (it is the witness-fabric order key, already + /// monotonic); this layer does not re-key by `(CycleId, stream_position)`. + pub stream_position: u64, + /// The mailbox this landing is on behalf of. + pub owner: MailboxId, + /// The SoA row this update writes (coalescing folds same-row updates). + pub row: u64, + /// The lifecycle move this thought cast; applied post-SEAL only. `None` = a + /// no-step landing. + pub paired_move: Option, + /// The update payload — a descriptor (a `NodeRowPacket` slice in production; + /// bytes here). Never a semantic witness. + pub payload: Vec, +} + +/// A landing read back from a SEALED cycle — its slot plus the version its cycle +/// sealed into (shared by every landing of that cycle). Only ever produced for +/// SEALED cycles; returned in the stored (already canonical) order, NEVER re-sorted. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LandedSlot { + /// The version the cycle sealed into (the cheap coarse timeline). + pub version: DatasetVersion, + pub slot: SweepSlot, +} + +/// A durable write that did not land (the cycle seal failed / was fenced). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WriteFailed(pub String); + +impl std::fmt::Display for WriteFailed { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "cycle seal did not land: {}", self.0) + } +} +impl std::error::Error for WriteFailed {} + +/// A **detached, frozen, already-deinterlaced** cycle image — the loom's finished +/// fabric, ready for the single WAL append. Built by [`DetachedCycleBatch::freeze`] +/// (temporal deinterlace + per-row coalesce) BEFORE [`WalSink::commit_cycle`], so +/// the durable operation is a pure atomic append of an already-coherent frame. +/// Detached = a snapshot, never a borrow of live mutable SoA. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DetachedCycleBatch { + pub frame: CycleFrame, + /// Landings in canonical `stream_position` order (the temporal deinterlace + /// already ran — the WAL never sees worker-completion order). + pub landings: Vec, + /// The coalesced final image: `row -> last payload in stream order`. + pub image: std::collections::BTreeMap>, +} + +impl DetachedCycleBatch { + /// Freeze concurrently-produced casts into the ordered, coalesced cycle image: + /// (1) [`order_cycle_stably`] stable-orders by `stream_position` — + /// completion order never becomes storage order (physical race); (2) fold + /// same-row updates into owned rows (later stream position wins). The result is + /// detached from any live SoA and ready for exactly one WAL append. + #[must_use] + pub fn freeze(frame: CycleFrame, mut casts: Vec) -> Self { + order_cycle_stably(&mut casts, |s| s.stream_position); + let mut image = std::collections::BTreeMap::new(); + for s in &casts { + image.insert(s.row, s.payload.clone()); + } + Self { + frame, + landings: casts, + image, + } + } +} + +/// Why a persist / seal / step operation produced no lifecycle advance. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PersistError { + /// The cycle seal did not land — nothing published, so no step. This is the + /// RETRYABLE class (a fenced / failed append may succeed on retry). + Write(WriteFailed), + /// A cast was staged against a different cycle than the frame — a caller + /// programming error, PERMANENT and never retryable (distinct from [`Write`] + /// so retry logic never loops on it). [`Write`]: PersistError::Write + CycleMismatch { + cast_cycle: CycleId, + frame_cycle: CycleId, + }, + /// The paired move is not a legal Rubicon edge from the owner's current phase. + Illegal(RubiconTransitionError), + /// A move minted for a different mailbox than the landing's owner — refused so + /// a cross-owner move never becomes durable (write-on-behalf never crosses + /// owners), and again as a recovery corruption guard. + OwnerMismatch { + move_owner: MailboxId, + landing_owner: MailboxId, + }, + /// An above-watermark landing's move `from` does not match the owner's current + /// phase — a gap/corruption in the sealed replay chain (phase equality is a + /// corruption guard, NOT the idempotence key — the watermark is). + StalePhase { + owner_phase: KanbanColumn, + move_from: KanbanColumn, + }, +} + +impl std::fmt::Display for PersistError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Write(e) => write!(f, "{e}"), + Self::CycleMismatch { + cast_cycle, + frame_cycle, + } => write!( + f, + "cast cycle {cast_cycle:?} != frame cycle {frame_cycle:?}" + ), + Self::Illegal(e) => { + write!(f, "illegal Rubicon transition {:?} -> {:?}", e.from, e.to) + } + Self::OwnerMismatch { + move_owner, + landing_owner, + } => write!( + f, + "move for mailbox {move_owner} on a landing owned by {landing_owner}" + ), + Self::StalePhase { + owner_phase, + move_from, + } => write!( + f, + "stale/corrupt replay: owner at {owner_phase:?}, move.from {move_from:?}" + ), + } + } +} +impl std::error::Error for PersistError { + /// Expose the wrapped cause so callers can walk the error chain — the two + /// wrapping variants ([`Write`](PersistError::Write) / + /// [`Illegal`](PersistError::Illegal)) forward to their inner error; the + /// self-describing variants have no deeper cause. + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Write(e) => Some(e), + Self::Illegal(e) => Some(e), + Self::CycleMismatch { .. } | Self::OwnerMismatch { .. } | Self::StalePhase { .. } => { + None + } + } + } +} + +/// The WAL seam — **one durable append per cycle**. [`commit_cycle`](WalSink::commit_cycle) +/// takes an ALREADY-deinterlaced, ALREADY-frozen [`DetachedCycleBatch`] (the loom +/// ran before the WAL) and performs the single atomic append → one +/// [`DatasetVersion`], visible all-or-nothing. Reads return committed landings +/// only, in the STORED canonical order — this seam NEVER sorts (order is a +/// write-side property, fixed before the append). +/// +/// `&self` (shared) and **no owner borrow**: the persistence path never holds the +/// SoA owner across I/O. +/// +/// **Futures are not `Send`** (native `async fn` in trait, no `Send` bound): callers +/// drive these on the crate's single-task persistence path and must NOT +/// `tokio::spawn` them onto a multi-threaded runtime. If a concrete sink ever needs +/// to be spawned, give it explicit `impl Future + Send` methods there. +#[allow(async_fn_in_trait)] +pub trait WalSink { + /// The SINGLE amortized WAL append for a whole cycle: commit `batch` (already + /// ordered + coalesced) against the sealed predecessor `base`, publishing + /// exactly one new [`DatasetVersion`] atomically. This is the ONLY durable op — + /// 64k thoughts → one append, not 64k appends. `async` because the concrete WAL + /// append is async. + async fn commit_cycle( + &self, + base: DatasetVersion, + batch: DetachedCycleBatch, + ) -> Result; + + /// Read back COMMITTED landings (never an uncommitted cycle), in the STORED + /// canonical order — this seam does NOT sort. Optionally only cycles after + /// `from_version`. + async fn scan_sealed( + &self, + from_version: Option, + ) -> Result, WriteFailed>; + + /// The CHEAP coarse timeline: `(CycleId, DatasetVersion)` per committed cycle, + /// WITHOUT replaying any landing. The read a downstream time-series consumer + /// (e.g. `stockfish-rs`, another session) does — a lookup of the version table + /// over already-coherent frames. + async fn versions(&self) -> Result, WriteFailed>; +} + +/// Deinterlace + freeze a whole cycle's casts and commit it in ONE WAL append. +/// The loom ([`order_cycle_stably`]) and the freeze run in +/// [`DetachedCycleBatch::freeze`] BEFORE the single [`WalSink::commit_cycle`], so +/// completion order never reaches storage. Rejects a cross-owner move (and a +/// cycle-id mismatch) before the batch is built. Returns the one published version. +pub async fn persist_cycle( + sink: &S, + frame: CycleFrame, + casts: Vec, +) -> Result { + for c in &casts { + if c.cycle != frame.cycle { + // Permanent caller error — NOT a retryable Write (retry would loop). + return Err(PersistError::CycleMismatch { + cast_cycle: c.cycle, + frame_cycle: frame.cycle, + }); + } + if let Some(mv) = c.paired_move { + if mv.mailbox != c.owner { + return Err(PersistError::OwnerMismatch { + move_owner: mv.mailbox, + landing_owner: c.owner, + }); + } + } + } + // Loom-before-WAL: order + coalesce + detach, THEN one atomic append. + let batch = DetachedCycleBatch::freeze(frame, casts); + sink.commit_cycle(frame.base_version, batch) + .await + .map_err(PersistError::Write) +} + +/// The result of a [`recover_and_apply`] pass: the moves applied this pass, and the +/// new stream-position **watermark** the caller must persist WITH the owner's +/// sealed cycle state so the next recovery is idempotent. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Recovered { + pub applied: Vec, + /// The highest `stream_position` now accounted for (per owner), or the input + /// watermark when nothing was applied. Same cross-cycle scope as + /// [`SweepSlot::stream_position`] — monotonic per owner across sealed cycles, + /// so it is a durable watermark over the whole multi-cycle `scan_sealed` stream, + /// not a per-cycle counter. + pub watermark: Option, +} + +/// **Crash recovery — post-SEAL only.** Given SEALED landings read back via +/// [`WalSink::scan_sealed`] (an unsealed cycle is never here — steps are eligible +/// only after the seal), an `owner` reconstructed at its durable phase, and the +/// durable `applied_through` watermark (`stream_position`) persisted alongside that +/// phase, re-apply the owner's PENDING tail in canonical `stream_position` order. +/// +/// The landings arrive already ordered (write-side deinterlace at seal), so no +/// sort is done here. The watermark — not phase equality — is the idempotence key +/// (the Rubicon lifecycle is cyclic). Above the watermark the chain must be +/// contiguous: a non-matching `from` is a gap/corruption ([`PersistError::StalePhase`]). +/// +/// **On error the accumulated [`Recovered`] is returned WITH the error** +/// (`Err((partial, cause))`). The function mutates `owner` as it walks the chain, +/// so a mid-chain failure has already advanced the owner's phase for the applied +/// prefix; returning the partial lets the caller persist the watermark it earned. +/// Discarding it would leave the watermark below the applied prefix, replaying those +/// moves against an already-advanced owner (a permanent `StalePhase` stall in the +/// acyclic case, a double-apply in a cyclic lap). +pub fn recover_and_apply( + owner: &mut O, + sealed: &[LandedSlot], + applied_through: Option, +) -> Result { + let mut applied = Vec::new(); + let mut watermark = applied_through; + let me = owner.mailbox_id(); + // Already in canonical stream order from the write side; filter to this owner. + for ls in sealed.iter().filter(|ls| ls.slot.owner == me) { + if applied_through.is_some_and(|hw| ls.slot.stream_position <= hw) { + continue; + } + match ls.slot.paired_move { + None => watermark = Some(ls.slot.stream_position), + Some(mv) => { + if mv.mailbox != me { + return Err(( + Recovered { applied, watermark }, + PersistError::OwnerMismatch { + move_owner: mv.mailbox, + landing_owner: me, + }, + )); + } + if mv.from != owner.phase() { + return Err(( + Recovered { applied, watermark }, + PersistError::StalePhase { + owner_phase: owner.phase(), + move_from: mv.from, + }, + )); + } + match owner.try_advance_phase(mv.to) { + Ok(step) => { + applied.push(step); + watermark = Some(ls.slot.stream_position); + } + Err(e) => { + return Err((Recovered { applied, watermark }, PersistError::Illegal(e))) + } + } + } + } + } + Ok(Recovered { applied, watermark }) +} + +#[cfg(test)] +mod tests { + use super::*; + use lance_graph_contract::kanban::{ExecTarget, KanbanColumn}; + use lance_graph_contract::soa_view::{MailboxSoaOwner, MailboxSoaView}; + use std::collections::BTreeMap; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::Mutex; + + // ── Minimal in-RAM owner ──────────────────────────────────────────────────── + struct FakeOwner { + id: MailboxId, + phase: KanbanColumn, + cycle: u32, + } + impl MailboxSoaView for FakeOwner { + fn mailbox_id(&self) -> MailboxId { + self.id + } + fn n_rows(&self) -> usize { + 0 + } + fn w_slot(&self) -> u8 { + 0 + } + fn current_cycle(&self) -> u32 { + self.cycle + } + fn phase(&self) -> KanbanColumn { + self.phase + } + fn energy(&self) -> &[f32] { + &[] + } + fn edges_raw(&self) -> &[u64] { + &[] + } + fn meta_raw(&self) -> &[u32] { + &[] + } + fn entity_type(&self) -> &[u16] { + &[] + } + } + impl MailboxSoaOwner for FakeOwner { + fn advance_phase(&mut self, to: KanbanColumn) -> KanbanMove { + let from = self.phase; + self.phase = to; + self.cycle = self.cycle.wrapping_add(1); + KanbanMove { + mailbox: self.id, + from, + to, + witness_chain_position: self.cycle, + exec: ExecTarget::Native, + } + } + } + fn owner(phase: KanbanColumn) -> FakeOwner { + FakeOwner { + id: 42, + phase, + cycle: 5, + } + } + + // ── The WAL sink fake ──────────────────────────────────────────────────────── + struct SealedCycle { + frame: CycleFrame, + version: DatasetVersion, + /// The landings AS COMMITTED (already write-side ordered before the append). + landings: Vec, + /// The coalesced final image: row -> last value in stream order. + image: BTreeMap>, + } + struct FakeWalSink { + succeed: bool, + sealed: Mutex>, + next_version: AtomicU64, + /// The number of physical WAL appends — must be ONE per committed cycle. + wal_writes: AtomicU64, + } + impl FakeWalSink { + fn new() -> Self { + Self { + succeed: true, + sealed: Mutex::new(Vec::new()), + next_version: AtomicU64::new(1), + wal_writes: AtomicU64::new(0), + } + } + fn failing() -> Self { + Self { + succeed: false, + ..Self::new() + } + } + fn wal_writes(&self) -> u64 { + self.wal_writes.load(Ordering::SeqCst) + } + fn version_count(&self) -> usize { + self.sealed.lock().unwrap().len() + } + fn image_of(&self, cycle: CycleId) -> Option>> { + self.sealed + .lock() + .unwrap() + .iter() + .find(|s| s.frame.cycle == cycle) + .map(|s| s.image.clone()) + } + /// Test-only: inject a committed cycle whose landings are DELIBERATELY out + /// of order, to prove `scan_sealed` does not re-sort (order is a write-side + /// property, fixed before the append — never a read-time repair). + fn inject_unordered_committed(&self, frame: CycleFrame, landings: Vec) { + let v = DatasetVersion(self.next_version.fetch_add(1, Ordering::SeqCst)); + self.sealed.lock().unwrap().push(SealedCycle { + frame, + version: v, + landings, + image: BTreeMap::new(), + }); + } + } + impl WalSink for FakeWalSink { + async fn commit_cycle( + &self, + base: DatasetVersion, + batch: DetachedCycleBatch, + ) -> Result { + if !self.succeed { + return Err(WriteFailed("cycle fenced".into())); + } + let mut sealed = self.sealed.lock().unwrap(); + // Optimistic-concurrency fence: a commit MUST target the current sealed + // head (`Vn`). A stale/in-flight `base` (a sibling that read an older + // predecessor) is rejected — the epistemic horizon enforced, not assumed. + let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); + if base != head { + return Err(WriteFailed(format!( + "stale base {base:?}: sealed head is {head:?}" + ))); + } + // The batch arrives ALREADY deinterlaced + coalesced (loom before WAL). + // THE single amortized WAL append for the whole cycle. + self.wal_writes.fetch_add(1, Ordering::SeqCst); + let version = DatasetVersion(self.next_version.fetch_add(1, Ordering::SeqCst)); + sealed.push(SealedCycle { + frame: batch.frame, + version, + landings: batch.landings, + image: batch.image, + }); + Ok(version) + } + async fn scan_sealed( + &self, + from_version: Option, + ) -> Result, WriteFailed> { + // Returned in STORED order — no sort. (The order was fixed at seal.) + Ok(self + .sealed + .lock() + .unwrap() + .iter() + .filter(|s| from_version.is_none_or(|f| s.version > f)) + .flat_map(|s| { + s.landings.iter().map(|slot| LandedSlot { + version: s.version, + slot: slot.clone(), + }) + }) + .collect()) + } + async fn versions(&self) -> Result, WriteFailed> { + Ok(self + .sealed + .lock() + .unwrap() + .iter() + .map(|s| (s.frame.cycle, s.version)) + .collect()) + } + } + + fn mv(owner: MailboxId, from: KanbanColumn, to: KanbanColumn) -> KanbanMove { + KanbanMove { + mailbox: owner, + from, + to, + witness_chain_position: 0, + exec: ExecTarget::Elixir, + } + } + /// A landing for `owner` at `stream_position`, moving `from→to`, writing `row`. + fn slot( + owner: MailboxId, + cycle: u64, + stream_position: u64, + row: u64, + m: Option<(KanbanColumn, KanbanColumn)>, + ) -> SweepSlot { + SweepSlot { + cycle: CycleId(cycle), + stream_position, + owner, + row, + paired_move: m.map(|(f, t)| mv(owner, f, t)), + payload: vec![stream_position as u8], + } + } + + // ── FALSIFIER (headline): one WAL write per cycle — the amortization ───────── + #[tokio::test] + async fn a_whole_cycle_of_casts_is_one_wal_write_one_version() { + let sink = FakeWalSink::new(); + let frame = CycleFrame::new(CycleId(1), DatasetVersion(0)); + // 100 concurrent thoughts stage into an owned cast vector — building it + // touches NO WAL (staging is the caller's, not the sink's). + let casts: Vec = (0..100u64).map(|i| slot(42, 1, i, i, None)).collect(); + assert_eq!( + sink.wal_writes(), + 0, + "staging writes no WAL — pure amortization" + ); + assert_eq!(sink.version_count(), 0, "no version before the seal"); + // persist_cycle is the SINGLE amortized WAL write for the whole cycle. + let version = persist_cycle(&sink, frame, casts).await.unwrap(); + assert_eq!(sink.wal_writes(), 1, "100 casts → exactly ONE WAL write"); + assert_eq!(version, DatasetVersion(1), "→ exactly one version"); + assert_eq!(sink.version_count(), 1); + } + + // ── FALSIFIER: write-side order under scrambled completion (physical race) ─── + #[tokio::test] + async fn scrambled_completion_lands_in_canonical_stream_order_at_write_time() { + let sink = FakeWalSink::new(); + // Thoughts "finish" (stage) in scrambled CPU order: stream 2, 0, 3, 1. + let sealed = persist_cycle( + &sink, + CycleFrame::new(CycleId(5), DatasetVersion(0)), + vec![ + slot(42, 5, 2, 20, None), + slot(42, 5, 0, 10, None), + slot(42, 5, 3, 30, None), + slot(42, 5, 1, 11, None), + ], + ) + .await + .unwrap(); + // scan_sealed does NOT sort; the order came from the WRITE side (seal). + let order: Vec = sink + .scan_sealed(None) + .await + .unwrap() + .iter() + .filter(|l| l.version == sealed) + .map(|l| l.slot.stream_position) + .collect(); + assert_eq!( + order, + vec![0, 1, 2, 3], + "canonical stream order was fixed at WRITE time, not repaired on read", + ); + } + + /// The complement: `scan_sealed` itself performs NO sort — order is purely a + /// write-side property. Inject a deliberately out-of-order sealed cycle and + /// prove scan returns it AS-STORED (a read-time sort would "fix" it). + #[tokio::test] + async fn scan_sealed_does_not_repair_order_on_read() { + let sink = FakeWalSink::new(); + sink.inject_unordered_committed( + CycleFrame::new(CycleId(9), DatasetVersion(0)), + vec![ + slot(42, 9, 3, 30, None), + slot(42, 9, 1, 10, None), + slot(42, 9, 2, 20, None), + ], + ); + let order: Vec = sink + .scan_sealed(None) + .await + .unwrap() + .iter() + .map(|l| l.slot.stream_position) + .collect(); + assert_eq!( + order, + vec![3, 1, 2], + "scan_sealed returns stored order verbatim — it never sorts (order is write-side)", + ); + } + + // ── FALSIFIER: per-row coalescing (not last-chunk-wins) ────────────────────── + #[tokio::test] + async fn same_row_updates_coalesce_distinct_rows_survive() { + let sink = FakeWalSink::new(); + persist_cycle( + &sink, + CycleFrame::new(CycleId(3), DatasetVersion(0)), + vec![ + // Two updates to ROW 7 (stream 0 then 2) — coalesce to the later. + SweepSlot { + payload: vec![0xA0], + ..slot(42, 3, 0, 7, None) + }, + // A different ROW 8 — survives independently. + SweepSlot { + payload: vec![0xB0], + ..slot(42, 3, 1, 8, None) + }, + SweepSlot { + payload: vec![0xA1], + ..slot(42, 3, 2, 7, None) + }, + ], + ) + .await + .unwrap(); + let image = sink.image_of(CycleId(3)).unwrap(); + assert_eq!(image.len(), 2, "two distinct rows in the final image"); + assert_eq!( + image.get(&7), + Some(&vec![0xA1]), + "row 7 coalesced to the later update" + ); + assert_eq!(image.get(&8), Some(&vec![0xB0]), "row 8 survived, disjoint"); + } + + // ── FALSIFIER: no partial visibility — an unsealed cycle is invisible ──────── + #[tokio::test] + async fn an_unsealed_cycle_is_invisible_epistemic_horizon() { + let sink = FakeWalSink::new(); + // A cycle's casts are staged in an owned vector held by the caller — NOT in + // the sink. Until the single atomic commit_cycle runs, nothing is visible: + // an open cycle's output is NOT readable as Vn input (read Vn / write Vn+1). + let casts = vec![slot( + 42, + 9, + 0, + 0, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + )]; + assert_eq!( + sink.scan_sealed(None).await.unwrap().len(), + 0, + "before the seal, no landing is visible", + ); + assert!( + sink.versions().await.unwrap().is_empty(), + "no version before seal" + ); + // The seal is all-or-nothing: the whole cycle appears at once, never partially. + persist_cycle(&sink, CycleFrame::new(CycleId(9), DatasetVersion(0)), casts) + .await + .unwrap(); + assert_eq!( + sink.scan_sealed(None).await.unwrap().len(), + 1, + "visible only after seal" + ); + } + + // ── FALSIFIER: sealed read horizon — a cycle reads the sealed predecessor ──── + #[tokio::test] + async fn a_cycle_reads_the_sealed_predecessor_not_an_in_flight_sibling() { + let sink = FakeWalSink::new(); + // Cycle 1 seals → V1 (the sealed head advances to V1). + let s1 = persist_cycle( + &sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![slot(42, 1, 0, 0, None)], + ) + .await + .unwrap(); + assert_eq!(s1, DatasetVersion(1)); + // A sibling that read the STALE predecessor V0 (an in-flight base, not the + // sealed head V1) is FENCED — the sink rejects the commit, proving the + // horizon is enforced, not merely restated by the caller. + let stale = persist_cycle( + &sink, + CycleFrame::new(CycleId(2), DatasetVersion(0)), + vec![slot(99, 2, 0, 0, None)], + ) + .await; + assert!( + matches!(stale, Err(PersistError::Write(_))), + "committing against a stale/in-flight base is fenced, not silently accepted", + ); + assert_eq!( + sink.version_count(), + 1, + "the fenced commit published nothing" + ); + // Committing against the SEALED head V1 succeeds and publishes exactly V2. + let s2 = persist_cycle( + &sink, + CycleFrame::new(CycleId(2), s1), + vec![slot(99, 2, 0, 0, None)], + ) + .await + .unwrap(); + assert_eq!(s2, DatasetVersion(2), "and publishes exactly its own Vn+1"); + } + + // ── FALSIFIER: cheap time-series — version table only, no landing replay ───── + #[tokio::test] + async fn downstream_time_series_reads_the_version_table_not_the_landings() { + let sink = FakeWalSink::new(); + for c in 1..=3u64 { + persist_cycle( + &sink, + CycleFrame::new(CycleId(c), DatasetVersion(c - 1)), + vec![slot(42, c, 0, 0, None)], + ) + .await + .unwrap(); + } + assert_eq!( + sink.versions().await.unwrap(), + vec![ + (CycleId(1), DatasetVersion(1)), + (CycleId(2), DatasetVersion(2)), + (CycleId(3), DatasetVersion(3)), + ], + "history scales with CYCLES; a time-series consumer just looks up the version table", + ); + } + + // ── Recovery over sealed landings (post-seal only) ─────────────────────────── + #[tokio::test] + async fn recovery_replays_an_owners_chain_in_stream_order_skipping_others() { + let sink = FakeWalSink::new(); + persist_cycle( + &sink, + CycleFrame::new(CycleId(4), DatasetVersion(0)), + vec![ + slot( + 42, + 4, + 0, + 0, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + ), + slot( + 99, + 4, + 1, + 1, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + ), + slot( + 42, + 4, + 2, + 2, + Some((KanbanColumn::CognitiveWork, KanbanColumn::Evaluation)), + ), + ], + ) + .await + .unwrap(); + let sealed = sink.scan_sealed(None).await.unwrap(); + let mut o = owner(KanbanColumn::Planning); + let rec = recover_and_apply(&mut o, &sealed, None).unwrap(); + assert_eq!( + rec.applied.iter().map(|m| m.to).collect::>(), + vec![KanbanColumn::CognitiveWork, KanbanColumn::Evaluation], + "owner 42's chain in stream order — owner 99's interleaved landing skipped", + ); + assert_eq!(o.phase(), KanbanColumn::Evaluation); + } + + #[tokio::test] + async fn cyclic_recovery_is_idempotent_only_with_the_watermark() { + assert_eq!(KanbanColumn::Plan.next_phases(), &[KanbanColumn::Planning]); + let sink = FakeWalSink::new(); + persist_cycle( + &sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![ + slot( + 42, + 1, + 0, + 0, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + ), + slot( + 42, + 1, + 1, + 1, + Some((KanbanColumn::CognitiveWork, KanbanColumn::Evaluation)), + ), + slot( + 42, + 1, + 2, + 2, + Some((KanbanColumn::Evaluation, KanbanColumn::Plan)), + ), + slot( + 42, + 1, + 3, + 3, + Some((KanbanColumn::Plan, KanbanColumn::Planning)), + ), + ], + ) + .await + .unwrap(); + let sealed = sink.scan_sealed(None).await.unwrap(); + let mut o = owner(KanbanColumn::Planning); + + let rec = recover_and_apply(&mut o, &sealed, None).unwrap(); + assert_eq!(rec.applied.len(), 4, "the whole lap applied once"); + assert_eq!( + o.phase(), + KanbanColumn::Planning, + "back at Planning after a lap" + ); + + let good = recover_and_apply(&mut o, &sealed, rec.watermark).unwrap(); + assert!( + good.applied.is_empty(), + "watermark makes cyclic recovery idempotent" + ); + + let bug = recover_and_apply(&mut o, &sealed, None).unwrap(); + assert_eq!( + bug.applied.len(), + 4, + "without the watermark the cyclic lap replays" + ); + } + + // ── FALSIFIER: a no-step landing advances the watermark past itself ────────── + #[tokio::test] + async fn a_no_step_landing_advances_the_watermark_and_is_not_re_scanned() { + // A mixed chain: step@0, a NO-STEP landing@1, step@2. If the no-step + // landing did not advance the watermark, it would be re-scanned forever and + // block the watermark behind every following step. + let sink = FakeWalSink::new(); + persist_cycle( + &sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![ + slot( + 42, + 1, + 0, + 0, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + ), + slot(42, 1, 1, 1, None), // no-step landing at position 1 + slot( + 42, + 1, + 2, + 2, + Some((KanbanColumn::CognitiveWork, KanbanColumn::Evaluation)), + ), + ], + ) + .await + .unwrap(); + let sealed = sink.scan_sealed(None).await.unwrap(); + let mut o = owner(KanbanColumn::Planning); + let rec = recover_and_apply(&mut o, &sealed, None).unwrap(); + assert_eq!(rec.applied.len(), 2, "the two real steps applied"); + assert_eq!( + rec.watermark, + Some(2), + "the watermark reaches the last landing, past the no-step at 1", + ); + // A second pass with that watermark applies nothing (no re-scan of the no-step). + let again = recover_and_apply(&mut o, &sealed, rec.watermark).unwrap(); + assert!(again.applied.is_empty(), "idempotent — nothing re-applied"); + } + + // ── FALSIFIER: a mid-chain failure returns the watermark it already earned ─── + #[tokio::test] + async fn a_mid_chain_failure_returns_the_applied_prefix_watermark() { + // step@0 is valid and advances the owner; step@1 has a `from` that no longer + // matches (corruption) → StalePhase. The Err must carry the partial Recovered + // so the caller can persist the watermark for the prefix that DID apply. + let sink = FakeWalSink::new(); + persist_cycle( + &sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![ + slot( + 42, + 1, + 0, + 0, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + ), + // Above-watermark landing whose `from` (Planning) mismatches the + // owner's now-current phase (CognitiveWork) → StalePhase. + slot( + 42, + 1, + 1, + 1, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + ), + ], + ) + .await + .unwrap(); + let sealed = sink.scan_sealed(None).await.unwrap(); + let mut o = owner(KanbanColumn::Planning); + let (partial, err) = + recover_and_apply(&mut o, &sealed, None).expect_err("the mid-chain mismatch must fail"); + assert!(matches!(err, PersistError::StalePhase { .. })); + assert_eq!( + partial.applied.len(), + 1, + "step 0 applied before the failure" + ); + assert_eq!( + partial.watermark, + Some(0), + "the returned watermark covers exactly the applied prefix (position 0)", + ); + assert_eq!( + o.phase(), + KanbanColumn::CognitiveWork, + "the owner's phase advanced for the applied prefix — the reason the \ + partial watermark must be persistable", + ); + } + + // ── FALSIFIER: the watermark spans cycles (stream_position is cross-cycle) ─── + #[tokio::test] + async fn recovery_watermark_spans_multiple_sealed_cycles() { + // stream_position is monotonic per owner ACROSS cycles: cycle 1 carries + // position 0, cycle 2 carries position 1. Recovery over the multi-cycle + // scan_sealed stream must treat the watermark as cross-cycle. + let sink = FakeWalSink::new(); + persist_cycle( + &sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![slot( + 42, + 1, + 0, + 0, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + )], + ) + .await + .unwrap(); + persist_cycle( + &sink, + CycleFrame::new(CycleId(2), DatasetVersion(1)), + vec![slot( + 42, + 2, + 1, + 1, + Some((KanbanColumn::CognitiveWork, KanbanColumn::Evaluation)), + )], + ) + .await + .unwrap(); + let sealed = sink.scan_sealed(None).await.unwrap(); + assert_eq!(sealed.len(), 2, "landings from BOTH sealed cycles"); + + // A fresh recovery applies the whole cross-cycle chain; watermark = 1. + let mut o = owner(KanbanColumn::Planning); + let rec = recover_and_apply(&mut o, &sealed, None).unwrap(); + assert_eq!( + rec.applied.iter().map(|m| m.to).collect::>(), + vec![KanbanColumn::CognitiveWork, KanbanColumn::Evaluation], + "both cycles' steps applied in cross-cycle stream order", + ); + assert_eq!( + rec.watermark, + Some(1), + "watermark reaches the cycle-2 landing" + ); + + // A watermark from cycle 1 (position 0) skips ONLY cycle 1, still applies + // cycle 2 — proving the watermark is not per-cycle. + let mut o2 = owner(KanbanColumn::CognitiveWork); + let rec2 = recover_and_apply(&mut o2, &sealed, Some(0)).unwrap(); + assert_eq!( + rec2.applied.iter().map(|m| m.to).collect::>(), + vec![KanbanColumn::Evaluation], + "cycle-1 landing skipped by watermark 0; cycle-2 landing still applied", + ); + } + + // ── FALSIFIER: a cast for the wrong cycle is a PERMANENT error, not Write ──── + #[tokio::test] + async fn a_cast_for_the_wrong_cycle_is_a_permanent_cycle_mismatch_not_write() { + // A cast whose cycle != frame.cycle is a caller programming error — it must + // NOT surface as the RETRYABLE Write class (retry would loop forever). + let sink = FakeWalSink::new(); + let r = persist_cycle( + &sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![slot(42, 2, 0, 0, None)], // cast cycle 2 ≠ frame cycle 1 + ) + .await; + assert!( + matches!( + r, + Err(PersistError::CycleMismatch { + cast_cycle: CycleId(2), + frame_cycle: CycleId(1), + }), + ), + "wrong-cycle cast is CycleMismatch (permanent), never Write (retryable)", + ); + assert_eq!( + sink.wal_writes(), + 0, + "nothing written on a wrong-cycle cast" + ); + } + + #[tokio::test] + async fn a_cross_owner_move_is_rejected_before_the_cycle_persists() { + let bad = SweepSlot { + paired_move: Some(mv(99, KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + ..slot(42, 1, 0, 0, None) + }; + let sink = FakeWalSink::new(); + assert!(matches!( + persist_cycle( + &sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![bad] + ) + .await, + Err(PersistError::OwnerMismatch { + move_owner: 99, + landing_owner: 42 + }), + )); + assert_eq!(sink.wal_writes(), 0, "nothing written on a bad cast"); + assert_eq!(sink.version_count(), 0); + } + + #[tokio::test] + async fn a_fenced_cycle_writes_nothing() { + let sink = FakeWalSink::failing(); + let r = persist_cycle( + &sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![slot( + 42, + 1, + 0, + 0, + Some((KanbanColumn::Planning, KanbanColumn::CognitiveWork)), + )], + ) + .await; + assert!(matches!(r, Err(PersistError::Write(_)))); + assert_eq!(sink.wal_writes(), 0, "no WAL write, no version, no step"); + assert_eq!(sink.scan_sealed(None).await.unwrap().len(), 0); + } + + #[test] + fn cycle_frame_is_storage_only() { + // The frame carries ONLY storage identity — no rung / projection / branch + // / semantic tags are minted in this layer (they live upstream). + let f = CycleFrame::new(CycleId(1), DatasetVersion(7)); + assert_eq!(f.cycle, CycleId(1)); + assert_eq!(f.base_version, DatasetVersion(7)); + } + + #[test] + fn order_cycle_stably_is_stable_and_generic() { + // Scrambled → ordered by the supplied key; equal keys keep arrival order. + let mut rows = vec![(3u64, "d"), (1u64, "b1"), (2u64, "c"), (1u64, "b2")]; + order_cycle_stably(&mut rows, |r| r.0); + assert_eq!(rows, vec![(1, "b1"), (1, "b2"), (2, "c"), (3, "d")]); + } +} diff --git a/crates/lance-graph-planner/src/temporal.rs b/crates/lance-graph-planner/src/temporal.rs index 65874d31..8e6ea722 100644 --- a/crates/lance-graph-planner/src/temporal.rs +++ b/crates/lance-graph-planner/src/temporal.rs @@ -43,6 +43,30 @@ //! so the single-server body can ignore them and the cross-server policy (peer- //! Raft / cluster bus) wakes them up with **no breaking signature change** — //! avoiding the `emitted_at_millis: u64` (decision #4) non-`Option` trap. +//! +//! ## Two temporal LAYERS (operator-ruled — do not conflate) +//! +//! `temporal.rs` deinterlaces on **two distinct axes**, and both are needed +//! before a durable kanban step can be trusted after a crash: +//! +//! - **Layer 1 — CAUSAL deinterlacing** ([`local_trajectories`], +//! [`LocalCausalRow`]): the global durable log interleaves *every* owner's +//! writes — `A@s0, C@s0, B@s0, A@s1`. Splitting it back into per-owner LOCAL +//! chains — owner A's own `[A@s0, A@s1]`, with B's and C's rows *removed* from +//! A's timeline — is what makes crash-replay in cast order possible. Answers +//! *"what is owner A's own trajectory?"* +//! - **Layer 2 — EPISTEMIC projection** ([`classify`], [`deinterlace`]): given a +//! trajectory, which rows may a reader at a given rung/horizon dispatch on +//! (`Contemporary` / `Anachronistic` / `Spoiler` / `Unknowable`). Answers +//! *"what may this reader know?"* +//! +//! They compose: layer 1 reconstructs an owner's local causal chain from the +//! interleaved log; layer 2 filters that chain by the reader's horizon. A crash +//! recovery reads the durable witnesses, runs layer 1 to find each owner's +//! pending tail, and re-applies in cast order (`persist_sink::recover_and_apply`). + +use lance_graph_contract::collapse_gate::MailboxId; +use std::collections::BTreeMap; /// A Lance dataset version — the storage frame's clock tick. pub type LanceVersion = u64; @@ -351,6 +375,78 @@ where out } +// ───────────────────────────────────────────────────────────────────────────── +// LAYER 1 — CAUSAL DEINTERLACING (global interleaved durable log → per-owner +// local trajectories). Distinct from the layer-2 epistemic projection above: +// layer 2 asks *what a reader may see*; layer 1 asks *what one owner actually +// did, in its own order*. The interleaved global log +// A@s0, C@s0, B@s0, A@s1 +// yields owner A's LOCAL chain [A@s0, A@s1] — B's and C's rows are REMOVED from +// A's timeline (deinterlaced away), not merely reordered. This is the read +// crash-recovery runs to find each owner's pending durable tail. +// ───────────────────────────────────────────────────────────────────────────── + +/// A row in the global durable log carrying its owner-local causal coordinates — +/// WHO wrote it and WHERE in that owner's own cast sequence. +/// +/// The layer-1 counterpart of [`DeinterlaceRow`]: where `DeinterlaceRow` exposes +/// the *epistemic* clocks (lance version / knowable-from / HLC), this exposes the +/// *causal* coordinates (owner + cast sequence) that split the interleaved stream +/// into per-owner local chains. A durable persistence witness +/// (`persist_sink::DurableWitness`) is the production implementor. +pub trait LocalCausalRow { + /// The mailbox that owns this row — the deinterlace grouping key. + fn owner(&self) -> MailboxId; + /// The owner-local ordering key WITHIN one owner's trajectory. Cross-owner + /// values are never compared; only rows sharing an `owner()` are ordered + /// against each other. + /// + /// PRECONDITION: **unique and monotonic per owner, ACROSS process restarts.** + /// [`local_trajectories`]'s sort is stable, so two rows of one owner sharing a + /// `cast_seq` keep their global arrival order — a tie-break that makes replay + /// order depend on scan order. A per-process counter (e.g. `BatchWriter`'s + /// `CastId`, which resets to 0 on restart) is therefore NOT a valid key: + /// different writer lifetimes collide. Use a durable-log position instead + /// (`persist_sink::LandedWitness` keys on `DurableCoordinate::log_order`, the + /// WAL entry position, which never resets). + fn cast_seq(&self) -> u64; +} + +/// Layer-1 causal deinterlacing: split a globally-interleaved durable stream into +/// per-owner LOCAL trajectories, each ordered by the owner's own `cast_seq`. +/// +/// The global order `A@s0, C@s0, B@s0, A@s1` yields +/// `{A: [A@s0, A@s1], B: [B@s0], C: [C@s0]}` — every owner's interleaved +/// neighbours are removed from its timeline, and each timeline is put back into +/// the owner's own cast order (the crash-replay order). Returns a `BTreeMap` so +/// owners iterate deterministically. +#[must_use] +pub fn local_trajectories(global: &[R]) -> BTreeMap> { + let mut by_owner: BTreeMap> = BTreeMap::new(); + for row in global { + by_owner.entry(row.owner()).or_default().push(row.clone()); + } + for rows in by_owner.values_mut() { + rows.sort_by_key(LocalCausalRow::cast_seq); + } + by_owner +} + +/// Owner `owner`'s local trajectory alone — the crash-recovery read for ONE +/// owner. Filters the interleaved global stream to that owner's rows and orders +/// them by `cast_seq`. Equivalent to `local_trajectories(global).remove(&owner)` +/// but without materialising the other owners' chains. +#[must_use] +pub fn local_trajectory_of(global: &[R], owner: MailboxId) -> Vec { + let mut chain: Vec = global + .iter() + .filter(|r| r.owner() == owner) + .cloned() + .collect(); + chain.sort_by_key(LocalCausalRow::cast_seq); + chain +} + #[cfg(test)] mod tests { use super::*; @@ -403,6 +499,117 @@ mod tests { } } + /// A minimal [`LocalCausalRow`] — owner + owner-local cast sequence, plus a + /// `tag` to make the surviving order observable. + #[derive(Clone, Debug, PartialEq)] + struct CausalRow { + owner: MailboxId, + seq: u64, + tag: &'static str, + } + impl LocalCausalRow for CausalRow { + fn owner(&self) -> MailboxId { + self.owner + } + fn cast_seq(&self) -> u64 { + self.seq + } + } + + /// LAYER-1 FALSIFIER (operator): the interleaved global log + /// `A@s0, C@s0, B@s0, A@s1` must yield owner A's LOCAL chain `[A@s0, A@s1]` — + /// C's and B's interleaved rows REMOVED from A's timeline, not reordered into + /// it. Vacuity guard: A's chain must be strictly shorter than the global log + /// (the other owners' rows were actually dropped, not merely sorted). + #[test] + fn layer1_deinterlaces_interleaved_global_log_into_owner_local_chain() { + const A: MailboxId = 40; + const B: MailboxId = 41; + const C: MailboxId = 42; + // Global durable-log order, exactly the operator's example. + let global = vec![ + CausalRow { + owner: A, + seq: 0, + tag: "A@s0", + }, + CausalRow { + owner: C, + seq: 0, + tag: "C@s0", + }, + CausalRow { + owner: B, + seq: 0, + tag: "B@s0", + }, + CausalRow { + owner: A, + seq: 1, + tag: "A@s1", + }, + ]; + + let a_chain = local_trajectory_of(&global, A); + assert_eq!( + a_chain.iter().map(|r| r.tag).collect::>(), + vec!["A@s0", "A@s1"], + "owner A's local trajectory is its own two casts, deinterlaced", + ); + assert!( + a_chain.len() < global.len(), + "the other owners' interleaved rows were REMOVED from A's timeline, not reordered", + ); + // The same via the all-owners split. + let all = local_trajectories(&global); + assert_eq!(all.len(), 3, "three distinct owners"); + assert_eq!( + all[&A].iter().map(|r| r.tag).collect::>(), + vec!["A@s0", "A@s1"] + ); + assert_eq!( + all[&B].iter().map(|r| r.tag).collect::>(), + vec!["B@s0"] + ); + assert_eq!( + all[&C].iter().map(|r| r.tag).collect::>(), + vec!["C@s0"] + ); + } + + /// LAYER-1 FALSIFIER (operator): within ONE owner, replay order is the cast + /// sequence, even when the durable log stored the casts out of order (a WAL + /// that interleaved a later batch ahead of an earlier one). Falsifiable: the + /// global log is deliberately seq-descending, so a no-op (identity) pass + /// would produce the wrong order. + #[test] + fn layer1_orders_one_owners_chain_by_cast_seq_not_log_order() { + const A: MailboxId = 7; + let global = vec![ + CausalRow { + owner: A, + seq: 2, + tag: "third", + }, + CausalRow { + owner: A, + seq: 0, + tag: "first", + }, + CausalRow { + owner: A, + seq: 1, + tag: "second", + }, + ]; + let chain = local_trajectory_of(&global, A); + assert_eq!( + chain.iter().map(|r| r.tag).collect::>(), + vec!["first", "second", "third"], + "replay follows cast_seq, not the (scrambled) durable-log order", + ); + } + #[test] fn for_rung_policy() { assert_eq!(EpistemicMode::for_rung(0), EpistemicMode::Strict);