diff --git a/Dockerfile b/Dockerfile index 0080d62..05d766d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -43,8 +43,11 @@ RUN chmod +x /usr/local/bin/entrypoint.sh # already exists. Without this the volume is created owned by root, the # embedding model download fails with EACCES, and semantic search silently # degrades to keyword-only — a working server that quietly answers worse. +# The state directory gets the same treatment, and for the same reason: it is a +# named volume mount point, so it must exist and be owned by openindex or the +# volume lands root-owned and the audit trail silently records nothing. RUN useradd --create-home --uid 10001 openindex \ - && mkdir -p /home/openindex/model-cache \ + && mkdir -p /home/openindex/model-cache /home/openindex/.local/state/open-index \ && chown -R openindex /brain /home/openindex USER openindex diff --git a/docker-compose.yml b/docker-compose.yml index db80563..4dd9823 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -32,6 +32,11 @@ x-brain-service: &brain-service # on every recreate — turning each restart into a fresh download. # OPEN_INDEX_EMBEDDING_CACHE (below) points it here. - model-cache:/home/openindex/model-cache + # The retrieval audit trail (which query returned which document, and the + # trace id it ran under). It lives in the user's state directory, which is + # inside the container — so without this volume every redeploy silently + # discards the history you would want precisely when debugging one. + - analytics-state:/home/openindex/.local/state/open-index environment: &brain-env OPEN_INDEX_TOKEN: ${OPEN_INDEX_TOKEN:-} # What the startup banner advertises. Defaulted because auto-detection is @@ -140,3 +145,4 @@ services: volumes: opensearch-data: model-cache: + analytics-state: diff --git a/open_index/analytics.py b/open_index/analytics.py index a578e82..edad68b 100644 --- a/open_index/analytics.py +++ b/open_index/analytics.py @@ -2,18 +2,43 @@ Usage can contain raw queries, so state lives outside the public brain checkout under ~/.local/state/open-index rather than beside source-controlled files. + +Two levels are recorded. `context_fetches` is one row per read — the question. +`retrieval_results` is one row per document that read returned — the answer, +with the score and the reason it matched. The second is what makes an agent's +memory debuggable: "this turn retrieved that document, ranked third, on a +semantic match at 0.71" is answerable, and "why did it think that?" stops being +guesswork. + +Both carry the caller's `trace_id` when one was supplied, so a turn can be +followed from the agent's side back into the index. """ from __future__ import annotations import hashlib import json +import os import sqlite3 import threading from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional +# How many reads to keep. Per-result rows multiply volume by the page size, and +# this is a local debugging aid, not a warehouse. OPEN_INDEX_ANALYTICS_MAX +# raises it, or 0 disables pruning for a deployment that ships the file +# somewhere durable. +_DEFAULT_MAX_FETCHES = 50_000 + + +def _as_float(value: Any) -> Optional[float]: + """Best-effort float. Analytics must never fail a read it is describing.""" + try: + return None if value is None else float(value) + except (TypeError, ValueError): + return None + class AnalyticsStore: """Record and aggregate the context that CLI/MCP clients retrieve.""" @@ -25,6 +50,11 @@ def __init__(self, brain_root: Optional[Path]): state_home.mkdir(parents=True, exist_ok=True) self.path = state_home / f"{slug}.db" self._lock = threading.Lock() + try: + self._max_fetches = int( + os.environ.get("OPEN_INDEX_ANALYTICS_MAX", _DEFAULT_MAX_FETCHES)) + except ValueError: + self._max_fetches = _DEFAULT_MAX_FETCHES self._conn = sqlite3.connect(str(self.path), check_same_thread=False) self._conn.row_factory = sqlite3.Row self._conn.executescript( @@ -45,10 +75,49 @@ def __init__(self, brain_root: Optional[Path]): ); CREATE INDEX IF NOT EXISTS idx_context_fetches_at ON context_fetches(fetched_at); + + -- One row per document returned. Deliberately not a foreign key + -- with ON DELETE CASCADE: pruning deletes from both tables in one + -- transaction, and a hard constraint would turn an analytics + -- bookkeeping slip into a failed read. + CREATE TABLE IF NOT EXISTS retrieval_results ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + fetch_id INTEGER NOT NULL, + rank INTEGER NOT NULL, + entity_id TEXT NOT NULL, + doc_type TEXT, + score REAL, + keyword_score REAL, + semantic_score REAL, + match_type TEXT + ); + CREATE INDEX IF NOT EXISTS idx_retrieval_fetch + ON retrieval_results(fetch_id); + CREATE INDEX IF NOT EXISTS idx_retrieval_entity + ON retrieval_results(entity_id); """ ) + self._migrate() self._conn.commit() + def _migrate(self) -> None: + """Add columns an older state database predates. + + These files live in the user's state directory and outlive any single + version, so a new column has to arrive by ALTER rather than by assuming + CREATE TABLE ran with it. + """ + existing = { + row["name"] + for row in self._conn.execute("PRAGMA table_info(context_fetches)") + } + if "trace_id" not in existing: + self._conn.execute("ALTER TABLE context_fetches ADD COLUMN trace_id TEXT") + self._conn.execute( + "CREATE INDEX IF NOT EXISTS idx_context_fetches_trace " + "ON context_fetches(trace_id)" + ) + def record( self, *, @@ -62,23 +131,131 @@ def record( result_doc_types: Optional[dict[str, int]] = None, success: bool = True, error: Optional[str] = None, - ) -> None: + trace_id: Optional[str] = None, + results: Optional[list[dict[str, Any]]] = None, + ) -> Optional[int]: + """Record one read, and the documents it returned. Returns the fetch id. + + `results` are the rows the caller actually received, in the order they + were received: rank is position, not score order, because that is what + the agent saw. + """ with self._lock: - self._conn.execute( + cur = self._conn.execute( """ INSERT INTO context_fetches ( fetched_at, source, operation, query, doc_types, entity_id, - result_count, result_doc_types, duration_ms, success, error - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + result_count, result_doc_types, duration_ms, success, error, + trace_id + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( datetime.now(timezone.utc).isoformat(), source, operation, query, json.dumps(doc_types or []), entity_id, result_count, json.dumps(result_doc_types or {}), round(duration_ms, 2), - int(success), error, + int(success), error, trace_id, ), ) + fetch_id = cur.lastrowid + if results: + self._conn.executemany( + """ + INSERT INTO retrieval_results ( + fetch_id, rank, entity_id, doc_type, score, + keyword_score, semantic_score, match_type + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + ( + fetch_id, i, str(r.get("id") or ""), r.get("doc_type"), + _as_float(r.get("score")), + _as_float((r.get("match") or {}).get("keyword_score")), + _as_float((r.get("match") or {}).get("semantic_score")), + (r.get("match") or {}).get("type"), + ) + for i, r in enumerate(results, start=1) + if r.get("id") + ], + ) self._conn.commit() + self._prune_locked() + return fetch_id + + def _prune_locked(self) -> None: + """Keep the state file bounded. + + Per-result rows multiply volume by the page size — a few thousand + searches is tens of thousands of rows — and this is a debugging aid on + someone's laptop, not a warehouse. Oldest fetches go first, with their + results, so a trace is either wholly present or wholly gone rather than + surviving as a fetch with no documents. + """ + if self._max_fetches <= 0: + return + row = self._conn.execute("SELECT COUNT(*) FROM context_fetches").fetchone() + if row[0] <= self._max_fetches: + return + cutoff = self._conn.execute( + "SELECT id FROM context_fetches ORDER BY id DESC LIMIT 1 OFFSET ?", + (self._max_fetches - 1,), + ).fetchone() + if cutoff is None: + return + self._conn.execute("DELETE FROM retrieval_results WHERE fetch_id < ?", (cutoff[0],)) + self._conn.execute("DELETE FROM context_fetches WHERE id < ?", (cutoff[0],)) + self._conn.commit() + + def results_for(self, fetch_ids: list[int]) -> dict[int, list[dict[str, Any]]]: + """The documents each of those fetches returned, keyed by fetch id.""" + if not fetch_ids: + return {} + placeholders = ",".join("?" * len(fetch_ids)) + with self._lock: + rows = self._conn.execute( + f"SELECT * FROM retrieval_results WHERE fetch_id IN ({placeholders}) " + f"ORDER BY fetch_id, rank", + fetch_ids, + ).fetchall() + out: dict[int, list[dict[str, Any]]] = {} + for row in rows: + out.setdefault(row["fetch_id"], []).append(dict(row)) + return out + + def by_trace(self, trace_id: str) -> list[dict[str, Any]]: + """Every read made under one trace id, each with the documents it returned.""" + if not trace_id: + return [] + with self._lock: + fetches = self._conn.execute( + "SELECT * FROM context_fetches WHERE trace_id = ? ORDER BY id", + (trace_id,), + ).fetchall() + events = [dict(row) for row in fetches] + results = self.results_for([e["id"] for e in events]) + for event in events: + event["results"] = results.get(event["id"], []) + return events + + def retrievals_of(self, entity_id: str, limit: int = 50) -> list[dict[str, Any]]: + """Which queries returned this document, most recent first. + + The question asked when a document keeps turning up where it should not: + not "what did this query return" but "what is retrieving this". + """ + with self._lock: + rows = self._conn.execute( + """ + SELECT r.rank, r.score, r.keyword_score, r.semantic_score, + r.match_type, f.id AS fetch_id, f.fetched_at, f.source, + f.operation, f.query, f.trace_id + FROM retrieval_results r + JOIN context_fetches f ON f.id = r.fetch_id + WHERE r.entity_id = ? + ORDER BY r.fetch_id DESC LIMIT ? + """, + (entity_id, limit), + ).fetchall() + return [dict(row) for row in rows] def recent(self, limit: int = 100) -> list[dict[str, Any]]: with self._lock: @@ -132,16 +309,30 @@ def summary(self) -> dict[str, Any]: class NullAnalyticsStore: - """No-op fallback when the user's local state directory is not writable.""" + """No-op fallback when the user's local state directory is not writable. + + Mirrors the real store's surface exactly. Anything missing here becomes an + AttributeError on a machine where analytics happen to be unavailable — a + failure in the path whose entire purpose is to not fail. + """ path = None - def record(self, **_: Any) -> None: + def record(self, **_: Any) -> Optional[int]: return None def recent(self, limit: int = 100) -> list[dict[str, Any]]: return [] + def by_trace(self, trace_id: str) -> list[dict[str, Any]]: + return [] + + def retrievals_of(self, entity_id: str, limit: int = 50) -> list[dict[str, Any]]: + return [] + + def results_for(self, fetch_ids: list[int]) -> dict[int, list[dict[str, Any]]]: + return {} + def summary(self) -> dict[str, Any]: return { "available": False, diff --git a/open_index/brain.py b/open_index/brain.py index 0f3deb8..0e66e61 100644 --- a/open_index/brain.py +++ b/open_index/brain.py @@ -19,6 +19,7 @@ from open_index.storage import get_backend from open_index.storage.base import SearchBackend, SearchResults from open_index.analytics import AnalyticsStore, NullAnalyticsStore +from open_index.tracing import current_trace_id @dataclass @@ -306,6 +307,7 @@ def search( source=source, operation="search", started=started, query=query, doc_types=doc_types, result_count=results.total, result_doc_types=results.doc_type_counts, + results=results.results, ) return results @@ -331,10 +333,14 @@ def search( results.total = len(kept) results.doc_type_counts = counts if source: + # `kept`, not the pre-filter rows: the trail must show what the + # caller actually received, or a document dropped by a confidence + # or validity filter would look like it was handed over. self._record_fetch( source=source, operation="search", started=started, query=query, doc_types=doc_types, result_count=results.total, result_doc_types=results.doc_type_counts, + results=kept, ) return results @@ -356,12 +362,24 @@ def get_entity( source=source, operation="get_entity", started=started, entity_id=entity_id, result_count=int(entity is not None), result_doc_types=({entity.doc_type: 1} if entity else {}), + # A lookup returns a document too, so it belongs in the same + # per-result trail as a search — otherwise a trace shows the + # searches an agent ran but not the documents it then opened. + results=([{"id": entity.id, "doc_type": entity.doc_type, + "score": None, + "match": {"type": "lookup", "keyword_score": None, + "semantic_score": None}}] + if entity else []), ) return entity def record_fetch(self, *, started: float, **event: Any) -> None: """Keep analytics best-effort so context access remains the priority.""" try: + # The trace id is ambient: it enters once at the edge (an HTTP + # header, an MCP tool argument) rather than being threaded through + # every read. An explicit one still wins. + event.setdefault("trace_id", current_trace_id()) self.analytics.record(duration_ms=(perf_counter() - started) * 1000, **event) except Exception: pass @@ -376,6 +394,14 @@ def analytics_summary(self) -> dict[str, Any]: def analytics_events(self, limit: int = 100) -> list[dict[str, Any]]: return self.analytics.recent(limit) + def analytics_by_trace(self, trace_id: str) -> list[dict[str, Any]]: + """Every read made under one trace id, with the documents each returned.""" + return self.analytics.by_trace(trace_id) + + def retrievals_of(self, entity_id: str, limit: int = 50) -> list[dict[str, Any]]: + """Which queries returned this document, most recent first.""" + return self.analytics.retrievals_of(entity_id, limit) + def counts(self) -> dict[str, int]: return self.backend.counts() @@ -502,6 +528,29 @@ def _guide_read_section() -> list[str]: "- `get_entity(entity_id)` — one entity plus its incoming *and* outgoing edges.\n" "- Start broad with a query, then narrow with `doc_types`. To enumerate a type,\n" " search with an empty query and a `doc_types` filter.", + "\n## Choosing how to match", + "`search_brain(mode=...)` decides *which* documents can come back at all:\n" + "- `hybrid` (default) — keyword matches plus the nearest by meaning.\n" + "- `keyword` — literal terms only. Use for an exact name, code or id, where a\n" + " near-miss is worse than no answer.\n" + "- `semantic` — meaning only. Use when the right words may not appear in the\n" + " document at all.\n\n" + "Every result carries `match`, saying why it came back — `keyword`, `semantic`,\n" + "`both`, `filter` or `none` — with each arm's score. Read it before trusting a\n" + "result: a semantic-only hit at a low score is a guess, not a fact.", + "\n## Exact filtering", + '`search_brain(filters={"field": "value"})` is a hard predicate, not a ranking\n' + "hint: a document that does not match cannot be returned at any score. Use it\n" + "whenever the answer must be scoped to one account, tenant or user.\n\n" + "Only fields declared `filterable` can be filtered, and filtering on any other\n" + "field is an error rather than being ignored — so a filter never silently\n" + "returns everything. The error names the fields you *can* filter.", + "\n## Making a retrieval debuggable", + "Pass `trace_id=\"\"` to `search_brain` and\n" + "`get_entity`. Every document returned is then recoverable by that id later,\n" + "with its rank, score and match type — which is how a human works out\n" + "afterwards what this index actually fed you. Cheap to pass, impossible to\n" + "reconstruct if you did not.", "\n## Retrieval workflow", '1. Start with `search_brain(query="...")` using the user\'s domain terms.\n' "2. Narrow with `doc_types=[...]` when the concept is known.\n" diff --git a/open_index/cli.py b/open_index/cli.py index fcf5c88..348dc2c 100644 --- a/open_index/cli.py +++ b/open_index/cli.py @@ -380,6 +380,39 @@ def ui( serve_ui(host=host, port=port) +@app.command() +def trace( + trace_id: str = typer.Argument(..., help="The trace id to look up."), + brain: str = BrainOpt, +): + """Show what one turn retrieved: each query, and the documents it returned. + + The debugging path when an agent behaved oddly — not "what does this index + contain" but "what did it actually hand over, ranked how, and why". + """ + b = _open_brain(brain) + events = b.analytics_by_trace(trace_id) + if not events: + typer.secho(f"nothing recorded under trace '{trace_id}'", + fg=typer.colors.YELLOW) + typer.echo(" Either that turn made no reads, or it did not send the id " + "(X-Trace-Id header, or the trace_id tool argument).") + raise typer.Exit(1) + + typer.secho(f"{len(events)} read(s) under {trace_id}", fg=typer.colors.GREEN) + for ev in events: + context = ev["query"] or ev["entity_id"] or "—" + typer.echo(f"\n {ev['operation']} {context}" + f" ({ev['duration_ms']} ms, via {ev['source']})") + if not ev["results"]: + typer.echo(" returned nothing") + continue + for r in ev["results"]: + score = "—" if r["score"] is None else f"{r['score']:g}" + typer.echo(f" {r['rank']:>2}. {r['entity_id']:<38} " + f"{(r['match_type'] or '—'):<9} score={score}") + + @app.command() def mcp( brain: str = BrainOpt, diff --git a/open_index/mcp_server.py b/open_index/mcp_server.py index a51dd91..34189f0 100644 --- a/open_index/mcp_server.py +++ b/open_index/mcp_server.py @@ -53,6 +53,20 @@ def _load_server_class(): ) from exc +def _trace_scope(trace_id: Optional[str]): + """Bind a caller-supplied trace id, or leave any ambient one alone. + + nullcontext rather than trace(None) when absent: entering the context with + None would *clear* a trace already set at the transport edge, so an omitted + argument would silently detach the read from its turn. + """ + from contextlib import nullcontext + + from open_index.tracing import trace + + return trace(trace_id) if trace_id else nullcontext() + + def build_server(brain: Brain, read_only: bool = False): """Construct an MCP server bound to an open brain. @@ -101,6 +115,7 @@ def search_brain( limit: int = 20, mode: str = "hybrid", filters: Optional[dict[str, Any]] = None, + trace_id: Optional[str] = None, ) -> str: """Search the brain. @@ -123,11 +138,17 @@ def search_brain( Each result carries `match`, saying why it came back: type is "keyword", "semantic", "both", "filter" or "none", with the normalised score from each arm. + + `trace_id` ties this read to the turn that caused it. Pass your own + request or conversation id and every document returned here is + recoverable later by that id, with its rank, score and match type — + which is how you work out afterwards what the index actually fed you. """ - results = brain.search( - query=query, doc_types=doc_types, limit=limit, source="mcp", - mode=mode, filters=filters, - ) + with _trace_scope(trace_id): + results = brain.search( + query=query, doc_types=doc_types, limit=limit, source="mcp", + mode=mode, filters=filters, + ) return json.dumps( { "query": query, @@ -141,10 +162,15 @@ def search_brain( ) @server.tool() - def get_entity(entity_id: str) -> str: + def get_entity(entity_id: str, trace_id: Optional[str] = None) -> str: """Fetch a single entity by id (e.g. "product:checkout"), including its - outgoing and incoming relationships with their edge meanings.""" - entity = brain.get_entity(entity_id, source="mcp") + outgoing and incoming relationships with their edge meanings. + + `trace_id` ties this lookup to the turn that caused it, so it appears + alongside the searches in that trace rather than the trail showing what + was searched but not what was then opened.""" + with _trace_scope(trace_id): + entity = brain.get_entity(entity_id, source="mcp") if entity is None: return json.dumps({"error": f"no entity '{entity_id}'"}) payload = entity.to_json() diff --git a/open_index/tracing.py b/open_index/tracing.py new file mode 100644 index 0000000..122fe7b --- /dev/null +++ b/open_index/tracing.py @@ -0,0 +1,77 @@ +"""The id that ties one agent turn to the retrievals it caused. + +An agent asks a question, the index answers with documents, and the agent then +does something surprising. Working out why means knowing *which* retrieval fed +that turn — so every recorded fetch carries a trace id, and looking one up +returns the queries, the documents, and the scores behind them. + +The id is ambient rather than a parameter on every method: it enters once, at +the edge, and `Brain` reads it wherever it records. A ContextVar is the right +shape for that — it is per-task and per-thread, so two concurrent requests never +see each other's id, which a module global would not survive. + +Two edges set it: + + HTTP an `X-Trace-Id` request header, via `trace_from_headers` + MCP an explicit `trace_id` tool argument, because an MCP tool has no + ambient request to read + +Nothing generates one. An absent trace id is recorded as NULL and everything +still works; inventing one would create ids that correlate nothing, which is +worse than none at all. +""" + +from __future__ import annotations + +import re +from contextlib import contextmanager +from contextvars import ContextVar +from typing import Iterator, Optional + +_current: ContextVar[Optional[str]] = ContextVar("open_index_trace_id", default=None) + +# Trace ids arrive from outside and are stored and displayed, so they are bounded +# and restricted to characters that cannot break out of a log line or a page. +# Anything else is dropped rather than sanitised: a mangled id would silently +# fail to correlate with whatever the caller thinks it sent. +_VALID = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") + +TRACE_HEADER = "x-trace-id" + + +def normalize(value: Optional[str]) -> Optional[str]: + """A usable trace id, or None. Never raises — a bad id must not fail a read.""" + if not value: + return None + value = value.strip() + return value if _VALID.match(value) else None + + +def current_trace_id() -> Optional[str]: + return _current.get() + + +def set_trace_id(value: Optional[str]) -> None: + _current.set(normalize(value)) + + +@contextmanager +def trace(value: Optional[str]) -> Iterator[Optional[str]]: + """Bind a trace id for the duration of a block, then restore the previous. + + Restoring matters on a server: the worker outlives the request, and a leaked + id would attribute the next caller's retrievals to the previous one. + """ + token = _current.set(normalize(value)) + try: + yield _current.get() + finally: + _current.reset(token) + + +def trace_from_headers(headers) -> Optional[str]: + """Read the trace id from request headers (case-insensitive).""" + try: + return normalize(headers.get(TRACE_HEADER)) + except Exception: + return None diff --git a/open_index/ui/templates/analytics.html b/open_index/ui/templates/analytics.html index 3c03931..f304fc0 100644 --- a/open_index/ui/templates/analytics.html +++ b/open_index/ui/templates/analytics.html @@ -17,6 +17,65 @@

