diff --git a/CHANGELOG.md b/CHANGELOG.md index 28e72d26..5830442e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +- Deterministic retrieval ordering: every equal-score tie (RRF fusion, graph and temporal stages, FTS/vector stage runs) now resolves through a content-stable cascade (event date, content length, text, capture fingerprint) instead of falling through to row ids — re-ingesting the same content yields byte-identical packs (two-seed divergence 7/40 → 0/40); disclosed as `fusion.tie_break: content_stable_v1` in retrieval traces. +- Benchmark harness gains a disclosed `--accept-rollups` step (default off, fingerprint-stamped) that review-accepts consolidation roll-up proposals through the real acceptance path, modeling the product's human review workflow; offline measurement found current roll-up grouping quality too noisy to help the benchmark, so the flag stays off and grouping quality is queued as product work. +- Grounding for products: the context-pack entity note now recognizes quantity-qualified and possessive compound entities ("30-gallon tank", "my snake plant") with conservative false-positive guards, and a new answer-verification library seam (`vnext_answer_verification`) lets integrators opt into post-generation grounding checks; nothing changes by default. + - Time-aware retrieval: temporal anchors parsed from the query ("two weeks ago", "in March 2023", "between X and Y") join RRF fusion as one more ranked list against event dates — never a hard filter, dormant on date-free queries, honest `temporal_anchor` trace stage. - Coverage mode for aggregation-shaped recall ("how many…", "list every…"): query-surface intent gate, capped clause decomposition, and instance-diversity fusion keyed on `(source_id, source_chunk_id)` so distinct instances fill the slots; dormant path byte-identical. - Consolidation roll-up cards: the merge engine proposes review-gated cards that pre-aggregate same-topic instances with per-instance dates, values, and speaker provenance; accepted cards are first-class recallable memories, members stay individually recallable, nothing auto-promotes, zero added commit-path work. diff --git a/apps/api/src/alicebot_api/mcp_tools.py b/apps/api/src/alicebot_api/mcp_tools.py index 98ee3a6e..af8272ff 100644 --- a/apps/api/src/alicebot_api/mcp_tools.py +++ b/apps/api/src/alicebot_api/mcp_tools.py @@ -1118,8 +1118,14 @@ def _handle_alice_recall(context: MCPRuntimeContext, arguments: Mapping[str, obj # graph stage. Only present when the query matched entities. payload["entities"] = matched_entities if debug: + from alicebot_api.vnext_retrieval import TIE_BREAK_CONTENT_STABLE + payload["retrieval"] = { - "fusion": {"algorithm": "reciprocal_rank_fusion", "k": RRF_K}, + "fusion": { + "algorithm": "reciprocal_rank_fusion", + "k": RRF_K, + "tie_break": TIE_BREAK_CONTENT_STABLE, + }, "vector_stage": vector_stage, "context_depth": context_depth, "budget_strategy": budget_strategy, @@ -2506,6 +2512,16 @@ def _handle_alice_context_pack(context: MCPRuntimeContext, arguments: Mapping[st recent_changes = pack.get("recent_changes") if isinstance(recent_changes, list) and recent_changes: payload["recent_changes"] = recent_changes + # -- entity grounding passthrough (vnext_grounding; single block) ------- + # pack["grounding"] exists only when a salient query entity has ZERO + # corpus support (see vnext_grounding); the compact view must not + # silently drop it or the honesty statistic never reaches the tool + # caller. Absent for every ungated query, so ordinary responses are + # byte-identical. + grounding = pack.get("grounding") + if isinstance(grounding, Mapping) and grounding: + payload["grounding"] = dict(grounding) + # -- end entity grounding passthrough ------------------------------------ if debug: payload["query_interpretation"] = dict(interpretation) payload["trace"] = pack.get("trace") diff --git a/apps/api/src/alicebot_api/vnext_answer_verification.py b/apps/api/src/alicebot_api/vnext_answer_verification.py new file mode 100644 index 00000000..8cf72460 --- /dev/null +++ b/apps/api/src/alicebot_api/vnext_answer_verification.py @@ -0,0 +1,328 @@ +"""Product answer-grounding verification: an OPT-IN library seam, wired nowhere. + +This module adapts the LongMemEval harness verification concept +(``eval/longmemeval/verification.py``) for product use: after a chat +layer has drafted an answer from an Alice context pack, an integrator +MAY ask a model to list the answer's concrete claims that the pack does +not support, and then gate or annotate the answer with the verdict. + +NOTHING calls this module by default. No API route, MCP tool, or +retrieval path imports it; adding it to a deployment changes no +behavior until an integrator explicitly calls +:func:`verify_answer_grounding` in their own answering loop. The +harness copy stays byte-frozen and independent — this is a separate +product surface with its own template and its own tests. + +Design constraints carried over from the harness (and asserted by +``tests/unit/test_vnext_answer_verification.py``): + +- The verifier sees ONLY the pack-derived context block, the question, + and the answer under test. There is no "expected answer" anywhere in + this module's inputs, and no code path reads benchmark labels. +- Fail-open: a provider error or an unparseable reply yields a verdict + with ``grounded=True`` and the failure recorded, so verification can + only ever be a disclosed, inspectable filter — never a hidden crash + or a silent rewrite. :func:`apply_answer_grounding_gate` returns the + answer byte-identical unless a CLEAN verdict found a load-bearing + ungrounded claim. +- Disclosure: every verdict's ``to_record()`` carries the template + fingerprint and the verifier provider/model when known, so any run + that used the gate can prove which prompt and model produced it. + +Typical opt-in wiring (integrator code, not Alice code):: + + pack = VNextRetrievalService(store).compile_context_pack(request) + answer = my_chat_layer.answer(question, pack) + verdict = verify_answer_grounding(answer, pack, provider) + final, gated = apply_answer_grounding_gate(answer, verdict) +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +import hashlib +import time + +from alicebot_api.vnext_repositories import JsonObject + + +VERIFY_TEMPERATURE = 0.0 + +# What the gate substitutes when an integrator opts into gating and the +# verdict found a load-bearing fabrication. Deliberately plain: it +# claims nothing about WHY, only that stored memory does not back the +# drafted answer. +WITHHELD_ANSWER_TEXT = ( + "I don't have stored memories that support an answer to this; " + "the drafted answer contained specifics my memory does not back." +) + +# The product verifier prompt. Same claim taxonomy as the harness +# template (LOAD-BEARING vs INCIDENTAL, strict line format) but a +# separate constant: the harness template is byte-frozen for benchmark +# integrity and must never be coupled to product wording changes. +ANSWER_VERIFIER_PROMPT_TEMPLATE = ( + "You are verifying that an answer is grounded in its source context. " + "Use only the context below; do not use outside knowledge.\n\n" + "Context:\n{context}\n\n" + "Question:\n{question}\n\n" + "Answer to verify:\n{answer}\n\n" + "List every specific factual claim in the answer (a name, number, date, " + "place, or other concrete value) that the context does not state or " + "directly support. Label a claim LOAD-BEARING if it is the direct answer " + "to the question; label it INCIDENTAL otherwise. Use one line per claim, " + "exactly:\n" + "UNGROUNDED LOAD-BEARING: \n" + "UNGROUNDED INCIDENTAL: \n" + "If every claim is supported by the context, or the answer only says the " + "information is unavailable, reply with exactly: GROUNDED" +) + +ANSWER_VERIFIER_TEMPLATE_SHA256 = hashlib.sha256( + ANSWER_VERIFIER_PROMPT_TEMPLATE.encode("utf-8") +).hexdigest() + +GROUNDED_TOKEN = "GROUNDED" +_LOAD_BEARING_PREFIX = "ungrounded load-bearing:" +_INCIDENTAL_PREFIX = "ungrounded incidental:" + +# Context-block rendering caps: the verifier needs the same evidence +# the answer was drafted from, not an unbounded dump. +_MAX_CONTEXT_ITEMS_PER_SECTION = 40 +_MAX_ITEM_CHARS = 700 + + +@dataclass(frozen=True, slots=True) +class UngroundedClaim: + text: str + load_bearing: bool + + def to_record(self) -> JsonObject: + return {"text": self.text, "load_bearing": self.load_bearing} + + +@dataclass(frozen=True, slots=True) +class AnswerGroundingVerdict: + """Outcome of one product verification call; always recorded in full.""" + + grounded: bool + ungrounded_claims: tuple[UngroundedClaim, ...] + raw_response: str | None + error: str | None + parse_note: str | None + latency_seconds: float | None + provider: str | None + model: str | None + + @property + def gate_should_withhold(self) -> bool: + """Withhold only on a CLEAN verdict with load-bearing failures (fail-open).""" + return self.error is None and not self.grounded + + def to_record(self) -> JsonObject: + return { + "grounded": self.grounded, + "ungrounded_claims": [claim.to_record() for claim in self.ungrounded_claims], + "raw_response": self.raw_response, + "error": self.error, + "parse_note": self.parse_note, + "latency_seconds": ( + round(self.latency_seconds, 3) if self.latency_seconds is not None else None + ), + "provider": self.provider, + "model": self.model, + "template_sha256": ANSWER_VERIFIER_TEMPLATE_SHA256, + } + + +def _clip(text: str) -> str: + text = " ".join(str(text).split()) + if len(text) > _MAX_ITEM_CHARS: + return text[: _MAX_ITEM_CHARS - 1] + "…" + return text + + +def _pack_rows(pack: Mapping[str, object], key: str) -> list[Mapping[str, object]]: + rows = pack.get(key) + if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes)): + return [] + return [row for row in rows if isinstance(row, Mapping)][:_MAX_CONTEXT_ITEMS_PER_SECTION] + + +def pack_question(pack: Mapping[str, object]) -> str: + """The query the pack was compiled for (``query_interpretation.query``).""" + interpretation = pack.get("query_interpretation") + if isinstance(interpretation, Mapping): + query = interpretation.get("query") + if isinstance(query, str) and query.strip(): + return query.strip() + return "" + + +def render_pack_context_block(pack: Mapping[str, object]) -> str: + """Deterministic text rendering of a context pack for the verifier. + + Mirrors what an answering layer can actually see: memory facts + (title + canonical text), supporting-evidence excerpts, source + titles, and the pack's own grounding notes. Read-only over the pack; + no store access, no model calls. + """ + lines: list[str] = [] + for memory in _pack_rows(pack, "relevant_memories"): + title = str(memory.get("title") or "").strip() + body = str(memory.get("canonical_text") or "").strip() + if title and body: + lines.append(f"- {_clip(title)}: {_clip(body)}") + elif title or body: + lines.append(f"- {_clip(title or body)}") + for evidence in _pack_rows(pack, "supporting_evidence"): + excerpt = str(evidence.get("excerpt") or evidence.get("text") or "").strip() + if excerpt: + lines.append(f"- Evidence: {_clip(excerpt)}") + for source in _pack_rows(pack, "sources"): + title = str(source.get("title") or "").strip() + if title: + lines.append(f"- Source: {_clip(title)}") + grounding = pack.get("grounding") + if isinstance(grounding, Mapping): + unsupported = grounding.get("unsupported_entities") + if isinstance(unsupported, Sequence) and not isinstance(unsupported, (str, bytes)): + for name in unsupported: + lines.append(f'- Note: no stored memories mention "{name}".') + if not lines: + return "(no stored memories were retrieved for this question)" + return "\n".join(lines) + + +def build_answer_verifier_prompt( + *, question: str, answer_text: str, context_block: str +) -> str: + """Fill the product verifier template. Context + question + answer only.""" + return ANSWER_VERIFIER_PROMPT_TEMPLATE.format( + context=context_block, + question=question, + answer=answer_text, + ) + + +def parse_verifier_reply(text: str) -> tuple[bool, tuple[UngroundedClaim, ...], str | None]: + """Deterministic parse of the strict reply format. + + Returns ``(grounded, claims, parse_note)``. ``grounded`` is False + only when at least one LOAD-BEARING line is present. A reply with + neither the GROUNDED token nor any UNGROUNDED line fails open as + grounded, with a ``parse_note`` recorded for transparency. + """ + claims: list[UngroundedClaim] = [] + for line in str(text).splitlines(): + stripped = line.strip().lstrip("-*").strip() + lowered = stripped.casefold() + if lowered.startswith(_LOAD_BEARING_PREFIX): + claim_text = stripped[len(_LOAD_BEARING_PREFIX) :].strip() + if claim_text: + claims.append(UngroundedClaim(text=claim_text, load_bearing=True)) + elif lowered.startswith(_INCIDENTAL_PREFIX): + claim_text = stripped[len(_INCIDENTAL_PREFIX) :].strip() + if claim_text: + claims.append(UngroundedClaim(text=claim_text, load_bearing=False)) + parse_note = None + if not claims and GROUNDED_TOKEN.casefold() not in str(text).casefold(): + parse_note = "unrecognized verifier reply; failing open as grounded" + grounded = not any(claim.load_bearing for claim in claims) + return grounded, tuple(claims), parse_note + + +def _call_chat(chat_config: object, prompt: str) -> str: + """Route one prompt through whatever chat seam the integrator handed us. + + Accepts a ``BrainModelProvider``-shaped object (``.chat(prompt=..., + temperature=...)``, e.g. the providers in + ``vnext_model_intelligence``) or a bare ``callable(prompt) -> str``. + """ + chat = getattr(chat_config, "chat", None) + if callable(chat): + return str(chat(prompt=prompt, temperature=VERIFY_TEMPERATURE)) + if callable(chat_config): + return str(chat_config(prompt)) + raise TypeError( + "chat_config must expose .chat(prompt=..., temperature=...) or be callable(prompt)" + ) + + +def verify_answer_grounding( + answer_text: str, + pack: Mapping[str, object], + chat_config: object, +) -> AnswerGroundingVerdict: + """One product verification call. Never raises: failures fail open. + + The prompt sent through ``chat_config`` contains only the + pack-derived context block, the pack's own question, and + ``answer_text`` — there is no expected-answer input at all. + """ + provider = getattr(chat_config, "provider", None) + model = getattr(chat_config, "model", None) + prompt = build_answer_verifier_prompt( + question=pack_question(pack), + answer_text=str(answer_text), + context_block=render_pack_context_block(pack), + ) + started = time.monotonic() + try: + reply = _call_chat(chat_config, prompt) + except Exception as exc: # noqa: BLE001 - fail-open: verification must never break answering + return AnswerGroundingVerdict( + grounded=True, + ungrounded_claims=(), + raw_response=None, + error=f"{type(exc).__name__}: {exc}", + parse_note=None, + latency_seconds=time.monotonic() - started, + provider=str(provider) if provider is not None else None, + model=str(model) if model is not None else None, + ) + grounded, claims, parse_note = parse_verifier_reply(reply) + return AnswerGroundingVerdict( + grounded=grounded, + ungrounded_claims=claims, + raw_response=str(reply).strip(), + error=None, + parse_note=parse_note, + latency_seconds=time.monotonic() - started, + provider=str(provider) if provider is not None else None, + model=str(model) if model is not None else None, + ) + + +def apply_answer_grounding_gate( + answer_text: str, + verdict: AnswerGroundingVerdict, + *, + withheld_text: str = WITHHELD_ANSWER_TEXT, +) -> tuple[str, bool]: + """``(final_text, gate_applied)``: withhold only on clean load-bearing failures. + + A grounded verdict (or any error/fail-open verdict) returns + ``answer_text`` byte-identical with ``gate_applied=False``. + """ + if verdict.gate_should_withhold: + return withheld_text, True + return answer_text, False + + +__all__ = [ + "ANSWER_VERIFIER_PROMPT_TEMPLATE", + "ANSWER_VERIFIER_TEMPLATE_SHA256", + "AnswerGroundingVerdict", + "GROUNDED_TOKEN", + "UngroundedClaim", + "VERIFY_TEMPERATURE", + "WITHHELD_ANSWER_TEXT", + "apply_answer_grounding_gate", + "build_answer_verifier_prompt", + "pack_question", + "parse_verifier_reply", + "render_pack_context_block", + "verify_answer_grounding", +] diff --git a/apps/api/src/alicebot_api/vnext_grounding.py b/apps/api/src/alicebot_api/vnext_grounding.py index 679269e7..f9afde9d 100644 --- a/apps/api/src/alicebot_api/vnext_grounding.py +++ b/apps/api/src/alicebot_api/vnext_grounding.py @@ -11,8 +11,10 @@ events, no model calls). The honesty constraints are load-bearing: - Salience is decided from the QUERY SURFACE ONLY (capitalized spans, - quoted titles, domains, @handles, mid-sentence capitalized tokens). - Nothing downstream may key off benchmark labels or query metadata. + quoted titles, domains, @handles, mid-sentence capitalized tokens, + and attribute-qualified lowercase compounds like "30-gallon tank" or + "my snake plant"). Nothing downstream may key off benchmark labels + or query metadata. - The claim "no stored memories mention X" is only made when EVERY available check misses: the entity substrate (names AND aliases via ``find_entities_by_names``) and cheap one-row FTS existence probes @@ -111,6 +113,73 @@ # is "Emma", never "Emma's". _POSSESSIVE_SUFFIX_RE = re.compile(r"['’]s$") +# -- lowercase attribute-qualified nouns (round-3 loss forensics) ------------- +# +# A query can name a specific THING without any capital letter when the +# noun carries an explicit qualifier: "my 30-gallon tank", "my snake +# plant", "my soccer team". Bare lowercase nouns are NEVER salient; +# conservatism lives in three stacked gates: +# +# 1. an explicit qualifier SHAPE — a measured quantity ("30-gallon") +# or a possessive determiner plus a noun modifier ("my snake ..."), +# 2. a small curated HEAD-NOUN lexicon (kept-thing and activity-group +# heads), so ordinary object talk ("my credit card", "my phone +# number") can never fire, +# 3. modifier hygiene — generic adjectives ("new", "favorite", +# "work"), blocklisted words, and digit-led modifiers never qualify +# the possessive shape. +# +# Both rules admit the full compound surface ("30-gallon tank", "snake +# plant"), so the FTS probe ORs every token variant: any corpus mention +# of "tank(s)", "plant(s)", "snakes", ... counts as support and the +# note stays suppressed — the safe direction. + +# Physical measurement units only. Time units (day, week, minute) are +# deliberately absent: "5-day trip" qualifies an EVENT, not a thing. +_MEASURE_UNITS = ( + "gallon", "litre", "liter", "quart", "pint", + "ounce", "oz", "pound", "lb", "gram", "kilogram", "kg", + "inch", "foot", "meter", "metre", "acre", + "watt", "volt", "amp", +) +_QUANTIFIED_NOUN_RE = re.compile( + r"\b\d+(?:\.\d+)?[- ](?:" + "|".join(_MEASURE_UNITS) + r")s?[- ]([a-z][a-z-]{2,})\b" +) + +# Possessive-compound heads: kept things (a specific plant/tank in the +# user's life) and activity groups ("my soccer team", "my book club"). +# Curated and short on purpose; growing it requires a false-positive +# table entry per addition (see test_vnext_grounding.py). +_COMPOUND_HEAD_NOUNS = frozenset( + { + "plant", "plants", "tree", "trees", + "tank", "tanks", "aquarium", "terrarium", + "team", "teams", "practice", "league", "club", "clubs", + "lesson", "lessons", "class", "classes", + } +) + +# Modifiers that describe rather than name: possessive compounds built +# on these are ordinary object talk ("my new team", "my work club"), +# never a specific named thing. +_GENERIC_COMPOUND_MODIFIERS = frozenset( + { + "new", "old", "own", "first", "last", "next", "previous", + "current", "former", "favorite", "favourite", "usual", + "regular", "recent", "little", "big", "small", "large", + "best", "worst", "whole", "entire", "main", "other", + "second", "third", "local", "nearby", "long", "short", + "early", "late", "morning", "evening", "afternoon", + "weekly", "daily", "monthly", "weekend", + "work", "home", "school", "office", "company", "family", + "dream", "future", "online", "virtual", + } +) + +_POSSESSIVE_COMPOUND_RE = re.compile( + r"\b(?:[Mm]y|[Oo]ur)\s+([a-z][a-z-]{2,})\s+([a-z]+)\b" +) + def _overlaps(covered: list[tuple[int, int]], start: int, end: int) -> bool: return any(start < c_end and end > c_start for c_start, c_end in covered) @@ -218,6 +287,41 @@ def admit(position: int, surface: str) -> None: continue admit(match.start(), surface) + # 5. Quantity-qualified lowercase nouns ("my 30-gallon tank"): a + # measured quantity is a naming qualifier, so the full compound + # is salient. Physical units only; time-qualified events never + # fire (see the lexicon comment above). + for match in _QUANTIFIED_NOUN_RE.finditer(query): + if _overlaps(covered, match.start(), match.end()): + continue + head = match.group(1).casefold() + if head in ENTITY_EXTRACTION_BLOCKLIST or head in _GENERIC_COMPOUND_MODIFIERS: + continue + covered.append(match.span()) + admit(match.start(), match.group()) + + # 6. Possessive noun-noun compounds ("my snake plant", "my soccer + # team"): possessive determiner + specific modifier + curated + # head. All three gates must pass; "my hamster" (no modifier), + # "my credit card" (head not curated), and "my work team" + # (generic modifier) never fire. + for match in _POSSESSIVE_COMPOUND_RE.finditer(query): + if _overlaps(covered, match.start(), match.end()): + continue + modifier = match.group(1).casefold() + head = match.group(2).casefold() + if head not in _COMPOUND_HEAD_NOUNS: + continue + if ( + modifier in ENTITY_EXTRACTION_BLOCKLIST + or modifier in _GENERIC_COMPOUND_MODIFIERS + or modifier in _QUERY_LEADING_STOPWORDS + or modifier in _COMPOUND_HEAD_NOUNS + ): + continue + covered.append(match.span()) + admit(match.start(), f"{match.group(1)} {match.group(2)}") + found.sort(key=lambda item: item[0]) return tuple(surface for _position, _normalized, surface in found[:MAX_GROUNDING_ENTITIES]) @@ -241,7 +345,17 @@ def add(token: str) -> None: seen.add(token) tokens.append(token) - for token in _PROBE_TOKEN_RE.findall(name): + raw_tokens = _PROBE_TOKEN_RE.findall(name) + # Pure-number tokens are dropped when the name has word tokens: a + # bare "30" on an unrelated receipt is not a mention of the + # "30-gallon tank", and letting it count as support would silently + # suppress a truthful note. The word tokens still OR together in + # the safe direction (any hit suppresses the note). All-number + # names keep their digits -- they are the only probe surface. + has_word_token = any(not token.isdigit() for token in raw_tokens) + for token in raw_tokens: + if has_word_token and token.isdigit(): + continue add(token) if len(token) < _VARIANT_MIN_TOKEN_CHARS: continue diff --git a/apps/api/src/alicebot_api/vnext_retrieval.py b/apps/api/src/alicebot_api/vnext_retrieval.py index 144dedd6..af389004 100644 --- a/apps/api/src/alicebot_api/vnext_retrieval.py +++ b/apps/api/src/alicebot_api/vnext_retrieval.py @@ -6,7 +6,11 @@ packer, and depth tiers only switch stages and sections on or off. NO depth tier, strategy, or section performs LLM synthesis, summarization, or any other model call — the pack is a pure function of stored rows -plus the request (house no-fake-intelligence rule). +plus the request (house no-fake-intelligence rule). Equal-score ordering +ties resolve through a content-stable cascade (``content_stable_tiebreak``, +disclosed in the trace as ``fusion.tie_break``) so re-ingesting the same +content — new uuids, same rows — reproduces the same pack composition; +the id remains the final total-order key. Budget strategies (``VNextRetrievalRequest.budget_strategy``) change the greedy packer's section order; ``recent_first`` and ``facts_first`` @@ -559,6 +563,124 @@ def _graph_memory_admissible( return True +# -- content-stable tie-breaks ------------------------------------------------- +# Equal-score ordering decisions (equal RRF fused scores, equal graph-stage +# timestamps, equal temporal-boost distances) used to fall straight through +# to the row id — a uuid minted at ingest. Within one store that is +# deterministic, but across two ingests of the SAME content the uuids +# differ, so every such tie was a coin flip re-rolled per ingest (measured +# as pure pack churn between identical-config benchmark runs). The cascade +# below decides those ties on row CONTENT instead: older event/session date +# first, then longer content, then the content text, then a content +# fingerprint; the id stays as the FINAL key — the total-order guarantee — +# but no longer casts the deciding vote between near-equals. Every key is a +# pure function of values the row already carries: no randomness, no store +# lookups, no clock reads. Deliberately NO write-clock fallbacks +# (created_at/first_seen_at/captured_at): a wall-clock key that collides in +# one ingest but not another would re-introduce seed-dependent ordering. +TIE_BREAK_CONTENT_STABLE = "content_stable_v1" +# Rows with no parseable content event signal sort after any dated row. +_TIEBREAK_UNDATED = datetime(9999, 12, 31, tzinfo=UTC) +# Content-honest event signals only: explicit validity start (memories), +# the source's own creation time, then connector-stamped metadata dates — +# the same signals search_memories_by_time and _source_event_time trust, +# minus their write-time fallbacks (see the seed-dependence note above). +_TIEBREAK_EVENT_KEYS = ("valid_from", "source_created_at") +_TIEBREAK_TEXT_KEYS = ("canonical_text", "text", "summary", "title") + + +def _tiebreak_event_time(item: JsonObject) -> datetime | None: + """Content-stamped event/session time of a row, or None.""" + for key in _TIEBREAK_EVENT_KEYS: + parsed = parse_event_datetime(item.get(key)) + if parsed is not None: + return parsed + metadata = item.get("metadata_json") + if isinstance(metadata, Mapping): + for key in SOURCE_EVENT_METADATA_KEYS: + parsed = parse_event_datetime(metadata.get(key)) + if parsed is not None: + return parsed + return None + + +def _tiebreak_content_text(item: JsonObject) -> str: + for key in _TIEBREAK_TEXT_KEYS: + value = item.get(key) + if isinstance(value, str) and value.strip(): + return value + return "" + + +def _tiebreak_content_fingerprint(item: JsonObject) -> str: + """Stable content-derived discriminator for rows whose text also ties. + + Sources carry ``content_hash`` (digest of the captured content); capture- + written memories carry the originating capture's digest and chunk index + in metadata. Two rows with identical text extracted from different + captures stay distinguishable by content, not by uuid. + """ + content_hash = item.get("content_hash") + if isinstance(content_hash, str) and content_hash: + return content_hash + metadata = item.get("metadata_json") + if isinstance(metadata, Mapping): + capture_hash = metadata.get("capture_content_hash") + if isinstance(capture_hash, str) and capture_hash: + return f"{capture_hash}:{metadata.get('source_chunk_index')}" + return "" + + +def content_stable_tiebreak(item: JsonObject) -> tuple[datetime, int, str, str]: + """Content-stable sort key for equal-score ties (ascending sorts first). + + Cascade: older content-stamped event/session date first (undated rows + last), longer content first, then the content text itself, then the + content fingerprint. Callers append the id as the final key. + """ + text = _tiebreak_content_text(item) + return ( + _tiebreak_event_time(item) or _TIEBREAK_UNDATED, + -len(text), + text, + _tiebreak_content_fingerprint(item), + ) + + +def _stabilize_scored_rows( + rows: Sequence[JsonObject], + *, + score_key: str = "fts_score", + descending: bool = True, +) -> list[JsonObject]: + """Reorder equal-score runs of a scored stage list content-stably. + + The stores' FTS stages order by ``fts_score DESC`` and then write-clock + columns with the row id as the last key, so rows with EQUAL scores + (identical term statistics — restated facts, near-duplicate turns) can + arrive in ingest-dependent order: their relative rank re-rolls per + ingest even though the content is identical. The Postgres vector stage + is worse — ``ORDER BY embedding_vector <=> query`` alone, so equal + distances get undefined database order within a single store. Distinct + scores keep the store's order exactly (``descending`` says which way the + stage ranks); equal scores fall through the content-stable cascade with + the id as the final total-order key. Lists where any row lacks a + numeric score are returned unchanged — an unknown store contract keeps + its own order. + """ + if not all(isinstance(row.get(score_key), (int, float)) for row in rows): + return list(rows) + sign = -1.0 if descending else 1.0 + return sorted( + rows, + key=lambda row: ( + sign * float(row[score_key]), # type: ignore[arg-type] + *content_stable_tiebreak(row), + str(row.get("id")), + ), + ) + + def reciprocal_rank_fusion( ranked_lists: Mapping[str, Sequence[JsonObject]], *, @@ -568,7 +690,8 @@ def reciprocal_rank_fusion( Each item scores ``sum(1 / (k + rank))`` over the stages it appears in. Returns ``(item, rrf_score, stage_ranks)`` tuples ordered by descending - score with a deterministic id tie-break. + score; equal scores fall through the content-stable cascade + (``content_stable_tiebreak``) with the id as the final total-order key. """ if k < 1: raise VNextRetrievalValidationError("reciprocal rank fusion k must be at least 1") @@ -582,7 +705,10 @@ def reciprocal_rank_fusion( items[item_id] = row scores[item_id] = scores.get(item_id, 0.0) + 1.0 / (k + rank) stage_ranks.setdefault(item_id, {})[stage] = rank - ordered_ids = sorted(items, key=lambda item_id: (-scores[item_id], item_id)) + ordered_ids = sorted( + items, + key=lambda item_id: (-scores[item_id], *content_stable_tiebreak(items[item_id]), item_id), + ) return [(items[item_id], scores[item_id], stage_ranks[item_id]) for item_id in ordered_ids] @@ -982,8 +1108,8 @@ def _memory_fts_rows( # Store predates the match_any kwarg; keep the strict # (empty) result rather than guessing. return [], fts_source - return list(rows), f"{fts_source}_or_fallback" - return list(rows), fts_source + return _stabilize_scored_rows(rows), f"{fts_source}_or_fallback" + return _stabilize_scored_rows(rows), fts_source rows = self.store.search_memories( query=query, domains=domains or None, @@ -1021,7 +1147,9 @@ def _memory_vector_rows( ) except (VNextEmbeddingConfigurationError, VNextEmbeddingProviderError) as exc: return [], f"disabled: query embedding failed ({exc})" - return list(rows), VECTOR_STAGE_ENABLED + # Ascending stage: smaller distance ranks first. Equal distances + # (identical texts embed identically) stabilize content-first. + return _stabilize_scored_rows(rows, score_key="vector_distance", descending=False), VECTOR_STAGE_ENABLED def _memory_graph_rows( self, @@ -1106,8 +1234,9 @@ def _memory_graph_rows( ) ranked.append((observed_at, recency, str(row.get("id")), row)) # Deterministic order: edge observed_at DESC, memory recency DESC, - # id ASC on ties (id-ascending pre-sort survives the stable sort). - ranked.sort(key=lambda entry: entry[2]) + # then the content-stable cascade with id ASC as the final key (the + # ascending pre-sort survives the stable reverse timestamp sort). + ranked.sort(key=lambda entry: (*content_stable_tiebreak(entry[3]), entry[2])) ranked.sort(key=lambda entry: (entry[0], entry[1]), reverse=True) rows = [entry[3] for entry in ranked[:limit]] return rows, GRAPH_STAGE_ENABLED, matched_entities @@ -1223,6 +1352,10 @@ def _resolve_sources(source_ids: list[str]) -> list[JsonObject]: chunk_rows = [] else: chunk_fts_source = f"{chunk_fts_source}_or_fallback" + # Equal-score chunk runs decide WHICH parent source enters the + # deduplicated chunk list first; stabilize them content-first so + # the list is ingest-invariant (distinct scores keep store order). + chunk_rows = _stabilize_scored_rows(chunk_rows) ordered_source_ids: list[str] = [] seen_source_ids: set[str] = set() for row in chunk_rows: @@ -1284,7 +1417,10 @@ def _resolve_sources(source_ids: list[str]) -> list[JsonObject]: if event is None or not (anchor.window_start <= event < anchor.window_end): continue dated.append((abs((event - center).total_seconds()), source_id, row)) - dated.sort(key=lambda entry: (entry[0], entry[1])) + # Distance ties are the norm here (day-resolution session dates + # share a window distance), so they fall through the + # content-stable cascade before the id total-order key. + dated.sort(key=lambda entry: (entry[0], *content_stable_tiebreak(entry[2]), entry[1])) temporal_sources = [row for _distance, _source_id, row in dated] ranked_lists[SOURCE_STAGE_TEMPORAL] = temporal_sources @@ -1725,7 +1861,13 @@ def compile_context_pack(self, request: VNextRetrievalRequest) -> JsonObject: "created_by_agent_ids": list(created_by_agent_ids), "run_id": filter_run_id, }, - "fusion": {"algorithm": "reciprocal_rank_fusion", "k": RRF_K}, + "fusion": { + "algorithm": "reciprocal_rank_fusion", + "k": RRF_K, + # Honest disclosure: equal-score ties resolve on row content + # (see content_stable_tiebreak), no longer on the raw id. + "tie_break": TIE_BREAK_CONTENT_STABLE, + }, "vector_stage": vector_stage, "context_depth": depth, "budget_strategy": strategy, @@ -2106,6 +2248,7 @@ def _walk_supersession_chain( "SUPERSESSION_STAGE_ENABLED", "TEMPORAL_STAGE_DISABLED_NO_STORE_SUPPORT", "TEMPORAL_STAGE_ENABLED", + "TIE_BREAK_CONTENT_STABLE", "TOKEN_ESTIMATE_CHARS_PER_TOKEN", "VALID_TO_UNBOUNDED_YEAR", "VECTOR_STAGE_DISABLED_NO_PROVIDER", @@ -2116,6 +2259,7 @@ def _walk_supersession_chain( "VNextRetrievalStore", "VNextRetrievalValidationError", "classify_query", + "content_stable_tiebreak", "entity_name_candidates", "estimate_item_tokens", "normalize_query", diff --git a/docs/alpha/mcp-tools.md b/docs/alpha/mcp-tools.md index a209176e..7545ab63 100644 --- a/docs/alpha/mcp-tools.md +++ b/docs/alpha/mcp-tools.md @@ -127,6 +127,56 @@ honored and audited as `unauthenticated_local`. the retrieval trace — which stages ran, candidate counts, and whether vector search was active or degraded to full-text (and why). +## Grounding + +`alice_context_pack` reports when the query names something the stored +corpus has never seen. If a salient query entity — a capitalized name +("Marcus Chen"), a quoted title ("Sapiens"), a domain, an @handle, or an +attribute-qualified thing ("my 30-gallon tank", "my snake plant", "my +soccer team") — has **zero** corpus support, the response carries a +`grounding` field: + +```json +"grounding": {"unsupported_entities": ["Zorblatt Nine"], "checked": 2} +``` + +The check is deliberately conservative, in both directions: + +- Salience comes from the query surface only. Generic nouns, acronyms, + sentence-initial capitals, and bare lowercase nouns ("my hamster") are + never checked; lowercase things need an explicit qualifier (a measured + quantity or a possessive noun-noun compound with a curated head noun). +- "Unsupported" is only claimed when every available check misses: the + entity table (names and aliases) plus cheap one-row full-text probes + over source chunks and memories, where **any** token variant counts as + support ("Hawaiian" supports "Hawaii"; "tank" alone supports + "30-gallon tank"). Bare numbers never fabricate support: "30" on an + unrelated receipt is not a mention of the 30-gallon tank. + +The field is absent for every ordinary query — fully supported entities, +no salient entities, or a store that cannot be checked all leave the +response unchanged. It never filters or blocks retrieval; it is a +statistic the caller can use to avoid synthesizing answers about things +memory has never seen. With `"debug": true` the same record also appears +in the retrieval trace. It is skipped at `context_depth: "minimal"` to +keep that tier's cheapest-call promise. `alice_recall` does not compute +it (recall does not compile a context pack). + +**Answer verification (opt-in library seam, not a tool).** For +integrators who generate answers from a context pack, +`alicebot_api.vnext_answer_verification` provides +`verify_answer_grounding(answer_text, pack, chat_config)`: it asks a +model of your choosing (any `BrainModelProvider`-shaped object with +`.chat(prompt=..., temperature=...)`, or a bare `callable(prompt) -> +str`) to list concrete claims in the answer that the pack does not +support, and returns a verdict object. Nothing in the API or MCP server +calls it — no behavior changes unless your answering loop invokes it. +The verdict is fail-open (provider errors and unparseable replies leave +the answer untouched) and self-disclosing (`to_record()` carries the +prompt-template fingerprint and the verifier provider/model). +`apply_answer_grounding_gate(answer_text, verdict)` then withholds the +answer only on a clean verdict with a load-bearing unsupported claim. + ## Embeddings Semantic search activates when an OpenAI-compatible embeddings endpoint is diff --git a/eval/longmemeval/adapter.py b/eval/longmemeval/adapter.py index 9c78603a..e968d177 100644 --- a/eval/longmemeval/adapter.py +++ b/eval/longmemeval/adapter.py @@ -42,6 +42,7 @@ from alicebot_api.sqlite_store import SQLiteVNextStore, ensure_sqlite_user, sqlite_user_connection from alicebot_api.vnext_capture import SourceCaptureInput, VNextCaptureService +from alicebot_api.vnext_memory_commit import VNextMemoryCommitService from alicebot_api.vnext_retrieval import ( VECTOR_STAGE_ENABLED, VNextRetrievalRequest, @@ -49,6 +50,7 @@ query_terms, ) from alicebot_api.vnext_fact_keys import attach_memory_fact_keys +from alicebot_api.vnext_rollups import RollupOptions, VNextRollupService from alicebot_api.vnext_temporal_query import parse_event_datetime from longmemeval.dataset import LongMemEvalQuestion, SessionTurn @@ -67,6 +69,15 @@ EMPTY_CONTEXT_PLACEHOLDER = "(no relevant chat history was retrieved)" +# Review-accept reason recorded on every roll-up acceptance the harness +# performs (--accept-rollups); it lands verbatim in the acceptance metadata, +# the promotion revision, and the agent.memory_consolidation_accepted event, +# so accepted-by-the-harness cards are always distinguishable offline. +ROLLUP_ACCEPTANCE_REASON = ( + "LongMemEval harness --accept-rollups: review-accepted consolidation roll-up proposal " + "(models the product's human review workflow)." +) + # Official LongMemEval reading templates (src/generation/run_generation.py in # xiaowu0162/LongMemEval), verbatim; the history slot receives Alice's # retrieved context block instead of the full haystack. @@ -88,6 +99,34 @@ _WORD_PATTERN = re.compile(r"[a-z0-9][a-z0-9_-]+") +@dataclass(frozen=True, slots=True) +class RollupAcceptanceStats: + """One post-ingest consolidation roll-up pass + review acceptance. + + Produced only when the disclosed ``--accept-rollups`` step ran, and then + serialized into the checkpoint row's ``ingest.rollups`` block, so every + scored run carries per-store visibility of how many cards were proposed + and accepted (and which memory rows they are). + """ + + groupable_memory_count: int + proposal_count: int + accepted_count: int + accepted_memory_ids: tuple[str, ...] + skipped: tuple[str, ...] + rollup_seconds: float + + def to_record(self) -> dict[str, object]: + return { + "groupable_memory_count": self.groupable_memory_count, + "proposal_count": self.proposal_count, + "accepted_count": self.accepted_count, + "accepted_memory_ids": list(self.accepted_memory_ids), + "skipped": list(self.skipped), + "rollup_seconds": round(self.rollup_seconds, 3), + } + + @dataclass(frozen=True, slots=True) class IngestStats: session_count: int @@ -97,9 +136,13 @@ class IngestStats: candidate_memory_count: int promoted_memory_count: int ingest_seconds: float + # None whenever the --accept-rollups step did not run, and then the + # record below is byte-identical to the pre-roll-up schema (no new key), + # so off-flag checkpoint rows do not drift. + rollups: RollupAcceptanceStats | None = None def to_record(self) -> dict[str, object]: - return { + record: dict[str, object] = { "session_count": self.session_count, "source_count": self.source_count, "duplicate_count": self.duplicate_count, @@ -108,6 +151,9 @@ def to_record(self) -> dict[str, object]: "promoted_memory_count": self.promoted_memory_count, "ingest_seconds": round(self.ingest_seconds, 3), } + if self.rollups is not None: + record["rollups"] = self.rollups.to_record() + return record @dataclass(frozen=True, slots=True) @@ -425,7 +471,7 @@ def __init__(self, question: LongMemEvalQuestion, store: SQLiteVNextStore) -> No # -- ingest ------------------------------------------------------------ - def ingest(self) -> IngestStats: + def ingest(self, *, accept_rollups: bool = False) -> IngestStats: started = time.monotonic() capture = VNextCaptureService(self.store, actor_type="system") source_count = 0 @@ -461,7 +507,8 @@ def ingest(self) -> IngestStats: candidate_count += result.candidate_memory_count if result.source_id is not None and result.source_id not in self._source_sessions: self._source_sessions[result.source_id] = (session_id, date) - promoted = self._promote_candidate_memories() + promoted = self._promote_candidate_memories(stamp_session_dates=accept_rollups) + rollups = self._consolidate_and_accept_rollups() if accept_rollups else None return IngestStats( session_count=session_count, source_count=source_count, @@ -470,21 +517,48 @@ def ingest(self) -> IngestStats: candidate_memory_count=candidate_count, promoted_memory_count=promoted, ingest_seconds=time.monotonic() - started, + rollups=rollups, ) - def _promote_candidate_memories(self) -> int: + def _session_date_for_memory(self, memory: dict[str, object]) -> str | None: + metadata = memory.get("metadata_json") + source_id = str(metadata.get("source_id") or "") if isinstance(metadata, dict) else "" + if not source_id: + return None + session_id, date = self._session_label(source_id) + return date if session_id and date != "undated" else None + + def _promote_candidate_memories(self, *, stamp_session_dates: bool = False) -> int: """Accept capture candidates so Alice's search stages can see them. Mirrors the store-level review-accept patch (``status: active``) used by the product's review flow; without it, ``search_memories*`` only matches pre-existing active/accepted rows and the memory stages would run empty for every question. + + ``stamp_session_dates`` (only set by the disclosed --accept-rollups + step) additionally records each memory's originating session date in + ``metadata_json.session_date`` — the per-instance date hook the + roll-up pass documents (``vnext_rollups._member_date``). In the + product, memories are created in real time so their ``created_at`` + IS the content date; benchmark replay compresses years of sessions + into one ingest moment, so without the stamp every roll-up instance + would carry today's date instead of its session's. The stamp is + provenance the ingest already wrote on the source row's metadata; + no benchmark label is involved. Off flag, the promotion patch is + byte-identical to the pre-roll-up harness (``{"status": "active"}``). """ promoted = 0 for memory in self.store.list_memories(status="candidate"): + patch: dict[str, object] = {"status": "active"} + if stamp_session_dates: + session_date = self._session_date_for_memory(memory) + metadata = memory.get("metadata_json") + if session_date is not None and isinstance(metadata, dict): + patch["metadata_json"] = {**metadata, "session_date": session_date} updated = self.store.update_memory( memory_id=str(memory["id"]), - patch={"status": "active"}, + patch=patch, actor_type="system", ) # Promotion is also the derived-retrieval-key moment (mirrors @@ -494,6 +568,78 @@ def _promote_candidate_memories(self) -> int: promoted += 1 return promoted + def _consolidate_and_accept_rollups(self) -> RollupAcceptanceStats: + """Run the consolidation roll-up pass, then review-accept every card. + + This models the product's intended human workflow (the user accepts + the review proposals the consolidation run produced), the same way + the harness already mirrors review-accept promotion for candidate + memories in ``_promote_candidate_memories``. Two real product code + paths, no store-level shortcuts: + + * Proposal — the roll-up portion of the scheduled consolidation + workflow. ``VNextConsolidationService.generate_memory_consolidation`` + invokes ``VNextRollupService.propose_rollups`` (see + apps/api/src/alicebot_api/vnext_consolidation.py); the harness + calls that same service entry point with the scheduled workflow's + argument shape: deterministic generation (no model call), default + ``RollupOptions``, default sensitivity scope, candidate creation + on. The full ``generate_memory_consolidation`` wrapper cannot run + against the per-question SQLite on-ramp store because it also + writes the consolidation report artifact (``create_artifact``), a + surface ``SQLiteVNextStore`` does not implement; the roll-up pass + itself is store-complete. The near-duplicate exclusion the + wrapper derives from embedding-based clustering is empty here for + the same reason keyless replay is FTS-only: ``_cluster_memories`` + skips with ``no_embedding_provider_configured``, so the scheduled + pass would also pass no clusters; the roll-up pass's own pairwise + Jaccard duplicate-group guard still applies. + * Acceptance — every proposed card goes through + ``VNextMemoryCommitService.accept_consolidation_candidate``, the + exact service call the review console's + ``POST /v0/vnext/memories/accept-consolidation`` endpoint + (apps/api/src/alicebot_api/main.py), the ``alicebot vnext + memories accept-consolidation`` CLI (cli.py), and the MCP review + tool (mcp_tools.py) all delegate to, with ``identity=None`` — the + human-reviewer shape. Acceptance therefore does everything the + product does: policy events, promotion to ``active``, + best-effort embedding, entity linking, the ``promoted`` revision, + and the ``agent.memory_consolidation_accepted`` event. + + Members are never superseded (first-time roll-up proposals carry an + empty ``proposed_supersede``), so every individual memory stays + active and recallable exactly as before; the cards are additive. + """ + started = time.monotonic() + outcome = VNextRollupService(self.store).propose_rollups( + domains=None, + sensitivity_allowed=["public", "internal", "private", "unknown"], + options=RollupOptions(), + create_candidate_memories=True, + generated_by="system", + trace_id=None, + generation_mode="deterministic", + exclude_member_id_sets=[], + ) + commit = VNextMemoryCommitService(self.store) # duck-typed store, like the CLI on-ramp + accepted_ids: list[str] = [] + for candidate_id in outcome.candidate_ids: + payload = commit.accept_consolidation_candidate( + candidate_id, + reason=ROLLUP_ACCEPTANCE_REASON, + identity=None, + ) + if payload.get("status") == "accepted": + accepted_ids.append(candidate_id) + return RollupAcceptanceStats( + groupable_memory_count=outcome.groupable_count, + proposal_count=len(outcome.proposals), + accepted_count=len(accepted_ids), + accepted_memory_ids=tuple(accepted_ids), + skipped=tuple(str(reason) for reason in outcome.skipped), + rollup_seconds=time.monotonic() - started, + ) + # -- retrieval ---------------------------------------------------------- def retrieve( @@ -775,7 +921,9 @@ def question_run(question: LongMemEvalQuestion, db_path: str | Path) -> Iterator "LME_USER_ID", "MAX_ITEMS_ENV", "QuestionRun", + "ROLLUP_ACCEPTANCE_REASON", "RetrievalOutcome", + "RollupAcceptanceStats", "build_answer_prompt", "collapse_intra_turn_blank_lines", "context_char_budget_from_env", diff --git a/eval/longmemeval/runner.py b/eval/longmemeval/runner.py index f2eb8f38..8bba885c 100644 --- a/eval/longmemeval/runner.py +++ b/eval/longmemeval/runner.py @@ -95,6 +95,12 @@ class RunnerConfig: # Disclosed post-generation grounding gate (see longmemeval/verification.py); # off by default and always visible in the config fingerprint. verify_grounding: bool = False + # Disclosed post-ingest consolidation step (see adapter.py, + # QuestionRun._consolidate_and_accept_rollups): propose roll-up cards via + # the product's consolidation pass and review-accept them through the + # real acceptance path. Off by default so the default replay stays + # byte-identical to published runs; always in the config fingerprint. + accept_rollups: bool = False @property def mode(self) -> str: @@ -139,6 +145,10 @@ def config_fingerprint( # verifier model when enabled) always feed the fingerprint digest. "verify_grounding": config.verify_grounding, "verifier_model": verifier.redacted() if config.verify_grounding and verifier is not None else None, + # Roll-up acceptance can never run undisclosed either: the flag feeds + # the digest, and per-store proposed/accepted counts land in each + # checkpoint row's ingest.rollups block. + "accept_rollups": config.accept_rollups, "generation_temperature": GENERATION_TEMPERATURE, "max_items": config.max_items, "context_char_budget": config.context_char_budget, @@ -266,7 +276,7 @@ def run_question( try: _cleanup_store(db_path) with question_run(question, db_path) as run: - ingest_stats = run.ingest() + ingest_stats = run.ingest(accept_rollups=config.accept_rollups) outcome = run.retrieve( max_items=config.max_items, context_char_budget=config.context_char_budget, @@ -472,6 +482,15 @@ def build_arg_parser() -> argparse.ArgumentParser: "(context + question + answer only; never the gold answer) converts answers with " "ungrounded load-bearing claims to the abstention phrasing; recorded in the fingerprint", ) + parser.add_argument( + "--accept-rollups", + action="store_true", + help="disclosed post-ingest consolidation step: run the product's roll-up " + "consolidation pass after candidate promotion and review-accept the proposed " + "cards through the real acceptance path (accept_consolidation_candidate), " + "modeling the product's human review workflow; recorded in the fingerprint " + "and per-store counts in each checkpoint row's ingest.rollups block", + ) parser.add_argument("--max-items", type=int, default=None, help=f"context-pack max_items (default: ${'{'}ALICE_LME_MAX_ITEMS{'}'} or {DEFAULT_MAX_ITEMS})") parser.add_argument( "--context-char-budget", @@ -519,6 +538,7 @@ def _resolve_config(args: argparse.Namespace, *, question_ids: tuple[str, ...] | report_path=args.report or RESULTS_DIR / f"{stem}_report.json", keep_stores=args.keep_stores, verify_grounding=args.verify_grounding, + accept_rollups=args.accept_rollups, ) diff --git a/eval/longmemeval/test_harness.py b/eval/longmemeval/test_harness.py index 391cd20e..1f9bc315 100644 --- a/eval/longmemeval/test_harness.py +++ b/eval/longmemeval/test_harness.py @@ -971,6 +971,257 @@ def test_question_run_ingests_and_retrieves_evidence(tmp_path: Path) -> None: assert outcome.retrieval_seconds >= 0.0 +# -- roll-up acceptance (post-ingest consolidation, real acceptance path) ----------- + +_ROLLUP_QUESTION_TEXT = "How many hours have I spent playing Stardew Valley in total?" + + +def _rollup_question(): + """Aggregation-shaped fixture: three sessions each asserting one distinct + Stardew Valley play instance (same entity, distinct dates/amounts), plus + one unrelated filler session.""" + return parse_question( + { + "question_id": "q_rollup_agg", + "question_type": "multi-session", + "question": _ROLLUP_QUESTION_TEXT, + "answer": "47 hours", + "question_date": "2023/06/01 (Thu) 10:00", + "haystack_dates": [ + "2023/05/01 (Mon) 10:00", + "2023/05/08 (Mon) 11:00", + "2023/05/15 (Mon) 09:30", + "2023/05/22 (Mon) 16:00", + ], + "haystack_session_ids": ["s_sv1", "s_sv2", "s_sv3", "s_filler"], + "haystack_sessions": [ + [ + {"role": "user", "content": "My Stardew Valley playthrough was about 30 hours over the spring break."}, + {"role": "assistant", "content": "That sounds like a relaxing break."}, + ], + [ + {"role": "user", "content": "The Stardew Valley harvest festival grind was another 12 hours of my weekend."}, + {"role": "assistant", "content": "Festival grinding pays off eventually."}, + ], + [ + {"role": "user", "content": "Stardew Valley multiplayer with my cousin was 5 hours of pure chaos on Friday."}, + {"role": "assistant", "content": "Multiplayer farms get chaotic fast."}, + ], + [ + {"role": "user", "content": "My sourdough starter needs feeding twice a day, which is a commitment."}, + {"role": "assistant", "content": "Daily feeding keeps it healthy."}, + ], + ], + "answer_session_ids": ["s_sv1", "s_sv2", "s_sv3"], + } + ) + + +def _rollup_cards(store) -> list[dict[str, object]]: + return [ + row + for row in store.list_memories(status="active") + if isinstance(row.get("metadata_json"), dict) + and row["metadata_json"].get("candidate_kind") == "memory_rollup" + ] + + +def test_accept_rollups_uses_real_acceptance_path(tmp_path: Path) -> None: + """The step must go through accept_consolidation_candidate — the same + service call the review console endpoint delegates to — never a + store-level status patch. The evidence: acceptance metadata, the + 'promoted' revision, and the agent.memory_consolidation_accepted event + that only the real path writes.""" + question = _rollup_question() + with adapter.question_run(question, tmp_path / "q.sqlite3") as run: + stats = run.ingest(accept_rollups=True) + assert stats.rollups is not None + assert stats.rollups.proposal_count >= 1 + assert stats.rollups.accepted_count == stats.rollups.proposal_count + assert len(stats.rollups.accepted_memory_ids) == stats.rollups.accepted_count + + cards = _rollup_cards(run.store) + assert {str(card["id"]) for card in cards} == set(stats.rollups.accepted_memory_ids) + stardew = [card for card in cards if "Stardew Valley" in str(card.get("canonical_text"))] + assert len(stardew) == 1 + card = stardew[0] + text = str(card["canonical_text"]) + # The deterministic instance list carries the aggregation needles: + # one instance per session, with its amount and REAL session date + # (stamped at promotion; without the stamp every instance would + # show the ingest wall-clock date). + assert "3 instances in total" in text + for needle in ("30 hours", "12 hours", "5 hours", "2023/05/01", "2023/05/08", "2023/05/15"): + assert needle in text + + # Real-path acceptance evidence on the card itself. + metadata = card["metadata_json"] + assert metadata["review_required"] is False + accepted = metadata["consolidation"]["accepted"] + assert accepted["actor_type"] == "user" # identity=None, the human-reviewer shape + assert accepted["reason"] == adapter.ROLLUP_ACCEPTANCE_REASON + assert accepted["superseded_member_ids"] == [] # members stay active + revisions = run.store.list_revisions(str(card["id"])) + assert any( + revision.get("revision_type") == "promoted" + and revision.get("action") == "agentic_memory_consolidation_accept" + for revision in revisions + ) + events = run.store.list_events(target_type="memory", target_id=str(card["id"])) + assert any(event.get("event_type") == "agent.memory_consolidation_accepted" for event in events) + + # Member memories stay active and individually recallable. + member_ids = [str(instance["memory_id"]) for instance in card["value"]["rollup"]["instances"]] + assert len(member_ids) == 3 + for member_id in member_ids: + member = run.store.get_memory(member_id) + assert member is not None and member["status"] == "active" + # Promotion stamped the member's originating session date. + assert str(member["metadata_json"]["session_date"]).startswith("2023/05/") + + # The cashed check: the accepted card wins a context-pack slot for + # the aggregation query and its instance list reaches the prompt. + outcome = run.retrieve(max_items=16, context_char_budget=12_000) + assert str(card["id"]) in outcome.memory_ids + assert "3 instances in total" in outcome.context_block + + # Ingest record discloses the counts (checkpoint visibility). + record = stats.to_record() + assert record["rollups"]["accepted_count"] == stats.rollups.accepted_count + assert record["rollups"]["proposal_count"] == stats.rollups.proposal_count + + +def test_accept_rollups_off_is_byte_identical_and_dormant(tmp_path: Path) -> None: + """Flag off (the default): the ingest record keeps the exact pre-roll-up + key set, no roll-up rows exist, and promotion writes the exact old patch + (no session_date stamp).""" + question = _rollup_question() + with adapter.question_run(question, tmp_path / "q.sqlite3") as run: + stats = run.ingest() + assert stats.rollups is None + assert sorted(stats.to_record()) == [ + "candidate_memory_count", + "chunk_count", + "duplicate_count", + "ingest_seconds", + "promoted_memory_count", + "session_count", + "source_count", + ] + assert _rollup_cards(run.store) == [] + assert run.store.list_memories(status="candidate") == [] # nothing left un-promoted + for memory in run.store.list_memories(status="active"): + assert "session_date" not in memory["metadata_json"] + + +def test_accept_rollups_pass_is_deterministic_and_idempotent(tmp_path: Path) -> None: + question = _rollup_question() + with adapter.question_run(question, tmp_path / "a.sqlite3") as run_a: + stats_a = run_a.ingest(accept_rollups=True) + cards_a = sorted(str(card["canonical_text"]) for card in _rollup_cards(run_a.store)) + # Idempotency: re-running the pass on the same store proposes and + # accepts nothing new (groups are already covered by accepted cards). + again = run_a._consolidate_and_accept_rollups() + assert again.proposal_count == 0 + assert again.accepted_count == 0 + assert cards_a == sorted(str(card["canonical_text"]) for card in _rollup_cards(run_a.store)) + with adapter.question_run(question, tmp_path / "b.sqlite3") as run_b: + stats_b = run_b.ingest(accept_rollups=True) + cards_b = sorted(str(card["canonical_text"]) for card in _rollup_cards(run_b.store)) + # Determinism across fresh ingests: same inputs, same cards (grouping + # keys are content-derived; instance order comes from stamped session + # dates, not wall-clock create times). + assert cards_a == cards_b + assert stats_a.rollups is not None and stats_b.rollups is not None + assert stats_a.rollups.proposal_count == stats_b.rollups.proposal_count + assert stats_a.rollups.skipped == stats_b.rollups.skipped + + +def test_fingerprint_records_accept_rollups_flag(tmp_path: Path) -> None: + def config_with(accept_rollups: bool) -> runner.RunnerConfig: + return runner.RunnerConfig( + variant="s", + dataset_path=SYNTHETIC_FIXTURE_PATH, + limit=None, + question_ids=None, + question_ids_file=None, + resume=False, + dry_run=True, + cot=False, + workers=1, + max_items=8, + context_char_budget=12_000, + work_dir=tmp_path, + checkpoint_path=tmp_path / "c.jsonl", + report_path=tmp_path / "r.json", + keep_stores=False, + accept_rollups=accept_rollups, + ) + + off = runner.config_fingerprint(config_with(False), model=None, judge=None) + on = runner.config_fingerprint(config_with(True), model=None, judge=None) + assert off["accept_rollups"] is False + assert on["accept_rollups"] is True + # The step can never run undisclosed: the flag feeds the digest. + assert off["digest"] != on["digest"] + # And the CLI default is off, so the default replay path is unchanged. + args = runner.build_arg_parser().parse_args([]) + assert args.accept_rollups is False + + +def test_runner_dry_run_with_accept_rollups_records_ingest_stats(tmp_path: Path) -> None: + checkpoint_path = tmp_path / "checkpoint.jsonl" + report_path = tmp_path / "report.json" + exit_code = runner.main( + [ + "--dry-run", + "--accept-rollups", + "--dataset-file", + str(SYNTHETIC_FIXTURE_PATH), + "--work-dir", + str(tmp_path / "work"), + "--checkpoint", + str(checkpoint_path), + "--report", + str(report_path), + "--workers", + "1", + ] + ) + assert exit_code == runner.EXIT_OK + report = json.loads(report_path.read_text(encoding="utf-8")) + assert report["config"]["accept_rollups"] is True + records = runner.load_checkpoint(checkpoint_path) + for record in records.values(): + rollups = record["ingest"]["rollups"] + assert rollups["accepted_count"] == rollups["proposal_count"] + assert isinstance(rollups["accepted_memory_ids"], list) + assert len(rollups["accepted_memory_ids"]) == rollups["accepted_count"] + + # Control: without the flag the ingest record has no rollups block. + off_checkpoint = tmp_path / "off_checkpoint.jsonl" + exit_code = runner.main( + [ + "--dry-run", + "--dataset-file", + str(SYNTHETIC_FIXTURE_PATH), + "--work-dir", + str(tmp_path / "work_off"), + "--checkpoint", + str(off_checkpoint), + "--report", + str(tmp_path / "off_report.json"), + "--workers", + "1", + ] + ) + assert exit_code == runner.EXIT_OK + off_report = json.loads((tmp_path / "off_report.json").read_text(encoding="utf-8")) + assert off_report["config"]["accept_rollups"] is False + for record in runner.load_checkpoint(off_checkpoint).values(): + assert "rollups" not in record["ingest"] + + def test_retrieval_outcome_record_carries_pack_provenance(tmp_path: Path) -> None: """Checkpoint rows must make flips offline-attributable: retrieved source session ids, selected memory ids, and a digest of the exact rendered @@ -996,6 +1247,27 @@ def test_retrieval_outcome_record_carries_pack_provenance(tmp_path: Path) -> Non assert all(isinstance(memory_id, str) and memory_id for memory_id in memory_ids) +def test_two_seed_ingest_renders_byte_identical_context(tmp_path: Path) -> None: + """Churn hardening: re-ingesting the same haystack (fresh uuids, fresh + write clocks) must reproduce the retrieved context block byte for byte. + The rendering is uuid-free (session labels + dates + content only) and + the retrieval tie cascade is content-stable, so pack composition is a + pure function of the ingested content — the paired-run coin flips the + loss forensics traced to id tie-breaks cannot re-roll.""" + for index, question in enumerate(load_dataset(SYNTHETIC_FIXTURE_PATH)): + outcomes = [] + for seed in ("seed_a", "seed_b"): + with adapter.question_run(question, tmp_path / f"q{index}_{seed}.sqlite3") as run: + run.ingest() + outcomes.append(run.retrieve(max_items=8, context_char_budget=12_000)) + first, second = outcomes + assert first.source_session_ids == second.source_session_ids + assert first.context_sha256 == second.context_sha256 + assert first.context_block == second.context_block + assert first.memory_count == second.memory_count + assert first.excerpt_count == second.excerpt_count + + def test_knowledge_update_shaped_pack_renders_correction_above_annotated_stale_fact( tmp_path: Path, ) -> None: diff --git a/tests/unit/test_mcp.py b/tests/unit/test_mcp.py index 237d41df..1f5e93a2 100644 --- a/tests/unit/test_mcp.py +++ b/tests/unit/test_mcp.py @@ -1611,6 +1611,40 @@ def fake_pack_payload_flat(_context, _arguments): assert flat["token_report"] == report +def test_alice_context_pack_forwards_the_grounding_statistic(monkeypatch, core_surface) -> None: + # pack["grounding"] exists only when a salient query entity has zero + # corpus support; the compact view must forward it, and its absence + # must leave the response schema untouched (ungated byte-identical). + base_pack = { + "context_pack_id": "pack-1", + "query_interpretation": {"query": "Did Zorblatt Nine ship?", "query_type": "recall"}, + "relevant_memories": [], + "open_loops": [], + "sources": [], + "trace_id": "trace-1", + } + + def gated_pack(_context, _arguments): + return { + **base_pack, + "grounding": {"unsupported_entities": ["Zorblatt Nine"], "checked": 1}, + } + + monkeypatch.setattr(mcp_tools_module, "_vnext_context_pack_payload", gated_pack) + gated = call_mcp_tool( + _mcp_context(), name="alice_context_pack", arguments={"query": "Did Zorblatt Nine ship?"} + ) + assert gated["grounding"] == {"unsupported_entities": ["Zorblatt Nine"], "checked": 1} + + monkeypatch.setattr( + mcp_tools_module, "_vnext_context_pack_payload", lambda _c, _a: dict(base_pack) + ) + ungated = call_mcp_tool( + _mcp_context(), name="alice_context_pack", arguments={"query": "Did Zorblatt Nine ship?"} + ) + assert "grounding" not in ungated + + def _trusted_identity_arguments() -> dict[str, object]: return { "agent_id": "hermes", diff --git a/tests/unit/test_retrieval_stability.py b/tests/unit/test_retrieval_stability.py new file mode 100644 index 00000000..614093e9 --- /dev/null +++ b/tests/unit/test_retrieval_stability.py @@ -0,0 +1,364 @@ +"""Retrieval ordering stability: churn-hardening regression pins. + +Loss forensics on identical-config LongMemEval runs measured a handful of +pure-churn answer flips per run: equal-score ordering decisions fell +through directly to the row id — a uuid minted at ingest — so re-ingesting +the SAME content re-rolled every near-equal coin flip (bistable packs +under the slot budget). These tests pin the fix at three levels: + +* unit — the ``content_stable_tiebreak`` cascade itself, its use in + ``reciprocal_rank_fusion``, and the equal-score stage-list + stabilization (``fts_score`` / ``vector_distance`` runs); +* same store — repeated ``compile_context_pack`` calls return identical + packs modulo the per-call pack/trace uuids; +* two seeds — the same content captured into two fresh SQLite stores + (different uuids, different write clocks) compiles packs with identical + content projections (session labels, memory texts), including under a + constructed exact RRF tie between two sources. + +The id remains the FINAL total-order key everywhere: rows whose content +keys are byte-identical still order deterministically within one store. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +import json +from pathlib import Path + +from alicebot_api.sqlite_store import SQLiteVNextStore, ensure_sqlite_user, sqlite_user_connection +from alicebot_api.vnext_capture import SourceCaptureInput, VNextCaptureService +from alicebot_api.vnext_retrieval import ( + SOURCE_STAGE_TEMPORAL, + TIE_BREAK_CONTENT_STABLE, + VNextRetrievalRequest, + VNextRetrievalService, + content_stable_tiebreak, + reciprocal_rank_fusion, +) +from alicebot_api.vnext_retrieval import _stabilize_scored_rows # tie-break internals under test +from alicebot_api.vnext_temporal_query import TemporalAnchor + + +USER_ID = "33333333-3333-4333-8333-333333333333" + + +# -- cascade unit tests --------------------------------------------------------- + + +def test_cascade_prefers_older_content_event_date() -> None: + older = {"id": "zzz", "canonical_text": "short", "valid_from": "2023-01-05T00:00:00Z"} + newer = {"id": "aaa", "canonical_text": "much much longer text", "valid_from": "2023-09-01T00:00:00Z"} + # Older event date wins even though the newer row has the longer text + # and the lexicographically smaller id. + assert sorted([newer, older], key=content_stable_tiebreak) == [older, newer] + + +def test_cascade_reads_source_created_at_and_metadata_session_date() -> None: + by_column = {"id": "b", "title": "x", "source_created_at": "2023-03-01T00:00:00Z"} + by_metadata = {"id": "a", "title": "x", "metadata_json": {"session_date": "2023/04/02 (Sun) 10:00"}} + assert sorted([by_metadata, by_column], key=content_stable_tiebreak) == [by_column, by_metadata] + + +def test_cascade_puts_undated_rows_after_dated_rows() -> None: + dated = {"id": "zzz", "canonical_text": "t", "valid_from": "2024-12-31T00:00:00Z"} + undated = {"id": "aaa", "canonical_text": "a very long text indeed"} + assert sorted([undated, dated], key=content_stable_tiebreak) == [dated, undated] + + +def test_cascade_prefers_longer_then_lexicographic_text() -> None: + longer = {"id": "zzz", "canonical_text": "abcdef"} + shorter_b = {"id": "yyy", "canonical_text": "abc"} + shorter_a = {"id": "xxx", "canonical_text": "abb"} + assert sorted([shorter_b, shorter_a, longer], key=content_stable_tiebreak) == [longer, shorter_a, shorter_b] + + +def test_cascade_fingerprint_distinguishes_identical_text() -> None: + from_capture_one = { + "id": "zzz", + "canonical_text": "same text", + "metadata_json": {"capture_content_hash": "aaaa1111", "source_chunk_index": 2}, + } + from_capture_two = { + "id": "aaa", + "canonical_text": "same text", + "metadata_json": {"capture_content_hash": "bbbb2222", "source_chunk_index": 0}, + } + # Identical text from different captures orders by the capture digest + # (content), not by the uuid. + assert sorted([from_capture_two, from_capture_one], key=content_stable_tiebreak) == [ + from_capture_one, + from_capture_two, + ] + source_one = {"id": "zzz", "title": "same title", "content_hash": "aaaa"} + source_two = {"id": "aaa", "title": "same title", "content_hash": "bbbb"} + assert sorted([source_two, source_one], key=content_stable_tiebreak) == [source_one, source_two] + + +# -- reciprocal rank fusion ties ----------------------------------------------- + + +def test_rrf_equal_scores_resolve_by_content_not_id() -> None: + # Exact fused tie: each row is rank 1 of its own stage. The content + # winner (older session date) carries the LOSING id on purpose. + content_winner = {"id": "zzz", "title": "x", "metadata_json": {"session_date": "2023/01/01 (Sun) 00:00"}} + content_loser = {"id": "aaa", "title": "x", "metadata_json": {"session_date": "2023/06/01 (Thu) 00:00"}} + fused = reciprocal_rank_fusion({"chunk_fts": [content_winner], "title_recency": [content_loser]}, k=60) + assert [item["id"] for item, _score, _ranks in fused] == ["zzz", "aaa"] + scores = [score for _item, score, _ranks in fused] + assert scores[0] == scores[1] # guard: the constructed tie is exact + + +def test_rrf_identical_content_ties_fall_back_to_id() -> None: + # Byte-identical content: the id keeps the total order deterministic. + fused = reciprocal_rank_fusion( + {"fts": [{"id": "b", "title": "same"}], "vector": [{"id": "a", "title": "same"}]}, + k=60, + ) + assert [item["id"] for item, _score, _ranks in fused] == ["a", "b"] + + +# -- scored stage-list stabilization --------------------------------------------- + + +def test_stabilize_reorders_equal_scores_content_first_and_keeps_distinct_scores() -> None: + top = {"id": "m3", "canonical_text": "clearly the best match", "fts_score": 9.0} + tied_newer = {"id": "a-first", "canonical_text": "note", "valid_from": "2024-05-01T00:00:00Z", "fts_score": 4.0} + tied_older = {"id": "z-last", "canonical_text": "note", "valid_from": "2022-05-01T00:00:00Z", "fts_score": 4.0} + rows = [top, tied_newer, tied_older] + stabilized = _stabilize_scored_rows(rows) + # Distinct score keeps its store rank; the equal-score pair reorders to + # the older-dated row first despite its losing id. + assert [row["id"] for row in stabilized] == ["m3", "z-last", "a-first"] + + +def test_stabilize_supports_ascending_scores_for_vector_distance() -> None: + near = {"id": "m1", "canonical_text": "text", "vector_distance": 0.1} + tied_long = {"id": "z", "canonical_text": "longer text wins", "vector_distance": 0.5} + tied_short = {"id": "a", "canonical_text": "short", "vector_distance": 0.5} + stabilized = _stabilize_scored_rows( + [near, tied_short, tied_long], score_key="vector_distance", descending=False + ) + assert [row["id"] for row in stabilized] == ["m1", "z", "a"] + + +def test_stabilize_leaves_unscored_lists_untouched() -> None: + rows = [{"id": "b", "canonical_text": "y"}, {"id": "a", "canonical_text": "zzzz"}] + assert _stabilize_scored_rows(rows) == rows # no fts_score: store order kept + + +# -- graph and temporal stage ties ----------------------------------------------- + + +class _GraphTieStore: + """Minimal entity-hop surface: two memories tied on every timestamp.""" + + def __init__(self) -> None: + observed = "2024-01-01T00:00:00Z" + self.entities = [{"id": "entity-1", "name": "meridian", "entity_type": "project", "mention_count": 3}] + self.edges = [ + {"from_type": "memory", "to_type": "entity", "from_id": memory_id, "to_id": "entity-1", + "edge_type": "mentions", "observed_at": observed} + for memory_id in ("zzz-memory", "aaa-memory") + ] + shared = { + "status": "active", + "domain": "project", + "sensitivity": "internal", + "updated_at": "2024-01-02T00:00:00Z", + "created_at": "2024-01-02T00:00:00Z", + } + self.memories = { + # The content winner (older valid_from) carries the losing id. + "zzz-memory": {"id": "zzz-memory", "canonical_text": "meridian kickoff", + "valid_from": "2023-01-01T00:00:00Z", **shared}, + "aaa-memory": {"id": "aaa-memory", "canonical_text": "meridian kickoff", + "valid_from": "2023-08-01T00:00:00Z", **shared}, + } + + def find_entities_by_names(self, names: tuple[str, ...]) -> list[dict[str, object]]: + return list(self.entities) + + def list_edges(self, *, from_id: str | None = None, to_id: str | None = None) -> list[dict[str, object]]: + if to_id is not None: + return [edge for edge in self.edges if edge["to_id"] == to_id] + return [] + + def get_memory(self, memory_id: str) -> dict[str, object] | None: + return self.memories.get(memory_id) + + +def test_graph_stage_timestamp_ties_resolve_by_content_not_id() -> None: + store = _GraphTieStore() + rows, stage, _entities = VNextRetrievalService(store, embedding_provider=None)._memory_graph_rows( + query="meridian", + domains=[], + sensitivity_allowed=["public", "internal", "private", "unknown"], + limit=8, + ) + assert stage == "enabled" + assert [row["id"] for row in rows] == ["zzz-memory", "aaa-memory"] + + +class _TemporalTieStore: + """Sources-only surface: two same-day sources tie on window distance.""" + + def __init__(self) -> None: + self.sources = [ + # Store order deliberately leads with the content LOSER (newer + # within the same day is not distinguishable here: identical + # event date, so the longer title must win, not the id). + {"id": "aaa-source", "title": "short", "domain": "unknown", "sensitivity": "internal", + "metadata_json": {"session_date": "2023/05/10 (Wed) 00:00"}}, + {"id": "zzz-source", "title": "short but longer", "domain": "unknown", "sensitivity": "internal", + "metadata_json": {"session_date": "2023/05/10 (Wed) 00:00"}}, + ] + + def search_sources(self, *, query: str, domains=None, sensitivity_allowed=None, limit: int = 8): + return list(self.sources)[:limit] + + def list_provenance_links(self, *, target_type: str, target_id: str): + return [] + + +def test_temporal_source_boost_distance_ties_resolve_by_content() -> None: + store = _TemporalTieStore() + service = VNextRetrievalService(store, embedding_provider=None) + anchor = TemporalAnchor( + window_start=datetime(2023, 5, 1, tzinfo=UTC), + window_end=datetime(2023, 6, 1, tzinfo=UTC), + parsed_from="in May 2023", + ) + ranked_lists, _record = service._source_stage_lists( + query="what happened", + domains=[], + sensitivity_allowed=["public", "internal", "private", "unknown"], + limit=8, + winning_memories=[], + anchor=anchor, + ) + temporal = ranked_lists[SOURCE_STAGE_TEMPORAL] + # Equal distance from the window center: the longer-titled source wins + # (content cascade), not the lexicographically smaller uuid. + assert [row["id"] for row in temporal] == ["zzz-source", "aaa-source"] + + +# -- same-store and two-seed pack stability --------------------------------------- + +_SESSIONS = ( + # (session_id, date, turns) — chat-shaped content captured through the + # REAL capture service. Crafted so retrieval fuses multiple ranked + # lists: session text carries the query terms (chunk/memory FTS) and + # one source title carries them too (title_recency lexical list), + # yielding an exact cross-list RRF tie between two sources. + ( + "session-cycling", + "2023/05/20 (Sat) 02:21", + "[USER]: My favorite bicycle is a blue Brompton folding bike.\n\n" + "[ASSISTANT]: Noted! Folding bikes are great for commuting.", + ), + ( + "session-trip", + "2023/05/21 (Sun) 09:00", + "[USER]: The bicycle trip to Ghent is planned for late September.\n\n" + "[ASSISTANT]: A bicycle trip in autumn sounds lovely.", + ), + ( + "session-unrelated", + "2023/06/02 (Fri) 12:00", + "[USER]: My tomato plants are finally ripening this week.\n\n" + "[ASSISTANT]: Fresh tomatoes are the best part of summer.", + ), +) + +_QUERY = "What bicycle does the user ride?" + + +def _ingest_synthetic_store(db_path: Path) -> None: + with sqlite_user_connection(db_path, USER_ID) as conn: + ensure_sqlite_user(conn, USER_ID, "stability@alice.local", "Stability Tests") + store = SQLiteVNextStore(conn, USER_ID) + capture = VNextCaptureService(store, actor_type="system") + for session_id, date, text in _SESSIONS: + capture.capture_source( + SourceCaptureInput( + source_type="chat_session", + title=f"Chat session {session_id} on {date}", + raw_text=f"Chat session {session_id} on {date}.\n\n{text}", + connector_name="stability-test", + external_id=f"stability/{session_id}", + domain="unknown", + sensitivity="internal", + metadata_json={"session_id": session_id, "session_date": date}, + ) + ) + for memory in store.list_memories(status="candidate"): + store.update_memory(memory_id=str(memory["id"]), patch={"status": "active"}, actor_type="system") + + +def _compile_pack(db_path: Path) -> dict[str, object]: + with sqlite_user_connection(db_path, USER_ID) as conn: + store = SQLiteVNextStore(conn, USER_ID) + service = VNextRetrievalService(store) + return service.compile_context_pack( + VNextRetrievalRequest(query=_QUERY, max_items=8, include_sources=True, actor_type="system") + ) + + +def _normalized_pack(pack: dict[str, object]) -> str: + """Canonical JSON with the per-call uuids and clock-derived note removed.""" + scrubbed = json.loads(json.dumps(pack, sort_keys=True, default=str)) + scrubbed.pop("context_pack_id", None) + scrubbed.pop("trace_id", None) + trace = scrubbed.get("trace") or {} + trace.pop("trace_id", None) + return json.dumps(scrubbed, sort_keys=True) + + +def _content_projection(pack: dict[str, object]) -> dict[str, object]: + """uuid-free composition: session labels and memory texts, in pack order.""" + sources = pack.get("sources") or [] + memories = pack.get("relevant_memories") or [] + return { + "source_sessions": [ + (source.get("metadata_json") or {}).get("session_id") for source in sources + ], + "memory_texts": [memory.get("canonical_text") for memory in memories], + "fusion_tie_break": (pack.get("trace") or {}).get("fusion", {}).get("tie_break"), + } + + +def test_same_store_repeated_packs_are_identical_modulo_pack_uuids(tmp_path: Path) -> None: + db_path = tmp_path / "store.sqlite3" + _ingest_synthetic_store(db_path) + first = _compile_pack(db_path) + second = _compile_pack(db_path) + assert _normalized_pack(first) == _normalized_pack(second) + + +def test_two_seed_ingest_compiles_identical_pack_composition(tmp_path: Path) -> None: + seed_a = tmp_path / "seed_a.sqlite3" + seed_b = tmp_path / "seed_b.sqlite3" + _ingest_synthetic_store(seed_a) + _ingest_synthetic_store(seed_b) + + pack_a = _compile_pack(seed_a) + pack_b = _compile_pack(seed_b) + + projection_a = _content_projection(pack_a) + projection_b = _content_projection(pack_b) + assert projection_a == projection_b + assert projection_a["fusion_tie_break"] == TIE_BREAK_CONTENT_STABLE + # The synthetic corpus retrieves real content on both sides. + assert projection_a["source_sessions"], "expected pack sources for the synthetic corpus" + assert projection_a["memory_texts"], "expected pack memories for the synthetic corpus" + + # Guard: the corpus really does produce at least one exact fused tie + # among selected source candidates, so this test would have been a + # coin flip under the old id tie-break rather than trivially stable. + trace_scores = [ + record["rrf_score"] + for record in (pack_a.get("trace") or {}).get("selected", []) + if record.get("target_type") == "source" + ] + assert len(trace_scores) != len(set(trace_scores)), "expected an exact fused source tie" diff --git a/tests/unit/test_sqlite_onramp.py b/tests/unit/test_sqlite_onramp.py index 64553a2b..eb762f5f 100644 --- a/tests/unit/test_sqlite_onramp.py +++ b/tests/unit/test_sqlite_onramp.py @@ -206,7 +206,11 @@ def test_capture_review_approve_recall_explain_flow(sqlite_context) -> None: assert recall["count"] >= 1 assert recall["results"][0]["id"] == memory_id assert recall["results"][0]["provenance_count"] == 1 - assert recall["retrieval"]["fusion"] == {"algorithm": "reciprocal_rank_fusion", "k": 60} + assert recall["retrieval"]["fusion"] == { + "algorithm": "reciprocal_rank_fusion", + "k": 60, + "tie_break": "content_stable_v1", + } assert recall["retrieval"]["stages"]["fts"]["candidate_count"] >= 1 audit = call_mcp_tool(sqlite_context, name="alice_explain", arguments={"memory_id": memory_id}) diff --git a/tests/unit/test_vnext_answer_verification.py b/tests/unit/test_vnext_answer_verification.py new file mode 100644 index 00000000..49bc5391 --- /dev/null +++ b/tests/unit/test_vnext_answer_verification.py @@ -0,0 +1,255 @@ +"""Tests for the product answer-grounding verification seam. + +The hard requirements under test: + +- DORMANT BY DEFAULT: no other ``alicebot_api`` module imports this + seam; deploying it changes nothing until an integrator calls it. +- Judge-neutral inputs: the verifier prompt is built from the + pack-derived context, the pack's question, and the answer under test. + There is no expected-answer input and no benchmark-label vocabulary + anywhere in the module. +- Fail-open: provider errors and unparseable replies yield a grounded + verdict with the failure recorded; the gate then returns the answer + byte-identical. +- Disclosure: every verdict record carries the template fingerprint and + the provider/model when known. +""" + +from __future__ import annotations + +from pathlib import Path + +import alicebot_api +from alicebot_api.vnext_answer_verification import ( + ANSWER_VERIFIER_TEMPLATE_SHA256, + WITHHELD_ANSWER_TEXT, + apply_answer_grounding_gate, + build_answer_verifier_prompt, + pack_question, + parse_verifier_reply, + render_pack_context_block, + verify_answer_grounding, +) + + +_PACK = { + "context_pack_id": "pack-1", + "query_interpretation": {"query": "Did Marcus approve the launch?"}, + "relevant_memories": [ + {"title": "Launch timing", "canonical_text": "The launch moves to next quarter."}, + ], + "supporting_evidence": [{"excerpt": "Marcus said: ship it next quarter."}], + "sources": [{"title": "Planning session 12"}], +} + + +class _RecordingProvider: + provider = "fake_provider" + model = "fake-model-1" + + def __init__(self, reply: str) -> None: + self.reply = reply + self.prompts: list[str] = [] + self.temperatures: list[float] = [] + + def chat(self, *, prompt: str, temperature: float) -> str: + self.prompts.append(prompt) + self.temperatures.append(temperature) + return self.reply + + +# -- dormant by default --------------------------------------------------------------- + + +def test_no_other_alicebot_module_imports_the_seam() -> None: + package_dir = Path(alicebot_api.__file__).parent + importers = [ + path.name + for path in sorted(package_dir.rglob("*.py")) + if path.name != "vnext_answer_verification.py" + and "vnext_answer_verification" in path.read_text(encoding="utf-8") + ] + assert importers == [] + + +def test_module_is_independent_of_the_harness_and_label_free() -> None: + source = ( + Path(alicebot_api.__file__).parent / "vnext_answer_verification.py" + ).read_text(encoding="utf-8") + # Product code never imports the byte-frozen harness component... + assert "from longmemeval" not in source + assert "import longmemeval" not in source + # ...and no code path can key off benchmark labels. + for label in ("question_type", "abstention_label", "gold_answer", "expected_answer"): + assert label not in source + + +# -- prompt construction (judge-neutral inputs only) ---------------------------------- + + +def test_prompt_contains_context_question_and_answer_only() -> None: + prompt = build_answer_verifier_prompt( + question="Did Marcus approve the launch?", + answer_text="Yes, in March.", + context_block="- Launch timing: The launch moves to next quarter.", + ) + assert "Did Marcus approve the launch?" in prompt + assert "Yes, in March." in prompt + assert "launch moves to next quarter" in prompt + + +def test_pack_question_reads_the_query_interpretation() -> None: + assert pack_question(_PACK) == "Did Marcus approve the launch?" + assert pack_question({}) == "" + assert pack_question({"query_interpretation": {"query": " "}}) == "" + + +def test_render_pack_context_block_covers_all_sections() -> None: + pack = dict(_PACK) + pack["grounding"] = {"unsupported_entities": ["Zorblatt Nine"], "checked": 1} + block = render_pack_context_block(pack) + assert "- Launch timing: The launch moves to next quarter." in block + assert "- Evidence: Marcus said: ship it next quarter." in block + assert "- Source: Planning session 12" in block + assert '- Note: no stored memories mention "Zorblatt Nine".' in block + + +def test_render_pack_context_block_empty_pack_is_explicit() -> None: + assert render_pack_context_block({}) == ( + "(no stored memories were retrieved for this question)" + ) + + +# -- reply parsing -------------------------------------------------------------------- + + +def test_parse_grounded_reply() -> None: + grounded, claims, note = parse_verifier_reply("GROUNDED") + assert grounded is True + assert claims == () + assert note is None + + +def test_parse_load_bearing_reply_is_not_grounded() -> None: + grounded, claims, note = parse_verifier_reply( + "UNGROUNDED LOAD-BEARING: the March approval date\n" + "UNGROUNDED INCIDENTAL: the meeting room name" + ) + assert grounded is False + assert [(claim.text, claim.load_bearing) for claim in claims] == [ + ("the March approval date", True), + ("the meeting room name", False), + ] + assert note is None + + +def test_incidental_only_reply_stays_grounded() -> None: + grounded, claims, _note = parse_verifier_reply( + "UNGROUNDED INCIDENTAL: the meeting room name" + ) + assert grounded is True + assert len(claims) == 1 + + +def test_unrecognized_reply_fails_open_with_a_note() -> None: + grounded, claims, note = parse_verifier_reply("I think the answer looks fine.") + assert grounded is True + assert claims == () + assert note == "unrecognized verifier reply; failing open as grounded" + + +# -- verify_answer_grounding ---------------------------------------------------------- + + +def test_verify_routes_through_the_provider_chat_seam() -> None: + provider = _RecordingProvider("GROUNDED") + + verdict = verify_answer_grounding("Yes, next quarter.", _PACK, provider) + + assert verdict.grounded is True + assert verdict.error is None + assert verdict.provider == "fake_provider" + assert verdict.model == "fake-model-1" + assert provider.temperatures == [0.0] + prompt = provider.prompts[0] + assert "Did Marcus approve the launch?" in prompt + assert "Yes, next quarter." in prompt + assert "The launch moves to next quarter." in prompt + + +def test_verify_accepts_a_bare_callable() -> None: + prompts: list[str] = [] + + def chat(prompt: str) -> str: + prompts.append(prompt) + return "UNGROUNDED LOAD-BEARING: the March date" + + verdict = verify_answer_grounding("Yes, in March.", _PACK, chat) + + assert verdict.grounded is False + assert verdict.gate_should_withhold is True + assert verdict.provider is None and verdict.model is None + assert len(prompts) == 1 + + +def test_provider_error_fails_open_and_is_recorded() -> None: + class _Boom: + provider = "boom" + model = "boom-1" + + def chat(self, *, prompt: str, temperature: float) -> str: + raise RuntimeError("no endpoint configured") + + verdict = verify_answer_grounding("Yes.", _PACK, _Boom()) + + assert verdict.grounded is True + assert verdict.gate_should_withhold is False + assert verdict.error == "RuntimeError: no endpoint configured" + assert verdict.raw_response is None + + +def test_invalid_chat_config_fails_open_with_a_type_error_recorded() -> None: + verdict = verify_answer_grounding("Yes.", _PACK, object()) + assert verdict.grounded is True + assert verdict.error is not None and verdict.error.startswith("TypeError:") + + +def test_verdict_record_discloses_template_fingerprint_and_model() -> None: + provider = _RecordingProvider("GROUNDED") + record = verify_answer_grounding("Yes.", _PACK, provider).to_record() + assert record["template_sha256"] == ANSWER_VERIFIER_TEMPLATE_SHA256 + assert record["provider"] == "fake_provider" + assert record["model"] == "fake-model-1" + assert record["grounded"] is True + assert record["ungrounded_claims"] == [] + + +# -- the opt-in gate ------------------------------------------------------------------ + + +def test_gate_withholds_only_on_clean_load_bearing_failures() -> None: + provider = _RecordingProvider("UNGROUNDED LOAD-BEARING: the March date") + verdict = verify_answer_grounding("Yes, in March.", _PACK, provider) + + final, applied = apply_answer_grounding_gate("Yes, in March.", verdict) + assert applied is True + assert final == WITHHELD_ANSWER_TEXT + + custom, applied_custom = apply_answer_grounding_gate( + "Yes, in March.", verdict, withheld_text="No supported answer." + ) + assert applied_custom is True + assert custom == "No supported answer." + + +def test_gate_returns_the_answer_byte_identical_when_grounded_or_failed() -> None: + grounded = verify_answer_grounding("Yes.", _PACK, _RecordingProvider("GROUNDED")) + assert apply_answer_grounding_gate("Yes.", grounded) == ("Yes.", False) + + incidental = verify_answer_grounding( + "Yes.", _PACK, _RecordingProvider("UNGROUNDED INCIDENTAL: room name") + ) + assert apply_answer_grounding_gate("Yes.", incidental) == ("Yes.", False) + + errored = verify_answer_grounding("Yes.", _PACK, object()) + assert apply_answer_grounding_gate("Yes.", errored) == ("Yes.", False) diff --git a/tests/unit/test_vnext_grounding.py b/tests/unit/test_vnext_grounding.py index 181f833d..1865684b 100644 --- a/tests/unit/test_vnext_grounding.py +++ b/tests/unit/test_vnext_grounding.py @@ -105,6 +105,62 @@ def test_domain_and_dedupe_and_appearance_order() -> None: assert names == ("Marcus Chen", "type3.capital", "Lisbon") +# -- salience: attribute-qualified lowercase compounds (round 3) --------------------- + + +@pytest.mark.parametrize( + ("query", "expected"), + [ + # quantifier-qualified: a measured quantity names a specific thing + ("How many fish are there in my 30-gallon tank?", ("30-gallon tank",)), + ("Is the 10 gallon aquarium cycling yet?", ("10 gallon aquarium",)), + ("Where did I put the 5-pound dumbbell?", ("5-pound dumbbell",)), + # possessive noun-noun compounds with curated heads + ("How often do I water my snake plant?", ("snake plant",)), + ("When is my soccer team playing next?", ("soccer team",)), + ("Did I skip my karate practice last week?", ("karate practice",)), + ("Has our book club picked the next read?", ("book club",)), + ("When does my pottery class meet?", ("pottery class",)), + ], +) +def test_qualified_lowercase_compounds_are_salient(query: str, expected: tuple[str, ...]) -> None: + assert salient_query_entities(query) == expected + + +@pytest.mark.parametrize( + "query", + [ + # bare lowercase nouns: no qualifier shape, never salient + "What is the name of my hamster?", + "the fish tank at the office needs cleaning", + "did I clean the tank yesterday?", + # possessive + generic object talk: head not in the curated lexicon + "Did my credit card payment go through?", + "What is my phone number?", + "Where did I buy my new tennis racket from?", + "what did I write in my journal entry?", + # generic/descriptive modifiers never qualify + "How is my work team doing?", + "my favorite team lost again", + "when does my new class start?", + "is my old plant still alive?", + "did my first practice go okay?", + "our local club meets on Mondays", + # time-quantified events are not things ("5-day trip") + "How many shirts did I pack for my 5-day trip?", + "I went on a 30-minute walk before lunch.", + # unit words without the full quantity-unit-noun shape + "How many miles per gallon was my car getting?", + "I bought a gallon of milk and a pound of coffee.", + # blocklisted or stopword modifiers can never qualify + "what about my may lessons?", + "did my the team win?", + ], +) +def test_generic_lowercase_nouns_are_never_salient(query: str) -> None: + assert salient_query_entities(query) == () + + def test_salient_entities_are_capped() -> None: query = ( "Did Alice Chen, Bob Marley, Carol Danvers, Dave Grohl, " @@ -217,6 +273,33 @@ def test_morphological_variant_counts_as_support_on_sqlite(conn) -> None: assert corpus_support(("Hawaiian",), reverse) == {"Hawaiian": True} +def test_pure_number_tokens_never_fabricate_support_on_sqlite(conn) -> None: + # "30" on an unrelated receipt is not a mention of the 30-gallon + # tank; a bare-number hit must not silently suppress a truthful note. + store = _store(conn) + _seed_chunk(store, "the receipt total was 30 dollars at the market") + assert corpus_support(("30-gallon tank",), store) == {"30-gallon tank": False} + + +def test_word_tokens_of_a_quantified_compound_still_count_as_support(conn) -> None: + # Any word-token variant hit suppresses the note -- the safe direction. + store = _store(conn) + _seed_chunk(store, "cleaned the tank filter after feeding the fish") + assert corpus_support(("30-gallon tank",), store) == {"30-gallon tank": True} + + unit_only = _store(conn) + _seed_chunk(unit_only, "bought two gallons of water for the trip") + assert corpus_support(("30-gallon tank",), unit_only) == {"30-gallon tank": True} + + +def test_all_number_names_keep_their_digit_probe_surface(conn) -> None: + # A purely numeric name has no word tokens; its digits remain the + # only probe surface, so a corpus mention still counts as support. + store = _store(conn) + _seed_chunk(store, "the 991 arrived at the dealership on Friday") + assert corpus_support(("991",), store) == {"991": True} + + def test_corpus_support_empty_and_uncheckable_inputs(conn) -> None: assert corpus_support((), _store(conn)) == {} # A bare object exposes no probe surface: "cannot check", never a claim. @@ -327,6 +410,22 @@ def test_grounding_payload_lists_only_unsupported_entities(conn) -> None: assert grounding == {"unsupported_entities": ["Marcus Chen"], "checked": 2} +def test_grounding_fires_for_qualified_lowercase_compound(conn) -> None: + store = _store(conn) + _seed_chunk(store, "the receipt total was 30 dollars at the market") + grounding = compute_query_grounding( + store, "How many fish are there in my 30-gallon tank?" + ) + assert grounding == {"unsupported_entities": ["30-gallon tank"], "checked": 1} + + supported = _store(conn) + _seed_chunk(supported, "set up the new tank for the goldfish") + assert ( + compute_query_grounding(supported, "How many fish are there in my 30-gallon tank?") + is None + ) + + # -- context pack integration (gated; ungated path byte-identical) ------------------- diff --git a/tests/unit/test_vnext_retrieval.py b/tests/unit/test_vnext_retrieval.py index fb0b56ed..3a9d5b36 100644 --- a/tests/unit/test_vnext_retrieval.py +++ b/tests/unit/test_vnext_retrieval.py @@ -35,6 +35,7 @@ SUPERSESSION_STAGE_ENABLED, TEMPORAL_STAGE_DISABLED_NO_STORE_SUPPORT, TEMPORAL_STAGE_ENABLED, + TIE_BREAK_CONTENT_STABLE, VECTOR_STAGE_DISABLED_NO_PROVIDER, VECTOR_STAGE_ENABLED, VNextRetrievalRequest, @@ -528,7 +529,11 @@ def test_context_pack_includes_memories_sources_open_loops_provenance_and_trace( assert pack["trace"]["candidate_count"] == 3 assert pack["trace"]["selected_count"] == 3 assert pack["trace"]["vector_stage"] == VECTOR_STAGE_DISABLED_NO_PROVIDER - assert pack["trace"]["fusion"] == {"algorithm": "reciprocal_rank_fusion", "k": RRF_K} + assert pack["trace"]["fusion"] == { + "algorithm": "reciprocal_rank_fusion", + "k": RRF_K, + "tie_break": TIE_BREAK_CONTENT_STABLE, + } assert pack["trace"]["stages"]["fts"] == {"source": "postgres_fts", "candidate_count": 1} assert pack["trace"]["stages"]["vector"] == { "status": VECTOR_STAGE_DISABLED_NO_PROVIDER,