Wave-3 decomposition: SQLite engine tasks (open/opts, seam+tx, mechanisms, integration, review gate) + core value-constructor pre-work; plan synced for waves-1-2 review + ADR-023 follow-through

tasks/sqlite-engine-streams.md

@@ -0,0 +1,83 @@
+---
+id: sqlite-engine-streams
+name: SQLite engine — streams (`StreamHandle`, reads, subscribe, trim)
+status: pending
+depends_on: [sqlite-engine-seam-tx]
+scope: moderate
+risk: medium
+impact: component
+level: implementation
+tags: [wave-3, sqlite-engine]
+---
+
+## Description
+
+Implement `Store::stream` and the full `StreamHandle` trait over the
+substrate's stream ops (`stream_publish`, `stream_read_since`,
+`stream_save_offset`, `stream_get_offset` — inherited near-verbatim;
+the `topic` column carries the contract's `stream` field, the one name
+delta, mapped at this layer):
+
+- `stream(name)` — validated constructor returning the boxed handle.
+- `publish` / `publish_with_key` — auto-commit publishes through the
+  writer slot; `encode_payload` at the seam (`Codec` propagates);
+  return the assigned offset.
+- `read_since(offset, limit)` / `read_from_consumer(consumer, limit)`
+  — reader-pool reads, offset ASC (global FIFO); map rows into
+  `StreamEvent` via the core constructor task's `from_row`.
+  **Extent guard (ADR-023 §2)**: `limit <= 0` yields the empty `Vec`
+  at the trait-impl entry, before the substrate — the `LIMIT -1`
+  dialect artifact dies here (the substrate passes `limit` through
+  contract-blind; the guard is the engine's obligation).
+- `save_offset(consumer, offset)` / `get_offset(consumer)` — monotone
+  saves (substrate's upsert already refuses regressions silently);
+  `get_offset` of an absent consumer = 0.
+- `trim_to(horizon)` — `DELETE FROM __alkstore_stream WHERE topic = ?
+  AND offset <= ?` inside the writer-slot lease; boundary argument is
+  **total** (ADR-023 §2: a negative horizon deletes nothing,
+  idempotently — no guard, pass through); returns the deleted count.
+- `subscribe(consumer)` — the durable receiver: attach, read to tail
+  from the stored offset, then wake-driven re-reads off the watcher
+  fanout (same bridge shape as `listen`, but yielding
+  `Result<StreamEvent>` per the `EventReceiver` trait — `Err` carries
+  `Database` only; `None` = closed on watcher death, terminal).
+  Receiver-form `save_offset()` saves the last-yielded event's offset
+  through the same monotone op (ADR-019 §6); `offset()` reports the
+  receiver's position.
+- Consumer identifiers (`consumer`) take the non-empty rule only;
+  stream names the shared-namespace rules.
+
+## Acceptance Criteria
+
+- [ ] Full `StreamHandle` impl; publish/read/save/get/trim/subscribe
+      all work against a temp-file store
+- [ ] Extent guard: `read_since`/`read_from_consumer` with
+      `limit <= 0` return the empty `Vec` (never the whole stream) —
+      pinned per ADR-023 §2's backlog row
+- [ ] `trim_to` exact-boundary (`<=`), negative horizon deletes
+      nothing, surviving offsets never renumbered
+- [ ] Monotone saves: regression saves are silent no-ops; direct and
+      receiver forms compose without regression
+- [ ] `subscribe` replays from the stored offset, then delivers
+      post-attach events wake-driven; watcher death closes it
+      (`None`, terminal); `recv()`'s `Err` carries `Database` only
+- [ ] Key round-trips exactly (`None` vs `Some`); `stream` field
+      carries the contract name (not `topic`)
+- [ ] `cargo test -p alkstore-sqlite`, clippy `-D warnings`, fmt clean
+
+## References
+
+- docs/architecture/core-contract.md (streams section)
+- docs/architecture/decisions/015-streams-depth.md
+- docs/architecture/decisions/019-mechanism-handle-surfaces.md §6
+- docs/architecture/decisions/023-fourth-review-round.md §2
+- docs/architecture/engine-sqlite.md (Mapping the contract: streams)
+- alkstore-sqlite/src/substrate/ops.rs (stream_* fns)
+
+## Notes
+
+> To be filled by implementation agent
+
+## Summary
+
+> To be filled on completion