Analytics

Stored in ~/.local/state/open-index/, outside the brain checkout. Search text and entity ids stay on that machine.

+
+

Trace lookup

+

Every read carries the caller's X-Trace-Id header + (or the trace_id argument, over MCP). Paste one to see what that + turn retrieved — each query, each document, its rank, score and why it matched.

+
+ + +
+
+ +{% if trace_events is not none %} + {% if not trace_events %} +
Nothing recorded under {{ trace_query }}. + Either that turn made no reads, or it did not send the id.
+ {% else %} +

{{ trace_events | length }} read{{ "" if trace_events | length == 1 else "s" }} + under {{ trace_query }}.

+ {% for ev in trace_events %} +
+ + {{ ev.operation }} + + {{ ev.query or ev.entity_id or "—" }} · {{ ev.results | length }} returned + · {{ ev.duration_ms }} ms · {{ ev.source }} + + +
+ {% if ev.results %} +
+ + + + + {% for r in ev.results %} + + + + + + + + + + {% endfor %} + +
#documenttypematchscorekeywordmeaning
{{ r.rank }}{{ r.entity_id }}{{ r.doc_type or "—" }}{{ r.match_type or "—" }}{{ "—" if r.score is none else r.score }}{{ "—" if r.keyword_score is none else r.keyword_score }}{{ "—" if r.semantic_score is none else r.semantic_score }}
+
+ {% else %} +

