diff --git a/AGENTS.md b/AGENTS.md index d7b98ff..b73a0e6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,6 +42,7 @@ Semantic search over agent session transcripts. Independent project, originally - **Hosted embeddings deferred** — SESF-6 keeps embedding fully local (MLX). No hosted/OpenAI setup steps, credentials, or collections exist. Future hosted path would require a separate identity/collection to avoid vector mixing. - **Optional LaunchAgent for multi-harness startup** — `./sessionflow-server.sh install-agent` writes `~/Library/LaunchAgents/cc.lbruton.sessionflow.plist` so the server starts at login before any harness hook races. OPTIONAL — `start/stop/status/restart` behavior is unchanged when not installed. `agent-status` / `uninstall-agent` manage it. `setup.sh` honors `SESSIONFLOW_INSTALL_AGENT=1` for non-interactive installs. - **Optional hourly backfill LaunchAgent (SESF-7)** — `./sessionflow-server.sh install-backfill-agent` writes `cc.lbruton.sessionflow-backfill.plist` with a `StartInterval` (defaults to 3600s; override via `SESSIONFLOW_BACKFILL_INTERVAL_SECONDS`). It runs `python cleanup.py backfill enqueue --provider claude_code_cli --mode incremental`, decoupling catch-up indexing from MCP session activity. MCP-client SessionStart hooks should connect to the already-running server rather than racing to start it. +- **Retrieval eval (SESF-82)** — `eval_retrieval.py` talks to the running server over MCP and never embeds. The real query set and reports live in the private companion (`DocVault/Projects/SessionFlow/eval/` in Devops), never in this public repo; `tests/eval/fixture_queries.jsonl` is synthetic. `run --env KEY=VALUE` refuses when `/health` `ranking` disagrees, so ranking env must reach the server process. `sweep --yes` does that with `launchctl setenv` and `kickstart -k`, then unsets and restarts. A plist `EnvironmentVariables` entry overrides `setenv`. `freshness` writes a `fact` with `topic_key` `eval:freshness:` and deletes it; probes on one host take turns on `~/.sessionflow/eval-freshness.lock`. `SESSIONFLOW_SEARCH_LOG=1` (off by default) logs searches and follow-ups to `~/.sessionflow/search-log.jsonl` (`search_log.py`). `install-eval-agent` needs `SESSIONFLOW_EVAL_QUERIES` and `SESSIONFLOW_EVAL_OUT`. - **Schema drift (SESF-11)** — `_ensure_collection` refuses to start when the live Milvus schema differs from `_expected_schema_fields()`. Drop+recreate is opt-in via `SESSIONFLOW_AUTO_MIGRATE_SCHEMA=1`, or run `python cleanup.py migrate-schema` explicitly (destructive). - **`SESSIONFLOW_BACKFILL_MAX_TURNS_PER_RUN` (SESF-12)** — default raised from 200 to 100,000 so a long-running server isn't silently capped after a few minutes. The budget logs a single WARN the first time the cap is hit so operators don't have to grep for "0 turns" lines. - **FTS5 thread affinity (SESF-13)** — `FTSIndex` keeps per-thread persistent connections (`threading.local`). Server-mode connections opened on the embed executor and request threads are isolated, and cross-thread `close_all()` is a no-op rather than a noisy WARN. diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b815b4..14d7816 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ have no git tags or GitHub Releases. For changes before the fork, see [UPSTREAM- ## [Unreleased] +## [0.9.0] - 2026-10-30 + +### Added + +- `eval_retrieval.py` scores a labelled JSONL query set against the running server: recall@5, MRR and nDCG@10 per query group, plus a current-first rate for superseded decisions and the top hit score for no-answer queries. `run`, `sweep` (restarts the LaunchAgent once per env value; needs `--yes`), `freshness` (a canary memory, deleted after the probe), `usage` and `nightly` subcommands (SESF-82). +- `SESSIONFLOW_SEARCH_LOG=1` appends each search, `get_memory` and `get_turns` call to `~/.sessionflow/search-log.jsonl` with redacted queries and result IDs in rank order. It is off by default (SESF-82). +- `/health` reports the hybrid ranking settings under `ranking` (SESF-82). +- `sessionflow-server.sh install-eval-agent`, `uninstall-eval-agent` and `eval-agent-status` manage a nightly LaunchAgent that runs `eval_retrieval.py nightly` (SESF-82). + ## [0.8.1] - 2026-10-30 ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md index 16aa204..e35ef9f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -42,6 +42,7 @@ Semantic search over Claude Code session transcripts. Independent project, origi - **Hosted embeddings deferred** — SESF-6 keeps embedding fully local (MLX). No hosted/OpenAI setup steps, credentials, or collections exist. Future hosted path would require a separate identity/collection to avoid vector mixing. - **Optional LaunchAgent for multi-harness startup** — `./sessionflow-server.sh install-agent` writes `~/Library/LaunchAgents/cc.lbruton.sessionflow.plist` so the server starts at login before any harness hook races. OPTIONAL — `start/stop/status/restart` behavior is unchanged when not installed. `agent-status` / `uninstall-agent` manage it. `setup.sh` honors `SESSIONFLOW_INSTALL_AGENT=1` for non-interactive installs. - **Optional hourly backfill LaunchAgent (SESF-7)** — `./sessionflow-server.sh install-backfill-agent` writes `cc.lbruton.sessionflow-backfill.plist` with a `StartInterval` (defaults to 3600s; override via `SESSIONFLOW_BACKFILL_INTERVAL_SECONDS`). It runs `python cleanup.py backfill enqueue --provider claude_code_cli --mode incremental`, decoupling catch-up indexing from MCP session activity. MCP-client SessionStart hooks should connect to the already-running server rather than racing to start it. +- **Retrieval eval (SESF-82)** — `eval_retrieval.py` talks to the running server over MCP and never embeds. The real query set and reports live in the private companion (`DocVault/Projects/SessionFlow/eval/` in Devops), never in this public repo; `tests/eval/fixture_queries.jsonl` is synthetic. `run --env KEY=VALUE` refuses when `/health` `ranking` disagrees, so ranking env must reach the server process. `sweep --yes` does that with `launchctl setenv` and `kickstart -k`, then unsets and restarts. A plist `EnvironmentVariables` entry overrides `setenv`. `freshness` writes a `fact` with `topic_key` `eval:freshness:` and deletes it; probes on one host take turns on `~/.sessionflow/eval-freshness.lock`. `SESSIONFLOW_SEARCH_LOG=1` (off by default) logs searches and follow-ups to `~/.sessionflow/search-log.jsonl` (`search_log.py`). `install-eval-agent` needs `SESSIONFLOW_EVAL_QUERIES` and `SESSIONFLOW_EVAL_OUT`. - **Schema drift (SESF-11)** — `_ensure_collection` refuses to start when the live Milvus schema differs from `_expected_schema_fields()`. Drop+recreate is opt-in via `SESSIONFLOW_AUTO_MIGRATE_SCHEMA=1`, or run `python cleanup.py migrate-schema` explicitly (destructive). - **`SESSIONFLOW_BACKFILL_MAX_TURNS_PER_RUN` (SESF-12)** — default raised from 200 to 100,000 so a long-running server isn't silently capped after a few minutes. The budget logs a single WARN the first time the cap is hit so operators don't have to grep for "0 turns" lines. - **FTS5 thread affinity (SESF-13)** — `FTSIndex` keeps per-thread persistent connections (`threading.local`). Server-mode connections opened on the embed executor and request threads are isolated, and cross-thread `close_all()` is a no-op rather than a noisy WARN. diff --git a/README.md b/README.md index 51dcf1c..dae711e 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,7 @@ Restart Claude Code after setup to load the MCP server. | `SESSIONFLOW_EXPIRE_DAYS` | `365` | Prune turns older than this many days. | | `SESSIONFLOW_WATCH` | `true` | Turn the transcript file watcher on or off. | | `SESSIONFLOW_URL` | `http://127.0.0.1:7102` | Server URL that the hooks call. | +| `SESSIONFLOW_SEARCH_LOG` | off | Set to `1` to log each search and follow-up call to `~/.sessionflow/search-log.jsonl`. Queries are redacted. | ## Running @@ -93,6 +94,19 @@ Several harnesses can start at the same time, and their `SessionStart` hooks the The `start`, `stop`, `restart` and `status` commands work the same with or without the LaunchAgent. `setup.sh` offers to install it. Set `SESSIONFLOW_INSTALL_AGENT=1` to install it without a prompt. +### Retrieval eval + +`eval_retrieval.py` scores a labelled query set against the running server. Each JSONL line names an `id`, a `group`, a `tool`, a `query`, optional `args` and the `expected` IDs (`memory_id` or `session_id:turn_index`). `tests/eval/fixture_queries.jsonl` shows the format. + +```bash +./venv/bin/python eval_retrieval.py run --queries queries.jsonl --sort-by hybrid --sort-by relevance --out reports/ +./venv/bin/python eval_retrieval.py sweep --queries queries.jsonl --var SESSIONFLOW_RECENCY_WEIGHT --values 0,0.15,0.3,0.5 --out reports/ --yes +./venv/bin/python eval_retrieval.py freshness # Time a canary memory until search finds it +./venv/bin/python eval_retrieval.py usage --days 7 # Roll up the search usage log +``` + +`sweep` restarts the LaunchAgent once per value. `SESSIONFLOW_EVAL_QUERIES= SESSIONFLOW_EVAL_OUT= ./sessionflow-server.sh install-eval-agent` runs `nightly` every day at 03:30. + ## Memories SessionFlow also stores memories: notes an agent writes on purpose, such as session summaries, decisions and handoffs. Postgres holds them verbatim. The Milvus `memories` collection and a local `memories_fts.db` file are search indexes that SessionFlow rebuilds from Postgres. Memory is optional. Without `SESSIONFLOW_MEMORY_DSN`, the server serves transcripts only and `/health` reports memory as not configured. diff --git a/eval_retrieval.py b/eval_retrieval.py new file mode 100644 index 0000000..cf9b7ab --- /dev/null +++ b/eval_retrieval.py @@ -0,0 +1,877 @@ +"""Retrieval eval for a running SessionFlow server. + +Runs a labelled JSONL query set through ``search_all_sessions`` or ``search_memories`` +over the local MCP endpoint and scores recall@5, MRR and nDCG@10 per query group. +Also measures memory index freshness with a canary memory and rolls up the opt-in +search usage log (``search_log``). Standard library only; it never embeds. + +Subcommands: ``run``, ``sweep``, ``freshness``, ``usage``, ``nightly``. +""" + +from __future__ import annotations + +import argparse +import fcntl +import json +import math +import os +import secrets +import socket +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Callable, Iterable, Optional + +DEFAULT_URL = os.environ.get("SESSIONFLOW_URL", "http://127.0.0.1:7102/mcp/") +PROTOCOL_VERSION = "2025-06-18" +LAUNCH_AGENT_LABEL = "cc.lbruton.sessionflow" + +GROUPS = ("identifier", "paraphrase", "buried-fact", "temporal", "cross-project", + "no-answer", "superseded") +TOOLS = ("search_all_sessions", "search_memories") +SORT_BY = ("hybrid", "relevance", "recency") +FETCH_N = 10 +CANARY_TOPIC = "eval:freshness" +CANARY_PROJECT = "sessionflow" +REFORMULATION_WINDOW_S = 60 +# MCP clientInfo name of this harness; the usage rollup skips its calls. +EVAL_CLIENT_NAME = "eval-retrieval" + +# Ranking env vars the server reports under /health "ranking", and their keys there. +RANKING_ENV = { + "SESSIONFLOW_RECENCY_WEIGHT": ("recency_weight", float), + "SESSIONFLOW_RECENCY_DECAY_DAYS": ("recency_decay_days", int), +} +METRICS = (("recall_at_5", "recall@5"), ("mrr", "MRR"), ("ndcg_at_10", "nDCG@10")) + + +class EvalError(Exception): + """A server call, query-set line or setting the eval cannot use.""" + + +# --- Query set ----------------------------------------------------------------- + + +def load_queries(path: Path) -> list[dict]: + """Read and check a JSONL query set; blank lines are skipped. + + Raises: + ValueError: a line is not JSON or breaks the schema; the message names the line. + """ + queries, seen = [], set() + for number, raw in enumerate(Path(path).read_text(encoding="utf-8").splitlines(), 1): + if not raw.strip(): + continue + try: + q = json.loads(raw) + except ValueError: + raise ValueError(f"line {number}: not JSON") from None + problem = _query_problem(q) + if problem is None and q["id"] in seen: + problem = f"duplicate id {q['id']!r}" + if problem: + raise ValueError(f"line {number}: {problem}") + seen.add(q["id"]) + queries.append(q) + return queries + + +def _query_problem(q) -> Optional[str]: + if not isinstance(q, dict): + return "not an object" + for key in ("id", "query"): + if not isinstance(q.get(key), str) or not q[key].strip(): + return f"{key} must be a non-empty string" + if q.get("group") not in GROUPS: + return f"group must be one of {', '.join(GROUPS)}" + if q.get("tool") not in TOOLS: + return f"tool must be one of {', '.join(TOOLS)}" + if not isinstance(q.get("args", {}), dict): + return "args must be an object" + expected = q.get("expected") + if not isinstance(expected, list) or not all(isinstance(e, str) for e in expected): + return "expected must be a list of IDs" + if (q["group"] == "no-answer") != (not expected): + return "expected is empty for no-answer queries and only for them" + if q["group"] == "superseded" and not q.get("superseded"): + return "superseded queries list the IDs they overturn in superseded" + return None + + +# --- Scoring ------------------------------------------------------------------- + + +def result_ids(tool: str, payload: dict) -> list[str]: + """Return result IDs in rank order: memory IDs, or ``session_id:turn_index``.""" + rows = payload.get("results") or [] + if tool == "search_memories": + return [str(r.get("memory_id")) for r in rows] + return [f"{r.get('session_id')}:{r.get('turn_index')}" for r in rows] + + +def top_score(tool: str, payload: dict) -> Optional[float]: + """Return the top transcript hit's cosine similarity; memories carry no score.""" + rows = payload.get("results") or [] + if tool != "search_all_sessions" or not rows or rows[0].get("distance") is None: + return None + return 1.0 - float(rows[0]["distance"]) + + +def score_query(q: dict, ranked: list[str], top_score: Optional[float] = None) -> dict: + """Score one query's ranked IDs against its expected IDs. + + A no-answer query has no recall, MRR or nDCG; its top hit score shows how + clearly the server could report no match. + """ + expected = list(dict.fromkeys(q["expected"])) + wanted = set(expected) + rank = next((i for i, rid in enumerate(ranked, 1) if rid in wanted), None) + scored = { + "id": q["id"], "group": q["group"], "tool": q["tool"], "query": q["query"], + "expected": expected, "ranked": ranked[:FETCH_N], "rank": rank, + "recall_at_5": None, "rr": None, "ndcg_at_10": None, + "top_score": top_score, "current_first": None, + } + if q.get("superseded"): + scored["superseded"] = list(q["superseded"]) + if expected: + scored["recall_at_5"] = len(wanted & set(ranked[:5])) / len(wanted) + scored["rr"] = 1.0 / rank if rank else 0.0 + scored["ndcg_at_10"] = _ndcg(ranked[:10], wanted) + if q.get("superseded"): + old = set(q["superseded"]) + old_rank = next((i for i, rid in enumerate(ranked, 1) if rid in old), None) + scored["current_first"] = rank is not None and (old_rank is None or rank < old_rank) + return scored + + +def _ndcg(ranked: list[str], wanted: set) -> float: + seen, dcg = set(), 0.0 + for i, rid in enumerate(ranked): + if rid in wanted and rid not in seen: + seen.add(rid) + dcg += 1.0 / math.log2(i + 2) + ideal = sum(1.0 / math.log2(i + 2) for i in range(min(len(wanted), 10))) + return dcg / ideal if ideal else 0.0 + + +def _mean(values: Iterable) -> Optional[float]: + values = [float(v) for v in values if v is not None] + return sum(values) / len(values) if values else None + + +def _group_summary(rows: list[dict]) -> dict: + return { + "n": len(rows), + "recall_at_5": _mean(r["recall_at_5"] for r in rows), + "mrr": _mean(r["rr"] for r in rows), + "ndcg_at_10": _mean(r["ndcg_at_10"] for r in rows), + "current_first": _mean(r.get("current_first") for r in rows), + "top_score": _mean(r.get("top_score") for r in rows), + "errors": sum(1 for r in rows if r.get("error")), + } + + +def summarize(scored: list[dict]) -> dict: + """Return per-group means, plus ``all``; a metric with no values is None.""" + out = {} + for group in GROUPS: + rows = [r for r in scored if r["group"] == group] + if rows: + out[group] = _group_summary(rows) + out["all"] = _group_summary(scored) + return out + + +def build_report(scored: list[dict], run: dict) -> dict: + """Return the result document: run settings, group summaries and every query.""" + return {"run": {**run, "query_count": len(scored)}, "groups": summarize(scored), + "queries": scored} + + +# --- Reports ------------------------------------------------------------------- + + +def _fmt(value) -> str: + return "-" if value is None else f"{value:.3f}" + + +def render_markdown(report: dict, title: str = "Retrieval eval") -> str: + """Return a Markdown summary table of one report.""" + run = report["run"] + lines = [f"## {title}", ""] + for key in ("started_at", "server_version", "host", "sort_by", "queries_file", "query_count"): + if run.get(key) is not None: + lines.append(f"- {key}: {run[key]}") + if run.get("env"): + lines.append("- env: " + ", ".join(f"{k}={v}" for k, v in run["env"].items())) + if run.get("ranking"): + lines.append("- ranking: " + ", ".join(f"{k}={v}" for k, v in run["ranking"].items())) + lines += ["", "| group | n | recall@5 | MRR | nDCG@10 | current first | top score | errors |", + "| --- | --- | --- | --- | --- | --- | --- | --- |"] + for group, s in report["groups"].items(): + lines.append(f"| {group} | {s['n']} | {_fmt(s['recall_at_5'])} | {_fmt(s['mrr'])} | " + f"{_fmt(s['ndcg_at_10'])} | {_fmt(s['current_first'])} | " + f"{_fmt(s['top_score'])} | {s['errors']} |") + misses = [q for q in report["queries"] if q["expected"] and not q["rank"]] + if misses: + lines += ["", "Misses: " + ", ".join(q["id"] for q in misses)] + return "\n".join(lines) + "\n" + + +def render_comparison(reports: dict, metric: str = "mrr") -> str: + """Return one Markdown table of ``metric`` per group, one column per labelled report.""" + labels = list(reports) + groups = [g for g in (*GROUPS, "all") if any(g in r["groups"] for r in reports.values())] + lines = ["| group | " + " | ".join(labels) + " |", + "| --- |" + " --- |" * len(labels)] + for group in groups: + cells = [_fmt(reports[label]["groups"].get(group, {}).get(metric)) for label in labels] + lines.append(f"| {group} | " + " | ".join(cells) + " |") + return "\n".join(lines) + "\n" + + +def write_report(report: dict, out_dir: Path, stem: str) -> tuple[Path, Path]: + """Write ``.json`` and ``.md`` into ``out_dir`` and return both paths.""" + out_dir = Path(out_dir) + out_dir.mkdir(parents=True, exist_ok=True) + json_path, md_path = out_dir / f"{stem}.json", out_dir / f"{stem}.md" + json_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + md_path.write_text(render_markdown(report, title=stem), encoding="utf-8") + return json_path, md_path + + +# --- Ranking settings ---------------------------------------------------------- + + +def parse_env(pairs: Iterable[str]) -> dict: + """Turn repeated ``KEY=VALUE`` arguments into a dict. + + Raises: + ValueError: an argument has no ``=``. + """ + env = {} + for pair in pairs or []: + key, sep, value = pair.partition("=") + if not sep or not key: + raise ValueError(f"expected KEY=VALUE, got {pair!r}") + env[key] = value + return env + + +def check_ranking(env: dict, health: dict) -> tuple[list[str], list[str]]: + """Compare the expected env with the server's /health ``ranking``. + + Returns: + ``(problems, unverified)``: mismatches that make the run invalid, and env + names the server does not report, which the run records unchecked. + """ + problems, unverified = [], [] + ranking = health.get("ranking") + for name, value in env.items(): + if name not in RANKING_ENV: + unverified.append(name) + continue + if not isinstance(ranking, dict): + problems.append("the server reports no ranking settings in /health; restart it on this version") + continue + key, cast = RANKING_ENV[name] + try: + want = cast(value) + except ValueError: + problems.append(f"{name}={value} is not a valid {cast.__name__}") + continue + have = ranking.get(key) + if have is None or not math.isclose(float(have), float(want), abs_tol=1e-9): + problems.append(f"{name}={value} expected, but the server applies {key}={have}") + return problems, unverified + + +# --- MCP client ---------------------------------------------------------------- + + +class McpClient: + """Stateful MCP client for the SessionFlow HTTP endpoint (initialize, call, delete).""" + + def __init__(self, url: str = DEFAULT_URL, timeout: float = 60.0): + """Store the endpoint; a base URL such as ``http://127.0.0.1:7102`` gets ``/mcp/``.""" + parts = urllib.parse.urlsplit(url) + if not parts.path.rstrip("/"): + url = urllib.parse.urlunsplit(parts._replace(path="/mcp/")) + self.url = url + self.timeout = timeout + self.session_id: Optional[str] = None + self._next_id = 0 + + def _headers(self) -> dict: + headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"} + if self.session_id: + headers["Mcp-Session-Id"] = self.session_id + headers["MCP-Protocol-Version"] = PROTOCOL_VERSION + return headers + + def _post(self, body: dict) -> dict: + req = urllib.request.Request(self.url, data=json.dumps(body).encode(), method="POST", + headers=self._headers()) + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + raw = resp.read().decode("utf-8", "replace") + ctype = resp.headers.get("Content-Type", "") + if resp.headers.get("Mcp-Session-Id"): + self.session_id = resp.headers["Mcp-Session-Id"] + if not raw: + return {} + if ctype.startswith("text/event-stream"): + events = [line[5:].strip() for line in raw.splitlines() if line.startswith("data:")] + raw = next((e for e in reversed(events) if e), "") + try: + return json.loads(raw) + except ValueError: + raise EvalError("the server reply is not JSON") from None + + def _request(self, method: str, params: dict) -> dict: + self._next_id += 1 + return self._post({"jsonrpc": "2.0", "id": self._next_id, "method": method, "params": params}) + + def _open(self) -> None: + self.session_id = None + self._request("initialize", {"protocolVersion": PROTOCOL_VERSION, "capabilities": {}, + "clientInfo": {"name": EVAL_CLIENT_NAME, "version": "1"}}) + self._post({"jsonrpc": "2.0", "method": "notifications/initialized"}) + + def call(self, name: str, arguments: dict) -> dict: + """Call one tool with ``format: json`` and return its parsed JSON payload. + + A 404 means the server dropped the session (restart or idle expiry); the + client opens a new one and retries once. + + Raises: + EvalError: the server is unreachable or the tool returned an error. + """ + params = {"name": name, "arguments": {**arguments, "format": "json"}} + for attempt in (1, 2): + try: + if not self.session_id: + self._open() + reply = self._request("tools/call", params) + break + except urllib.error.HTTPError as exc: + if exc.code == 404 and attempt == 1: + self.session_id = None + continue + raise EvalError(f"{name}: HTTP {exc.code}") from None + except (urllib.error.URLError, OSError) as exc: + raise EvalError(f"{name}: server unreachable ({getattr(exc, 'reason', exc)})") from None + if "error" in reply: + raise EvalError(f"{name}: JSON-RPC error {reply['error'].get('message')}") + result = reply.get("result") or {} + content = result.get("content") or [] + text = content[0].get("text", "") if content else "" + if result.get("isError"): + raise EvalError(f"{name}: {text}") + try: + return json.loads(text) + except ValueError: + raise EvalError(f"{name}: {text.strip()[:200]}") from None + + def health(self) -> dict: + """Return the /health JSON, also while the server answers 503 during startup.""" + url = self.url.split("/mcp", 1)[0] + "/health" + try: + with urllib.request.urlopen(url, timeout=self.timeout) as resp: + return json.loads(resp.read()) + except urllib.error.HTTPError as exc: + try: + return json.loads(exc.read()) + except ValueError: + raise EvalError(f"/health: HTTP {exc.code}") from None + except (urllib.error.URLError, OSError, ValueError) as exc: + raise EvalError(f"/health: {getattr(exc, 'reason', exc)}") from None + + def close(self) -> None: + """Ask the server to drop the session; it expires idle sessions anyway.""" + if not self.session_id: + return + req = urllib.request.Request(self.url, method="DELETE", headers=self._headers()) + try: + with urllib.request.urlopen(req, timeout=5): + pass + except (urllib.error.URLError, OSError): + pass + self.session_id = None + + +def _host(health: dict) -> str: + return ((health.get("memory") or {}).get("index") or {}).get("host") or socket.gethostname() + + +# --- Query runs ---------------------------------------------------------------- + + +def run_queries(client, queries: list[dict], sort_by: str) -> list[dict]: + """Run every query and score it; a failed call scores as a miss and keeps its error.""" + scored = [] + for q in queries: + args = {**q.get("args", {}), "query": q["query"], "n": FETCH_N, "sort_by": sort_by} + try: + payload = client.call(q["tool"], args) + row = score_query(q, result_ids(q["tool"], payload), top_score(q["tool"], payload)) + except EvalError as exc: + row = score_query(q, []) + row["error"] = str(exc) + scored.append(row) + return scored + + +def evaluate(client, queries: list[dict], *, sort_by: str, env: dict, queries_file: str) -> dict: + """Check the server's ranking against ``env``, run the set and return the report. + + Raises: + EvalError: the server applies different ranking settings than ``env`` names. + """ + health = client.health() + problems, unverified = check_ranking(env, health) + if problems: + raise EvalError("; ".join(problems)) + started = datetime.now(timezone.utc).isoformat() + scored = run_queries(client, queries, sort_by) + run = {"started_at": started, "finished_at": datetime.now(timezone.utc).isoformat(), + "server_version": health.get("version"), "host": _host(health), "sort_by": sort_by, + "env": env, "env_unverified": unverified, "ranking": health.get("ranking"), + "queries_file": queries_file} + return build_report(scored, run) + + +# --- Freshness probe ----------------------------------------------------------- + + +def make_token() -> str: + """Return a unique canary token: letters and digits, one FTS term, no secret-like entropy.""" + return f"sfcanary{time.time_ns()}{secrets.randbelow(10_000):04d}" + + +def canary_text(token: str) -> str: + """Return the canary memory text that carries ``token``.""" + return f"Freshness canary {token}, written by eval_retrieval.py and deleted after the probe." + + +def default_lock_path() -> Path: + """Return the file lock that serializes freshness probes on this host.""" + return Path.home() / ".sessionflow" / "eval-freshness.lock" + + +def freshness_probe(client, *, project: str = CANARY_PROJECT, agent_id: str = "claude-code", + poll_s: float = 1.0, timeout_s: float = 120.0, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.monotonic, + lock_path: Optional[Path] = None) -> dict: + """Write a canary memory, poll ``search_memories`` until it appears, then delete it. + + The ``topic_key`` names the host, so a later probe there replaces a canary a + failed delete left behind. ``lock_path`` makes probes on one host wait for each + other, and a hit counts only when the record still carries this probe's token. + """ + if lock_path is None: + return _probe(client, project, agent_id, poll_s, timeout_s, sleep, clock) + lock_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + with open(lock_path, "a") as handle: + fcntl.flock(handle, fcntl.LOCK_EX) + try: + return _probe(client, project, agent_id, poll_s, timeout_s, sleep, clock) + finally: + fcntl.flock(handle, fcntl.LOCK_UN) + + +def _delete_exact_canary(client, project: str, token: str) -> bool: + """Delete memories whose text equals this token's canary text; True when one was deleted.""" + text = canary_text(token) + try: + found = client.call("search_memories", {"query": token, "project": project, "n": 5}) + except EvalError: + return False + deleted = False + for row in found.get("results") or []: + if row.get("text") == text: + try: + client.call("delete_memory", {"memory_id": row["memory_id"], "project": project}) + deleted = True + except EvalError: + pass + return deleted + + +def _probe(client, project, agent_id, poll_s, timeout_s, sleep, clock) -> dict: + token = make_token() + result = {"host": None, "memory_id": None, "add_outcome": None, "add_index": None, + "time_to_hit_s": None, "polls": 0, "channels": None, "health": None, + "deleted": False} + start = clock() + try: + added = client.call("add_memory", { + "text": canary_text(token), "agent_id": agent_id, "project": project, "type": "fact", + "topic_key": f"{CANARY_TOPIC}:{socket.gethostname()}", "extra": {"source": "eval"}}) + except EvalError as exc: + # The add may have committed before the error; delete only an exact match. + result["error"] = str(exc) + result["deleted"] = _delete_exact_canary(client, project, token) + return result + result.update(memory_id=added.get("memory_id"), add_outcome=added.get("outcome"), + add_index=added.get("index")) + try: + try: + health = client.health() + memory = health.get("memory") or {} + index = memory.get("index") or {} + result["host"] = _host(health) + result["health"] = {"ack_lag": memory.get("ack_lag"), + "last_tick_age_s": index.get("last_tick_age_s"), + "stale": index.get("stale")} + except EvalError as exc: + result["host"] = socket.gethostname() + result["health_error"] = str(exc) + while True: + result["polls"] += 1 + try: + found = client.call("search_memories", {"query": token, "project": project, "n": 5}) + except EvalError as exc: + found, result["search_error"] = {}, str(exc) + if any(r.get("memory_id") == result["memory_id"] and token in (r.get("text") or "") + for r in found.get("results") or []): + result["time_to_hit_s"] = clock() - start + result["channels"] = found.get("channels") + break + if clock() - start >= timeout_s: + break + sleep(poll_s) + finally: + try: + client.call("delete_memory", {"memory_id": result["memory_id"], "project": project}) + result["deleted"] = True + except EvalError as exc: + result["delete_error"] = str(exc) + return result + + +def render_freshness(result: dict) -> str: + """Return a Markdown summary of one freshness probe.""" + hit = result.get("time_to_hit_s") + lines = ["## Freshness", "", + f"- host: {result.get('host')}", + f"- time to hit: {'timed out' if hit is None else f'{hit:.1f} s'} " + f"after {result.get('polls')} polls", + f"- add index outcome: {result.get('add_index')}", + f"- channels at hit: {', '.join(result.get('channels') or []) or '-'}", + f"- health: {json.dumps(result.get('health'))}", + f"- canary deleted: {result.get('deleted')}"] + for key in ("error", "health_error", "search_error", "delete_error"): + if result.get(key): + lines.append(f"- {key}: {result[key]}") + return "\n".join(lines) + "\n" + + +# --- Usage rollup -------------------------------------------------------------- + + +def read_usage_log(path: Path) -> list[dict]: + """Return the usage log entries; unreadable lines are skipped.""" + entries = [] + try: + text = Path(path).read_text(encoding="utf-8") + except OSError: + return entries + for line in text.splitlines(): + try: + entry = json.loads(line) + except ValueError: + continue + if isinstance(entry, dict) and "ts" in entry: + entries.append(entry) + return entries + + +def usage_rollup(entries: list[dict], *, now: datetime, days: int = 7) -> dict: + """Roll up the usage log over the last ``days``. + + A follow-up counts as top 3 when it opens one of the first three IDs of the + same client's latest search. A search counts as reformulated when the same + client searches again within 60 s with no follow-up in between. Calls from + this harness (client ``eval-retrieval:...``) are skipped and counted. + """ + cutoff = now - timedelta(days=days) + timed, skipped = [], 0 + for e in entries: + if str(e.get("client") or "").startswith(f"{EVAL_CLIENT_NAME}:"): + skipped += 1 + continue + try: + ts = datetime.fromisoformat(e["ts"]) + except (TypeError, ValueError): + continue + if ts >= cutoff: + timed.append((ts, e)) + timed.sort(key=lambda pair: pair[0]) + latest: dict = {} + searches = follow_ups = top3 = reformulated = 0 + for ts, e in timed: + client = e.get("client") + prev = latest.get(client) + if e.get("kind") == "search": + searches += 1 + if prev and not prev["followed"] and \ + (ts - prev["ts"]).total_seconds() <= REFORMULATION_WINDOW_S: + reformulated += 1 + latest[client] = {"ts": ts, "ids": list(e.get("ids") or []), "followed": False} + elif e.get("kind") == "follow_up": + follow_ups += 1 + if prev: + prev["followed"] = True + if e.get("id") in prev["ids"][:3]: + top3 += 1 + return { + "days": days, "searches": searches, "follow_ups": follow_ups, + "follow_up_top3": top3, "reformulated_searches": reformulated, + "follow_up_top3_share": top3 / follow_ups if follow_ups else None, + "reformulation_share": reformulated / searches if searches else None, + "eval_calls_skipped": skipped, + } + + +# --- Server restarts for sweeps ------------------------------------------------ + + +def _launchctl(*args: str) -> str: + done = subprocess.run(["launchctl", *args], check=True, capture_output=True, text=True) + return done.stdout + + +def _launchctl_getenv(name: str) -> Optional[str]: + """Return the launchd value of ``name``, or None when it is unset.""" + value = _launchctl("getenv", name).strip() + return value or None + + +def restore_server(var: str, prior: Optional[str], label: str, url: str) -> list[str]: + """Put ``var`` back to its value before the sweep, restart and wait; return the failures. + + Each step runs even when an earlier one fails, so a failed ``setenv`` still + leaves a restarted server. + """ + errors = [] + try: + if prior is None: + _launchctl("unsetenv", var) + else: + _launchctl("setenv", var, prior) + except (subprocess.CalledProcessError, OSError) as exc: + errors.append(f"launchctl env for {var}: {exc}") + try: + _restart_server(label) + except (subprocess.CalledProcessError, OSError) as exc: + errors.append(f"restart {label}: {exc}") + expected = {var: prior} if prior is not None and var in RANKING_ENV else {} + try: + wait_ready(McpClient(url), expected) + except EvalError as exc: + errors.append(str(exc)) + return errors + + +def _restart_server(label: str) -> None: + _launchctl("kickstart", "-k", f"gui/{os.getuid()}/{label}") + + +def wait_ready(client, env: dict, timeout_s: float = 300.0) -> dict: + """Wait until /health is ok and applies ``env``; return the health JSON. + + Raises: + EvalError: the server is not ready with ``env`` within ``timeout_s``. + """ + deadline = time.monotonic() + timeout_s + last = "no answer" + while time.monotonic() < deadline: + try: + health = client.health() + problems, _ = check_ranking(env, health) + if health.get("status") == "ok" and not problems: + return health + last = "; ".join(problems) or f"status {health.get('status')}" + except EvalError as exc: + last = str(exc) + time.sleep(2) + raise EvalError(f"the server was not ready within {timeout_s:.0f} s: {last}") + + +# --- CLI ----------------------------------------------------------------------- + + +def _local_date() -> str: + return datetime.now().astimezone().date().isoformat() + + +def _cmd_run(args) -> int: + queries = load_queries(args.queries) + env = parse_env(args.env) + client = McpClient(args.url) + try: + for sort_by in args.sort_by: + report = evaluate(client, queries, sort_by=sort_by, env=env, queries_file=str(args.queries)) + stem = f"{_local_date()}-{sort_by}" + "".join(f"-{k}={v}" for k, v in env.items()) + if args.out: + json_path, _ = write_report(report, args.out, stem) + print(f"Wrote {json_path}", file=sys.stderr) + print(render_markdown(report, title=stem)) + finally: + client.close() + return 0 + + +def _cmd_sweep(args) -> int: + if not args.yes: + print(f"sweep restarts the {args.label} LaunchAgent once per value and once to restore it. " + "Pass --yes to run it.", file=sys.stderr) + return 2 + queries = load_queries(args.queries) + values = [v.strip() for v in args.values.split(",") if v.strip()] + prior = _launchctl_getenv(args.var) + reports = {} + client = McpClient(args.url) + cleanup_errors: list[str] = [] + try: + for value in values: + env = {args.var: value} + _launchctl("setenv", args.var, value) + _restart_server(args.label) + wait_ready(client, env) + client.session_id = None + report = evaluate(client, queries, sort_by=args.sort_by, env=env, + queries_file=str(args.queries)) + label = f"{args.var}={value}" + reports[label] = report + write_report(report, args.out, f"{_local_date()}-sweep-{args.sort_by}-{label}") + finally: + client.close() + cleanup_errors = restore_server(args.var, prior, args.label, args.url) + for error in cleanup_errors: + print(f"eval_retrieval: restore failed: {error}", file=sys.stderr) + parts = [f"## Sweep of {args.var} ({args.sort_by})", ""] + for metric, name in METRICS: + parts += [f"### {name}", "", render_comparison(reports, metric)] + text = "\n".join(parts) + out = Path(args.out) / f"{_local_date()}-sweep-{args.sort_by}-{args.var}.md" + out.write_text(text, encoding="utf-8") + print(text) + print(f"Wrote {out}", file=sys.stderr) + return 1 if cleanup_errors else 0 + + +def _cmd_freshness(args) -> int: + client = McpClient(args.url) + try: + result = freshness_probe(client, timeout_s=args.timeout, lock_path=default_lock_path()) + finally: + client.close() + print(render_freshness(result)) + if args.out: + Path(args.out).mkdir(parents=True, exist_ok=True) + path = Path(args.out) / f"{_local_date()}-freshness.json" + path.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8") + print(f"Wrote {path}", file=sys.stderr) + return 0 if result.get("deleted") and result.get("time_to_hit_s") is not None else 1 + + +def _cmd_usage(args) -> int: + import search_log + + path = args.log or search_log.log_path() + roll = usage_rollup(read_usage_log(path), now=datetime.now(timezone.utc), days=args.days) + print(json.dumps(roll, indent=2)) + return 0 + + +def _cmd_nightly(args) -> int: + queries = load_queries(args.queries) + client = McpClient(args.url) + reports, status = {}, 0 + try: + for sort_by in ("hybrid", "relevance"): + reports[sort_by] = evaluate(client, queries, sort_by=sort_by, env={}, + queries_file=str(args.queries)) + freshness = freshness_probe(client, lock_path=default_lock_path()) + finally: + client.close() + if not freshness.get("deleted") or freshness.get("time_to_hit_s") is None: + status = 1 + stem = f"{_local_date()}-nightly" + out = Path(args.out) + out.mkdir(parents=True, exist_ok=True) + doc = {"date": _local_date(), "reports": reports, "freshness": freshness} + (out / f"{stem}.json").write_text(json.dumps(doc, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8") + md = [f"# Nightly retrieval eval {_local_date()}", ""] + md += [render_markdown(r, title=f"Queries, {sort_by}") for sort_by, r in reports.items()] + md.append(render_freshness(freshness)) + (out / f"{stem}.md").write_text("\n".join(md), encoding="utf-8") + print(f"Wrote {out / stem}.json and .md", file=sys.stderr) + return status + + +def build_parser() -> argparse.ArgumentParser: + """Return the CLI argument parser.""" + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--url", default=DEFAULT_URL, help="MCP endpoint (default: %(default)s)") + sub = parser.add_subparsers(dest="command", required=True) + + run = sub.add_parser("run", help="Score a query set against the running server.") + run.add_argument("--queries", type=Path, required=True) + run.add_argument("--sort-by", choices=SORT_BY, action="append", + help="Ranking to score; repeat for several (default: hybrid).") + run.add_argument("--env", action="append", default=[], metavar="KEY=VALUE", + help="Ranking env the server must apply; the run refuses on a mismatch.") + run.add_argument("--out", type=Path, help="Write -.json and .md here.") + run.set_defaults(func=_cmd_run) + + sweep = sub.add_parser("sweep", help="Restart the server once per env value and compare.") + sweep.add_argument("--queries", type=Path, required=True) + sweep.add_argument("--var", required=True, help="Env var to sweep, for example SESSIONFLOW_RECENCY_WEIGHT.") + sweep.add_argument("--values", required=True, help="Comma-separated values, for example 0,0.15,0.3,0.5.") + sweep.add_argument("--sort-by", choices=SORT_BY, default="hybrid") + sweep.add_argument("--out", type=Path, required=True) + sweep.add_argument("--label", default=LAUNCH_AGENT_LABEL, help="Server LaunchAgent label.") + sweep.add_argument("--yes", action="store_true", help="Allow the server restarts.") + sweep.set_defaults(func=_cmd_sweep) + + fresh = sub.add_parser("freshness", help="Time a canary memory until search finds it.") + fresh.add_argument("--timeout", type=float, default=120.0) + fresh.add_argument("--out", type=Path) + fresh.set_defaults(func=_cmd_freshness) + + usage = sub.add_parser("usage", help="Roll up the search usage log.") + usage.add_argument("--days", type=int, default=7) + usage.add_argument("--log", type=Path, help="Log file (default: ~/.sessionflow/search-log.jsonl).") + usage.set_defaults(func=_cmd_usage) + + nightly = sub.add_parser("nightly", help="Run hybrid, relevance and freshness into one dated report.") + nightly.add_argument("--queries", type=Path, required=True) + nightly.add_argument("--out", type=Path, required=True) + nightly.set_defaults(func=_cmd_nightly) + return parser + + +def main(argv: Optional[list[str]] = None) -> int: + """Run the CLI and return its exit code.""" + args = build_parser().parse_args(argv) + if getattr(args, "sort_by", None) is None and args.command == "run": + args.sort_by = ["hybrid"] + try: + return args.func(args) + except (EvalError, ValueError, subprocess.CalledProcessError) as exc: + print(f"eval_retrieval: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/http_server.py b/http_server.py index 672ec9a..7f8aeb3 100644 --- a/http_server.py +++ b/http_server.py @@ -971,6 +971,7 @@ async def health(request: Request) -> JSONResponse: "providers": _provider_health(deep=deep), "backfill": _backfill_status_payload(), "embedding": _embedding_status_payload(), + "ranking": rag_engine.ranking_settings(), "fts": fts_payload, }, status_code=503 if starting else 200) diff --git a/rag_engine.py b/rag_engine.py index 1e2f2f6..33e7e0f 100644 --- a/rag_engine.py +++ b/rag_engine.py @@ -1877,6 +1877,22 @@ def _semantic_scores(results: List[Dict]) -> List[float]: return [(s - lo) / span for s in raw] +def ranking_settings() -> Dict[str, Union[float, int]]: + """Return the hybrid ranking settings this process applies, read from the environment. + + ``SESSIONFLOW_RECENCY_WEIGHT`` outside ``[0, 1]`` and an unparsable value fall + back to the default; ``SESSIONFLOW_RECENCY_DECAY_DAYS`` is at least 1. + """ + return { + "recency_weight": _env_float( + "SESSIONFLOW_RECENCY_WEIGHT", RECENCY_WEIGHT_DEFAULT, 0.0, 1.0 + ), + "recency_decay_days": _env_int( + "SESSIONFLOW_RECENCY_DECAY_DAYS", RECENCY_DECAY_DAYS_DEFAULT, minimum=1 + ), + } + + def _rank_results(results: List[Dict], sort_by: str, n: int, now: Optional[datetime] = None, *, timestamp_key: str = "timestamp", @@ -1910,12 +1926,9 @@ def _with_tie(primary): reverse=True, ) else: # hybrid - weight = _env_float( - "SESSIONFLOW_RECENCY_WEIGHT", RECENCY_WEIGHT_DEFAULT, 0.0, 1.0 - ) - decay_days = _env_int( - "SESSIONFLOW_RECENCY_DECAY_DAYS", RECENCY_DECAY_DAYS_DEFAULT, minimum=1 - ) + settings = ranking_settings() + weight = settings["recency_weight"] + decay_days = settings["recency_decay_days"] semantic = _semantic_scores(results) for r, sem in zip(results, semantic, strict=True): rec = _recency_score(r.get(timestamp_key, ""), now, decay_days) diff --git a/search_log.py b/search_log.py new file mode 100644 index 0000000..8fa5a96 --- /dev/null +++ b/search_log.py @@ -0,0 +1,83 @@ +"""Opt-in usage log of search and follow-up tool calls, for the retrieval eval. + +``SESSIONFLOW_SEARCH_LOG=1`` turns it on. Each call appends one JSON line to +``~/.sessionflow/search-log.jsonl``. Queries pass through the secret redactor in +enforce mode, and a write failure is logged and never fails the tool call. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable, Optional, Union + +import secret_redaction + +logger = logging.getLogger(__name__) + +_TRUE = {"1", "true", "yes", "on"} +_lock = threading.Lock() + + +def enabled() -> bool: + """Return whether ``SESSIONFLOW_SEARCH_LOG`` turns the log on.""" + return os.getenv("SESSIONFLOW_SEARCH_LOG", "").strip().lower() in _TRUE + + +def log_path() -> Path: + """Return the log file path under the current home folder.""" + return Path.home() / ".sessionflow" / "search-log.jsonl" + + +def _append(entry: dict) -> None: + entry = {"ts": datetime.now(timezone.utc).isoformat(), **entry} + path = log_path() + try: + path.parent.mkdir(mode=0o700, exist_ok=True) + with _lock: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + # O_CREAT's mode applies only to a new file; tighten one created looser. + os.fchmod(fd, 0o600) + with os.fdopen(fd, "a", encoding="utf-8") as handle: + handle.write(json.dumps(entry) + "\n") + except (OSError, ValueError) as exc: + logger.warning("Search log write failed: %s", exc) + + +def record_search(tool: str, *, client: Optional[str], project: Union[str, list[str], None], + sort_by: Optional[str], + query: str, ids: Iterable[str]) -> None: + """Append one search call with its result IDs in rank order. + + The query is redacted with the built-in rules only; the operator allowlist + is not applied, so an allowlisted path may be masked in the log. + """ + if not enabled(): + return + redacted, _ = secret_redaction.redact(query or "", mode="enforce") + _append({"kind": "search", "tool": tool, "client": client, "project": project, + "sort_by": sort_by, "query": redacted, "ids": list(ids)}) + + +def record_follow_up(tool: str, *, client: Optional[str], item_id: str) -> None: + """Append one ``get_memory`` or ``get_turns`` call with the ID it opened.""" + if not enabled(): + return + _append({"kind": "follow_up", "tool": tool, "client": client, "id": item_id}) + + +async def log_search(tool: str, **kwargs) -> None: + """Run :func:`record_search` on a worker thread, so redaction and disk I/O stay off the event loop.""" + if enabled(): + await asyncio.to_thread(record_search, tool, **kwargs) + + +async def log_follow_up(tool: str, **kwargs) -> None: + """Run :func:`record_follow_up` on a worker thread.""" + if enabled(): + await asyncio.to_thread(record_follow_up, tool, **kwargs) diff --git a/sessionflow-server.sh b/sessionflow-server.sh index d1dc2ec..f500526 100755 --- a/sessionflow-server.sh +++ b/sessionflow-server.sh @@ -55,6 +55,15 @@ BACKFILL_AGENT_STDOUT="$SERVER_DIR/backfill-agent.out.log" BACKFILL_AGENT_STDERR="$SERVER_DIR/backfill-agent.err.log" BACKFILL_INTERVAL_SECONDS="${SESSIONFLOW_BACKFILL_INTERVAL_SECONDS:-3600}" +# Nightly retrieval eval agent. The query set and reports live outside the +# public repo, so install-eval-agent requires both paths. +EVAL_AGENT_LABEL="cc.lbruton.sessionflow-eval" +EVAL_AGENT_PLIST="$LAUNCH_AGENT_DIR/${EVAL_AGENT_LABEL}.plist" +EVAL_AGENT_STDOUT="$SERVER_DIR/eval-agent.out.log" +EVAL_AGENT_STDERR="$SERVER_DIR/eval-agent.err.log" +EVAL_HOUR="${SESSIONFLOW_EVAL_HOUR:-3}" +EVAL_MINUTE="${SESSIONFLOW_EVAL_MINUTE:-30}" + # Claude Desktop's stdio bridge. Installed as a copy on the internal # disk because Desktop's MCP child cannot read the external volume (TCC). MCP_BRIDGE_SOURCE="$SCRIPT_DIR/mcp-remote-wait.sh" @@ -602,6 +611,139 @@ do_backfill_agent_status() { fi } +xml_escape() { + # sed, because bash 3.2 and 5.2 treat & in a ${s//x/y} replacement differently. + printf '%s' "$1" | sed -e 's/&/\&/g' -e 's//\>/g' +} + +write_eval_agent_plist() { + mkdir -p "$LAUNCH_AGENT_DIR" + local python script_dir queries out url + python=$(xml_escape "$PYTHON") + script_dir=$(xml_escape "$SCRIPT_DIR") + queries=$(xml_escape "$SESSIONFLOW_EVAL_QUERIES") + out=$(xml_escape "$SESSIONFLOW_EVAL_OUT") + url=$(xml_escape "${SESSIONFLOW_URL:-http://127.0.0.1:${SESSIONFLOW_PORT:-7102}}") + cat > "$EVAL_AGENT_PLIST" < + + + + Label + ${EVAL_AGENT_LABEL} + + ProgramArguments + + ${python} + ${script_dir}/eval_retrieval.py + --url + ${url} + nightly + --queries + ${queries} + --out + ${out} + + StartCalendarInterval + + Hour + ${EVAL_HOUR} + Minute + ${EVAL_MINUTE} + + RunAtLoad + + EnvironmentVariables + + HOME + ${HOME} + PATH + /opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin + + StandardOutPath + ${EVAL_AGENT_STDOUT} + StandardErrorPath + ${EVAL_AGENT_STDERR} + WorkingDirectory + ${script_dir} + + +PLIST +} + +do_install_eval_agent() { + if [ -z "${SESSIONFLOW_EVAL_QUERIES:-}" ] || [ -z "${SESSIONFLOW_EVAL_OUT:-}" ]; then + echo "[sessionflow] Set SESSIONFLOW_EVAL_QUERIES (query set JSONL) and SESSIONFLOW_EVAL_OUT (report folder)." >&2 + exit 1 + fi + # The agent runs from the repo folder, so relative paths would resolve there. + case "$SESSIONFLOW_EVAL_QUERIES" in /*) ;; *) + echo "[sessionflow] SESSIONFLOW_EVAL_QUERIES must be an absolute path." >&2; exit 1 ;; + esac + case "$SESSIONFLOW_EVAL_OUT" in /*) ;; *) + echo "[sessionflow] SESSIONFLOW_EVAL_OUT must be an absolute path." >&2; exit 1 ;; + esac + if [ ! -f "$SESSIONFLOW_EVAL_QUERIES" ]; then + echo "[sessionflow] No query set at $SESSIONFLOW_EVAL_QUERIES" >&2 + exit 1 + fi + if [ ! -x "$PYTHON" ]; then + echo "[sessionflow] No project venv interpreter at $PYTHON; run setup.sh first." >&2 + exit 1 + fi + local target + target=$(launch_agent_target) + if launchctl print "${target}/${EVAL_AGENT_LABEL}" >/dev/null 2>&1; then + launchctl bootout "$target" "$EVAL_AGENT_PLIST" 2>/dev/null || true + fi + write_eval_agent_plist + if ! plutil -lint "$EVAL_AGENT_PLIST" >/dev/null; then + echo "[sessionflow] The eval plist at $EVAL_AGENT_PLIST is not valid; not loading it." >&2 + exit 1 + fi + if launchctl bootstrap "$target" "$EVAL_AGENT_PLIST" 2>/dev/null; then + echo "[sessionflow] Eval LaunchAgent installed at $EVAL_AGENT_PLIST" >&2 + printf '[sessionflow] Runs daily at %02d:%02d\n' "$EVAL_HOUR" "$EVAL_MINUTE" >&2 + else + launchctl enable "${target}/${EVAL_AGENT_LABEL}" 2>/dev/null || true + echo "[sessionflow] Eval plist written at $EVAL_AGENT_PLIST" >&2 + echo "[sessionflow] launchctl bootstrap reported a soft failure; tried enable." >&2 + fi +} + +do_uninstall_eval_agent() { + local target + target=$(launch_agent_target) + if [ -f "$EVAL_AGENT_PLIST" ]; then + launchctl bootout "$target" "$EVAL_AGENT_PLIST" 2>/dev/null || true + rm -f "$EVAL_AGENT_PLIST" + echo "[sessionflow] Eval LaunchAgent removed ($EVAL_AGENT_PLIST)" >&2 + else + launchctl bootout "${target}/${EVAL_AGENT_LABEL}" 2>/dev/null || true + echo "[sessionflow] No eval plist found at $EVAL_AGENT_PLIST" >&2 + fi +} + +do_eval_agent_status() { + local target + target=$(launch_agent_target) + if [ -f "$EVAL_AGENT_PLIST" ]; then + echo "[sessionflow] Eval plist: $EVAL_AGENT_PLIST (present)" >&2 + else + echo "[sessionflow] Eval plist: $EVAL_AGENT_PLIST (missing)" >&2 + fi + if launchctl print "${target}/${EVAL_AGENT_LABEL}" >/dev/null 2>&1; then + echo "[sessionflow] Eval LaunchAgent: loaded in ${target}" >&2 + launchctl print "${target}/${EVAL_AGENT_LABEL}" \ + | grep -E '^[[:space:]]*(state|last exit code|pid)[[:space:]]*=' \ + | sed 's/^[[:space:]]*/ /' >&2 || true + exit 0 + else + echo "[sessionflow] Eval LaunchAgent: not loaded in ${target}" >&2 + exit 1 + fi +} + do_agent_status() { local target target=$(launch_agent_target) @@ -680,12 +822,15 @@ case "${1:-start}" in install-backfill-agent) do_install_backfill_agent ;; uninstall-backfill-agent) do_uninstall_backfill_agent ;; backfill-agent-status) do_backfill_agent_status ;; + install-eval-agent) do_install_eval_agent ;; + uninstall-eval-agent) do_uninstall_eval_agent ;; + eval-agent-status) do_eval_agent_status ;; install-mcp-bridge) do_install_mcp_bridge ;; mcp-bridge-status) do_mcp_bridge_status ;; install-headers-helper) do_install_headers_helper ;; headers-helper-status) do_headers_helper_status ;; *) - echo "Usage: $0 {start|stop|status|restart|install-agent|uninstall-agent|agent-status|install-backfill-agent|uninstall-backfill-agent|backfill-agent-status|install-mcp-bridge|mcp-bridge-status|install-headers-helper|headers-helper-status}" >&2 + echo "Usage: $0 {start|stop|status|restart|install-agent|uninstall-agent|agent-status|install-backfill-agent|uninstall-backfill-agent|backfill-agent-status|install-eval-agent|uninstall-eval-agent|eval-agent-status|install-mcp-bridge|mcp-bridge-status|install-headers-helper|headers-helper-status}" >&2 exit 1 ;; esac diff --git a/tests/conftest.py b/tests/conftest.py index 93ef302..0c7b961 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -449,6 +449,7 @@ async def get_issue_timeline_async(*args: object, **kwargs: object) -> list: backfill_fts=lambda *args, **kwargs: 0, get_model=lambda *args, **kwargs: object(), get_model_name=lambda *args, **kwargs: "embeddinggemma", + ranking_settings=lambda: {"recency_weight": 0.3, "recency_decay_days": 7}, ) before = set(sys.modules) monkeypatch.setitem(sys.modules, "rag_engine", module) diff --git a/tests/eval/fixture_queries.jsonl b/tests/eval/fixture_queries.jsonl new file mode 100644 index 0000000..16b4aa0 --- /dev/null +++ b/tests/eval/fixture_queries.jsonl @@ -0,0 +1,7 @@ +{"id": "q-ident-1", "group": "identifier", "tool": "search_all_sessions", "query": "SYN-12 parser offset", "args": {"project_root": "/tmp/synthetic"}, "expected": ["s1:10"]} +{"id": "q-para-1", "group": "paraphrase", "tool": "search_memories", "query": "why the queue drains on restart", "args": {"project": "synthetic"}, "expected": ["m-a", "m-b"]} +{"id": "q-buried-1", "group": "buried-fact", "tool": "search_all_sessions", "query": "the timeout picked at the end of the long debugging turn", "args": {"project_root": "/tmp/synthetic"}, "expected": ["s2:40"]} +{"id": "q-temporal-1", "group": "temporal", "tool": "search_all_sessions", "query": "what changed in the watcher this week", "args": {"project_root": "/tmp/synthetic"}, "expected": ["s3:5"]} +{"id": "q-cross-1", "group": "cross-project", "tool": "search_all_sessions", "query": "shared deploy script across repos", "args": {"project_root": "*"}, "expected": ["s4:1"]} +{"id": "q-none-1", "group": "no-answer", "tool": "search_all_sessions", "query": "kubernetes helm chart for the billing service", "args": {"project_root": "/tmp/synthetic"}, "expected": []} +{"id": "q-super-1", "group": "superseded", "tool": "search_memories", "query": "which embedding model do we use", "args": {"project": "synthetic"}, "expected": ["m-new"], "superseded": ["m-old"]} diff --git a/tests/test_eval_retrieval.py b/tests/test_eval_retrieval.py new file mode 100644 index 0000000..abfc03f --- /dev/null +++ b/tests/test_eval_retrieval.py @@ -0,0 +1,435 @@ +"""Retrieval eval harness: query-set loading, metric math, reports, usage rollup, freshness probe.""" + +import json +import math +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +import eval_retrieval as ev +import secret_redaction + +FIXTURE = Path(__file__).parent / "eval" / "fixture_queries.jsonl" + + +def _queries(): + return {q["id"]: q for q in ev.load_queries(FIXTURE)} + + +# --- Query set ----------------------------------------------------------------- + + +def test_fixture_set_loads_every_group(): + queries = ev.load_queries(FIXTURE) + assert {q["group"] for q in queries} == set(ev.GROUPS) + assert all(q["tool"] in ev.TOOLS for q in queries) + + +@pytest.mark.parametrize("line,needle", [ + ({"id": "a", "group": "bogus", "tool": "search_memories", "query": "x", "expected": []}, "group"), + ({"id": "a", "group": "identifier", "tool": "get_turns", "query": "x", "expected": []}, "tool"), + ({"id": "a", "group": "identifier", "tool": "search_memories", "query": "x"}, "expected"), + ({"id": "a", "group": "identifier", "tool": "search_memories", "expected": ["m"]}, "query"), + ({"id": "a", "group": "no-answer", "tool": "search_memories", "query": "x", "expected": ["m"]}, + "no-answer"), + ({"id": "a", "group": "superseded", "tool": "search_memories", "query": "x", "expected": ["m"]}, + "superseded"), +]) +def test_load_queries_rejects_bad_lines_with_line_number(tmp_path, line, needle): + path = tmp_path / "q.jsonl" + good = {"id": "ok", "group": "identifier", "tool": "search_memories", "query": "x", "expected": ["m"]} + path.write_text(json.dumps(good) + "\n" + json.dumps(line) + "\n") + with pytest.raises(ValueError) as exc: + ev.load_queries(path) + assert "line 2" in str(exc.value) + assert needle in str(exc.value) + + +def test_load_queries_rejects_duplicate_ids(tmp_path): + path = tmp_path / "q.jsonl" + line = {"id": "dup", "group": "identifier", "tool": "search_memories", "query": "x", "expected": ["m"]} + path.write_text(json.dumps(line) + "\n\n" + json.dumps(line) + "\n") + with pytest.raises(ValueError, match="line 3.*dup"): + ev.load_queries(path) + + +# --- Metric math --------------------------------------------------------------- + + +def test_single_hit_at_rank_two(): + q = _queries()["q-ident-1"] + s = ev.score_query(q, ["s9:1", "s1:10", "s9:2"]) + assert s["recall_at_5"] == 1.0 + assert s["rr"] == 0.5 + assert s["ndcg_at_10"] == pytest.approx(1 / math.log2(3)) + assert s["rank"] == 2 + + +def test_two_expected_one_inside_top_five(): + q = _queries()["q-para-1"] + s = ev.score_query(q, ["m-a", "x", "y", "z", "w", "m-b"]) + assert s["recall_at_5"] == 0.5 + assert s["rr"] == 1.0 + ideal = 1 + 1 / math.log2(3) + assert s["ndcg_at_10"] == pytest.approx((1 + 1 / math.log2(7)) / ideal) + + +def test_miss_scores_zero(): + s = ev.score_query(_queries()["q-buried-1"], ["s9:1", "s9:2"]) + assert (s["recall_at_5"], s["rr"], s["ndcg_at_10"], s["rank"]) == (0.0, 0.0, 0.0, None) + + +def test_hit_past_rank_ten_counts_only_for_rr(): + ranked = [f"x:{i}" for i in range(11)] + ["s2:40"] + s = ev.score_query(_queries()["q-buried-1"], ranked) + assert s["ndcg_at_10"] == 0.0 + assert s["recall_at_5"] == 0.0 + assert s["rr"] == pytest.approx(1 / 12) + + +def test_no_answer_has_no_recall_and_keeps_top_score(): + s = ev.score_query(_queries()["q-none-1"], ["s9:1"], top_score=0.41) + assert s["recall_at_5"] is None and s["rr"] is None and s["ndcg_at_10"] is None + assert s["top_score"] == 0.41 + + +def test_superseded_requires_current_above_old(): + q = _queries()["q-super-1"] + assert ev.score_query(q, ["m-old", "m-new"])["current_first"] is False + assert ev.score_query(q, ["m-new", "m-old"])["current_first"] is True + assert ev.score_query(q, ["m-new"])["current_first"] is True + assert ev.score_query(q, ["m-old"])["current_first"] is False + + +def test_summarize_means_per_group_and_overall(): + qs = _queries() + scored = [ + ev.score_query(qs["q-ident-1"], ["s9:1", "s1:10"]), + ev.score_query(qs["q-para-1"], ["m-a", "x", "y", "z", "w", "m-b"]), + ev.score_query(qs["q-none-1"], [], top_score=None), + ev.score_query(qs["q-super-1"], ["m-new", "m-old"]), + ] + summary = ev.summarize(scored) + assert summary["identifier"]["n"] == 1 + assert summary["identifier"]["mrr"] == 0.5 + assert summary["paraphrase"]["recall_at_5"] == 0.5 + assert summary["no-answer"]["recall_at_5"] is None + assert summary["superseded"]["current_first"] == 1.0 + # Overall means skip no-answer queries, which have no expected result. + assert summary["all"]["n"] == 4 + assert summary["all"]["mrr"] == pytest.approx((0.5 + 1.0 + 1.0) / 3) + + +# --- Tool payloads ------------------------------------------------------------- + + +def test_result_ids_and_top_score_for_turns(): + payload = {"results": [{"session_id": "s1", "turn_index": 10, "distance": 0.25}, + {"session_id": "s2", "turn_index": 3, "distance": None}]} + assert ev.result_ids("search_all_sessions", payload) == ["s1:10", "s2:3"] + assert ev.top_score("search_all_sessions", payload) == pytest.approx(0.75) + + +def test_result_ids_for_memories_have_no_score(): + payload = {"results": [{"memory_id": "m-1"}, {"memory_id": "m-2"}]} + assert ev.result_ids("search_memories", payload) == ["m-1", "m-2"] + assert ev.top_score("search_memories", payload) is None + assert ev.top_score("search_all_sessions", {"results": []}) is None + + +# --- Ranking settings ---------------------------------------------------------- + + +def test_parse_env_pairs(): + assert ev.parse_env(["SESSIONFLOW_RECENCY_WEIGHT=0.15", "A=b=c"]) == { + "SESSIONFLOW_RECENCY_WEIGHT": "0.15", "A": "b=c"} + with pytest.raises(ValueError): + ev.parse_env(["NOEQUALS"]) + + +def test_check_ranking_reports_mismatch_and_unverified(): + health = {"ranking": {"recency_weight": 0.3, "recency_decay_days": 7}} + assert ev.check_ranking({"SESSIONFLOW_RECENCY_WEIGHT": "0.3"}, health) == ([], []) + problems, unverified = ev.check_ranking( + {"SESSIONFLOW_RECENCY_WEIGHT": "0.15", "SESSIONFLOW_CANDIDATE_POOL": "200"}, health) + assert len(problems) == 1 and "0.15" in problems[0] + assert unverified == ["SESSIONFLOW_CANDIDATE_POOL"] + problems, _ = ev.check_ranking({"SESSIONFLOW_RECENCY_WEIGHT": "0.3"}, {}) + assert problems and "ranking" in problems[0] + + +# --- Reports ------------------------------------------------------------------- + + +def _report(): + qs = _queries() + scored = [ev.score_query(qs["q-ident-1"], ["s1:10"]), ev.score_query(qs["q-para-1"], ["m-b"])] + return ev.build_report(scored, run={"sort_by": "hybrid", "env": {}, "queries_file": "fixture"}) + + +def test_report_file_shape(tmp_path): + json_path, md_path = ev.write_report(_report(), tmp_path, "2026-10-30-hybrid") + data = json.loads(json_path.read_text()) + assert set(data) == {"run", "groups", "queries"} + assert data["run"]["sort_by"] == "hybrid" + assert data["run"]["query_count"] == 2 + assert {"identifier", "paraphrase", "all"} <= set(data["groups"]) + q = data["queries"][0] + assert {"id", "group", "tool", "query", "expected", "ranked", "rank", + "recall_at_5", "rr", "ndcg_at_10"} <= set(q) + md = md_path.read_text() + assert "| group | n | recall@5 | MRR | nDCG@10 |" in md + assert "| identifier | 1 | 1.000 | 1.000 | 1.000 |" in md + + +def test_comparison_table_has_one_column_per_label(): + md = ev.render_comparison({"w=0": _report(), "w=0.3": _report()}, metric="mrr") + header = md.splitlines()[0] + assert "w=0" in header and "w=0.3" in header + assert any(line.startswith("| identifier |") for line in md.splitlines()) + + +# --- Usage rollup -------------------------------------------------------------- + + +def _at(seconds): + base = datetime(2026, 10, 30, 12, 0, tzinfo=timezone.utc) + return (base + timedelta(seconds=seconds)).isoformat() + + +def test_usage_rollup_counts_top_three_follow_ups_and_reformulations(): + entries = [ + # Client a: search, open rank 2, then open rank 5. + {"ts": _at(0), "kind": "search", "tool": "search_memories", "client": "a", + "ids": ["m1", "m2", "m3", "m4", "m5"]}, + {"ts": _at(5), "kind": "follow_up", "tool": "get_memory", "client": "a", "id": "m2"}, + {"ts": _at(9), "kind": "follow_up", "tool": "get_memory", "client": "a", "id": "m5"}, + # Client b: search, re-search within 60 s with no follow-up, then a late search. + {"ts": _at(0), "kind": "search", "tool": "search_all_sessions", "client": "b", "ids": ["s:1"]}, + {"ts": _at(30), "kind": "search", "tool": "search_all_sessions", "client": "b", "ids": ["s:2"]}, + {"ts": _at(200), "kind": "search", "tool": "search_all_sessions", "client": "b", "ids": ["s:3"]}, + # Outside the window. + {"ts": _at(-9 * 86400), "kind": "search", "tool": "search_memories", "client": "c", "ids": []}, + ] + now = datetime(2026, 10, 30, 13, 0, tzinfo=timezone.utc) + roll = ev.usage_rollup(entries, now=now, days=7) + assert roll["searches"] == 4 + assert roll["follow_ups"] == 2 + assert roll["follow_up_top3_share"] == 0.5 + assert roll["reformulated_searches"] == 1 + assert roll["reformulation_share"] == 0.25 + + +def test_usage_rollup_skips_eval_clients(): + eval_client = f"{ev.EVAL_CLIENT_NAME}:1f" + entries = [ + {"ts": _at(i), "kind": "search", "tool": "search_memories", "client": eval_client, "ids": []} + for i in range(10) + ] + [{"ts": _at(20), "kind": "search", "tool": "search_memories", "client": "claude-code:2a", + "ids": ["m1"]}] + roll = ev.usage_rollup(entries, now=datetime(2026, 10, 30, 13, 0, tzinfo=timezone.utc), days=7) + assert roll["searches"] == 1 + assert roll["reformulated_searches"] == 0 + assert roll["eval_calls_skipped"] == 10 + + +def test_usage_rollup_empty_window(): + roll = ev.usage_rollup([], now=datetime.now(timezone.utc), days=7) + assert roll["searches"] == 0 + assert roll["follow_up_top3_share"] is None + assert roll["reformulation_share"] is None + + +# --- Freshness probe ----------------------------------------------------------- + + +class FakeClient: + def __init__(self, hit_after=3, delete_fails=False, stored_text=None): + self.calls = [] + self.hit_after = hit_after + self.delete_fails = delete_fails + self.searches = 0 + self.stored_text = stored_text + + def call(self, name, arguments): + self.calls.append((name, arguments)) + if name == "add_memory": + self.stored_text = self.stored_text or arguments["text"] + return {"memory_id": "canary-1", "outcome": "created", "index": "ok"} + if name == "search_memories": + self.searches += 1 + hit = self.hit_after is not None and self.searches >= self.hit_after + row = {"memory_id": "canary-1", "text": self.stored_text} + return {"results": [row] if hit else [], + "channels": ["fts"] if hit else []} + if name == "delete_memory": + if self.delete_fails: + raise ev.EvalError("delete failed") + return {"memory_id": "canary-1", "outcome": "deleted"} + raise AssertionError(name) + + def health(self): + return {"memory": {"ack_lag": {"milvus": 0}, "index": { + "host": "minitron", "last_tick_age_s": 1.5, "stale": False}}} + + +class FakeClock: + def __init__(self): + self.now = 0.0 + + def __call__(self): + return self.now + + def sleep(self, seconds): + self.now += seconds + + +def test_freshness_probe_reports_time_to_hit_and_deletes(): + client, clock = FakeClient(hit_after=3), FakeClock() + result = ev.freshness_probe(client, sleep=clock.sleep, clock=clock) + assert result["time_to_hit_s"] == pytest.approx(2.0) + assert result["polls"] == 3 + assert result["host"] == "minitron" + assert result["add_index"] == "ok" + assert result["health"]["last_tick_age_s"] == 1.5 + assert result["deleted"] is True + add = client.calls[0][1] + # One canary per host, so probes on two machines never share a record. + assert add["topic_key"] == f"{ev.CANARY_TOPIC}:{ev.socket.gethostname()}" + assert add["project"] == "sessionflow" and add["type"] == "fact" + token = client.calls[1][1]["query"] + assert token in add["text"] + assert client.calls[-1][0] == "delete_memory" + + +def test_freshness_probe_ignores_a_record_another_probe_rewrote(): + client, clock = FakeClient(hit_after=1, stored_text=ev.canary_text("sfcanary1")), FakeClock() + result = ev.freshness_probe(client, timeout_s=3, sleep=clock.sleep, clock=clock) + assert result["time_to_hit_s"] is None + + +def test_freshness_probes_on_one_host_take_turns(tmp_path): + import threading + + lock = tmp_path / "freshness.lock" + order = [] + + class SlowClient(FakeClient): + def call(self, name, arguments): + if name == "add_memory": + order.append("add") + if name == "delete_memory": + order.append("delete") + return super().call(name, arguments) + + def probe(): + ev.freshness_probe(SlowClient(hit_after=2), lock_path=lock, poll_s=0.05) + + threads = [threading.Thread(target=probe) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join(10) + assert order == ["add", "delete", "add", "delete"] + + +@pytest.mark.parametrize("given,want", [ + ("http://127.0.0.1:7102", "http://127.0.0.1:7102/mcp/"), + ("http://127.0.0.1:7102/", "http://127.0.0.1:7102/mcp/"), + ("http://127.0.0.1:7102/mcp", "http://127.0.0.1:7102/mcp"), + ("http://127.0.0.1:7102/mcp/", "http://127.0.0.1:7102/mcp/"), +]) +def test_server_url_points_at_the_mcp_endpoint(given, want): + assert ev.McpClient(given).url == want + + +def test_freshness_probe_times_out_and_still_deletes(): + client, clock = FakeClient(hit_after=None), FakeClock() + result = ev.freshness_probe(client, timeout_s=5, sleep=clock.sleep, clock=clock) + assert result["time_to_hit_s"] is None + assert result["deleted"] is True + assert client.calls[-1][0] == "delete_memory" + + +def test_freshness_probe_reports_failed_delete(): + client, clock = FakeClient(delete_fails=True), FakeClock() + result = ev.freshness_probe(client, sleep=clock.sleep, clock=clock) + assert result["deleted"] is False + assert "delete failed" in result["delete_error"] + + +def test_canary_text_passes_the_memory_secret_check(): + for _ in range(20): + text = ev.canary_text(ev.make_token()) + _, spans = secret_redaction.scan_spans(text, mode="report") + assert spans == [] + + +def test_failed_add_deletes_only_an_exact_canary(): + class AddFails(FakeClient): + def call(self, name, arguments): + if name == "add_memory": + self.calls.append((name, arguments)) + self.stored_text = arguments["text"] + raise ev.EvalError("timeout after commit") + if name == "search_memories": + self.calls.append((name, arguments)) + return {"results": [{"memory_id": "other", "text": "unrelated"}, + {"memory_id": "canary-1", "text": self.stored_text}]} + return super().call(name, arguments) + + client = AddFails() + result = ev.freshness_probe(client) + assert result["error"] == "timeout after commit" + assert result["deleted"] is True + deletes = [a for n, a in client.calls if n == "delete_memory"] + assert [d["memory_id"] for d in deletes] == ["canary-1"] + + +def test_restore_server_runs_every_step_and_reports_failures(monkeypatch): + import subprocess + + steps = [] + + def launchctl(*args): + steps.append(args) + raise subprocess.CalledProcessError(1, "launchctl") + + monkeypatch.setattr(ev, "_launchctl", launchctl) + monkeypatch.setattr(ev, "_restart_server", lambda label: steps.append(("restart", label))) + monkeypatch.setattr(ev, "wait_ready", lambda client, env: steps.append(("ready", env))) + errors = ev.restore_server("SESSIONFLOW_RECENCY_WEIGHT", "0.2", "lbl", "http://x/mcp/") + assert steps == [("setenv", "SESSIONFLOW_RECENCY_WEIGHT", "0.2"), ("restart", "lbl"), + ("ready", {"SESSIONFLOW_RECENCY_WEIGHT": "0.2"})] + assert len(errors) == 1 + + +def test_restore_server_unsets_a_value_that_was_unset(monkeypatch): + steps = [] + monkeypatch.setattr(ev, "_launchctl", lambda *args: steps.append(args) or "") + monkeypatch.setattr(ev, "_restart_server", lambda label: None) + monkeypatch.setattr(ev, "wait_ready", lambda client, env: steps.append(env)) + assert ev.restore_server("SESSIONFLOW_RECENCY_WEIGHT", None, "lbl", "http://x/mcp/") == [] + assert steps == [("unsetenv", "SESSIONFLOW_RECENCY_WEIGHT"), {}] + + +def test_eval_agent_plist_escapes_paths_and_keeps_the_url(tmp_path): + import plistlib + import subprocess + + script = Path(__file__).resolve().parent.parent / "sessionflow-server.sh" + plist = tmp_path / "eval.plist" + shell = ( + 'eval "$(sed -n "/^xml_escape()/,/^}/p;/^write_eval_agent_plist()/,/^}/p" "$0")"; ' + f'LAUNCH_AGENT_DIR="{tmp_path}"; EVAL_AGENT_PLIST="{plist}"; EVAL_AGENT_LABEL=l; ' + 'PYTHON=/p/python; SCRIPT_DIR=/repo; EVAL_HOUR=3; EVAL_MINUTE=30; ' + 'EVAL_AGENT_STDOUT=/o; EVAL_AGENT_STDERR=/e; ' + 'SESSIONFLOW_EVAL_QUERIES="/a & b/q.jsonl"; SESSIONFLOW_EVAL_OUT="/a /eval"; ' + 'SESSIONFLOW_PORT=7200; write_eval_agent_plist' + ) + subprocess.run(["bash", "-c", shell, str(script)], check=True, + env={"PATH": "/usr/bin:/bin", "HOME": str(tmp_path)}) + args = plistlib.loads(plist.read_bytes())["ProgramArguments"] + assert args[2:4] == ["--url", "http://127.0.0.1:7200"] + assert "/a & b/q.jsonl" in args and "/a /eval" in args diff --git a/tests/test_ranking_settings.py b/tests/test_ranking_settings.py new file mode 100644 index 0000000..884b79b --- /dev/null +++ b/tests/test_ranking_settings.py @@ -0,0 +1,53 @@ +"""The hybrid ranking settings are read from the environment and reported by /health.""" + +import pytest + +import rag_engine + + +def test_defaults(monkeypatch): + monkeypatch.delenv("SESSIONFLOW_RECENCY_WEIGHT", raising=False) + monkeypatch.delenv("SESSIONFLOW_RECENCY_DECAY_DAYS", raising=False) + assert rag_engine.ranking_settings() == { + "recency_weight": rag_engine.RECENCY_WEIGHT_DEFAULT, + "recency_decay_days": rag_engine.RECENCY_DECAY_DAYS_DEFAULT, + } + + +def test_env_values(monkeypatch): + monkeypatch.setenv("SESSIONFLOW_RECENCY_WEIGHT", "0.15") + monkeypatch.setenv("SESSIONFLOW_RECENCY_DECAY_DAYS", "14") + assert rag_engine.ranking_settings() == {"recency_weight": 0.15, "recency_decay_days": 14} + + +@pytest.mark.parametrize("value", ["2", "-1", "nan", "x"]) +def test_out_of_range_weight_falls_back(monkeypatch, value): + monkeypatch.setenv("SESSIONFLOW_RECENCY_WEIGHT", value) + assert rag_engine.ranking_settings()["recency_weight"] == rag_engine.RECENCY_WEIGHT_DEFAULT + + +@pytest.mark.parametrize("weight,order", [("1", ["new", "old"]), (None, ["old", "new"])]) +def test_hybrid_rank_uses_the_reported_weight(monkeypatch, weight, order): + from datetime import datetime, timezone + + if weight is None: + monkeypatch.delenv("SESSIONFLOW_RECENCY_WEIGHT", raising=False) + else: + monkeypatch.setenv("SESSIONFLOW_RECENCY_WEIGHT", weight) + rows = [{"_rrf_score": 1.0, "timestamp": "2020-01-01T00:00:00Z", "doc_id": "old"}, + {"_rrf_score": 0.5, "timestamp": "2026-10-30T00:00:00Z", "doc_id": "new"}] + now = datetime(2026, 10, 30, 12, tzinfo=timezone.utc) + ranked = rag_engine._rank_results(rows, "hybrid", 2, now=now) + assert [r["doc_id"] for r in ranked] == order + + +@pytest.mark.anyio +async def test_health_reports_ranking(stub_rag_engine, monkeypatch): + import importlib + import json + import sys + + monkeypatch.delitem(sys.modules, "http_server", raising=False) + http_server = importlib.import_module("http_server") + response = await http_server.health(None) + assert json.loads(response.body)["ranking"] == {"recency_weight": 0.3, "recency_decay_days": 7} diff --git a/tests/test_search_log.py b/tests/test_search_log.py new file mode 100644 index 0000000..42e74c4 --- /dev/null +++ b/tests/test_search_log.py @@ -0,0 +1,159 @@ +"""Search usage log: off by default, redacts queries, records search and follow-up calls.""" + +import asyncio +import importlib +import json +import stat +import types + +import pytest + +import search_log + +tools = importlib.import_module("tools") + +GITHUB_TOKEN = "ghp_" + "A1b2C3d4" * 4 + "Zz9Y" + + +@pytest.fixture +def home(tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.delenv("SESSIONFLOW_SEARCH_LOG", raising=False) + return tmp_path + + +def _lines(home): + path = home / ".sessionflow" / "search-log.jsonl" + if not path.exists(): + return [] + return [json.loads(line) for line in path.read_text().splitlines() if line] + + +def test_off_by_default_writes_nothing(home): + search_log.record_search("search_memories", client="c", project="p", sort_by="hybrid", + query="anything", ids=["m1"]) + search_log.record_follow_up("get_memory", client="c", item_id="m1") + assert not (home / ".sessionflow" / "search-log.jsonl").exists() + assert not (home / ".sessionflow").exists() + + +@pytest.mark.parametrize("value", ["0", "off", "false", ""]) +def test_falsy_values_keep_it_off(home, monkeypatch, value): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", value) + assert search_log.enabled() is False + + +def test_on_redacts_the_query_and_keeps_ids_in_order(home, monkeypatch): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", "1") + search_log.record_search("search_all_sessions", client="c", project="/repo", sort_by="relevance", + query=f"why did {GITHUB_TOKEN} leak", ids=["s:3", "s:1", "s:2"]) + raw = (home / ".sessionflow" / "search-log.jsonl").read_text() + assert GITHUB_TOKEN not in raw + (line,) = _lines(home) + assert line["kind"] == "search" + assert line["tool"] == "search_all_sessions" + assert line["ids"] == ["s:3", "s:1", "s:2"] + assert line["project"] == "/repo" and line["sort_by"] == "relevance" + assert "[REDACTED:" in line["query"] + assert line["ts"].endswith("+00:00") + + +def test_log_file_is_private(home, monkeypatch): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", "1") + search_log.record_follow_up("get_turns", client="c", item_id="s:1") + path = home / ".sessionflow" / "search-log.jsonl" + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + assert _lines(home)[0] == {**_lines(home)[0], "kind": "follow_up", "tool": "get_turns", "id": "s:1"} + + +def test_write_failure_does_not_raise(home, monkeypatch): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", "1") + (home / ".sessionflow").write_text("not a directory") + search_log.record_follow_up("get_memory", client=None, item_id="m1") + + +# --- Hooks in the MCP tools ---------------------------------------------------- + + +class StubService: + async def search(self, query, filters, n, sort_by="hybrid"): + return {"results": [{"memory_id": "m-2"}, {"memory_id": "m-1"}], "channels": ["fts"], + "incomplete": False} + + async def get(self, memory_id, project=None): + return {"memory_id": memory_id, "text": "t"} + + +def _call(name, arguments, client_name=None): + captured = {} + + class _Server: + if client_name: + request_context = types.SimpleNamespace(session=types.SimpleNamespace( + client_params=types.SimpleNamespace(clientInfo=types.SimpleNamespace(name=client_name)))) + + def list_tools(self): + return lambda fn: fn + + def call_tool(self): + def deco(fn): + captured["call"] = fn + return fn + return deco + + tools.register_tools(_Server()) + return asyncio.run(captured["call"](name, arguments)) + + +@pytest.fixture +def service(): + tools.set_memory_service(StubService()) + yield + tools.set_memory_service(None) + + +def test_memory_search_and_get_are_logged(home, monkeypatch, service): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", "1") + _call("search_memories", {"query": "queue drain", "project": "sessionflow"}) + _call("get_memory", {"memory_id": "m-1", "project": "sessionflow"}) + search, follow = _lines(home) + assert search["tool"] == "search_memories" and search["ids"] == ["m-2", "m-1"] + assert search["project"] == "sessionflow" and search["sort_by"] == "hybrid" + assert follow == {**follow, "kind": "follow_up", "tool": "get_memory", "id": "m-1"} + + +def test_get_turns_is_logged_as_a_turn_id(home, monkeypatch): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", "1") + monkeypatch.setattr(tools.rag_engine, "get_turns", lambda *a, **k: []) + _call("get_turns", {"session_id": "abc", "turn_index": 42}) + (line,) = _lines(home) + assert line["tool"] == "get_turns" and line["id"] == "abc:42" + + +def test_client_key_names_the_mcp_client(home, monkeypatch, service): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", "1") + _call("get_memory", {"memory_id": "m-1"}, client_name="eval-retrieval") + (line,) = _lines(home) + assert line["client"].startswith("eval-retrieval:") + + +def test_tools_write_nothing_when_off(home, service): + _call("search_memories", {"query": "queue drain", "project": "sessionflow"}) + assert _lines(home) == [] + + +def test_existing_loose_log_file_is_tightened(home, monkeypatch): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", "1") + path = home / ".sessionflow" / "search-log.jsonl" + path.parent.mkdir() + path.write_text("") + path.chmod(0o644) + search_log.record_follow_up("get_memory", client=None, item_id="m1") + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_lone_surrogate_does_not_raise(home, monkeypatch): + monkeypatch.setenv("SESSIONFLOW_SEARCH_LOG", "1") + search_log.record_search("search_memories", client=None, project="p", sort_by="hybrid", + query="bad \ud800 text", ids=[]) + assert len(_lines(home)) == 1 diff --git a/tools.py b/tools.py index 1ccdfab..9ad1df5 100644 --- a/tools.py +++ b/tools.py @@ -14,6 +14,7 @@ import client_roots import rag_engine import sanitize +import search_log from project_scope import ProjectScope, resolve_project_scope from provider_adapters import ( LEGAL_PROVIDERS, @@ -553,7 +554,7 @@ def _format_context(payload: dict) -> str: return "\n\n".join(parts) -async def _call_memory_tool(name: str, arguments: dict): +async def _call_memory_tool(name: str, arguments: dict, client: str | None = None): args = dict(arguments or {}) fmt = args.pop("format", "markdown") service = _memory_service @@ -598,6 +599,12 @@ async def _call_memory_tool(name: str, arguments: dict): message = getattr(exc, "message", None) or str(exc) or type(exc).__name__ details = getattr(exc, "details", None) return _memory_error(code, message, fmt, details if isinstance(details, dict) else None) + if name == "search_memories": + await search_log.log_search(name, client=client, project=project, + sort_by=args.get("sort_by", "hybrid"), query=args.get("query") or "", + ids=[r.get("memory_id") for r in payload.get("results") or []]) + elif name == "get_memory": + await search_log.log_follow_up(name, client=client, item_id=args["memory_id"]) if fmt == "json": text = _json_text(payload) return [types.TextContent(type="text", text=text)] @@ -840,10 +847,29 @@ async def _scope(arg: str | None, db: str) -> ProjectScope: scope = replace(scope, note=note) return scope + def _client_key() -> str | None: + """Return ``:`` for the usage log, which pairs calls by it. + + The client name lets the usage rollup skip the eval harness's own traffic. + """ + try: + session = server.request_context.session + except (LookupError, AttributeError): + return None + params = getattr(session, "client_params", None) + name = getattr(getattr(params, "clientInfo", None), "name", None) or "unknown" + return f"{name}:{id(session):x}" + + async def _log_turn_search(name: str, arguments: dict, scope: ProjectScope, results: list) -> None: + await search_log.log_search( + name, client=_client_key(), project=scope.filter_value(), + sort_by=arguments.get("sort_by", "hybrid"), query=arguments.get("query") or "", + ids=[f"{r.get('session_id')}:{r.get('turn_index')}" for r in results]) + @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: if name in MEMORY_TOOL_NAMES: - return await _call_memory_tool(name, arguments) + return await _call_memory_tool(name, arguments, client=_client_key()) db = get_db_path() current_project = _header_root() @@ -873,6 +899,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: issue_id=issue_id_arg, db_path=db, ) + await _log_turn_search(name, arguments, scope, results) return [types.TextContent(type="text", text=format_results(results, scope))] elif name == "search_all_sessions": @@ -909,6 +936,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: issue_id=issue_id_arg, db_path=db, ) + await _log_turn_search(name, arguments, scope, results) if arguments.get("format") == "json": payload = { "scope": {"roots": scope.filter_value(), "note": scope.note}, @@ -924,6 +952,9 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: context=arguments.get("context", 2), db_path=db, ) + await search_log.log_follow_up( + name, client=_client_key(), + item_id=f"{arguments['session_id']}:{arguments['turn_index']}") return [types.TextContent(type="text", text=format_turns(results))] elif name == "get_session_stats": diff --git a/version.py b/version.py index 95be475..9941aa1 100644 --- a/version.py +++ b/version.py @@ -1,3 +1,3 @@ """SessionFlow version. CHANGELOG.md names the same version in its newest released heading.""" -__version__ = "0.8.1" +__version__ = "0.9.0"