---
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