Returned nothing.

+ {% endif %} +
+
+ {% endfor %} + {% endif %} +{% endif %} +
{{ stats.total_fetches | comma }}
fetches
{{ stats.failed_fetches | comma }}
failed
diff --git a/open_index/ui/templates/entity.html b/open_index/ui/templates/entity.html index e860423..dd69ca4 100644 --- a/open_index/ui/templates/entity.html +++ b/open_index/ui/templates/entity.html @@ -74,6 +74,40 @@

Relationships ({{ links | length }})

{% endif %} +{% if retrievals %} +
+ What retrieved this ({{ retrievals | length }}) +
+

Queries that returned this document, most recent first — + the trail to follow when it keeps appearing in an agent's context and + should not.

+
+ + + + + {% for r in retrievals %} + + + + + + + + + + {% endfor %} + +
whenclientquery#matchscoretrace
{{ r.fetched_at[:19] }}{{ r.source }}{{ r.query or r.operation }}{{ r.rank }}{{ r.match_type or "—" }}{{ "—" if r.score is none else r.score }} + {% if r.trace_id %} + {{ r.trace_id }} + {% else %}—{% endif %} +
+
+
+
+{% endif %} +

Show on the map →

diff --git a/open_index/ui/web.py b/open_index/ui/web.py index 0b4eb80..3e607bc 100644 --- a/open_index/ui/web.py +++ b/open_index/ui/web.py @@ -234,6 +234,9 @@ def page_entity(request, name: str, brain: Brain, entity_id: str) -> dict[str, A fields=view.field_rows(entity), provenance=view.provenance_row(entity), links=view.neighbours(brain, entity_id), + # "What keeps retrieving this?" — the question when a document turns up + # in an agent's context where it should not. + retrievals=brain.retrievals_of(entity_id, limit=25), ) return ctx @@ -289,6 +292,12 @@ def page_analytics(request, name: str, brain: Brain) -> dict[str, Any]: ctx["stats"] = summary ctx["events"] = brain.analytics_events(limit=100) if summary.get( "total_fetches") else [] + + # Trace lookup: the whole point of recording the id is being able to ask + # "what did this turn actually retrieve?" afterwards. + wanted = (request.query_params.get("trace") or "").strip() + ctx["trace_query"] = wanted + ctx["trace_events"] = brain.analytics_by_trace(wanted) if wanted else None return ctx @@ -328,11 +337,34 @@ def page_jobs(request, name: str, brain: Brain) -> dict[str, Any]: def build_app(): """The Starlette app serving the explorer.""" from starlette.applications import Starlette + from starlette.middleware import Middleware + from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import JSONResponse, RedirectResponse from starlette.routing import Mount, Route from starlette.staticfiles import StaticFiles from starlette.templating import Jinja2Templates + from open_index.tracing import TRACE_HEADER, trace, trace_from_headers + + class TraceMiddleware(BaseHTTPMiddleware): + """Bind X-Trace-Id for the life of the request. + + Set here rather than in each handler so every read a page performs — + including ones several calls deep — is attributed to the same turn. The + context manager restores the previous value on the way out: a worker + outlives its request, and a leaked id would credit the next caller's + retrievals to the previous one. + """ + + async def dispatch(self, request, call_next): + with trace(trace_from_headers(request.headers)) as tid: + response = await call_next(request) + if tid: + # Echoed so a caller can confirm the id actually took, rather + # than discovering weeks later that a malformed one was dropped. + response.headers[TRACE_HEADER] = tid + return response + templates = Jinja2Templates(directory=str(TEMPLATES)) templates.env.filters["comma"] = lambda n: f"{n:,}" templates.env.filters["md"] = inline_markdown @@ -444,7 +476,7 @@ def healthz(request): routes.append(Route("/{name}/entity/{entity_id:path}", entity)) routes.append(Route("/{name}/api/graph", graph_json)) - return Starlette(routes=routes) + return Starlette(routes=routes, middleware=[Middleware(TraceMiddleware)]) def serve(host: str = "0.0.0.0", port: int = 8501) -> None: diff --git a/tests/test_retrieval_audit.py b/tests/test_retrieval_audit.py new file mode 100644 index 0000000..1f81469 --- /dev/null +++ b/tests/test_retrieval_audit.py @@ -0,0 +1,339 @@ +"""The retrieval audit trail: what was returned, why, and under whose trace. + +The question this exists to answer is not "what does the index contain" but +"what did it actually hand the agent, ranked how, and on what basis" — which +needs a row per returned document, not per read. +""" + +import shutil + +import pytest +from starlette.testclient import TestClient + +from open_index.brain import Brain +from open_index.tracing import current_trace_id, normalize, trace + +EXAMPLE = "examples/support-brain" + + +@pytest.fixture +def brain(tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path / "home")) # isolate the state db + (tmp_path / "home").mkdir() + d = tmp_path / "b" + shutil.copytree(EXAMPLE, d) + b = Brain.open(d) + b.index() + return b + + +# -- the trace id itself ------------------------------------------------------- + + +def test_a_trace_id_is_bound_for_the_block_and_restored_after(): + """A leaked id on a reused worker would credit the next caller's retrievals + to the previous one.""" + assert current_trace_id() is None + with trace("turn-1"): + assert current_trace_id() == "turn-1" + with trace("turn-2"): + assert current_trace_id() == "turn-2" + assert current_trace_id() == "turn-1" + assert current_trace_id() is None + + +@pytest.mark.parametrize("value", ["turn-1", "a.b:c-d", "A1", "x" * 128]) +def test_reasonable_ids_are_accepted(value): + assert normalize(value) == value + + +@pytest.mark.parametrize("value", ["", None, " ", "x" * 129, "bad id", "