ratatoskr/docs/contracts/bifrost_affect_provider.contract.md

contract_version, module, purpose, touches, language, complexity, estimated_loc, confidence, assumptions, open_questions, external_invariants, revisions

contract_version	module	purpose	touches	language	complexity	estimated_loc	confidence	assumptions	open_questions	external_invariants	revisions
2.1	ratatoskr.provider.affect_store	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.	src/ratatoskr/provider/affect_store.py tests/test_provider_affect.py	python	medium	180	0.85	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 (ctx). The idempotency actor is derivable from ctx (mirrors bifrost's reference `_ctx_actor(ctx)` — the dispatch subject/actor identity).	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; affect.* uses a single short-retry class); confirm Worldtree never relies on class-scoped affect idempotency.	source invariant_id ~/development/bifrost/docs/contracts/affect.contract.md INV-001 source invariant_id ~/development/bifrost/bifrost/reference_server/affect.py InMemoryAffectStore.emit	version at summary delta 1.1 2026-06-14 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. ADDED MODIFIED REMOVED 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 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) the 'same idempotency_key + different content hash -> LWW overwrite' clause (it was backwards: bifrost treats that as a conflict)

Context

ratatoskr is the v1 Bifrost consumer — the durable persistence Worldtree writes Tier-3 agent state into. This contract specifies the affect plane slice: the first tracer-bullet through the whole consumer wire (handshake → 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. 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.

Data flow

• In: Worldtree's post-turn affect emit → POST /bifrost/affect-call → library validates envelope + per-dispatch JWT → store.emit(snapshot, *, idempotency_key, ctx, idempotency_class=None).
• 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: 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 affect.contract.md INV-001 so this contract stands alone). The store's OWN logic references ONLY snapshot["agent_id"] and snapshot["end_user_id"]. It MUST NOT index, attribute-access, validate, compare, or branch on pad / valence / persona_baselines / emitted_at. Mechanically serializing the whole dict (json.dumps) and hashing the bytes is explicitly PERMITTED — that 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 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 (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 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 — 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

SQLite in WAL mode (concurrent readers, single writer). emit writes are serialized by the per-(agent_id, end_user_id) primary key; last-write-wins is the upsert itself. No cross-row coordination — affect rows are independent.

Division of labor (library vs store)

A crisp line, since the responsibilities interleave: the bifrost library owns the entire wire — envelope validation, per-dispatch JWT verification, scope 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.

Constraints

• [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; 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.

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] 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, arrived_at TEXT,
       PRIMARY KEY (agent_id, end_user_id))
  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; both tables queryable
  reopen [happy]: open existing file twice → no error, schema idempotent

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, 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:
  AffectInvalidArguments:
    local_handling: raise on missing/empty addressing keys or empty idempotency_key
    flow_control: abort
    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"/"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}; 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)

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

FN build_affect_provider_app(store: RatatoskrAffectStore, heimdall_key: bytes, consumer_id: str = "ratatoskr") -> Starlette
BRIEF: Wire the JWT verifier + registration and hand the store to bifrost's build_affect_app.
PRE: [PRE-001 hard] store.affect_supported is True -- assert getattr(store, "affect_supported", False) is True (INV-005)
PRE: [PRE-002 hard] heimdall_key is non-empty bytes -- assert
POST: [POST-001 return_value] returns a Starlette app exposing POST /bifrost/handshake and POST /bifrost/affect-call -- assert routes present
STEPS:
  1. [setup] SET verifier = JwtVerifier(algorithm="HS256", key_bytes=heimdall_key)
  2. [setup] SET registration = ConsumerRegistration(consumer_id=consumer_id)
  3. [sequential, flexibility=prescriptive] SET app = build_affect_app(store=store, verifier=verifier, registration=registration)
  4. [cleanup] RETURN app
TESTS:
  builds_app [happy,tracer]: valid store + key → Starlette app with the two routes
  non_advertising_store [adversarial]: store with affect_supported=False → rejected (PRE-001 or library raises affect.unsupported_capability)
  bad_key [error]: empty heimdall_key → raises at construction