feat(provider): SQLite affect store + ASGI wiring — #195 parity green
The first slice of ratatoskr-as-Tier-3-Bifrost-consumer: a SQLite-backed, conduit-opaque affect store Worldtree emits Tier-3 persona/affect snapshots into, plus the thin build_affect_provider_app wiring. The bifrost library owns the wire; this owns only the store + glue. - ratatoskr.provider.affect_store: open_affect_store / emit / get / build_affect_provider_app. Two-table SQLite schema (snapshots + idempotency). - Conduit-opaque (INV-001): reads only the two addressing keys; whole-blob hash only. LWW-by-arrival across distinct keys; replay-or-conflict idempotency (actor-scoped from ctx), raising bifrost's typed exceptions. - Conformance: #195 parity vs InMemoryAffectStore through the real dispatch_affect_call engine. 17 provider tests; 395 full suite. - Contract v1.1: idempotency model corrected to bifrost's actual semantics (caught by real-lib TDD; the artifact-only review structurally could not). Heid-panel reviewed (contract + code); idempotency-cache TTL pruning, memory.* plane, and the combined two-plane server deferred (see Out of scope).
This commit is contained in:
@@ -2,22 +2,43 @@
|
||||
contract_version: "2.1"
|
||||
module: "ratatoskr.provider.affect_store"
|
||||
purpose: "Affect-plane Bifrost consumer (v1 tracer): a SQLite-backed, conduit-opaque affect store + the ASGI app wiring Worldtree emits Tier-3 persona/affect snapshots into."
|
||||
touches:
|
||||
- src/ratatoskr/provider/affect_store.py
|
||||
- tests/test_provider_affect.py
|
||||
language: "python"
|
||||
complexity: "medium"
|
||||
estimated_loc: 150
|
||||
confidence: 0.8
|
||||
estimated_loc: 180
|
||||
confidence: 0.85
|
||||
assumptions:
|
||||
- "bifrost>=0.6.1 is installed (pending the gitea-index credential via infra-ops) and exposes build_affect_app, dispatch_affect_call, JwtVerifier, ConsumerRegistration per bifrost/docs/implementing-a-consumer.md @ 8df54ed."
|
||||
- "bifrost>=0.6.1 is installed and exposes build_affect_app, dispatch_affect_call, JwtVerifier, ConsumerRegistration, AffectInvalidArguments, AffectIdempotencyConflict per bifrost/docs/implementing-a-consumer.md @ 8df54ed and bifrost/reference_server/affect.py."
|
||||
- "The affect snapshot dict always carries string addressing keys 'agent_id' and 'end_user_id'; the bifrost wire validates the envelope before the store is called."
|
||||
- "A Heimdall HS256 key for consumer_id='ratatoskr' is provisioned (deploy-time, brokered via infra-ops); the store itself never sees raw auth — the library verifies per-dispatch JWTs and hands a DispatchContext."
|
||||
- "A Heimdall HS256 key for consumer_id='ratatoskr' is provisioned (deploy-time, brokered via infra-ops); the store itself never sees raw auth — the library verifies per-dispatch JWTs and hands a DispatchContext (ctx)."
|
||||
- "The idempotency actor is derivable from ctx (mirrors bifrost's reference `_ctx_actor(ctx)` — the dispatch subject/actor identity)."
|
||||
open_questions:
|
||||
- "SQLite file path + whether the affect plane shares one DB file with the memory plane or uses its own — deferred to the combined-server slice (guide §7) and the memory-plane contract."
|
||||
- "idempotency_class is accepted and ignored in v1 (reserved); confirm Worldtree never relies on class-scoped affect idempotency."
|
||||
- "idempotency_class is accepted and ignored in v1 (reserved; affect.* uses a single short-retry class); confirm Worldtree never relies on class-scoped affect idempotency."
|
||||
external_invariants:
|
||||
- source: ~/development/bifrost/docs/contracts/affect.contract.md
|
||||
invariant_id: "INV-001" # conduit opacity — the governing rule of the affect plane
|
||||
- source: ~/development/bifrost/docs/implementing-a-consumer.md
|
||||
invariant_id: "§6 affect v1 floor"
|
||||
- source: ~/development/bifrost/bifrost/reference_server/affect.py
|
||||
invariant_id: "InMemoryAffectStore.emit" # the executable reference for the wire semantics we parity-prove against
|
||||
revisions:
|
||||
- version: "1.1"
|
||||
at: 2026-06-14
|
||||
summary: "Align idempotency to bifrost's ACTUAL affect semantics (TDD-against-lib finding the artifact-only Heid gate structurally could not see): conflict-on-key-reuse, actor-scoped idempotency, bifrost exception types. Add get() read seam. Two-table schema. Defer idempotency-cache TTL pruning."
|
||||
delta:
|
||||
ADDED:
|
||||
- "INV-009 (idempotency-cache TTL pruning deferred to a follow-up)"
|
||||
- "get() function block (read-back seam; mirrors the reference store's get())"
|
||||
- "idempotency_conflict test"
|
||||
- "affect_idempotency table"
|
||||
MODIFIED:
|
||||
- "INV-008 — replay-noop + conflict-on-reuse (was: same-key-different-hash overwrites)"
|
||||
- "emit ERROR_ROUTING/STEPS/exceptions — AffectInvalidArguments + AffectIdempotencyConflict (was: ValueError)"
|
||||
- "Data flow at-rest — two tables (snapshot + idempotency)"
|
||||
- "basic_emit wording — semantic round-trip (was: byte-identical)"
|
||||
REMOVED:
|
||||
- "the 'same idempotency_key + different content hash -> LWW overwrite' clause (it was backwards: bifrost treats that as a conflict)"
|
||||
---
|
||||
|
||||
## Context
|
||||
@@ -29,7 +50,9 @@ per-dispatch JWT → dispatch → store → conformance), chosen first because
|
||||
`affect.*` has exactly one verb over an opaque blob, so it proves the pipes with
|
||||
minimal store complexity before the heavier `memory.*` plane. We implement
|
||||
**bifrost's own** `affect`-store shape (not worldtree-memory's), hand it to
|
||||
`bifrost.consumer.build_affect_app`, and mount the Starlette app.
|
||||
`bifrost.consumer.build_affect_app`, and mount the Starlette app. The wire
|
||||
semantics are parity-proven against `bifrost.consumer.testing.InMemoryAffectStore`
|
||||
(the executable reference).
|
||||
|
||||
The boundary is absolute (ADR-0001/0002/0009): **Worldtree appraises and decides
|
||||
affect; we only persist and round-trip it.** We run no affect logic.
|
||||
@@ -42,14 +65,24 @@ affect; we only persist and round-trip it.** We run no affect logic.
|
||||
- **Snapshot shape:** `{agent_id, end_user_id, pad, valence, persona_baselines,
|
||||
emitted_at}`. We read **only** `agent_id` + `end_user_id` (the addressing
|
||||
keys); the rest is opaque payload.
|
||||
- **At rest:** one row per `(agent_id, end_user_id)` in a SQLite table
|
||||
`affect_snapshots`, holding the **verbatim** snapshot JSON + a content hash +
|
||||
the last idempotency key + an `arrived_at` receive-timestamp that is
|
||||
**audit/debug only** (never used for ordering — LWW is the upsert itself — and
|
||||
never returned). Last-write-wins by arrival → the row is upserted.
|
||||
- **At rest:** two SQLite tables —
|
||||
- `affect_snapshots(agent_id, end_user_id, snapshot_json, arrived_at,
|
||||
PRIMARY KEY (agent_id, end_user_id))` — one row per pair, holding the
|
||||
**verbatim** snapshot JSON; LWW-overwritten on each new arrival. `arrived_at`
|
||||
is **audit/debug only** (never used for ordering, never returned).
|
||||
- `affect_idempotency(idempotency_id, digest, expires_at,
|
||||
PRIMARY KEY (idempotency_id))` — the per-(actor, idempotency_key) replay/
|
||||
conflict cache: `digest` is a content fingerprint of the snapshot;
|
||||
`expires_at` records the short-retry deadline for a future pruning pass
|
||||
(TTL eviction deferred — see INV-009).
|
||||
- **Out:** `{"stored": True}` ack (the library wraps it with the transport
|
||||
`{"success": True}` envelope).
|
||||
|
||||
**Async surface:** `emit` is `async def` (the bifrost consumer Protocol awaits
|
||||
it); `open_affect_store` and `get` are sync (no I/O await — `get` is a read-back
|
||||
seam). The `FN` lines below omit the `async` keyword only because the contract
|
||||
grammar's `FN <name>` form has no async marker.
|
||||
|
||||
## Invariants
|
||||
|
||||
- **INV-001** [hard]: **Conduit opacity** (inlined from bifrost
|
||||
@@ -61,36 +94,49 @@ affect; we only persist and round-trip it.** We run no affect logic.
|
||||
is non-semantic serialization, not a field read. The distinction the
|
||||
implementer must preserve: *serialize-the-whole-blob* = allowed;
|
||||
*read-a-payload-field-and-act-on-it* = forbidden.
|
||||
- **INV-002** [hard]: **Last-write-wins by ARRIVAL.** The most recently arrived
|
||||
emit for a `(agent_id, end_user_id)` is the persisted state. **Arrival order =
|
||||
- **INV-002** [hard]: **Last-write-wins by ARRIVAL across distinct emits.** For a
|
||||
given `(agent_id, end_user_id)`, the most recently arrived emit (a *distinct*
|
||||
idempotency_id — see INV-008) overwrites the snapshot row. **Arrival order =
|
||||
the order in which `emit`'s upsert transaction commits** (serialized under
|
||||
SQLite's single-writer model). `emitted_at` is NEVER compared — Worldtree
|
||||
throttles + sequences emits, so arrival order at the conduit is the intended
|
||||
semantics.
|
||||
- **INV-003** [hard]: The snapshot is persisted **verbatim** in the sense of
|
||||
**semantic round-trip**: the stored bytes are the store's canonical
|
||||
serialization (`json.dumps(..., sort_keys=True)`), and a read deserializes to a
|
||||
Python object EQUAL to the input snapshot (`snapshot_out == snapshot_in`).
|
||||
"Verbatim" does NOT promise byte-equality with the caller's original wire bytes
|
||||
— key order, whitespace, and numeric formatting may differ; only value-equality
|
||||
of the decoded object is guaranteed.
|
||||
serialization (`json.dumps(..., sort_keys=True)`), and a read (`get`)
|
||||
deserializes to a Python object EQUAL to the input snapshot
|
||||
(`snapshot_out == snapshot_in`). "Verbatim" does NOT promise byte-equality with
|
||||
the caller's original wire bytes — key order, whitespace, and numeric
|
||||
formatting may differ; only value-equality of the decoded object is guaranteed.
|
||||
- **INV-004** [hard]: `emit` returns exactly `{"stored": True}` on every
|
||||
successful persist (Worldtree's emitter validates `stored` specifically).
|
||||
successful persist AND on a recognized replay (Worldtree's emitter validates
|
||||
`stored` specifically).
|
||||
- **INV-005** [hard]: The store advertises `affect_supported = True`; it is the
|
||||
REQUIRED store — `build_affect_app(store=None, ...)` raises (no silent
|
||||
in-memory default).
|
||||
- **INV-006** [hard]: Authorization identity/scope is taken from `ctx`
|
||||
(DispatchContext), never from call arguments; the snapshot addressing keys are
|
||||
used ONLY as the persistence key, not as an auth claim.
|
||||
- **INV-007** [hard]: The upsert is atomic (single transaction); no partial affect
|
||||
row is ever observable.
|
||||
- **INV-008** [soft, recovery_window=1]: A replayed emit (same `idempotency_key`
|
||||
AND same content hash) **SHOULD** skip the redundant write as a best-effort
|
||||
optimization, but **MUST** still return `{"stored": True}`. Re-writing identical
|
||||
bytes is harmless under LWW, so the skip is an optimization, not a correctness
|
||||
requirement (the hard guarantee is the ack, INV-004). A same `idempotency_key`
|
||||
carrying a *different* content hash is **NOT** a replay — it is a normal arrival
|
||||
and overwrites per INV-002.
|
||||
- **INV-006** [hard]: Authorization identity/scope — and the **idempotency
|
||||
actor** — are taken from `ctx` (DispatchContext), never from the snapshot or
|
||||
other call arguments. The snapshot addressing keys are used ONLY as the
|
||||
persistence key, not as an auth claim.
|
||||
- **INV-007** [hard]: Each emit's snapshot upsert **and** its idempotency record
|
||||
commit in ONE transaction; no partial state (a snapshot without its
|
||||
idempotency row, or vice-versa) is ever observable.
|
||||
- **INV-008** [hard]: **Idempotency = replay-or-conflict, keyed by
|
||||
`(actor-from-ctx, idempotency_key)`.** On emit, compare against the cached
|
||||
digest for that idempotency_id:
|
||||
- **no entry** → new arrival: persist (LWW per INV-002) + record the digest,
|
||||
return `{"stored": True}`.
|
||||
- **entry, same digest** → **replay**: no second snapshot write, return
|
||||
`{"stored": True}`.
|
||||
- **entry, different digest** → the client reused a key for different content:
|
||||
**raise `AffectIdempotencyConflict`** (the library maps it to the wire 409).
|
||||
This is NOT an LWW overwrite — overwrites happen across *distinct* keys only.
|
||||
- **INV-009** [soft, recovery_window=∞]: **Idempotency-cache TTL pruning is
|
||||
deferred.** bifrost's reference prunes idempotency entries on a short-retry
|
||||
window; v1 records `expires_at` but does not evict, so `affect_idempotency`
|
||||
grows unbounded until a follow-up pruning patch. Wire-observable behavior is
|
||||
unaffected (replay/conflict still resolve correctly); only cache size is.
|
||||
`affect_snapshots` is already bounded to one row per `(agent_id, end_user_id)`.
|
||||
|
||||
## Concurrency
|
||||
|
||||
@@ -102,19 +148,21 @@ the upsert itself. No cross-row coordination — affect rows are independent.
|
||||
|
||||
A crisp line, since the responsibilities interleave: the **bifrost library** owns
|
||||
the entire wire — envelope validation, per-dispatch JWT verification, scope
|
||||
authorization, error mapping, capability negotiation, route exposure. **This
|
||||
contract** owns ONLY the store (`emit` + the SQLite persistence) and the thin
|
||||
`build_affect_provider_app` wiring. Any guarantee about the wire is the
|
||||
library's; this contract neither re-implements nor overrides it. `emit`'s
|
||||
defensive missing-key check (PRE-001) is belt-and-suspenders — the wire should
|
||||
already have rejected a malformed envelope; the store raises defensively rather
|
||||
than relying on it.
|
||||
authorization, error mapping (including mapping the store's `AffectInvalidArguments`
|
||||
/ `AffectIdempotencyConflict` to transport status), capability negotiation, route
|
||||
exposure. **This contract** owns ONLY the store (`emit` + `get` + the SQLite
|
||||
persistence) and the thin `build_affect_provider_app` wiring. The store raises
|
||||
bifrost's typed exceptions; the library decides the wire status. `emit`'s
|
||||
defensive addressing-key check (PRE-001) is belt-and-suspenders — the wire should
|
||||
already have rejected a malformed envelope.
|
||||
|
||||
## Integration points
|
||||
|
||||
- `bifrost.consumer.build_affect_app(store, verifier, registration)` → Starlette ASGI app.
|
||||
- `bifrost.reference_server.JwtVerifier(algorithm="HS256", key_bytes=...)`.
|
||||
- `bifrost.consumer.ConsumerRegistration(consumer_id="ratatoskr")`.
|
||||
- `bifrost.affect.AffectInvalidArguments` / `AffectIdempotencyConflict` — the typed
|
||||
exceptions the store raises; the library maps them to wire status.
|
||||
- **Conformance (tests only):** `bifrost.consumer.testing.InMemoryAffectStore`
|
||||
+ `bifrost.affect.dispatch_affect_call` — the #195 parity pattern.
|
||||
|
||||
@@ -123,63 +171,97 @@ than relying on it.
|
||||
- **[security]** Never read or log the affect payload (`pad`/`valence`/
|
||||
`persona_baselines`); opacity is a security + correctness boundary, not just a
|
||||
style choice.
|
||||
- **[compatibility]** Implement bifrost's affect-store shape exactly; never fork
|
||||
the wire/engine/auth. Custom behavior, if ever needed, goes through `Hooks` in
|
||||
a namespace OUTSIDE `affect.*` (ADR-0005, INV-008).
|
||||
- **[compatibility]** Implement bifrost's affect-store shape exactly; raise its
|
||||
typed exceptions; never fork the wire/engine/auth. Custom behavior, if ever
|
||||
needed, goes through `Hooks` in a namespace OUTSIDE `affect.*` (ADR-0005).
|
||||
- **[correctness]** Do not compare `emitted_at` anywhere (would both read the
|
||||
payload and break arrival-order LWW).
|
||||
|
||||
## Out of scope (deferred — do NOT flag as drift)
|
||||
|
||||
- **Idempotency-cache TTL pruning** (INV-009): `affect_idempotency` rows
|
||||
accumulate without eviction in v1; `expires_at` is recorded but not acted on.
|
||||
The short-retry-window pruning pass is a follow-up patch.
|
||||
- **The `memory.*` plane**: this slice is affect-only; the memory store + its
|
||||
Protocol are a later contract.
|
||||
- **The combined two-plane server** (guide §7): one handshake negotiating both
|
||||
memory + affect is deferred; `build_affect_provider_app` mounts affect alone.
|
||||
- **`affect.fetch` / `affect:read` / persona-baseline rehydrate**: RESERVED in
|
||||
v1; only `emit` + the test-only `get()` exist.
|
||||
- **`idempotency_class`**: accepted and ignored (affect.* uses a single
|
||||
short-retry class).
|
||||
- **WAL/concurrency hardening, deployment DB path, auth-key provisioning**:
|
||||
wiring/ops concerns, not this contract's function-block surface.
|
||||
|
||||
```contract
|
||||
FN open_affect_store(db_path: str) -> RatatoskrAffectStore
|
||||
BRIEF: Open the SQLite-backed affect store, creating the schema on first use.
|
||||
PRE: [PRE-001 hard] db_path is a writable path or ":memory:" -- guard clause
|
||||
POST: [POST-001 return_value] returned store has affect_supported is True -- assert store.affect_supported is True
|
||||
POST: [POST-002 state_change] table affect_snapshots exists -- assert schema present
|
||||
POST: [POST-002 state_change] tables affect_snapshots + affect_idempotency exist -- assert schema present
|
||||
STEPS:
|
||||
1. [setup] CONNECT sqlite3 to db_path; SET journal_mode=WAL (skip for ":memory:")
|
||||
2. [sequential, flexibility=prescriptive] CREATE TABLE IF NOT EXISTS affect_snapshots (
|
||||
agent_id TEXT NOT NULL, end_user_id TEXT NOT NULL,
|
||||
snapshot_json TEXT NOT NULL, content_hash TEXT NOT NULL,
|
||||
idempotency_key TEXT, arrived_at TEXT,
|
||||
snapshot_json TEXT NOT NULL, arrived_at TEXT,
|
||||
PRIMARY KEY (agent_id, end_user_id))
|
||||
3. [cleanup] RETURN RatatoskrAffectStore(conn)
|
||||
3. [sequential, flexibility=prescriptive] CREATE TABLE IF NOT EXISTS affect_idempotency (
|
||||
idempotency_id TEXT PRIMARY KEY, digest TEXT NOT NULL, expires_at REAL)
|
||||
4. [cleanup] RETURN RatatoskrAffectStore(conn)
|
||||
TESTS:
|
||||
fresh_db [happy,tracer]: open ":memory:" → store.affect_supported is True; affect_snapshots queryable
|
||||
fresh_db [happy,tracer]: open ":memory:" → store.affect_supported is True; both tables queryable
|
||||
reopen [happy]: open existing file twice → no error, schema idempotent
|
||||
```
|
||||
|
||||
```contract
|
||||
FN emit(self, snapshot: dict, *, idempotency_key: str, ctx: DispatchContext, idempotency_class: str | None = None) -> dict
|
||||
BRIEF: Persist a Worldtree affect snapshot verbatim, conduit-opaque, last-write-wins by arrival.
|
||||
PRE: [PRE-001 hard] snapshot["agent_id"] and snapshot["end_user_id"] are non-empty strings -- read both; raise ValueError if absent (defensive; the wire should prevent)
|
||||
PRE: [PRE-002 hard] idempotency_key is a non-empty string -- assert
|
||||
POST: [POST-001 return_value] returns {"stored": True} -- assert result == {"stored": True} (INV-004)
|
||||
POST: [POST-002 side_effect] exactly one row for (agent_id, end_user_id) holds the verbatim snapshot -- stored snapshot deserializes equal to input (INV-003)
|
||||
POST: [POST-003 state_change] a replayed identical emit SHOULD skip the second write (best-effort, INV-008); the ack is unconditional -- replay test asserts one logical write + {"stored": True}
|
||||
BRIEF: Persist a Worldtree affect snapshot verbatim — conduit-opaque, replay-or-conflict idempotent, last-write-wins by arrival across distinct keys.
|
||||
PRE: [PRE-001 hard] snapshot["agent_id"] and snapshot["end_user_id"] are non-empty strings -- else raise AffectInvalidArguments (defensive; the wire should prevent)
|
||||
PRE: [PRE-002 hard] idempotency_key is a non-empty string -- else raise AffectInvalidArguments
|
||||
POST: [POST-001 return_value] returns {"stored": True} on persist AND on recognized replay -- assert result == {"stored": True} (INV-004)
|
||||
POST: [POST-002 side_effect] after a new arrival, get(agent_id, end_user_id) deserializes equal to input -- (INV-003)
|
||||
POST: [POST-003 state_change] same idempotency_id + same digest → no second snapshot write, {"stored": True}; same idempotency_id + different digest → AffectIdempotencyConflict (INV-008)
|
||||
ERROR_ROUTING:
|
||||
KeyError:
|
||||
local_handling: raise ValueError("affect snapshot missing addressing key")
|
||||
AffectInvalidArguments:
|
||||
local_handling: raise on missing/empty addressing keys or empty idempotency_key
|
||||
flow_control: abort
|
||||
state_recovery: none
|
||||
state_recovery: none (no write performed)
|
||||
AffectIdempotencyConflict:
|
||||
local_handling: raise when idempotency_id is cached with a different digest
|
||||
flow_control: abort
|
||||
state_recovery: none (prior snapshot + idempotency row untouched)
|
||||
sqlite3.OperationalError:
|
||||
local_handling: let propagate (library maps to transport error)
|
||||
flow_control: abort
|
||||
state_recovery: transaction rolled back — no partial row (INV-007)
|
||||
STEPS:
|
||||
1. [setup, flexibility=prescriptive] IF "agent_id" not in snapshot OR "end_user_id" not in snapshot: RAISE ValueError("affect snapshot missing addressing key") -- explicit presence-guard so PRE-001 surfaces ValueError, not a raw KeyError
|
||||
ELSE READ agent_id = snapshot["agent_id"]; end_user_id = snapshot["end_user_id"] -- the ONLY two fields read (INV-001)
|
||||
2. [sequential, flexibility=indicative] SET blob = json.dumps(snapshot, sort_keys=True, separators=(",", ":")); SET content_hash = sha256(blob).hexdigest() -- whole-blob hash is opacity-safe; no individual payload field is read
|
||||
3. [branch] IF a row exists for (agent_id, end_user_id) with content_hash == this content_hash AND idempotency_key == this idempotency_key:
|
||||
- RETURN {"stored": True} -- replay no-op (INV-008)
|
||||
4. [sequential, flexibility=prescriptive] UPSERT (agent_id, end_user_id, snapshot_json=blob, content_hash, idempotency_key, arrived_at=<receive wall-clock>) in ONE transaction -- LWW by arrival, unconditional overwrite (INV-002, INV-007). Do NOT compare emitted_at.
|
||||
1. [setup, flexibility=prescriptive] IF "agent_id"/"end_user_id" missing or not non-empty str: RAISE AffectInvalidArguments. IF not idempotency_key: RAISE AffectInvalidArguments. ELSE READ agent_id, end_user_id -- the ONLY snapshot fields read (INV-001)
|
||||
2. [sequential, flexibility=indicative] SET digest = sha256(json.dumps(snapshot, sort_keys=True, separators=(",", ":"))).hexdigest(); SET actor = ctx-derived actor (INV-006); SET idempotency_id = f"affect.emit|{actor}|{idempotency_key}" -- whole-blob hash is opacity-safe
|
||||
3. [branch] SELECT digest FROM affect_idempotency WHERE idempotency_id = ?:
|
||||
IF row exists AND stored digest == digest: RETURN {"stored": True} -- replay no-op (INV-008)
|
||||
IF row exists AND stored digest != digest: RAISE AffectIdempotencyConflict("idempotency key reused with different payload")
|
||||
4. [sequential, flexibility=prescriptive] BEGIN; UPSERT affect_snapshots (agent_id, end_user_id, snapshot_json=blob, arrived_at=<wall-clock>); UPSERT affect_idempotency (idempotency_id, digest, expires_at=<now + short_retry_ttl>); COMMIT -- LWW + idempotency record in ONE transaction (INV-002, INV-007). Do NOT compare emitted_at.
|
||||
5. [cleanup] RETURN {"stored": True} (INV-004)
|
||||
TESTS:
|
||||
basic_emit [happy,tracer]: valid snapshot → {"stored": True}; row round-trips byte-identical
|
||||
opacity [adversarial]: snapshot carrying arbitrary extra/unknown payload fields → persists + round-trips verbatim + returns stored:True (store never validates or branches on payload); AND two snapshots for the same key differing ONLY in payload address the SAME row (behavioral opacity — does not rely on attribute-access booby-trapping, which dict __getitem__/json.dumps would not trigger)
|
||||
lww_by_arrival [scenario]: emit A, then emit B (different payload, OLDER emitted_at) for same key → stored state is B; emitted_at never compared
|
||||
replay_noop [happy]: identical emit twice (same key + hash) → single row, {"stored": True} both times, one write
|
||||
missing_key [adversarial]: snapshot without "end_user_id" → raises ValueError; no row written
|
||||
parity_vs_reference [scenario]: drive identical affect.emit envelopes through dispatch_affect_call against InMemoryAffectStore and RatatoskrAffectStore → {success,...}+{stored} wire bodies agree (#195)
|
||||
basic_emit [happy,tracer]: valid snapshot → {"stored": True}; get() round-trips semantically equal (out == in)
|
||||
opacity [adversarial]: snapshot carrying arbitrary extra/unknown payload fields → persists + round-trips verbatim + returns stored:True (store never validates or branches on payload); AND two snapshots for the same key differing ONLY in payload address the SAME row (behavioral opacity — not attribute-access booby-trapping, which dict __getitem__/json.dumps would not trigger)
|
||||
lww_by_arrival [scenario]: emit A then emit B (DISTINCT idempotency keys, different payload, OLDER emitted_at on B) for same (agent,user) → get() == B; emitted_at never compared
|
||||
replay_noop [happy]: same idempotency_key + same payload twice → {"stored": True} both; one snapshot row, get() == payload
|
||||
idempotency_conflict [adversarial]: same idempotency_key + DIFFERENT payload → second emit raises AffectIdempotencyConflict; first snapshot unchanged
|
||||
missing_key [adversarial]: snapshot without "end_user_id" → raises AffectInvalidArguments; no row written
|
||||
parity_vs_reference [scenario]: drive identical affect.emit envelopes (happy + conflict) through dispatch_affect_call against InMemoryAffectStore and RatatoskrAffectStore → (status, body) tuples agree (#195)
|
||||
```
|
||||
|
||||
```contract
|
||||
FN get(self, agent_id: str, end_user_id: str) -> dict | None
|
||||
BRIEF: Read-back of the stored snapshot (tests / future rehydrate-seed). NOT a wire verb — affect.fetch is RESERVED in v1.
|
||||
POST: [POST-001 return_value] returns the verbatim snapshot for the key, or None if absent -- (INV-003)
|
||||
STEPS:
|
||||
1. [sequential] SELECT snapshot_json FROM affect_snapshots WHERE agent_id = ? AND end_user_id = ?
|
||||
2. [cleanup] RETURN json.loads(snapshot_json) IF row else None
|
||||
TESTS:
|
||||
get_absent [boundary]: no row for key → None
|
||||
get_after_emit [happy]: returns the emitted snapshot, deserialized equal
|
||||
```
|
||||
|
||||
```contract
|
||||
|
||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "ratatoskr"
|
||||
version = "0.17.0"
|
||||
version = "0.17.1"
|
||||
description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
"""Tier-3 Bifrost consumer: durable memory.* + affect.* persistence provider.
|
||||
|
||||
Contracts: docs/contracts/bifrost_affect_provider.contract.md (affect plane, v1).
|
||||
"""
|
||||
@@ -0,0 +1,146 @@
|
||||
"""SQLite-backed, conduit-opaque affect store (Bifrost affect plane, v1).
|
||||
|
||||
Contract: docs/contracts/bifrost_affect_provider.contract.md
|
||||
|
||||
The store persists Worldtree's Tier-3 affect snapshots verbatim and round-trips
|
||||
them. It runs NO affect logic: it reads only the two addressing keys
|
||||
(`agent_id`, `end_user_id`) and treats `pad`/`valence`/`persona_baselines`/
|
||||
`emitted_at` as an opaque blob (INV-001). Idempotency is replay-or-conflict,
|
||||
keyed by (actor-from-ctx, idempotency_key) (INV-008), and snapshots are
|
||||
last-write-wins by arrival across distinct keys (INV-002).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import sqlite3
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from bifrost.affect import AffectIdempotencyConflict, AffectInvalidArguments
|
||||
from bifrost.consumer import ConsumerRegistration, build_affect_app
|
||||
from bifrost.reference_server import JwtVerifier
|
||||
|
||||
_SHORT_RETRY_TTL_SECONDS = 300
|
||||
|
||||
|
||||
def _ctx_actor(ctx: Any) -> str:
|
||||
"""Mirror bifrost reference `_ctx_actor`: idempotency actor = JWT `sub`.
|
||||
|
||||
The real DispatchContext exposes the `sub` claim as `session_id`; test
|
||||
contexts set `jwt_sub`. (INV-006: the actor comes from ctx, never the
|
||||
snapshot.)
|
||||
"""
|
||||
return str(getattr(ctx, "jwt_sub", None) or getattr(ctx, "session_id", ""))
|
||||
|
||||
|
||||
class RatatoskrAffectStore:
|
||||
"""The affect `MemoryDataStore`-shaped store handed to `build_affect_app`."""
|
||||
|
||||
affect_supported = True
|
||||
|
||||
def __init__(self, conn: sqlite3.Connection):
|
||||
self._conn = conn
|
||||
|
||||
async def emit(
|
||||
self,
|
||||
snapshot: dict,
|
||||
*,
|
||||
idempotency_key: str,
|
||||
ctx: Any,
|
||||
idempotency_class: str | None = None,
|
||||
) -> dict:
|
||||
del idempotency_class # reserved; affect.* uses a single short-retry class
|
||||
|
||||
# INV-001: read ONLY the two addressing keys; everything else is opaque.
|
||||
agent_id = snapshot.get("agent_id")
|
||||
end_user_id = snapshot.get("end_user_id")
|
||||
if not (
|
||||
isinstance(agent_id, str)
|
||||
and agent_id
|
||||
and isinstance(end_user_id, str)
|
||||
and end_user_id
|
||||
):
|
||||
raise AffectInvalidArguments("snapshot missing agent_id / end_user_id")
|
||||
if not (isinstance(idempotency_key, str) and idempotency_key):
|
||||
raise AffectInvalidArguments("idempotency_key required")
|
||||
|
||||
# Whole-blob serialize + hash is opacity-safe (not a field read).
|
||||
blob = json.dumps(snapshot, sort_keys=True, separators=(",", ":"))
|
||||
digest = hashlib.sha256(blob.encode()).hexdigest()
|
||||
idempotency_id = f"affect.emit|{_ctx_actor(ctx)}|{idempotency_key}"
|
||||
|
||||
# INV-008: replay (same digest) -> no-op; reuse with different digest -> conflict.
|
||||
cached = self._conn.execute(
|
||||
"SELECT digest FROM affect_idempotency WHERE idempotency_id = ?",
|
||||
(idempotency_id,),
|
||||
).fetchone()
|
||||
if cached is not None:
|
||||
if cached[0] == digest:
|
||||
return {"stored": True}
|
||||
raise AffectIdempotencyConflict(
|
||||
"idempotency key reused with different payload"
|
||||
)
|
||||
|
||||
# INV-002 + INV-007: LWW snapshot upsert + idempotency record, one transaction.
|
||||
now = time.time()
|
||||
with self._conn:
|
||||
self._conn.execute(
|
||||
"INSERT INTO affect_snapshots (agent_id, end_user_id, snapshot_json, arrived_at) "
|
||||
"VALUES (?, ?, ?, ?) "
|
||||
"ON CONFLICT(agent_id, end_user_id) DO UPDATE SET "
|
||||
"snapshot_json = excluded.snapshot_json, arrived_at = excluded.arrived_at",
|
||||
(agent_id, end_user_id, blob, str(now)),
|
||||
)
|
||||
self._conn.execute(
|
||||
"INSERT INTO affect_idempotency (idempotency_id, digest, expires_at) "
|
||||
"VALUES (?, ?, ?)",
|
||||
(idempotency_id, digest, now + _SHORT_RETRY_TTL_SECONDS),
|
||||
)
|
||||
return {"stored": True}
|
||||
|
||||
def get(self, agent_id: str, end_user_id: str) -> dict | None:
|
||||
"""Read-back of the stored snapshot (tests / future rehydrate-seed)."""
|
||||
row = self._conn.execute(
|
||||
"SELECT snapshot_json FROM affect_snapshots WHERE agent_id = ? AND end_user_id = ?",
|
||||
(agent_id, end_user_id),
|
||||
).fetchone()
|
||||
return json.loads(row[0]) if row is not None else None
|
||||
|
||||
|
||||
def open_affect_store(db_path: str) -> RatatoskrAffectStore:
|
||||
"""Open the SQLite-backed affect store, creating the schema on first use."""
|
||||
conn = sqlite3.connect(db_path)
|
||||
if db_path != ":memory:":
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute(
|
||||
"CREATE TABLE IF NOT EXISTS affect_snapshots ("
|
||||
"agent_id TEXT NOT NULL, end_user_id TEXT NOT NULL, "
|
||||
"snapshot_json TEXT NOT NULL, arrived_at TEXT, "
|
||||
"PRIMARY KEY (agent_id, end_user_id))"
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE TABLE IF NOT EXISTS affect_idempotency ("
|
||||
"idempotency_id TEXT PRIMARY KEY, digest TEXT NOT NULL, expires_at REAL)"
|
||||
)
|
||||
conn.commit()
|
||||
return RatatoskrAffectStore(conn)
|
||||
|
||||
|
||||
def build_affect_provider_app(
|
||||
store: RatatoskrAffectStore,
|
||||
heimdall_key: bytes,
|
||||
consumer_id: str = "ratatoskr",
|
||||
):
|
||||
"""Wire the JWT verifier + registration and hand the store to bifrost.
|
||||
|
||||
Returns a Starlette ASGI app exposing POST /bifrost/handshake and
|
||||
POST /bifrost/affect-call. The library owns the wire; this is the thin glue.
|
||||
"""
|
||||
if getattr(store, "affect_supported", False) is not True: # INV-005
|
||||
raise ValueError("store must advertise affect_supported=True")
|
||||
if not (isinstance(heimdall_key, bytes) and heimdall_key):
|
||||
raise ValueError("heimdall_key must be non-empty bytes")
|
||||
verifier = JwtVerifier(algorithm="HS256", key_bytes=heimdall_key)
|
||||
registration = ConsumerRegistration(consumer_id=consumer_id)
|
||||
return build_affect_app(store=store, verifier=verifier, registration=registration)
|
||||
@@ -0,0 +1,230 @@
|
||||
"""Tests for the Tier-3 Bifrost affect provider (ratatoskr.provider.affect_store).
|
||||
|
||||
Contract: docs/contracts/bifrost_affect_provider.contract.md
|
||||
Vertical tracer-first: basic_emit -> opacity -> lww -> replay -> conflict ->
|
||||
missing_key -> #195 parity vs InMemoryAffectStore.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import types
|
||||
|
||||
import pytest
|
||||
from bifrost.affect import AffectIdempotencyConflict, AffectInvalidArguments
|
||||
|
||||
from ratatoskr.provider.affect_store import build_affect_provider_app, open_affect_store
|
||||
|
||||
|
||||
def _ctx(sub: str = "sub-1"):
|
||||
# Mirrors bifrost's _ctx_actor: actor = jwt_sub (test ctx) or session_id.
|
||||
return types.SimpleNamespace(jwt_sub=sub)
|
||||
|
||||
|
||||
def _snapshot(agent: str = "a1", user: str = "u1", **payload):
|
||||
base = {
|
||||
"agent_id": agent,
|
||||
"end_user_id": user,
|
||||
"pad": {"p": 0.1, "a": 0.2, "d": 0.3},
|
||||
"valence": 0.5,
|
||||
"persona_baselines": {"warmth": 0.7},
|
||||
"emitted_at": "2026-06-14T00:00:00Z",
|
||||
}
|
||||
base.update(payload)
|
||||
return base
|
||||
|
||||
|
||||
def _row_count(store, table: str) -> int:
|
||||
return store._conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
|
||||
|
||||
|
||||
# --- open_affect_store ---
|
||||
|
||||
def test_open_advertises_capability_and_schema():
|
||||
store = open_affect_store(":memory:")
|
||||
assert store.affect_supported is True
|
||||
# both tables queryable
|
||||
store._conn.execute("SELECT * FROM affect_snapshots")
|
||||
store._conn.execute("SELECT * FROM affect_idempotency")
|
||||
|
||||
|
||||
def test_reopen_existing_file_is_idempotent(tmp_path):
|
||||
db = str(tmp_path / "affect.db")
|
||||
open_affect_store(db) # first open creates schema
|
||||
store = open_affect_store(db) # reopen: CREATE TABLE IF NOT EXISTS is a no-op
|
||||
assert store.affect_supported is True
|
||||
store._conn.execute("SELECT * FROM affect_snapshots")
|
||||
store._conn.execute("SELECT * FROM affect_idempotency")
|
||||
|
||||
|
||||
# --- emit ---
|
||||
|
||||
async def test_basic_emit_stores_and_round_trips():
|
||||
store = open_affect_store(":memory:")
|
||||
snap = _snapshot()
|
||||
result = await store.emit(snap, idempotency_key="k1", ctx=_ctx())
|
||||
assert result == {"stored": True}
|
||||
assert store.get("a1", "u1") == snap
|
||||
|
||||
|
||||
async def test_opacity_arbitrary_payload_round_trips_and_addressing_invariant():
|
||||
store = open_affect_store(":memory:")
|
||||
# arbitrary extra/unknown payload fields persist + round-trip verbatim
|
||||
snap = _snapshot(weird_field={"nested": [1, 2, 3]}, mystery="x")
|
||||
assert await store.emit(snap, idempotency_key="k1", ctx=_ctx()) == {"stored": True}
|
||||
assert store.get("a1", "u1") == snap
|
||||
# two snapshots for the same key differing ONLY in payload address the SAME row
|
||||
snap2 = _snapshot(weird_field={"nested": [9]}, mystery="y", valence=0.99)
|
||||
await store.emit(snap2, idempotency_key="k2", ctx=_ctx())
|
||||
assert store.get("a1", "u1") == snap2
|
||||
assert _row_count(store, "affect_snapshots") == 1 # same row overwritten
|
||||
|
||||
|
||||
async def test_lww_by_arrival_ignores_emitted_at():
|
||||
store = open_affect_store(":memory:")
|
||||
a = _snapshot(valence=0.1, emitted_at="2026-06-14T10:00:00Z")
|
||||
b = _snapshot(valence=0.9, emitted_at="2026-06-14T08:00:00Z") # OLDER emitted_at
|
||||
await store.emit(a, idempotency_key="ka", ctx=_ctx())
|
||||
await store.emit(b, idempotency_key="kb", ctx=_ctx()) # distinct key -> arrival wins
|
||||
assert store.get("a1", "u1") == b # later arrival wins despite older emitted_at
|
||||
|
||||
|
||||
async def test_replay_noop_same_key_same_payload():
|
||||
store = open_affect_store(":memory:")
|
||||
snap = _snapshot()
|
||||
assert await store.emit(snap, idempotency_key="k1", ctx=_ctx()) == {"stored": True}
|
||||
assert await store.emit(snap, idempotency_key="k1", ctx=_ctx()) == {"stored": True}
|
||||
assert store.get("a1", "u1") == snap
|
||||
assert _row_count(store, "affect_snapshots") == 1 # replay did not duplicate
|
||||
|
||||
|
||||
async def test_idempotency_conflict_same_key_different_payload():
|
||||
store = open_affect_store(":memory:")
|
||||
first = _snapshot(valence=0.1)
|
||||
await store.emit(first, idempotency_key="k1", ctx=_ctx())
|
||||
with pytest.raises(AffectIdempotencyConflict):
|
||||
await store.emit(_snapshot(valence=0.2), idempotency_key="k1", ctx=_ctx())
|
||||
assert store.get("a1", "u1") == first # prior snapshot untouched
|
||||
|
||||
|
||||
async def test_same_key_distinct_actor_is_not_a_conflict():
|
||||
# idempotency is actor-scoped (INV-006/-008): same key, different ctx actor
|
||||
store = open_affect_store(":memory:")
|
||||
await store.emit(_snapshot(valence=0.1), idempotency_key="k1", ctx=_ctx("sub-A"))
|
||||
# different actor, same key, different payload -> NOT a conflict (distinct id)
|
||||
assert await store.emit(
|
||||
_snapshot(valence=0.2), idempotency_key="k1", ctx=_ctx("sub-B")
|
||||
) == {"stored": True}
|
||||
|
||||
|
||||
async def test_missing_end_user_id_raises_and_no_write():
|
||||
store = open_affect_store(":memory:")
|
||||
bad = _snapshot()
|
||||
del bad["end_user_id"]
|
||||
with pytest.raises(AffectInvalidArguments):
|
||||
await store.emit(bad, idempotency_key="k1", ctx=_ctx())
|
||||
assert _row_count(store, "affect_snapshots") == 0
|
||||
|
||||
|
||||
async def test_missing_agent_id_raises_and_no_write():
|
||||
# PRE-001 guards BOTH addressing keys symmetrically.
|
||||
store = open_affect_store(":memory:")
|
||||
bad = _snapshot()
|
||||
del bad["agent_id"]
|
||||
with pytest.raises(AffectInvalidArguments):
|
||||
await store.emit(bad, idempotency_key="k1", ctx=_ctx())
|
||||
assert _row_count(store, "affect_snapshots") == 0
|
||||
|
||||
|
||||
async def test_empty_idempotency_key_raises():
|
||||
store = open_affect_store(":memory:")
|
||||
with pytest.raises(AffectInvalidArguments):
|
||||
await store.emit(_snapshot(), idempotency_key="", ctx=_ctx())
|
||||
|
||||
|
||||
# --- get ---
|
||||
|
||||
def test_get_absent_returns_none():
|
||||
store = open_affect_store(":memory:")
|
||||
assert store.get("nope", "nope") is None
|
||||
|
||||
|
||||
async def test_get_after_emit_returns_equal():
|
||||
store = open_affect_store(":memory:")
|
||||
snap = _snapshot()
|
||||
await store.emit(snap, idempotency_key="k1", ctx=_ctx())
|
||||
assert store.get("a1", "u1") == snap
|
||||
|
||||
|
||||
# --- build_affect_provider_app ---
|
||||
|
||||
def test_build_app_exposes_handshake_and_affect_routes():
|
||||
store = open_affect_store(":memory:")
|
||||
app = build_affect_provider_app(store, heimdall_key=b"secret-key")
|
||||
routes = {getattr(r, "path", None): r for r in app.routes}
|
||||
assert "/bifrost/handshake" in routes
|
||||
assert "/bifrost/affect-call" in routes
|
||||
assert "POST" in routes["/bifrost/affect-call"].methods # POST-001: the verb, not just the path
|
||||
|
||||
|
||||
def test_build_app_rejects_non_advertising_store():
|
||||
store = open_affect_store(":memory:")
|
||||
store.affect_supported = False
|
||||
with pytest.raises(ValueError):
|
||||
build_affect_provider_app(store, heimdall_key=b"k")
|
||||
|
||||
|
||||
def test_build_app_rejects_empty_key():
|
||||
store = open_affect_store(":memory:")
|
||||
with pytest.raises(ValueError):
|
||||
build_affect_provider_app(store, heimdall_key=b"")
|
||||
|
||||
|
||||
# --- #195 conformance: parity vs the reference store through the real engine ---
|
||||
|
||||
def _dispatch_ctx(*scopes: str, session_id: str = "actor-1"):
|
||||
return types.SimpleNamespace(
|
||||
scope=list(scopes), session_id=session_id, jwt_sub=session_id
|
||||
)
|
||||
|
||||
|
||||
def _env(snap: dict, key: str = "sess-1:1:affect") -> dict:
|
||||
return {
|
||||
"operation": "affect.emit",
|
||||
"args": snap,
|
||||
"idempotency_key": key,
|
||||
"idempotency_class": "short-retry",
|
||||
}
|
||||
|
||||
|
||||
def _ref_shaped_snapshot(*, pleasure: float = 0.5, emitted_at: str = "2026-06-14T12:00:00Z"):
|
||||
# Mirror the reference test's snapshot shape so the envelope validates.
|
||||
return {
|
||||
"agent_id": "agent-1",
|
||||
"end_user_id": "user-1",
|
||||
"pad": {"pleasure": pleasure, "arousal": 0.2, "dominance": -0.1},
|
||||
"valence": [{"entity_id": "e1", "regard": 0.7, "familiarity": 0.3}],
|
||||
"emitted_at": emitted_at,
|
||||
}
|
||||
|
||||
|
||||
async def test_parity_vs_reference_store_through_dispatch():
|
||||
from bifrost.affect import dispatch_affect_call
|
||||
from bifrost.consumer.testing import InMemoryAffectStore
|
||||
|
||||
ref = InMemoryAffectStore()
|
||||
mine = open_affect_store(":memory:")
|
||||
ctx = _dispatch_ctx("affect:write")
|
||||
snap = _ref_shaped_snapshot()
|
||||
|
||||
# happy persist: wire bodies must agree
|
||||
assert await dispatch_affect_call(_env(snap), ctx, ref) == await dispatch_affect_call(
|
||||
_env(snap), ctx, mine
|
||||
)
|
||||
# replay (same key + same payload): both no-op {stored: true}
|
||||
assert await dispatch_affect_call(_env(snap), ctx, ref) == await dispatch_affect_call(
|
||||
_env(snap), ctx, mine
|
||||
)
|
||||
# conflict (same key + different payload): both map to the same error envelope
|
||||
other = _ref_shaped_snapshot(pleasure=0.99)
|
||||
assert await dispatch_affect_call(_env(other), ctx, ref) == await dispatch_affect_call(
|
||||
_env(other), ctx, mine
|
||||
)
|
||||
Reference in New Issue
Block a user