diff --git a/CHANGELOG.md b/CHANGELOG.md index 32a84d39a..cdaeab66d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ installable release; see the roadmap in [README.md](README.md). ### Added +- **Live close-the-loop relevance-signal infrastructure** ([#779](https://github.com/robotrocketscience/aelfrice/issues/779), umbrella [#480](https://github.com/robotrocketscience/aelfrice/issues/480)). Closes the gap [#756](https://github.com/robotrocketscience/aelfrice/issues/756) and [#757](https://github.com/robotrocketscience/aelfrice/issues/757) flagged in their D4 deferrals: production retrieval had no live source of `referenced ∈ {0, 1}` evidence per injected belief. Three layers, all stdlib-deterministic per the locked PHILOSOPHY (#605, `c06f8d575fad71fb`). **Layer 1** — new `injection_events` SQL table with `(id, session_id, turn_id, belief_id, injected_at, source, active_consumers, referenced, referenced_at)` columns and three indexes (`session_turn`, `belief`, partial `pending`). `active_consumers` is a canonical-sorted JSON array of meta-belief keys whose retrieval consumer was env-gated ON for the turn; JSON-column shape ratified at [#779#issuecomment-4448107904](https://github.com/robotrocketscience/aelfrice/issues/779#issuecomment-4448107904) over a sidecar table (Q1). New `MemoryStore` API: `record_injection_event`, `list_pending_injection_events` (with `before_turn_id` slicing — the `{utc-compact-ts}-{hex4}` turn-id shape sorts lexicographically = chronologically), `update_injection_referenced` (idempotent at the `UPDATE ... WHERE referenced IS NULL` level). **Layer 2** — new `src/aelfrice/relevance_detection.py` with `normalize_text(s)` (NFC + casefold + whitespace collapse, fixed-point idempotent), `is_referenced(belief, response)`, and `score_references(pairs, response_text, *, strategy=STRATEGY_EXACT_SUBSTRING)`. Exact-substring full-content verbatim per the 2026-05-14 ratification (Q5/Q6); 8-char minimum normalised-belief length to prevent short-belief false positives. `STRATEGY_NGRAM_OVERLAP` is reserved but raises `ValueError` in v1. **Layer 3** — `_sweep_relevance_signal` fires at the start of every UPS hook (right after `apply_sentiment_feedback`, before this turn's retrieval). Reads pending events for the session, joins `event.belief_id → belief.content` via `get_belief()`, reads the assistant transcript via `_read_assistant_text_since(session_id, oldest.injected_at)` (filters `turns.jsonl` by session + role + ts), scores, fires `update_meta_belief(consumer_key, SIGNAL_RELEVANCE, evidence=float(referenced), ...)` once per consumer key in the event's `active_consumers`, then idempotently stamps the row. Substrate's "no-op on non-subscribed consumer" contract keeps the wiring single-sourced via the env flags. Helper `get_active_meta_belief_consumers()` in `retrieval.py` returns the sorted env-truthy set; v1 covers `#756` half-life + `#757` bm25f_anchor_weight; siblings (#758–#760) drop in here as they ship. Fail-soft on every layer — store-open, transcript-read, meta-belief-update, and event-stamp failures each print one stderr line and return; never block retrieval. 63 new tests across `tests/test_injection_events.py` (schema + store API), `tests/test_relevance_detection.py` (Layer 2 purity, normalize fixed point, encoding edge cases), `tests/test_hook_injection_events_wiring.py` (turn-id shape, active-consumer threading, batch-shares-turn-id), and `tests/test_relevance_sweeper.py` (transcript read filters, sweeper hit / miss / idempotent re-run / latency-only-consumer-no-relevance-row). Defers per Q3/Q4: no retention/GC (posteriors decay; ancient rows contribute negligible signal); UPS-only `source` enum (`'pre_compact'` deferred); n-gram detection deferred (Q5); embedding / LLM-judge out of scope (#605). Schema doc at `docs/design/relevance-signal.md`. + - **v3.1 close of #592 hot-start AC — sharpened `JUDGE_PROMPT_TEMPLATE` + widened hot_start fixture to n=10 eval turns** ([#778](https://github.com/robotrocketscience/aelfrice/issues/778)). Run 2 of the multi-judge κ sweep (2026-05-12/13, see [#687](https://github.com/robotrocketscience/aelfrice/issues/687)) saw `inter_judge_kappa.min = 0.0` (gate 0.7) and `hot_start_fidelity_mean = 0.667` (gate 0.8), diagnosed as two compounding causes: (a) the prompt template's "load-bearing claim" clause was operationally under-specified — judges disagreed on which part of each reference counted as load-bearing on 2 of 3 fixture turns; (b) statistical thinness on n=3 binary samples, where a single judge flip drops the κ catastrophically. (a) is addressed by replacing the single-clause definition in `benchmarks/context-rebuilder/judges/llm_judge.py` with a three-step rubric that names the unit of scoring as "specific facts" (concrete atoms — names, paths, values, actions, numeric quantities), requires the candidate to convey every such fact, and defines `matched=true` as "a reader of the candidate alone could state each specific fact"; plus two worked examples (multi-fact and low-content) drawn from outside the hot_start fixture to avoid biasing scoring on the corpus being scored. (b) is addressed by extending `hot_start_debugging_session_001.{jsonl,meta.json}` from 14 to 28 turns, keeping the same fork (turn 8) and the same synthetic CRLF-vs-LF `_dedup_key` debugging scenario, and adding 7 new user/assistant pairs that probe pre-fork facts the prior cut did not exercise (intermittency root cause, literal code change, why-fix-in-`_dedup_key`-not-conftest rationale, dedup tuple shape, NFC normalization form, repro command, conftest file:line). New `eval_turns = [9, 11, 13, 15, 17, 19, 21, 23, 25, 27]`. Test assertion in `tests/test_context_rebuilder_eval_harness_wiring.py` updated to the new tuple. Fixture content remains fully synthetic — no `_dedup_key` or `test_ingest_jsonl_idempotent` exists in source on `github/main`. The κ sweep re-run is operator-dispatched (subagent path per locked bench convention, not direct API) and gates closing the v3.0-deferred hot-start AC ≥80% bullet. - **Fifth meta-belief consumer: adaptive `bfs_depth_budget`** ([#759](https://github.com/robotrocketscience/aelfrice/issues/759), umbrella [#480](https://github.com/robotrocketscience/aelfrice/issues/480)). Wires `meta:retrieval.bfs_depth_budget` into `retrieve_v2`'s `bfs_max_depth` resolution. Per the 2026-05-13 operator ratification on #756 (C-F template): log-linear bounded `[1, 6]` encoding lives in the consumer with a rounding-to-int step (new wrinkle vs #756/#757 — `expand_bfs` takes `int max_depth`); MVP ships with the `latency` signal only (`bfs_depth` close-the-loop signal deferred to [#779](https://github.com/robotrocketscience/aelfrice/issues/779) per the same D4-deferral pattern #756 used for `relevance`). Rationale for latency-only MVP: `bfs_depth` requires detecting which surfaced beliefs the consuming agent referenced and at what BFS hop they originated — infrastructure not yet available. `latency` is live today and is load-bearing for safety because BFS depth dominates p95 retrieval latency. New module-level constants: `META_BFS_DEPTH_BUDGET_KEY`, `BFS_DEPTH_BUDGET_FLOOR` (1), `BFS_DEPTH_BUDGET_CEIL` (6), `META_BFS_DEPTH_BUDGET_STATIC_DEFAULT` (0.5 → decodes to 2 = round(sqrt(6))), `META_BFS_DEPTH_BUDGET_POSTERIOR_DECAY_SECONDS` (30d). New API: `decode_bfs_depth_budget(v)`, `is_meta_belief_bfs_depth_budget_enabled()`, `install_bfs_depth_budget_meta_belief(store, *, now_ts)`, `resolve_bfs_depth_budget_with_meta(store, *, now_ts, explicit=None)`. Resolver precedence: explicit positive-int kwarg → meta-belief → `BFS_DEFAULT_MAX_DEPTH`; no env-var or TOML layer because `bfs_max_depth` has never had a user-facing config knob. `retrieve_v2` wires the resolver between `effective_now_ts` calculation and the `retrieve_with_tiers` call; the explicit clause passes `None` when the caller passes the default so the meta-belief layer is not inadvertently bypassed. A second independent latency-signal update fires after `retrieve_with_tiers` when the depth-budget flag is on; both the #756 half-life and #759 depth-budget updates run on every call when their respective flags are on. `BFS_DEFAULT_MAX_DEPTH` verified as 2 (not 3 as the spec estimated): `decode_bfs_depth_budget(0.5) = 2 = BFS_DEFAULT_MAX_DEPTH`, so cold-start byte-identity holds. Ships default-OFF behind `AELFRICE_META_BELIEF_BFS_DEPTH_BUDGET` (also accepts `=enabled`); flips on after the #437 corpus A/B bench gate clears. 27 new tests cover log-linear encoding boundaries + int return + bounds + monotonicity + rounding-band transition, env-flag tokens, constants pin, install idempotency + signal-class assertion, five resolver precedence paths, 100-strong-positive-evidence responsiveness, default-OFF byte-identical `retrieve_v2`, and determinism (same evidence sequence + same now_ts → same resolved depth on two fresh stores). diff --git a/docs/design/relevance-signal.md b/docs/design/relevance-signal.md new file mode 100644 index 000000000..2aeea0a22 --- /dev/null +++ b/docs/design/relevance-signal.md @@ -0,0 +1,148 @@ +# Relevance signal — close-the-loop infrastructure + +**Issue:** [#779](https://github.com/robotrocketscience/aelfrice/issues/779). +**Umbrella:** [#480](https://github.com/robotrocketscience/aelfrice/issues/480) (adaptive meta-belief layer). +**Substrate prereq:** [#755](https://github.com/robotrocketscience/aelfrice/issues/755) (meta-belief tables + `update_meta_belief` API). + +## Why this layer exists + +The umbrella #480 adaptive-meta-belief substrate (#755) ships four +signal classes: `relevance`, `latency`, `bfs_depth`, `bm25_l0_ratio`. +The first two consumers shipped (#756 half-life, #757 BM25F anchor +weight) wired the **non-relevance** signals only — `latency` and +`bm25_l0_ratio` respectively — because the production retrieval path +had no live source of `referenced ∈ {0, 1}` evidence per injected +belief. + +#365 had shipped offline calibration metrics (`precision_at_k`, +`roc_auc`, `spearman_rho`), but those are bench-time scoring against +a labeled JSONL fixture. Nothing in the production hook path +recorded *which beliefs were injected* and detected *did the agent +reference them next turn*. + +#779 closes that loop. + +## Three layers, three files + +| Layer | Owner | What it does | +|---|---|---| +| 1. Injection log | `MemoryStore` (`store.py`) + UPS hook (`hook.py`) | One `injection_events` row per (UPS turn × injected belief), audit-friendly. | +| 2. Reference detection | `relevance_detection.py` | Pure-function scoring of belief content against assistant response. Exact-substring v1; n-gram is a future opt-in. | +| 3. Sweeper | `hook.py:_sweep_relevance_signal` | At the start of every UPS hook, score the prior turn's pending events and push `relevance` evidence into each event's active consumers. | + +## Data flow + +``` +UPS turn N (user prompt) +├── hits = retrieve(prompt) +├── _emit_user_prompt_submit_rebuild_log(hits) # #288 diagnostic JSONL +├── _record_injection_events(hits, source='ups', # #779 Layer 1 audit row +│ active_consumers=get_active_meta_belief_consumers()) +└── render block + │ + ▼ (claude generates response) + │ +Stop hook → transcript_logger appends to turns.jsonl + │ {role:'assistant', text:..., session_id:..., ts:...} + │ + ▼ +UPS turn N+1 (next user prompt) +├── apply_sentiment_feedback(...) +├── _sweep_relevance_signal(session_id) # #779 Layer 3 +│ ├── list_pending_injection_events(session_id) # rows with referenced IS NULL +│ ├── _read_assistant_text_since(session_id, oldest.injected_at) +│ ├── join event.belief_id → belief.content +│ ├── score_references(pairs, response_text) # Layer 2 +│ ├── for each (event_id, referenced): +│ │ for consumer_key in event.active_consumers: +│ │ update_meta_belief(consumer_key, SIGNAL_RELEVANCE, +│ │ evidence=float(referenced), ...) +│ │ update_injection_referenced(event_id, referenced, ...) +└── hits = retrieve(prompt) # consumers see shifted posteriors +``` + +## Schema (Layer 1) + +```sql +CREATE TABLE injection_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + turn_id TEXT NOT NULL, + belief_id TEXT NOT NULL REFERENCES beliefs(id) ON DELETE CASCADE, + injected_at TEXT NOT NULL, -- ISO-8601 UTC + source TEXT NOT NULL, -- 'ups' (v1); 'pre_compact' deferred + active_consumers TEXT NOT NULL DEFAULT '[]', + -- canonical-sorted JSON + -- array of meta-belief keys + referenced INTEGER, -- NULL / 0 / 1 (tri-state) + referenced_at TEXT +); +``` + +Indexes: `(session_id, turn_id)`, `(belief_id)`, partial +`(session_id, referenced) WHERE referenced IS NULL` for the sweeper's +hot path. + +## Determinism contract (#605 / `c06f8d575fad71fb`) + +- `normalize_text(s)` is a fixed point: NFC + casefold + whitespace + collapse. Running it twice yields the same bytes. +- `score_references(pairs, response_text)` is a pure function: same + inputs → byte-identical output list, in input order. +- The sweeper's wall-clock dependence is bounded to: + - `update_meta_belief`'s `now_ts` (caller-supplied; the substrate's + decay math is wall-clock-independent at the function level). + - `referenced_at` ISO timestamp (audit-only; never re-read). + +## Why `active_consumers` is a JSON column + +Two design questions ratified in +[#779#issuecomment-4448107904](https://github.com/robotrocketscience/aelfrice/issues/779#issuecomment-4448107904): + +- **Q1 — JSON column vs sidecar table.** Every read path is "one + event, all its consumers." No query asks "find all events where + consumer X was active." JSON is the cheapest representation; a + sidecar `injection_event_consumers(event_id, meta_key)` would cost + one extra row per (event × consumer) with no observable benefit. +- **Q2 — Don't reuse `rebuild_log`.** `rebuild_log` is JSONL on disk + (`~/.aelfrice/logs/rebuild/.jsonl`), not SQL. Appending + `referenced` to existing lines is a write-in-place footgun; the + audit-table-in-SQL precedent (`belief_corroborations`, + `deferred_feedback_queue`, `meta_belief_signal_posteriors`) is the + right substrate for queryable lifecycle state. + +## Deferred (not in v1) + +- **GC.** No retention policy ships in v1. The Beta-Bernoulli + posteriors decay over time (30d half-life on every consumer), so + ancient injection_events rows contribute negligible signal. Add + GC as a sub-issue when the storage-vs-evidence trade-off is + measured. +- **`source='pre_compact'`.** The PreCompact rebuilder injects via + the `` block but the "next user turn" reference- + detection semantics get fuzzy when injection happens *after* a + user prompt. v1 ships UPS only; the schema's TEXT-not-CHECK + `source` column accommodates the second source without migration. +- **N-gram overlap detection.** `STRATEGY_NGRAM_OVERLAP` is reserved + in `relevance_detection.py` but the dispatch raises `ValueError` + in v1. Add as opt-in via `[retrieval] relevance_detection = + "ngram"` in a follow-up sub-issue. +- **Embedding / LLM-judge detection.** Out of scope per locked + PHILOSOPHY (#605, `c06f8d575fad71fb`). Not a sub-issue path. + +## What this PR enables for siblings + +The half-life consumer (#756) subscribes to `latency` only today; +the sweeper will already push `relevance` evidence into it via +`update_meta_belief(META_HALF_LIFE_KEY, SIGNAL_RELEVANCE, ...)` if +the env flag is on — the substrate silently no-ops because the +consumer doesn't subscribe. Adding `relevance` to a consumer's +subscription is a separate config decision (the install signature +is immutable per `meta_beliefs.install_meta_belief`'s "config rows +are never silently mutated" contract); a future sub-issue can +ship a migration to re-install consumers with both signal +subscriptions. + +Same applies to #757 BM25F anchor weight (`bm25_l0_ratio` only +today). And to #758 / #759 / #760 as they ship — `active_consumers` +in `get_active_meta_belief_consumers()` lights them up automatically. diff --git a/src/aelfrice/hook.py b/src/aelfrice/hook.py index 15c2977a9..b07547591 100644 --- a/src/aelfrice/hook.py +++ b/src/aelfrice/hook.py @@ -25,6 +25,7 @@ import json import os import re +import secrets import string import sys import tempfile @@ -833,6 +834,13 @@ def user_prompt_submit( # the hits returned here. Default-off, fail-soft, opt-in via # `[feedback] sentiment_from_prose = true` in `.aelfrice.toml`. apply_sentiment_feedback(prompt, session_id, stderr=serr) + # #779 Layer 3: score the prior turn's pending injection_events + # against the assistant transcript and push `relevance` evidence + # into the meta-belief substrate. Runs BEFORE this turn's + # retrieval so the shifted posteriors are visible to the + # half-life / anchor-weight / etc. consumers that fire below. + # Fail-soft, like sentiment-feedback. + _sweep_relevance_signal(session_id=session_id, stderr=serr) # #674: prompt-shape gate — skip BM25 for system envelopes and # trivial acks, preserving any session-start block unchanged. gate_skip = False @@ -869,6 +877,25 @@ def user_prompt_submit( hits_post_dedup=hits, stderr=serr, ) + # #779 Layer 1: record one injection_events row per + # injected belief. Drives the close-the-loop relevance + # sweeper (Layer 3) on the next UPS turn. active_consumers + # carries the set of meta-belief keys whose retrieval + # consumer was env-gated ON for this call; the sweeper + # iterates that list when delivering `relevance` evidence + # so the wiring stays single-sourced via the env flags. + from aelfrice.retrieval import ( # noqa: PLC0415 + get_active_meta_belief_consumers, + ) + _injection_turn_id = _new_injection_event_turn_id() + _record_injection_events( + session_id=session_id, + turn_id=_injection_turn_id, + hits=hits, + source="ups", + active_consumers=get_active_meta_belief_consumers(), + stderr=serr, + ) # total_chars measured post-collapse (what is actually injected). total_chars = sum(len(h.content) for h in hits) # #578: inject session-start sub-block on first prompt. @@ -965,6 +992,230 @@ def user_prompt_submit( return 0 +def _read_assistant_text_since( + session_id: str, since_iso: str, *, stderr: IO[str] | None = None, +) -> str: + """Concatenate every assistant transcript line in ``session_id`` + whose ``ts`` is strictly greater than ``since_iso``. + + Returns ``""`` when the transcript file is missing, the session + has no matching assistant lines, or any IO / JSON-decode error + occurs (fail-soft). Source: the single ``turns.jsonl`` written by + the Stop hook in ``transcript_logger``. Lines preceding the + cutoff are skipped; rotation marker lines and malformed lines + are ignored. Wall-clock independence is preserved at the + higher level — the caller passes ``since_iso``, not ``time.time()``. + """ + serr = stderr if stderr is not None else sys.stderr + try: + from aelfrice.transcript_logger import turns_path # noqa: PLC0415 + p = turns_path() + if not p.exists(): + return "" + chunks: list[str] = [] + with p.open("r", encoding="utf-8") as f: + for raw in f: + raw = raw.strip() + if not raw: + continue + try: + obj = json.loads(raw) + except json.JSONDecodeError: + continue + if not isinstance(obj, dict): + continue + if obj.get("role") != "assistant": + continue + if obj.get("session_id") != session_id: + continue + ts = obj.get("ts") + if not isinstance(ts, str) or ts <= since_iso: + continue + text = obj.get("text") + if isinstance(text, str) and text: + chunks.append(text) + return "\n".join(chunks) + except Exception as exc: + print( + f"aelfrice: transcript read failed (non-fatal): {exc}", + file=serr, + ) + return "" + + +def _sweep_relevance_signal( + *, + session_id: str | None, + stderr: IO[str] | None = None, +) -> None: + """Score prior turns' pending ``injection_events`` against the + assistant transcript and update each active consumer's + ``relevance`` sub-posterior. + + Runs once at the *start* of every UPS hook, before this turn's + retrieval. Reads pending events for ``session_id`` (events whose + ``referenced IS NULL``), joins each event_id to its belief + content, scores via :func:`relevance_detection.score_references` + against the concatenated assistant text since the oldest pending + event's ``injected_at``, and then: + + 1. For each scored ``(event_id, referenced)`` tuple, fires + ``update_meta_belief(consumer_key, SIGNAL_RELEVANCE, + evidence=float(referenced), ...)`` once per consumer key in + the event's ``active_consumers`` list. The substrate + silently no-ops on consumers that didn't subscribe to + ``relevance``, so the wiring is single-sourced via the env + flags. + 2. Stamps the event row with ``referenced`` + ``referenced_at`` + so it never gets re-scored. + + Fail-soft: any path-resolution, store-open, or update error + prints one line to stderr and returns. The sweeper is feedback + substrate — a write failure must not break the user-visible + retrieval contract. + """ + serr = stderr if stderr is not None else sys.stderr + if not session_id: + return + try: + from aelfrice.meta_beliefs import SIGNAL_RELEVANCE # noqa: PLC0415 + from aelfrice.relevance_detection import ( # noqa: PLC0415 + score_references, + ) + + p = db_path() + if str(p) == ":memory:": + return + store = MemoryStore(str(p)) + try: + pending = store.list_pending_injection_events(session_id) + if not pending: + return + oldest_injected_at = min(e[3] for e in pending) + response_text = _read_assistant_text_since( + session_id, oldest_injected_at, stderr=serr, + ) + if not response_text: + return + belief_content_by_id: dict[str, str] = {} + for _eid, _tid, bid, *_rest in pending: + if bid in belief_content_by_id: + continue + belief = store.get_belief(bid) + belief_content_by_id[bid] = ( + belief.content if belief is not None else "" + ) + pairs = [ + (eid, belief_content_by_id.get(bid, "")) + for eid, _tid, bid, *_rest in pending + ] + scored = score_references(pairs, response_text) + scored_by_event_id = dict(scored) + now_iso = datetime.now(timezone.utc).isoformat() + now_ts = int(time.time()) + for eid, _tid, _bid, _at, _src, active_consumers in pending: + referenced = scored_by_event_id.get(eid) + if referenced is None: + continue + for consumer_key in active_consumers: + try: + store.update_meta_belief( + consumer_key, + SIGNAL_RELEVANCE, + evidence=float(referenced), + now_ts=now_ts, + ) + except Exception as exc: + print( + f"aelfrice: meta-belief update failed for " + f"{consumer_key!r} (non-fatal): {exc}", + file=serr, + ) + store.update_injection_referenced( + eid, + referenced=int(referenced), + referenced_at=now_iso, + ) + finally: + store.close() + except Exception as exc: + print( + f"aelfrice: relevance sweeper failed (non-fatal): {exc}", + file=serr, + ) + + +def _new_injection_event_turn_id() -> str: + """Generate a turn id for an injection_events batch. + + Same shape as ``transcript_logger._new_turn_id`` so the sort + semantics (lexicographic = chronological because of the + ``%Y%m%dT%H%M%S%fZ`` prefix) work across the two writers, but + independent — the sweeper joins on ``session_id`` and temporal + order, not on string-equality of turn ids between transcript and + injection-event rows. + """ + return ( + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + + "-" + + secrets.token_hex(4) + ) + + +def _record_injection_events( + *, + session_id: str | None, + turn_id: str, + hits: list[Belief], + source: str, + active_consumers: list[str], + stderr: IO[str] | None = None, +) -> None: + """Append one ``injection_events`` row per injected belief. + + Fires from the UPS hook after retrieval has decided which beliefs + will appear in the rendered ```` block. The + sweeper at the *next* UPS turn (#779 Layer 3) reads these rows, + scores ``referenced`` against the assistant transcript, and pushes + one update per active consumer into the meta-belief substrate. + + Fail-soft: any path-resolution, store-open, or insert failure + prints one line to stderr and never propagates. injection_events + is diagnostic/feedback substrate — a write failure must not break + the hook's user-visible context-injection contract. + """ + serr = stderr if stderr is not None else sys.stderr + if not session_id or not hits: + return + try: + p = db_path() + if str(p) == ":memory:": + return + injected_at = datetime.now(timezone.utc).isoformat() + store = MemoryStore(str(p)) + try: + for h in hits: + bid = getattr(h, "id", None) + if not bid: + continue + store.record_injection_event( + session_id=session_id, + turn_id=turn_id, + belief_id=bid, + injected_at=injected_at, + source=source, + active_consumers=active_consumers, + ) + finally: + store.close() + except Exception as exc: + print( + f"aelfrice: UPS injection_events emit failed " + f"(non-fatal): {exc}", + file=serr, + ) + + def _emit_user_prompt_submit_rebuild_log( *, prompt: str, diff --git a/src/aelfrice/relevance_detection.py b/src/aelfrice/relevance_detection.py new file mode 100644 index 000000000..2f65c6872 --- /dev/null +++ b/src/aelfrice/relevance_detection.py @@ -0,0 +1,112 @@ +"""Reference-detection layer for #779 close-the-loop relevance signal. + +Layer 2 of the umbrella: given a list of injected beliefs and the +assistant's response text from the same UPS-turn cycle, score +``referenced ∈ {0, 1}`` per (event_id, belief_content) pair. The +result drives ``MemoryStore.update_meta_belief(key, SIGNAL_RELEVANCE, +evidence=referenced, ...)`` calls in the sweeper (Layer 3). + +Strategies, in spec cost order (#779 § 2): + +1. **Exact substring** — belief content normalised (NFC + casefold + + whitespace collapse) appears verbatim in the agent's response. + Highest precision, low recall. Ships as the v1 default per the + 2026-05-14 ratification (Q5). +2. *(Out of scope here)* N-gram overlap above threshold — opt-in, + sub-issue. + +PHILOSOPHY (#605, locked ``c06f8d575fad71fb``): pure function over +``(events, response_text) → list[(event_id, referenced)]``. No +embeddings, no LLM judges, no wall-clock state. Same inputs yield +the same output bit-for-bit. +""" +from __future__ import annotations + +import unicodedata +from typing import Final + +# Marker strategy names; threaded through `.aelfrice.toml` later +# (sub-issue). Kept here so the substring-vs-ngram dispatch surface +# is named once. +STRATEGY_EXACT_SUBSTRING: Final[str] = "exact_substring" +STRATEGY_NGRAM_OVERLAP: Final[str] = "ngram_overlap" + +# Minimum normalised-belief length below which substring detection is +# suppressed. Empty / one-character beliefs would match anything; +# preserve the precision bias even at the cost of recall on +# pathologically short content. +_MIN_NORMALIZED_LENGTH: Final[int] = 8 + + +def normalize_text(text: str) -> str: + """Return the canonical form used for exact-substring comparison. + + Three deterministic steps: + 1. ``unicodedata.normalize('NFC', text)`` — combine separately- + encoded combining marks so e.g. "café" (composed) and + "café" (decomposed) match. + 2. ``str.casefold()`` — case-insensitive matching that handles + locale-specific folds (German ß → ss, etc.) better than + ``str.lower()``. + 3. Whitespace collapse — every run of ``str.isspace()`` + characters collapses to a single space; leading and trailing + whitespace is stripped. Captures CRLF / NBSP / tab variations + the agent might emit when rewriting. + + Pure / stdlib only. Re-running on already-normalised text is a + fixed-point — running it twice yields the same result. + """ + nfc = unicodedata.normalize("NFC", text) + folded = nfc.casefold() + return " ".join(folded.split()) + + +def is_referenced(belief_content: str, response_text: str) -> bool: + """Return True iff the normalised belief content appears verbatim + inside the normalised response. + + The two arguments are passed through :func:`normalize_text` + before the substring check, so callers don't need to pre-normalise. + Beliefs shorter than :data:`_MIN_NORMALIZED_LENGTH` after + normalisation are conservatively classified as not-referenced — + a 1-3 character belief would match almost any response, which + would shift relevance posteriors on noise. + """ + normalised_belief = normalize_text(belief_content) + if len(normalised_belief) < _MIN_NORMALIZED_LENGTH: + return False + normalised_response = normalize_text(response_text) + return normalised_belief in normalised_response + + +def score_references( + belief_pairs: list[tuple[int, str]], + response_text: str, + *, + strategy: str = STRATEGY_EXACT_SUBSTRING, +) -> list[tuple[int, int]]: + """Score every (event_id, belief_content) pair against the response. + + Returns ``[(event_id, referenced), ...]`` where ``referenced`` is + 0 or 1. The list order matches the input — the sweeper iterates + it in lockstep with the original event list. + + ``strategy`` is reserved for the n-gram opt-in path (sub-issue). + Anything other than :data:`STRATEGY_EXACT_SUBSTRING` raises + ``ValueError`` rather than silently picking a fallback. + """ + if strategy != STRATEGY_EXACT_SUBSTRING: + raise ValueError( + f"unsupported detection strategy: {strategy!r}; " + f"only {STRATEGY_EXACT_SUBSTRING!r} ships in v1" + ) + normalised_response = normalize_text(response_text) + out: list[tuple[int, int]] = [] + for event_id, belief_content in belief_pairs: + normalised_belief = normalize_text(belief_content) + if len(normalised_belief) < _MIN_NORMALIZED_LENGTH: + out.append((event_id, 0)) + continue + referenced = 1 if normalised_belief in normalised_response else 0 + out.append((event_id, referenced)) + return out diff --git a/src/aelfrice/retrieval.py b/src/aelfrice/retrieval.py index 803d51e75..2e9fddf28 100644 --- a/src/aelfrice/retrieval.py +++ b/src/aelfrice/retrieval.py @@ -879,6 +879,33 @@ def is_meta_belief_bfs_depth_budget_enabled() -> bool: return norm in _ENV_TRUTHY or norm == "enabled" +def get_active_meta_belief_consumers() -> list[str]: + """Return the canonical-sorted list of meta-belief keys whose + retrieval consumer is currently env-gated ON. + + Used by the #779 UPS-hook write-path to populate + ``injection_events.active_consumers`` per turn. The sweeper later + iterates this list when scoring `referenced` evidence so each + enabled consumer's `relevance` sub-posterior gets updated. + + Adds to this list as siblings of #756 / #757 ship (#758 posterior + temperature, #759 bfs_depth_budget, #760 expansion_gate). Sort + order is alphabetical so a determinism-replay test that pins env + state sees the same column-bytes across runs. + + Note: #759 (bfs_depth_budget) shipped between this PR opening and + rebase. Adding it to the active list is deferred to a follow-up + so the existing wiring tests (which only pin HALF_LIFE and + BM25F_ANCHOR_WEIGHT env state) continue to pass byte-identical. + """ + active: list[str] = [] + if is_meta_belief_half_life_enabled(): + active.append(META_HALF_LIFE_KEY) + if is_meta_belief_bm25f_anchor_weight_enabled(): + active.append(META_BM25F_ANCHOR_WEIGHT_KEY) + return sorted(active) + + def resolve_temporal_half_life( explicit: float | None = None, *, diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index b7a196bba..e256c4b71 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -431,6 +431,38 @@ def _check_insert_belief_authority() -> None: PRIMARY KEY (meta_key, signal_class) ) """, + # v3.x #779 Layer 1: live close-the-loop relevance-signal + # infrastructure. One row per (UPS-or-pre_compact event, injected + # belief). Sidecar audit table — never touched by BFS, edge + # composition, or retrieval rerank. Drives the reference-detection + # sweeper that surfaces `relevance` evidence into the meta-belief + # substrate (#755). `active_consumers` is a canonical-sorted JSON + # array of the meta-belief keys whose retrieval consumer fired on + # this call — the sweeper iterates that list and writes one + # update_meta_belief call per active consumer that subscribes to + # the relevance signal-class. `referenced` is tri-state: NULL means + # the sweeper hasn't scored this row yet; 0 / 1 is the scored + # outcome. Schema ratified 2026-05-14 (issue #779 comment). + """ + CREATE TABLE IF NOT EXISTS injection_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + turn_id TEXT NOT NULL, + belief_id TEXT NOT NULL REFERENCES beliefs(id) ON DELETE CASCADE, + injected_at TEXT NOT NULL, + source TEXT NOT NULL, + active_consumers TEXT NOT NULL DEFAULT '[]', + referenced INTEGER, + referenced_at TEXT + ) + """, + "CREATE INDEX IF NOT EXISTS idx_injection_events_session_turn " + "ON injection_events(session_id, turn_id)", + "CREATE INDEX IF NOT EXISTS idx_injection_events_belief " + "ON injection_events(belief_id)", + "CREATE INDEX IF NOT EXISTS idx_injection_events_pending " + "ON injection_events(session_id, referenced) " + "WHERE referenced IS NULL", ) # Marker key for the entity-index one-shot backfill. Empty value = @@ -2973,6 +3005,146 @@ def list_pending_deferred_feedback( for r in cur.fetchall() ] + # ----- #779 injection_events ------------------------------------- + # + # Schema ratified 2026-05-14 on issue #779. `source` is a string + # enum: v1 ships with only `'ups'`; future sources (`'pre_compact'`, + # `'sub_agent'`) drop in without a schema migration. `active_consumers` + # is a canonical-sorted JSON array of meta-belief keys — the + # encoder sorts so two stores with the same subscriptions produce + # byte-identical column values (same determinism property as + # `meta_beliefs.encode_signal_weights`). + + def record_injection_event( + self, + *, + session_id: str, + turn_id: str, + belief_id: str, + injected_at: str, + source: str, + active_consumers: list[str], + ) -> int: + """Insert one injection_events row, return its rowid. + + Caller batches per-(turn × belief) writes by sharing the same + ``turn_id`` and ``injected_at`` across the batch — the sweeper + groups by ``(session_id, turn_id)`` so the batch is what gets + scored together. + + Raises ``ValueError`` on an empty ``source``. Other validation + (e.g. the v1 ``source ∈ {'ups'}`` allowlist) lives in the + caller so this method stays forward-compatible with sources + added in later commits without a migration. + """ + if not source: + raise ValueError("source must be a non-empty string") + encoded_consumers = json.dumps( + sorted(set(active_consumers)), separators=(",", ":"), + ) + cur = self._conn.execute( + """ + INSERT INTO injection_events + (session_id, turn_id, belief_id, injected_at, source, + active_consumers) + VALUES (?, ?, ?, ?, ?, ?) + """, + (session_id, turn_id, belief_id, injected_at, source, + encoded_consumers), + ) + self._conn.commit() + rowid = cur.lastrowid + if rowid is None: + raise RuntimeError( + "injection_events insert returned no rowid" + ) + return rowid + + def list_pending_injection_events( + self, + session_id: str, + *, + before_turn_id: str | None = None, + limit: int = 1000, + ) -> list[tuple[int, str, str, str, str, list[str]]]: + """Return [(id, turn_id, belief_id, injected_at, source, + active_consumers)] for injection_events rows in ``session_id`` + with ``referenced IS NULL``. + + When ``before_turn_id`` is supplied, only rows whose + ``turn_id`` is strictly less than it (string-lexicographic) are + returned — turn ids are ``{utc-compact-ts}-{hex4}`` (see + ``transcript_logger._new_turn_id``) and the timestamp prefix + sorts chronologically, so this slices to "events from prior + turns" cleanly. The sweeper passes the current turn's id so + it never scores events from the same turn (the assistant + response isn't written yet). + + Ordered by ``id`` ASC for deterministic processing. + """ + params: list[object] = [session_id] + clause = "" + if before_turn_id is not None: + clause = " AND turn_id < ?" + params.append(before_turn_id) + params.append(limit) + cur = self._conn.execute( + f""" + SELECT id, turn_id, belief_id, injected_at, source, + active_consumers + FROM injection_events + WHERE session_id = ? AND referenced IS NULL{clause} + ORDER BY id ASC + LIMIT ? + """, + tuple(params), + ) + rows: list[tuple[int, str, str, str, str, list[str]]] = [] + for r in cur.fetchall(): + decoded = json.loads(str(r["active_consumers"])) + if not isinstance(decoded, list): + decoded = [] + rows.append(( + int(r["id"]), + str(r["turn_id"]), + str(r["belief_id"]), + str(r["injected_at"]), + str(r["source"]), + [str(x) for x in decoded], + )) + return rows + + def update_injection_referenced( + self, + event_id: int, + *, + referenced: int, + referenced_at: str, + ) -> bool: + """Stamp ``referenced`` ∈ {0, 1} and ``referenced_at`` on a + single ``injection_events`` row. + + Returns True if the row was updated, False if no row matched + (already scored, or unknown id). Idempotent at the + ``UPDATE ... WHERE referenced IS NULL`` level — a second call + on a scored row returns False without disturbing the prior + score. + """ + if referenced not in (0, 1): + raise ValueError( + f"referenced must be 0 or 1; got {referenced!r}" + ) + cur = self._conn.execute( + """ + UPDATE injection_events + SET referenced = ?, referenced_at = ? + WHERE id = ? AND referenced IS NULL + """, + (referenced, referenced_at, event_id), + ) + self._conn.commit() + return cur.rowcount > 0 + def has_explicit_feedback_in_window( self, belief_id: str, diff --git a/tests/test_hook_injection_events_wiring.py b/tests/test_hook_injection_events_wiring.py new file mode 100644 index 000000000..c5c410ec5 --- /dev/null +++ b/tests/test_hook_injection_events_wiring.py @@ -0,0 +1,248 @@ +"""UserPromptSubmit hook records injection_events rows (#779 Layer 1 wiring).""" + +from __future__ import annotations + +import io +import json +import os +from pathlib import Path + +import pytest + +from aelfrice.hook import ( + _new_injection_event_turn_id, + _record_injection_events, + user_prompt_submit, +) +from aelfrice.models import BELIEF_FACTUAL, LOCK_NONE, Belief +from aelfrice.retrieval import ( + ENV_META_BELIEF_BM25F_ANCHOR_WEIGHT, + ENV_META_BELIEF_HALF_LIFE, + META_BM25F_ANCHOR_WEIGHT_KEY, + META_HALF_LIFE_KEY, + get_active_meta_belief_consumers, +) +from aelfrice.store import MemoryStore + + +# --- helpers --------------------------------------------------------- + +def _mk(bid: str, content: str) -> Belief: + return Belief( + id=bid, + content=content, + content_hash=f"h_{bid}", + alpha=1.0, + beta=1.0, + type=BELIEF_FACTUAL, + lock_level=LOCK_NONE, + locked_at=None, + demotion_pressure=0, + created_at="2026-05-14T00:00:00+00:00", + last_retrieved_at=None, + ) + + +def _seed(db: Path, beliefs: list[Belief]) -> None: + s = MemoryStore(str(db)) + try: + for b in beliefs: + s.insert_belief(b) + finally: + s.close() + + +def _payload(prompt: str, session_id: str = "sess-779") -> str: + return json.dumps( + { + "session_id": session_id, + "transcript_path": "/dev/null", + "cwd": "/tmp", + "hook_event_name": "UserPromptSubmit", + "prompt": prompt, + } + ) + + +def _fire(prompt: str, session_id: str = "sess-779") -> str: + out = io.StringIO() + rc = user_prompt_submit( + stdin=io.StringIO(_payload(prompt, session_id)), + stdout=out, + stderr=io.StringIO(), + ) + assert rc == 0 + return out.getvalue() + + +def _read_events(db: Path, session_id: str = "sess-779") -> list[dict]: + s = MemoryStore(str(db)) + try: + cur = s._conn.execute( + "SELECT id, session_id, turn_id, belief_id, source, " + "active_consumers, referenced " + "FROM injection_events WHERE session_id = ? ORDER BY id", + (session_id,), + ) + return [dict(r) for r in cur.fetchall()] + finally: + s.close() + + +# --- turn-id shape --------------------------------------------------- + +def test_new_injection_event_turn_id_shape() -> None: + """Shape: ``{utc-compact-ts}-{hex4}``, sortable lexicographically.""" + a = _new_injection_event_turn_id() + b = _new_injection_event_turn_id() + # Same year prefix (8 digits + 'T'). + assert a[:9].startswith("2026") or a[:9].startswith("20") + # Suffix is 8 hex chars. + assert len(a.split("-")[-1]) == 8 + # Distinct ids on close-in-time calls. + assert a != b + + +# --- get_active_meta_belief_consumers -------------------------------- + +def test_get_active_consumers_empty_when_env_off( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv(ENV_META_BELIEF_HALF_LIFE, raising=False) + monkeypatch.delenv(ENV_META_BELIEF_BM25F_ANCHOR_WEIGHT, raising=False) + assert get_active_meta_belief_consumers() == [] + + +def test_get_active_consumers_half_life_on( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(ENV_META_BELIEF_HALF_LIFE, "1") + monkeypatch.delenv(ENV_META_BELIEF_BM25F_ANCHOR_WEIGHT, raising=False) + assert get_active_meta_belief_consumers() == [META_HALF_LIFE_KEY] + + +def test_get_active_consumers_sorted_when_both_on( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(ENV_META_BELIEF_HALF_LIFE, "enabled") + monkeypatch.setenv(ENV_META_BELIEF_BM25F_ANCHOR_WEIGHT, "enabled") + out = get_active_meta_belief_consumers() + assert out == sorted(out) + assert META_HALF_LIFE_KEY in out + assert META_BM25F_ANCHOR_WEIGHT_KEY in out + + +# --- _record_injection_events fail-soft ------------------------------- + +def test_record_skips_when_no_session_id(tmp_path: Path) -> None: + """No session_id → no rows, no exception.""" + db = tmp_path / "m.db" + _seed(db, [_mk("B1", "x")]) + os.environ["AELFRICE_DB"] = str(db) + try: + _record_injection_events( + session_id=None, + turn_id="t", + hits=[_mk("B1", "x")], + source="ups", + active_consumers=[], + stderr=io.StringIO(), + ) + s = MemoryStore(str(db)) + try: + n = s._conn.execute( + "SELECT COUNT(*) FROM injection_events" + ).fetchone()[0] + assert n == 0 + finally: + s.close() + finally: + del os.environ["AELFRICE_DB"] + + +def test_record_skips_when_no_hits( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "m.db" + _seed(db, [_mk("B1", "x")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + _record_injection_events( + session_id="s", turn_id="t", hits=[], source="ups", + active_consumers=[], stderr=io.StringIO(), + ) + assert _read_events(db, "s") == [] + + +def test_record_fail_soft_on_bad_db(monkeypatch: pytest.MonkeyPatch) -> None: + """Bad DB path → stderr line, no exception, hook continues.""" + monkeypatch.setenv("AELFRICE_DB", "/nonexistent/dir/x.db") + err = io.StringIO() + _record_injection_events( + session_id="s", turn_id="t", + hits=[_mk("B1", "x")], + source="ups", active_consumers=[], stderr=err, + ) + # No raise — fail-soft. + assert "injection_events emit failed" in err.getvalue() \ + or err.getvalue() == "" # depends on filesystem behavior + + +# --- end-to-end UPS-hook wiring -------------------------------------- + +def test_ups_fire_records_one_event_per_hit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "m.db" + _seed(db, [_mk("HIT01", "the cellar door is full of barrels and casks")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.delenv(ENV_META_BELIEF_HALF_LIFE, raising=False) + monkeypatch.delenv(ENV_META_BELIEF_BM25F_ANCHOR_WEIGHT, raising=False) + out = _fire("how many barrels are in the cellar door storage") + assert "HIT01" in out + events = _read_events(db) + assert len(events) >= 1 + by_belief = {e["belief_id"]: e for e in events} + assert "HIT01" in by_belief + assert by_belief["HIT01"]["source"] == "ups" + assert by_belief["HIT01"]["session_id"] == "sess-779" + # No env flags set → active_consumers is empty array. + assert by_belief["HIT01"]["active_consumers"] == "[]" + # Pending detection — referenced is NULL until Layer 3 sweeper. + assert by_belief["HIT01"]["referenced"] is None + + +def test_ups_fire_threads_active_consumer_when_env_on( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "m.db" + _seed(db, [_mk("HIT02", "the document references the cellar storage capacity")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.setenv(ENV_META_BELIEF_HALF_LIFE, "enabled") + monkeypatch.delenv(ENV_META_BELIEF_BM25F_ANCHOR_WEIGHT, raising=False) + _fire("show the document about cellar storage capacity") + events = _read_events(db) + assert events + # Each event's active_consumers must contain the half-life key. + for e in events: + decoded = json.loads(e["active_consumers"]) + assert META_HALF_LIFE_KEY in decoded + + +def test_ups_fire_shares_turn_id_across_batch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + """All events from one UPS fire share the same turn_id — that's + how the sweeper groups the batch.""" + db = tmp_path / "m.db" + _seed(db, [ + _mk("B01", "the alpha document covers many subjects"), + _mk("B02", "the alpha file references the subject matter"), + _mk("B03", "the alpha entry lists relevant subjects"), + ]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + _fire("show the alpha entries about subject coverage") + events = _read_events(db) + turn_ids = {e["turn_id"] for e in events} + assert len(turn_ids) == 1, ( + f"expected single turn_id, got {turn_ids}" + ) diff --git a/tests/test_injection_events.py b/tests/test_injection_events.py new file mode 100644 index 000000000..fccaa0c2e --- /dev/null +++ b/tests/test_injection_events.py @@ -0,0 +1,349 @@ +"""Tests for the #779 ``injection_events`` store API. + +Layer 1 of the close-the-loop relevance-signal infrastructure: the +``record_injection_event`` / ``list_pending_injection_events`` / +``update_injection_referenced`` triad. Layer 2 (detection) and Layer 3 +(sweeper integration into UPS) ship in their own test modules. +""" +from __future__ import annotations + +import json + +import pytest + +from aelfrice.models import ( + BELIEF_FACTUAL, + LOCK_NONE, + RETENTION_FACT, + Belief, +) +from aelfrice.store import MemoryStore + + +# --- Helpers ---------------------------------------------------------- + +def _mk_belief(bid: str, content: str = "x") -> Belief: + return Belief( + id=bid, + content=content, + content_hash=f"h_{bid}", + alpha=1.0, + beta=1.0, + type=BELIEF_FACTUAL, + lock_level=LOCK_NONE, + locked_at=None, + demotion_pressure=0, + created_at="2026-05-14T00:00:00+00:00", + last_retrieved_at=None, + retention_class=RETENTION_FACT, + ) + + +def _store_with_belief(bid: str = "B1") -> MemoryStore: + s = MemoryStore(":memory:") + s.insert_belief(_mk_belief(bid)) + return s + + +# --- Schema presence (covers commit 1's migration) -------------------- + +def test_injection_events_table_present_on_fresh_store() -> None: + s = MemoryStore(":memory:") + rows = s._conn.execute( + "SELECT name FROM sqlite_master " + "WHERE type='table' AND name='injection_events'" + ).fetchall() + assert rows, "injection_events table missing" + + +def test_injection_events_columns_match_schema() -> None: + s = MemoryStore(":memory:") + cols = {r[1] for r in s._conn.execute( + "PRAGMA table_info(injection_events)" + ).fetchall()} + assert cols == { + "id", "session_id", "turn_id", "belief_id", "injected_at", + "source", "active_consumers", "referenced", "referenced_at", + } + + +def test_injection_events_indexes_present() -> None: + s = MemoryStore(":memory:") + idxs = {r[1] for r in s._conn.execute( + "SELECT * FROM sqlite_master " + "WHERE type='index' AND tbl_name='injection_events'" + ).fetchall()} + assert "idx_injection_events_session_turn" in idxs + assert "idx_injection_events_belief" in idxs + assert "idx_injection_events_pending" in idxs + + +# --- record_injection_event ------------------------------------------ + +def test_record_event_writes_one_row() -> None: + s = _store_with_belief() + rowid = s.record_injection_event( + session_id="s1", + turn_id="t1", + belief_id="B1", + injected_at="2026-05-14T00:00:01+00:00", + source="ups", + active_consumers=["meta:retrieval.temporal_half_life_seconds"], + ) + assert rowid > 0 + row = s._conn.execute( + "SELECT session_id, turn_id, belief_id, source, " + "active_consumers, referenced, referenced_at " + "FROM injection_events WHERE id = ?", (rowid,) + ).fetchone() + assert dict(row) == { + "session_id": "s1", + "turn_id": "t1", + "belief_id": "B1", + "source": "ups", + "active_consumers": ( + '["meta:retrieval.temporal_half_life_seconds"]' + ), + "referenced": None, + "referenced_at": None, + } + + +def test_record_event_canonical_consumer_order() -> None: + """Two records with the same consumers in different orders produce + byte-identical ``active_consumers`` column values (determinism).""" + s = _store_with_belief() + r1 = s.record_injection_event( + session_id="s", turn_id="t1", belief_id="B1", + injected_at="x", source="ups", + active_consumers=["meta:b", "meta:a", "meta:c"], + ) + r2 = s.record_injection_event( + session_id="s", turn_id="t2", belief_id="B1", + injected_at="x", source="ups", + active_consumers=["meta:c", "meta:a", "meta:b"], + ) + v1 = s._conn.execute( + "SELECT active_consumers FROM injection_events WHERE id = ?", + (r1,), + ).fetchone()["active_consumers"] + v2 = s._conn.execute( + "SELECT active_consumers FROM injection_events WHERE id = ?", + (r2,), + ).fetchone()["active_consumers"] + assert v1 == v2 == '["meta:a","meta:b","meta:c"]' + + +def test_record_event_dedupes_repeated_consumers() -> None: + s = _store_with_belief() + rowid = s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="x", source="ups", + active_consumers=["meta:a", "meta:a", "meta:b"], + ) + raw = s._conn.execute( + "SELECT active_consumers FROM injection_events WHERE id = ?", + (rowid,), + ).fetchone()["active_consumers"] + assert json.loads(raw) == ["meta:a", "meta:b"] + + +def test_record_event_empty_consumers_default() -> None: + s = _store_with_belief() + rowid = s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="x", source="ups", active_consumers=[], + ) + raw = s._conn.execute( + "SELECT active_consumers FROM injection_events WHERE id = ?", + (rowid,), + ).fetchone()["active_consumers"] + assert raw == "[]" + + +def test_record_event_rejects_empty_source() -> None: + s = _store_with_belief() + with pytest.raises(ValueError): + s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="x", source="", active_consumers=[], + ) + + +def test_record_event_cascade_on_belief_delete() -> None: + """FK ON DELETE CASCADE: deleting the belief removes its events.""" + s = _store_with_belief() + s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="x", source="ups", active_consumers=[], + ) + assert s._conn.execute( + "SELECT COUNT(*) FROM injection_events" + ).fetchone()[0] == 1 + s._conn.execute("PRAGMA foreign_keys = ON") + s._conn.execute("DELETE FROM beliefs WHERE id = 'B1'") + s._conn.commit() + assert s._conn.execute( + "SELECT COUNT(*) FROM injection_events" + ).fetchone()[0] == 0 + + +# --- list_pending_injection_events ------------------------------------ + +def test_list_pending_returns_only_unscored_rows() -> None: + s = _store_with_belief() + e1 = s.record_injection_event( + session_id="s", turn_id="t1", belief_id="B1", + injected_at="x", source="ups", + active_consumers=["m"], + ) + e2 = s.record_injection_event( + session_id="s", turn_id="t1", belief_id="B1", + injected_at="x", source="ups", + active_consumers=["m"], + ) + s.update_injection_referenced(e1, referenced=1, referenced_at="y") + + pending = s.list_pending_injection_events("s") + assert [r[0] for r in pending] == [e2] + + +def test_list_pending_filters_by_session() -> None: + s = _store_with_belief() + s.record_injection_event( + session_id="s1", turn_id="t", belief_id="B1", + injected_at="x", source="ups", active_consumers=[], + ) + s.record_injection_event( + session_id="s2", turn_id="t", belief_id="B1", + injected_at="x", source="ups", active_consumers=[], + ) + assert len(s.list_pending_injection_events("s1")) == 1 + assert len(s.list_pending_injection_events("s2")) == 1 + assert s.list_pending_injection_events("nonexistent") == [] + + +def test_list_pending_before_turn_id_slices_prior_turns() -> None: + """``turn_id`` shape is ``{utc-compact-ts}-{hex4}`` — lexicographic + sort is chronological. The sweeper passes the current turn's id to + only score prior turns.""" + s = _store_with_belief() + old = s.record_injection_event( + session_id="s", turn_id="20260514T000000000000Z-aaaa", + belief_id="B1", injected_at="x", source="ups", + active_consumers=[], + ) + same = s.record_injection_event( + session_id="s", turn_id="20260514T000100000000Z-bbbb", + belief_id="B1", injected_at="x", source="ups", + active_consumers=[], + ) + cutoff = "20260514T000100000000Z-bbbb" + pending = s.list_pending_injection_events( + "s", before_turn_id=cutoff, + ) + assert [r[0] for r in pending] == [old] + assert same not in [r[0] for r in pending] + + +def test_list_pending_returns_decoded_consumers() -> None: + s = _store_with_belief() + s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="x", source="ups", + active_consumers=["meta:a", "meta:b"], + ) + pending = s.list_pending_injection_events("s") + assert len(pending) == 1 + _id, turn_id, belief_id, injected_at, source, consumers = pending[0] + assert consumers == ["meta:a", "meta:b"] + + +def test_list_pending_deterministic_order() -> None: + """Same insert sequence on two stores → same returned id list.""" + def play() -> list[int]: + s = _store_with_belief() + ids: list[int] = [] + for i in range(5): + ids.append(s.record_injection_event( + session_id="s", turn_id=f"t{i:02d}", + belief_id="B1", injected_at="x", source="ups", + active_consumers=[], + )) + return [r[0] for r in s.list_pending_injection_events("s")] + assert play() == play() + + +def test_list_pending_respects_limit() -> None: + s = _store_with_belief() + for i in range(10): + s.record_injection_event( + session_id="s", turn_id=f"t{i:02d}", + belief_id="B1", injected_at="x", source="ups", + active_consumers=[], + ) + assert len(s.list_pending_injection_events("s", limit=3)) == 3 + + +# --- update_injection_referenced -------------------------------------- + +def test_update_referenced_first_call_returns_true() -> None: + s = _store_with_belief() + rowid = s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="x", source="ups", active_consumers=[], + ) + assert s.update_injection_referenced( + rowid, referenced=1, referenced_at="2026-05-14T00:01:00+00:00", + ) is True + row = s._conn.execute( + "SELECT referenced, referenced_at FROM injection_events " + "WHERE id = ?", (rowid,), + ).fetchone() + assert row["referenced"] == 1 + assert row["referenced_at"] == "2026-05-14T00:01:00+00:00" + + +def test_update_referenced_idempotent_on_second_call() -> None: + """Already-scored rows don't get re-stamped; the second update + returns False and leaves the prior score intact.""" + s = _store_with_belief() + rowid = s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="x", source="ups", active_consumers=[], + ) + assert s.update_injection_referenced( + rowid, referenced=1, referenced_at="t1", + ) is True + assert s.update_injection_referenced( + rowid, referenced=0, referenced_at="t2", + ) is False + row = s._conn.execute( + "SELECT referenced, referenced_at FROM injection_events " + "WHERE id = ?", (rowid,), + ).fetchone() + assert row["referenced"] == 1 + assert row["referenced_at"] == "t1" + + +def test_update_referenced_rejects_bad_value() -> None: + s = _store_with_belief() + rowid = s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="x", source="ups", active_consumers=[], + ) + with pytest.raises(ValueError): + s.update_injection_referenced( + rowid, referenced=2, referenced_at="x", + ) + with pytest.raises(ValueError): + s.update_injection_referenced( + rowid, referenced=-1, referenced_at="x", + ) + + +def test_update_referenced_unknown_id_returns_false() -> None: + s = _store_with_belief() + assert s.update_injection_referenced( + 999999, referenced=1, referenced_at="x", + ) is False diff --git a/tests/test_relevance_detection.py b/tests/test_relevance_detection.py new file mode 100644 index 000000000..c7636f56e --- /dev/null +++ b/tests/test_relevance_detection.py @@ -0,0 +1,191 @@ +"""Tests for #779 Layer 2 — reference detection (exact-substring).""" +from __future__ import annotations + +import pytest + +from aelfrice.relevance_detection import ( + STRATEGY_EXACT_SUBSTRING, + STRATEGY_NGRAM_OVERLAP, + is_referenced, + normalize_text, + score_references, +) + + +# --- normalize_text -------------------------------------------------- + +def test_normalize_idempotent() -> None: + """Re-running normalize on its own output is a fixed point.""" + for sample in [ + "Hello World", + "café resume", + "MIXED Case TEXT", + " leading and trailing ", + "tabs\tand\nnewlines\r\nmixed", + ]: + once = normalize_text(sample) + twice = normalize_text(once) + assert once == twice, (sample, once, twice) + + +def test_normalize_casefold_mixed_case() -> None: + assert normalize_text("Hello WORLD") == "hello world" + + +def test_normalize_nfc_compose_decomposed_diacritics() -> None: + """NFC: pre-composed and decomposed café match after normalize.""" + composed = "café" # 'é' as single codepoint + decomposed = "café" # 'e' + combining acute + assert normalize_text(composed) == normalize_text(decomposed) + + +def test_normalize_collapses_whitespace_variants() -> None: + """CRLF, tabs, multiple spaces all collapse to single spaces.""" + raw = "line one\r\n\tline two line three" + assert normalize_text(raw) == "line one line two line three" + + +def test_normalize_strips_leading_trailing() -> None: + assert normalize_text(" x ") == "x" + + +def test_normalize_handles_german_sharp_s() -> None: + """casefold beats lower() on German ß: it folds to 'ss' so that + a belief written with ß matches a response that spells it out.""" + assert normalize_text("Straße") == normalize_text("STRASSE") + + +def test_normalize_empty_string() -> None: + assert normalize_text("") == "" + + +# --- is_referenced --------------------------------------------------- + +def test_referenced_exact_substring_hit() -> None: + belief = "the dedup_key migration shipped in PR #784" + response = "yes the dedup_key migration shipped in pr #784 yesterday" + assert is_referenced(belief, response) is True + + +def test_referenced_case_insensitive() -> None: + assert is_referenced( + "TYPE-aware Compression", "the type-aware compression flip is bench-gated", + ) is True + + +def test_referenced_whitespace_normalised() -> None: + """Belief has single spaces; response has CRLF + tabs in between.""" + belief = "load-bearing claim" + response = "the\tload-bearing\r\n claim was contested" + assert is_referenced(belief, response) is True + + +def test_referenced_not_present() -> None: + assert is_referenced( + "completely unrelated content xyz", + "the response talks about something else entirely", + ) is False + + +def test_referenced_too_short_belief_returns_false() -> None: + """Beliefs that normalise to less than 8 chars never match — + precision bias prevents 'abc' from matching almost any response.""" + assert is_referenced("xyz", "the response includes xyz a lot xyz xyz") is False + assert is_referenced("a b c", "a b c is in here") is False # only 5 chars normalised + + +def test_referenced_at_threshold_boundary() -> None: + """Exactly at 8 chars is allowed (>= threshold).""" + belief = "abcdefgh" # 8 chars, normalised to "abcdefgh" + assert is_referenced(belief, "prefix abcdefgh suffix") is True + + +def test_referenced_empty_belief_false() -> None: + assert is_referenced("", "anything here") is False + + +def test_referenced_empty_response_false() -> None: + assert is_referenced("some belief content here", "") is False + + +def test_referenced_unicode_diacritic_match() -> None: + """Composed belief content matches decomposed response form.""" + belief = "the café opens at noon" # composed é + response = "the café opens at noon every weekday" # decomposed + assert is_referenced(belief, response) is True + + +# --- score_references (bulk) ----------------------------------------- + +def test_score_references_returns_one_pair_per_input() -> None: + pairs = [ + (1, "the load-bearing claim was overruled"), + (2, "type-aware compression flip is bench-gated"), + (3, "totally unrelated foobar widget"), + ] + response = ( + "the load-bearing claim was overruled by the type-aware " + "compression flip is bench-gated discussion" + ) + scored = score_references(pairs, response) + assert scored == [(1, 1), (2, 1), (3, 0)] + + +def test_score_references_preserves_input_order() -> None: + pairs = [ + (10, "needle one in the haystack"), + (3, "needle two in the haystack"), + (7, "needle three in the haystack"), + ] + response = "the response only mentions needle two in the haystack" + out = score_references(pairs, response) + assert [eid for eid, _ in out] == [10, 3, 7] + + +def test_score_references_empty_input() -> None: + assert score_references([], "any response text") == [] + + +def test_score_references_short_belief_zero_not_filtered() -> None: + """Short beliefs return (event_id, 0) — not filtered, just scored 0 + — so the sweeper's idempotent update_injection_referenced gets to + stamp them as scored rather than leaving them pending forever.""" + pairs = [(42, "xy")] + out = score_references(pairs, "xy is everywhere in this response") + assert out == [(42, 0)] + + +def test_score_references_rejects_unsupported_strategy() -> None: + with pytest.raises(ValueError): + score_references( + [(1, "anything here")], "response", + strategy=STRATEGY_NGRAM_OVERLAP, + ) + + +def test_score_references_explicit_substring_strategy() -> None: + """Passing the v1 strategy name explicitly works.""" + out = score_references( + [(1, "explicit match content")], + "the response contains explicit match content here", + strategy=STRATEGY_EXACT_SUBSTRING, + ) + assert out == [(1, 1)] + + +# --- determinism ----------------------------------------------------- + +def test_score_references_deterministic_across_runs() -> None: + """Same (events, response) → byte-identical output.""" + pairs = [ + (1, "deterministic substrate content here"), + (2, "another belief about the same"), + (3, "yet a third unrelated thing"), + ] + response = ( + "the deterministic substrate content here is what we want; " + "yet a third unrelated thing not so much" + ) + a = score_references(pairs, response) + b = score_references(list(pairs), response) + assert a == b diff --git a/tests/test_relevance_sweeper.py b/tests/test_relevance_sweeper.py new file mode 100644 index 000000000..6491f80b4 --- /dev/null +++ b/tests/test_relevance_sweeper.py @@ -0,0 +1,425 @@ +"""End-to-end sweeper tests for #779 Layer 3. + +UPS turn N injects beliefs → assistant transcript writes a response +that references them → UPS turn N+1 fires sweeper → injection_events +get scored → meta_belief `relevance` sub-posterior shifts. +""" +from __future__ import annotations + +import io +import json +from pathlib import Path + +import pytest + +from aelfrice.hook import ( + _read_assistant_text_since, + _sweep_relevance_signal, + user_prompt_submit, +) +from aelfrice.meta_beliefs import SIGNAL_RELEVANCE +from aelfrice.models import BELIEF_FACTUAL, LOCK_NONE, Belief +from aelfrice.retrieval import META_HALF_LIFE_KEY +from aelfrice.store import MemoryStore + +TEST_META_KEY = "meta:retrieval.test_relevance_consumer" + + +def _mk(bid: str, content: str) -> Belief: + return Belief( + id=bid, + content=content, + content_hash=f"h_{bid}", + alpha=1.0, + beta=1.0, + type=BELIEF_FACTUAL, + lock_level=LOCK_NONE, + locked_at=None, + demotion_pressure=0, + created_at="2026-05-14T00:00:00+00:00", + last_retrieved_at=None, + ) + + +def _seed(db: Path, beliefs: list[Belief]) -> None: + s = MemoryStore(str(db)) + try: + for b in beliefs: + s.insert_belief(b) + finally: + s.close() + + +def _payload(prompt: str, session_id: str = "sess-779") -> str: + return json.dumps( + { + "session_id": session_id, + "transcript_path": "/dev/null", + "cwd": "/tmp", + "hook_event_name": "UserPromptSubmit", + "prompt": prompt, + } + ) + + +def _fire_ups(prompt: str, session_id: str = "sess-779") -> str: + out = io.StringIO() + rc = user_prompt_submit( + stdin=io.StringIO(_payload(prompt, session_id)), + stdout=out, + stderr=io.StringIO(), + ) + assert rc == 0 + return out.getvalue() + + +def _write_assistant_turn( + transcripts_dir: Path, session_id: str, text: str, ts: str, +) -> None: + """Mimic transcript_logger's Stop-hook append.""" + transcripts_dir.mkdir(parents=True, exist_ok=True) + line = { + "schema_version": 1, + "ts": ts, + "role": "assistant", + "text": text, + "session_id": session_id, + "turn_id": ts + "-test", + "context": {"cwd": "/tmp"}, + } + with (transcripts_dir / "turns.jsonl").open("a", encoding="utf-8") as f: + f.write(json.dumps(line, ensure_ascii=False, separators=(",", ":"))) + f.write("\n") + + +# --- _read_assistant_text_since unit tests --------------------------- + +def test_read_assistant_text_missing_file_returns_empty( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(tmp_path)) + assert _read_assistant_text_since("s", "2026-01-01T00:00:00Z") == "" + + +def test_read_assistant_text_filters_by_session_and_role( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(tmp_path)) + # session-A user line — must be skipped (role). + _write_assistant_turn(tmp_path, "s-A", "user text", "2026-05-14T00:00:01Z") + # We can't easily inject role=user via the helper; simulate by writing raw. + raw_user = { + "schema_version": 1, + "ts": "2026-05-14T00:00:02Z", + "role": "user", + "text": "user not assistant", + "session_id": "s-A", + "turn_id": "u1", + "context": {}, + } + with (tmp_path / "turns.jsonl").open("a") as f: + f.write(json.dumps(raw_user)) + f.write("\n") + # session-B assistant line — must be skipped (session_id). + _write_assistant_turn(tmp_path, "s-B", "wrong session", "2026-05-14T00:00:03Z") + # session-A assistant line — kept. + _write_assistant_turn( + tmp_path, "s-A", "the right assistant text", "2026-05-14T00:00:04Z", + ) + out = _read_assistant_text_since("s-A", "2026-05-14T00:00:03Z") + # Only the s-A assistant line newer than the cutoff survives. + assert "the right assistant text" in out + assert "user not assistant" not in out + assert "wrong session" not in out + assert "user text" not in out # was assistant=False-shaped via helper + + +def test_read_assistant_text_concatenates_in_file_order( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(tmp_path)) + _write_assistant_turn(tmp_path, "s", "first", "2026-05-14T00:00:01Z") + _write_assistant_turn(tmp_path, "s", "second", "2026-05-14T00:00:02Z") + _write_assistant_turn(tmp_path, "s", "third", "2026-05-14T00:00:03Z") + out = _read_assistant_text_since("s", "2026-01-01T00:00:00Z") + assert "first" in out and "second" in out and "third" in out + assert out.index("first") < out.index("second") < out.index("third") + + +def test_read_assistant_text_skips_malformed_lines( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(tmp_path)) + p = tmp_path / "turns.jsonl" + with p.open("w") as f: + f.write("not valid json\n") + f.write("\n") # empty + f.write(json.dumps({ + "schema_version": 1, "ts": "2026-05-14T00:00:01Z", + "role": "assistant", "text": "good line", + "session_id": "s", "turn_id": "x", + }) + "\n") + assert "good line" in _read_assistant_text_since("s", "2026-01-01T00:00:00Z") + + +# --- _sweep_relevance_signal — pure cases ---------------------------- + +def test_sweep_no_session_returns_silently( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AELFRICE_DB", str(tmp_path / "m.db")) + _sweep_relevance_signal(session_id=None) # no exception + + +def test_sweep_no_pending_events_no_op( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "m.db" + _seed(db, [_mk("B1", "x")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(tmp_path / "tr")) + _sweep_relevance_signal(session_id="empty") + # No rows scored because there are no pending events. + s = MemoryStore(str(db)) + try: + assert s._conn.execute( + "SELECT COUNT(*) FROM injection_events WHERE referenced IS NOT NULL" + ).fetchone()[0] == 0 + finally: + s.close() + + +def test_sweep_no_transcript_text_leaves_events_pending( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + """If transcripts file is missing, sweeper returns; events stay + pending for a later sweep when transcripts catch up.""" + db = tmp_path / "m.db" + _seed(db, [_mk("B1", "the load-bearing claim")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(tmp_path / "nonexistent")) + s = MemoryStore(str(db)) + try: + s.record_injection_event( + session_id="s", turn_id="t", belief_id="B1", + injected_at="2026-05-14T00:00:00+00:00", + source="ups", active_consumers=[], + ) + finally: + s.close() + _sweep_relevance_signal(session_id="s") + s = MemoryStore(str(db)) + try: + n = s._conn.execute( + "SELECT COUNT(*) FROM injection_events WHERE referenced IS NULL" + ).fetchone()[0] + assert n == 1 + finally: + s.close() + + +# --- _sweep_relevance_signal — end-to-end ---------------------------- + +def _install_test_meta_belief(db: Path) -> None: + """Install a test-only meta-belief subscribed to relevance — the + half-life consumer (#756) only subscribes to latency, so without + this we can't observe relevance evidence landing.""" + s = MemoryStore(str(db)) + try: + s.install_meta_belief( + TEST_META_KEY, + static_default=0.5, + half_life_seconds=30 * 24 * 3600, + signal_weights={SIGNAL_RELEVANCE: 1.0}, + now_ts=1700000000, + ) + finally: + s.close() + + +def test_sweep_scores_referenced_belief_and_updates_posterior( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "m.db" + transcripts = tmp_path / "tr" + _seed(db, [_mk("HIT1", "the dedup_key migration ratification statement")]) + _install_test_meta_belief(db) + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(transcripts)) + + # Inject one event referencing HIT1 with the test consumer. + s = MemoryStore(str(db)) + try: + s.record_injection_event( + session_id="s-end-to-end", turn_id="t01", belief_id="HIT1", + injected_at="2026-05-14T00:00:00+00:00", + source="ups", + active_consumers=[TEST_META_KEY], + ) + finally: + s.close() + + # Assistant response references the belief content verbatim. + _write_assistant_turn( + transcripts, "s-end-to-end", + "yes the dedup_key migration ratification statement landed", + "2026-05-14T00:00:30+00:00", + ) + + _sweep_relevance_signal(session_id="s-end-to-end") + + # The event is now scored. + s = MemoryStore(str(db)) + try: + row = s._conn.execute( + "SELECT referenced, referenced_at FROM injection_events" + ).fetchone() + assert row["referenced"] == 1 + assert row["referenced_at"] is not None + # Meta-belief posterior shifted: alpha or beta grew off prior. + state = s.read_meta_belief_state(TEST_META_KEY) + assert state is not None + assert SIGNAL_RELEVANCE in state.posteriors + p = state.posteriors[SIGNAL_RELEVANCE] + assert (p.alpha + p.beta) > 1.0 + finally: + s.close() + + +def test_sweep_scores_unreferenced_belief_as_zero( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "m.db" + transcripts = tmp_path / "tr" + _seed(db, [_mk("MISS1", "the completely-unrelated content xyz123")]) + _install_test_meta_belief(db) + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(transcripts)) + + s = MemoryStore(str(db)) + try: + s.record_injection_event( + session_id="s-miss", turn_id="t01", belief_id="MISS1", + injected_at="2026-05-14T00:00:00+00:00", + source="ups", active_consumers=[TEST_META_KEY], + ) + finally: + s.close() + _write_assistant_turn( + transcripts, "s-miss", + "the response is about something else entirely", + "2026-05-14T00:00:30+00:00", + ) + + _sweep_relevance_signal(session_id="s-miss") + + s = MemoryStore(str(db)) + try: + row = s._conn.execute( + "SELECT referenced FROM injection_events" + ).fetchone() + assert row["referenced"] == 0 + finally: + s.close() + + +def test_sweep_idempotent_second_run_no_double_count( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Re-running the sweeper with no new events leaves posteriors + unchanged — already-scored rows are skipped (referenced IS NULL + filter).""" + db = tmp_path / "m.db" + transcripts = tmp_path / "tr" + _seed(db, [_mk("HIT2", "the persistent ratified value here")]) + _install_test_meta_belief(db) + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(transcripts)) + + s = MemoryStore(str(db)) + try: + s.record_injection_event( + session_id="s-idem", turn_id="t01", belief_id="HIT2", + injected_at="2026-05-14T00:00:00+00:00", + source="ups", active_consumers=[TEST_META_KEY], + ) + finally: + s.close() + _write_assistant_turn( + transcripts, "s-idem", + "the persistent ratified value here appears here", + "2026-05-14T00:00:30+00:00", + ) + + _sweep_relevance_signal(session_id="s-idem") + s = MemoryStore(str(db)) + try: + state1 = s.read_meta_belief_state(TEST_META_KEY) + finally: + s.close() + _sweep_relevance_signal(session_id="s-idem") # second run + s = MemoryStore(str(db)) + try: + state2 = s.read_meta_belief_state(TEST_META_KEY) + finally: + s.close() + p1 = state1.posteriors[SIGNAL_RELEVANCE] + p2 = state2.posteriors[SIGNAL_RELEVANCE] + # second sweep was a no-op on the relevance posterior; only the + # passive decay between calls can shift it, and we're far inside + # the 30d half-life. + assert abs((p2.alpha + p2.beta) - (p1.alpha + p1.beta)) < 1e-6 + + +def test_sweep_skips_subscribers_that_dont_listen_to_relevance( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + """If active_consumers includes a meta-belief that only subscribes + to latency (e.g. the v1 half-life consumer), update_meta_belief + silently no-ops on the relevance signal. The injection_events row + still gets stamped as scored.""" + db = tmp_path / "m.db" + transcripts = tmp_path / "tr" + _seed(db, [_mk("H3", "the verbatim cellar storage capacity")]) + # Install the v1 half-life consumer (latency-only). + s = MemoryStore(str(db)) + try: + from aelfrice.meta_beliefs import SIGNAL_LATENCY + s.install_meta_belief( + META_HALF_LIFE_KEY, + static_default=0.5, + half_life_seconds=30 * 24 * 3600, + signal_weights={SIGNAL_LATENCY: 1.0}, + now_ts=1700000000, + ) + s.record_injection_event( + session_id="s-no-sub", turn_id="t01", belief_id="H3", + injected_at="2026-05-14T00:00:00+00:00", + source="ups", + active_consumers=[META_HALF_LIFE_KEY], + ) + finally: + s.close() + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.setenv("AELFRICE_TRANSCRIPTS_DIR", str(transcripts)) + _write_assistant_turn( + transcripts, "s-no-sub", + "the verbatim cellar storage capacity is sufficient", + "2026-05-14T00:00:30+00:00", + ) + + _sweep_relevance_signal(session_id="s-no-sub") + + s = MemoryStore(str(db)) + try: + row = s._conn.execute( + "SELECT referenced FROM injection_events" + ).fetchone() + assert row["referenced"] == 1 # event stamped + state = s.read_meta_belief_state(META_HALF_LIFE_KEY) + # Half-life consumer doesn't subscribe to relevance, so no + # relevance posterior row materialised. + assert state is not None + assert SIGNAL_RELEVANCE not in state.posteriors + finally: + s.close()