Wave-6 decomposition (mem engine): eleven tasks per the pg-wave rhythm — foundations (crate scaffold, MemStore::new isolation, guarded state core, wake routing, wasm32 gate from birth), seam-tx (the load-bearing tx overlay: eleven staged _tx methods, commit-merge + wake-at-commit — pg-fix-tx-wake's discipline structural from the start, drop = discard trivial, with_tx panic posture), mechanisms mutually parallel (notify-listen close-at-engine-drop/no-replay/never-coalesced; queues = ADR-010 machine driver-free with the equal-jitter curve and opts resolution as ADR-012 §2's third owner; streams log+offsets+trim; locks TTL registry, no busy-path analog), scheduler-outbox (tick/catch-up-64/leadership through the __alkstore_scheduler lock, engine-uniform per ADR-009 §4; outbox over queues with the 60/5/5 seam stamp), integration (suite third StoreFactory column green on host under a current-thread flavor, wasm32 acceptance items, engine-mem.md to implementation-complete), and the review-wave-6 gate (born-pinned audit: conformance read, discharge-map wiring, wasm items, three-way equivalence green, class-1 window discipline). Window riders: suite-harness-current-thread — the wave's one suite-side pre-task, past_stamp_sleep's blocking sleep goes async per review 003's properties.rs:1683 posture note, enabling the current-thread flavor; pg-fix-scheduler-tick-seam — review 003 Finding 1, the cfg(test) transient-fault seam driving the real tick_with_retries through retry-then-success and exhaustion arms (the last uncovered wave-4 hardening arm, sqlite-commit-error-arm pattern). Task hygiene: pg-fix-open-path frontmatter fixed (unquoted 'max_size: 0' colon-space inside the name scalar broke YAML parsing — taskgraph validate had been red one-error since the wave-4 fix-batch decomposition; also restores review-wave-4-fixes' dependency resolution). implementation.md: wave-6 table row → decomposed, task-set listing with the parallelism note (rider + pre-task independent of the mem chain; rider closes before the gate), wave-6 decomposition bullet in the review-rounds record. Docs-only change; verified taskgraph validate (60 tasks, no cycles), cargo fmt --check, cross-reference read

tasks/mem-engine-streams.md

@@ -0,0 +1,80 @@
+---
+id: mem-engine-streams
+name: Mem engine — streams (in-memory log, per-consumer offsets, trim)
+status: pending
+depends_on: [mem-engine-seam-tx, mem-engine-notify-listen]
+scope: moderate
+risk: medium
+impact: component
+level: implementation
+tags: [wave-6, mem-engine, streams]
+---
+
+## Description
+
+The streams mechanism over an in-memory event log + per-consumer
+offset table (engine-mem.md §Mapping the contract, streams row):
+
+- **`StreamHandle` surface** ([ADR-019](../architecture/decisions/019-mechanism-handle-surfaces.md)
+  §2): `publish` / `publish_with_key` (present key must be non-empty
+  → `InvalidName`), `read_since` / `read_from_consumer` (cursor reads,
+  `offset ASC`, extents clamp empty per
+  [ADR-023](../architecture/decisions/023-fourth-review-round.md) §2),
+  `save_offset` / `get_offset` (monotone saves — below-checkpoint
+  saves are silent no-ops, `Err` arm carries `Database` only),
+  `trim_to` (delete `offset <= horizon`, exact-boundary, returns
+  deleted count, negative horizon deletes nothing idempotently,
+  **wakes nothing**, surviving rows keep offsets, never renumbered),
+  and `subscribe` (durable consumption: attach, read to tail, resume
+  from saved offset; wake-driven with re-read — the wake substrate
+  delivers at commit; explicit saves only, no auto-checkpoint).
+- **Ordering**: engine-assigned monotonically increasing per-stream
+  offsets, global FIFO by offset — the ADR-015 ordering row, mem's own
+  counter (per-stream; absolute values need not equal the SQL engines'
+  — the equivalence rows pin relative order).
+- **`StreamEvent` shape** exactly as pinned
+  ([ADR-015](../architecture/decisions/015-streams-depth.md) §3):
+  `{ offset, stream, key, payload, created_at }` — `created_at` is
+  unix-seconds at publish, informational, from the clock accessor.
+  Reads from a trimmed-away region resume at the horizon's first
+  remaining row.
+- **Keyed publish rides the tx overlay** (staged `publish_with_key_tx`
+  landed in `mem-engine-seam-tx`); this task's non-tx forms, receiver,
+  and trim live here.
+- **Receiver close arms** ([ADR-021](../architecture/decisions/021-tx-reads-and-value-shape-fixes.md)
+  §5): `recv()`'s `Err` carries `Database` only (payloads cross raw;
+  `Codec` at `payload_as<T>`); close is terminal — the one close path
+  is engine drop (the notify-listen task's close arm, realized
+  identically); no reconnection concept (the pg stays-open-across-
+  reconnect row is absent from mem's column).
+
+## Acceptance Criteria
+
+- [ ] Full `StreamHandle` surface over the in-memory log +
+      per-consumer offsets
+- [ ] The ADR-015 properties unit-pinned: global FIFO, immutable
+      never-renumbered offsets (gaps legal post-trim), exact-boundary
+      trim semantics (reads resume at horizon; saved sub-horizon
+      offsets stay valid; trim wakes nothing), monotone save
+      composition (direct + receiver forms interleave)
+- [ ] Key round-trips exactly (`None` vs `Some`); empty-`Some`-key
+      rejected at every entry point
+- [ ] Extent/offset numeric-domain guards at entry — pinned
+- [ ] `cargo check --target wasm32-unknown-unknown -p alkstore-mem`
+      clean; workspace gates green
+
+## References
+
+- docs/architecture/core-contract.md §streams
+- docs/architecture/decisions/015-streams-depth.md;
+  019-mechanism-handle-surfaces.md §2/§6; 021 §5; 023 §2
+- docs/architecture/engine-mem.md §Mapping the contract (streams row)
+- tasks/pg-engine-streams.md (the structural twin's task of record)
+
+## Notes
+
+> To be filled by implementation agent
+
+## Summary
+
+> To be filled on completion