diff --git a/docs/contracts/bifrost_affect_provider.contract.md b/docs/contracts/bifrost_affect_provider.contract.md index 0955663..71796bf 100644 --- a/docs/contracts/bifrost_affect_provider.contract.md +++ b/docs/contracts/bifrost_affect_provider.contract.md @@ -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 ` 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=) 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=); UPSERT affect_idempotency (idempotency_id, digest, expires_at=); 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 diff --git a/pyproject.toml b/pyproject.toml index 83c117c..b97f706 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/src/ratatoskr/provider/__init__.py b/src/ratatoskr/provider/__init__.py new file mode 100644 index 0000000..c43b415 --- /dev/null +++ b/src/ratatoskr/provider/__init__.py @@ -0,0 +1,4 @@ +"""Tier-3 Bifrost consumer: durable memory.* + affect.* persistence provider. + +Contracts: docs/contracts/bifrost_affect_provider.contract.md (affect plane, v1). +""" diff --git a/src/ratatoskr/provider/affect_store.py b/src/ratatoskr/provider/affect_store.py new file mode 100644 index 0000000..b66e8d5 --- /dev/null +++ b/src/ratatoskr/provider/affect_store.py @@ -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) diff --git a/tests/test_provider_affect.py b/tests/test_provider_affect.py new file mode 100644 index 0000000..12263f6 --- /dev/null +++ b/tests/test_provider_affect.py @@ -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 + ) diff --git a/uv.lock b/uv.lock index d6c03e1..f43cc8b 100644 --- a/uv.lock +++ b/uv.lock @@ -1052,7 +1052,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.17.0" +version = "0.17.1" source = { editable = "." } dependencies = [ { name = "httpx" },