From 88e21da5db156bdb5fb3ec86a9fb7f607d738f45 Mon Sep 17 00:00:00 2001 From: lbruton Date: Fri, 30 Oct 2026 18:41:56 -0500 Subject: [PATCH 1/7] feat(SESF-82): retrieval eval harness, freshness probe and search usage log Nothing measured search quality or memory index freshness, so ranking settings were tuned by impression and stalls went unnoticed. - eval_retrieval.py scores a labelled JSONL query set over the local MCP endpoint: recall@5, MRR and nDCG@10 per group, a current-first rate for superseded decisions and the top hit score for no-answer queries. Subcommands run, sweep, freshness, usage and nightly. - run --env refuses when the server applies other ranking settings; /health reports them under "ranking" via rag_engine.ranking_settings. - sweep sets each value with launchctl setenv, restarts the LaunchAgent, and restores the server afterwards. It needs --yes. - freshness writes a canary fact (topic_key eval:freshness), polls search_memories until it appears, records the /health memory index state and deletes it. - search_log.py: SESSIONFLOW_SEARCH_LOG=1 appends search, get_memory and get_turns calls with redacted queries and ranked IDs. Off by default. - sessionflow-server.sh install-eval-agent runs nightly at 03:30. The real query set and reports stay in the private companion vault; tests/eval/fixture_queries.jsonl is synthetic. --- AGENTS.md | 1 + CHANGELOG.md | 7 + CLAUDE.md | 1 + README.md | 14 + eval_retrieval.py | 791 +++++++++++++++++++++++++++++++ http_server.py | 1 + rag_engine.py | 25 +- search_log.py | 67 +++ sessionflow-server.sh | 119 ++++- tests/conftest.py | 1 + tests/eval/fixture_queries.jsonl | 7 + tests/test_eval_retrieval.py | 308 ++++++++++++ tests/test_ranking_settings.py | 46 ++ tests/test_search_log.py | 130 +++++ tools.py | 29 +- 15 files changed, 1538 insertions(+), 9 deletions(-) create mode 100644 eval_retrieval.py create mode 100644 search_log.py create mode 100644 tests/eval/fixture_queries.jsonl create mode 100644 tests/test_eval_retrieval.py create mode 100644 tests/test_ranking_settings.py create mode 100644 tests/test_search_log.py diff --git a/AGENTS.md b/AGENTS.md index d7b98ff..0e302c0 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. `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..5bd8459 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ have no git tags or GitHub Releases. For changes before the fork, see [UPSTREAM- ## [Unreleased] +### 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..46bbb31 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. `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..472a9f1 --- /dev/null +++ b/eval_retrieval.py @@ -0,0 +1,791 @@ +"""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 json +import math +import os +import secrets +import socket +import subprocess +import sys +import time +import urllib.error +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 + +# 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; the session opens on the first call.""" + 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-retrieval", "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 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) -> dict: + """Write a canary memory, poll ``search_memories`` until it appears, then delete it. + + The canary's fixed ``topic_key`` makes a later probe replace one left behind by + a failed delete. The result names the host, the time to the first hit (None on + timeout), the add's index outcome and the /health memory index state. + """ + 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": CANARY_TOPIC, "extra": {"source": "eval"}}) + except EvalError as exc: + result["error"] = str(exc) + 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 result["memory_id"] in result_ids("search_memories", found): + 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. + """ + cutoff = now - timedelta(days=days) + timed = [] + for e in entries: + 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, + } + + +# --- Server restarts for sweeps ------------------------------------------------ + + +def _launchctl(*args: str) -> None: + subprocess.run(["launchctl", *args], check=True, capture_output=True, text=True) + + +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()] + reports = {} + client = McpClient(args.url) + 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() + _launchctl("unsetenv", args.var) + _restart_server(args.label) + wait_ready(McpClient(args.url), {}) + 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 0 + + +def _cmd_freshness(args) -> int: + client = McpClient(args.url) + try: + result = freshness_probe(client, timeout_s=args.timeout) + 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) + 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..8f99677 --- /dev/null +++ b/search_log.py @@ -0,0 +1,67 @@ +"""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 json +import logging +import os +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable, Optional + +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) + with os.fdopen(fd, "a", encoding="utf-8") as handle: + handle.write(json.dumps(entry, ensure_ascii=False) + "\n") + except OSError as exc: + logger.warning("Search log write failed: %s", exc) + + +def record_search(tool: str, *, client: Optional[str], project, 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}) diff --git a/sessionflow-server.sh b/sessionflow-server.sh index d1dc2ec..a122747 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,111 @@ do_backfill_agent_status() { fi } +write_eval_agent_plist() { + mkdir -p "$LAUNCH_AGENT_DIR" + cat > "$EVAL_AGENT_PLIST" < + + + + Label + ${EVAL_AGENT_LABEL} + + ProgramArguments + + ${PYTHON} + ${SCRIPT_DIR}/eval_retrieval.py + nightly + --queries + ${SESSIONFLOW_EVAL_QUERIES} + --out + ${SESSIONFLOW_EVAL_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 + if [ ! -f "$SESSIONFLOW_EVAL_QUERIES" ]; then + echo "[sessionflow] No query set at $SESSIONFLOW_EVAL_QUERIES" >&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 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 +794,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..4b916e7 --- /dev/null +++ b/tests/test_eval_retrieval.py @@ -0,0 +1,308 @@ +"""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_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): + self.calls = [] + self.hit_after = hit_after + self.delete_fails = delete_fails + self.searches = 0 + + def call(self, name, arguments): + self.calls.append((name, arguments)) + if name == "add_memory": + 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 + return {"results": [{"memory_id": "canary-1"}] 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] + assert add["topic_key"] == ev.CANARY_TOPIC + 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_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 == [] diff --git a/tests/test_ranking_settings.py b/tests/test_ranking_settings.py new file mode 100644 index 0000000..cd83ff7 --- /dev/null +++ b/tests/test_ranking_settings.py @@ -0,0 +1,46 @@ +"""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 + + +def test_hybrid_rank_uses_the_reported_weight(monkeypatch): + monkeypatch.setenv("SESSIONFLOW_RECENCY_WEIGHT", "0") + 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"}] + ranked = rag_engine._rank_results(rows, "hybrid", 2) + assert [r["doc_id"] for r in ranked] == ["old", "new"] + + +@pytest.mark.anyio +async def test_health_reports_ranking(stub_rag_engine): + import importlib + import json + import sys + + sys.modules.pop("http_server", None) + 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..0c27f3e --- /dev/null +++ b/tests/test_search_log.py @@ -0,0 +1,130 @@ +"""Search usage log: off by default, redacts queries, records search and follow-up calls.""" + +import asyncio +import importlib +import json +import stat + +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): + captured = {} + + class _Server: + 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_tools_write_nothing_when_off(home, service): + _call("search_memories", {"query": "queue drain", "project": "sessionflow"}) + assert _lines(home) == [] diff --git a/tools.py b/tools.py index 1ccdfab..5410323 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": + search_log.record_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": + search_log.record_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,23 @@ async def _scope(arg: str | None, db: str) -> ProjectScope: scope = replace(scope, note=note) return scope + def _client_key() -> str | None: + """Return a key for this tool call's MCP session, so the usage log can pair calls.""" + try: + return f"{id(server.request_context.session):x}" + except (LookupError, AttributeError): + return None + + def _log_turn_search(name: str, arguments: dict, scope: ProjectScope, results: list) -> None: + search_log.record_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 +893,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: issue_id=issue_id_arg, db_path=db, ) + _log_turn_search(name, arguments, scope, results) return [types.TextContent(type="text", text=format_results(results, scope))] elif name == "search_all_sessions": @@ -909,6 +930,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: issue_id=issue_id_arg, db_path=db, ) + _log_turn_search(name, arguments, scope, results) if arguments.get("format") == "json": payload = { "scope": {"roots": scope.filter_value(), "note": scope.note}, @@ -924,6 +946,9 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: context=arguments.get("context", 2), db_path=db, ) + search_log.record_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": From ff7b3ca577078e079789ef0d53c1de98e8480292 Mon Sep 17 00:00:00 2001 From: lbruton Date: Fri, 30 Oct 2026 18:43:57 -0500 Subject: [PATCH 2/7] chore(SESF-82): version 0.9.0 --- CHANGELOG.md | 2 ++ version.py | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5bd8459..14d7816 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ 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). 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" From af73db475b158168099a73f14a13a174ca0acf92 Mon Sep 17 00:00:00 2001 From: lbruton Date: Fri, 30 Oct 2026 18:53:42 -0500 Subject: [PATCH 3/7] fix(SESF-82): keep eval traffic out of the usage rollup The usage log keyed calls by MCP session only, so the harness's own queries counted as user reformulations. The client key now carries the MCP clientInfo name, and usage_rollup skips eval-retrieval clients and reports how many it skipped. --- eval_retrieval.py | 13 ++++++++++--- tests/test_eval_retrieval.py | 13 +++++++++++++ tests/test_search_log.py | 14 +++++++++++++- tools.py | 10 ++++++++-- 4 files changed, 44 insertions(+), 6 deletions(-) diff --git a/eval_retrieval.py b/eval_retrieval.py index 472a9f1..829c98e 100644 --- a/eval_retrieval.py +++ b/eval_retrieval.py @@ -37,6 +37,8 @@ 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 = { @@ -333,7 +335,7 @@ def _request(self, method: str, params: dict) -> dict: def _open(self) -> None: self.session_id = None self._request("initialize", {"protocolVersion": PROTOCOL_VERSION, "capabilities": {}, - "clientInfo": {"name": "eval-retrieval", "version": "1"}}) + "clientInfo": {"name": EVAL_CLIENT_NAME, "version": "1"}}) self._post({"jsonrpc": "2.0", "method": "notifications/initialized"}) def call(self, name: str, arguments: dict) -> dict: @@ -552,11 +554,15 @@ def usage_rollup(entries: list[dict], *, now: datetime, days: int = 7) -> dict: 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. + 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 = [] + 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): @@ -586,6 +592,7 @@ def usage_rollup(entries: list[dict], *, now: datetime, days: int = 7) -> dict: "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, } diff --git a/tests/test_eval_retrieval.py b/tests/test_eval_retrieval.py index 4b916e7..2e7123b 100644 --- a/tests/test_eval_retrieval.py +++ b/tests/test_eval_retrieval.py @@ -221,6 +221,19 @@ def test_usage_rollup_counts_top_three_follow_ups_and_reformulations(): 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 diff --git a/tests/test_search_log.py b/tests/test_search_log.py index 0c27f3e..b4583d3 100644 --- a/tests/test_search_log.py +++ b/tests/test_search_log.py @@ -4,6 +4,7 @@ import importlib import json import stat +import types import pytest @@ -83,10 +84,14 @@ async def get(self, memory_id, project=None): return {"memory_id": memory_id, "text": "t"} -def _call(name, arguments): +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 @@ -125,6 +130,13 @@ def test_get_turns_is_logged_as_a_turn_id(home, monkeypatch): 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) == [] diff --git a/tools.py b/tools.py index 5410323..5785a15 100644 --- a/tools.py +++ b/tools.py @@ -848,11 +848,17 @@ async def _scope(arg: str | None, db: str) -> ProjectScope: return scope def _client_key() -> str | None: - """Return a key for this tool call's MCP session, so the usage log can pair calls.""" + """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: - return f"{id(server.request_context.session):x}" + 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}" def _log_turn_search(name: str, arguments: dict, scope: ProjectScope, results: list) -> None: search_log.record_search( From 87798a7ee4a89c0f0cc15b3f282a48ee7b68e5e3 Mon Sep 17 00:00:00 2001 From: lbruton Date: Fri, 30 Oct 2026 18:59:06 -0500 Subject: [PATCH 4/7] fix(SESF-82): base server URLs and overlapping freshness probes A base SESSIONFLOW_URL such as http://127.0.0.1:7102 posted MCP requests to / and every query failed with 404; McpClient now appends /mcp/ when the URL has no path. Two probes sharing the canary topic_key replaced each other's record, so one could time out or take the other's record as its hit. The topic_key names the host, probes on one host take turns on a file lock, and a hit needs the probe's own token in the returned text. --- AGENTS.md | 2 +- CLAUDE.md | 2 +- eval_retrieval.py | 42 ++++++++++++++++++++++------- tests/test_eval_retrieval.py | 51 +++++++++++++++++++++++++++++++++--- 4 files changed, 83 insertions(+), 14 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 0e302c0..b73a0e6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,7 +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. `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`. +- **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/CLAUDE.md b/CLAUDE.md index 46bbb31..e35ef9f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -42,7 +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. `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`. +- **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/eval_retrieval.py b/eval_retrieval.py index 829c98e..ba7b687 100644 --- a/eval_retrieval.py +++ b/eval_retrieval.py @@ -11,6 +11,7 @@ from __future__ import annotations import argparse +import fcntl import json import math import os @@ -20,6 +21,7 @@ import sys import time import urllib.error +import urllib.parse import urllib.request from datetime import datetime, timedelta, timezone from pathlib import Path @@ -297,7 +299,10 @@ 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; the session opens on the first call.""" + """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 @@ -454,16 +459,34 @@ def canary_text(token: str) -> str: 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) -> dict: + 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 canary's fixed ``topic_key`` makes a later probe replace one left behind by - a failed delete. The result names the host, the time to the first hit (None on - timeout), the add's index outcome and the /health memory index state. + 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 _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, @@ -472,7 +495,7 @@ def freshness_probe(client, *, project: str = CANARY_PROJECT, agent_id: str = "c try: added = client.call("add_memory", { "text": canary_text(token), "agent_id": agent_id, "project": project, "type": "fact", - "topic_key": CANARY_TOPIC, "extra": {"source": "eval"}}) + "topic_key": f"{CANARY_TOPIC}:{socket.gethostname()}", "extra": {"source": "eval"}}) except EvalError as exc: result["error"] = str(exc) return result @@ -496,7 +519,8 @@ def freshness_probe(client, *, project: str = CANARY_PROJECT, agent_id: str = "c found = client.call("search_memories", {"query": token, "project": project, "n": 5}) except EvalError as exc: found, result["search_error"] = {}, str(exc) - if result["memory_id"] in result_ids("search_memories", found): + 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 @@ -692,7 +716,7 @@ def _cmd_sweep(args) -> int: def _cmd_freshness(args) -> int: client = McpClient(args.url) try: - result = freshness_probe(client, timeout_s=args.timeout) + result = freshness_probe(client, timeout_s=args.timeout, lock_path=default_lock_path()) finally: client.close() print(render_freshness(result)) @@ -721,7 +745,7 @@ def _cmd_nightly(args) -> int: 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) + 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: diff --git a/tests/test_eval_retrieval.py b/tests/test_eval_retrieval.py index 2e7123b..b1715f6 100644 --- a/tests/test_eval_retrieval.py +++ b/tests/test_eval_retrieval.py @@ -245,20 +245,23 @@ def test_usage_rollup_empty_window(): class FakeClient: - def __init__(self, hit_after=3, delete_fails=False): + 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 - return {"results": [{"memory_id": "canary-1"}] if hit else [], + 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: @@ -292,13 +295,55 @@ def test_freshness_probe_reports_time_to_hit_and_deletes(): assert result["health"]["last_tick_age_s"] == 1.5 assert result["deleted"] is True add = client.calls[0][1] - assert add["topic_key"] == ev.CANARY_TOPIC + # 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) From 27aee70e62ab63ce0e70b1127d396f141b7ab9b9 Mon Sep 17 00:00:00 2001 From: lbruton Date: Fri, 30 Oct 2026 19:05:17 -0500 Subject: [PATCH 5/7] fix(SESF-82): review fixes for the eval harness and usage log - sweep restores the launchd value it found, runs every restore step even when one fails, and exits 1 when the restore fails. - A freshness add that errors after committing deletes the canary whose text matches exactly. - The usage log runs on a worker thread, tightens an existing log file to 0600 and escapes non-ASCII, so an odd query cannot fail a call. - install-eval-agent requires absolute paths, XML-escapes them, passes --url from SESSIONFLOW_URL or SESSIONFLOW_PORT and refuses a plist that plutil rejects. - The recency-weight test fails when the env var is ignored, and the /health test restores http_server in sys.modules. --- eval_retrieval.py | 67 ++++++++++++++++++++++++++++++--- search_log.py | 24 ++++++++++-- sessionflow-server.sh | 37 +++++++++++++++--- tests/test_eval_retrieval.py | 69 ++++++++++++++++++++++++++++++++++ tests/test_ranking_settings.py | 19 +++++++--- tests/test_search_log.py | 17 +++++++++ tools.py | 14 +++---- 7 files changed, 219 insertions(+), 28 deletions(-) diff --git a/eval_retrieval.py b/eval_retrieval.py index ba7b687..cf9b7ab 100644 --- a/eval_retrieval.py +++ b/eval_retrieval.py @@ -486,6 +486,24 @@ def freshness_probe(client, *, project: str = CANARY_PROJECT, agent_id: str = "c 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, @@ -497,7 +515,9 @@ def _probe(client, project, agent_id, poll_s, timeout_s, sleep, clock) -> dict: "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")) @@ -623,8 +643,41 @@ def usage_rollup(entries: list[dict], *, now: datetime, days: int = 7) -> dict: # --- Server restarts for sweeps ------------------------------------------------ -def _launchctl(*args: str) -> None: - subprocess.run(["launchctl", *args], check=True, capture_output=True, text=True) +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: @@ -683,8 +736,10 @@ def _cmd_sweep(args) -> int: 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} @@ -699,9 +754,9 @@ def _cmd_sweep(args) -> int: write_report(report, args.out, f"{_local_date()}-sweep-{args.sort_by}-{label}") finally: client.close() - _launchctl("unsetenv", args.var) - _restart_server(args.label) - wait_ready(McpClient(args.url), {}) + 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)] @@ -710,7 +765,7 @@ def _cmd_sweep(args) -> int: out.write_text(text, encoding="utf-8") print(text) print(f"Wrote {out}", file=sys.stderr) - return 0 + return 1 if cleanup_errors else 0 def _cmd_freshness(args) -> int: diff --git a/search_log.py b/search_log.py index 8f99677..8fa5a96 100644 --- a/search_log.py +++ b/search_log.py @@ -7,13 +7,14 @@ 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 +from typing import Iterable, Optional, Union import secret_redaction @@ -40,13 +41,16 @@ def _append(entry: dict) -> None: 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, ensure_ascii=False) + "\n") - except OSError as exc: + 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, sort_by: Optional[str], +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. @@ -65,3 +69,15 @@ def record_follow_up(tool: str, *, client: Optional[str], item_id: str) -> None: 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 a122747..0fab0a8 100755 --- a/sessionflow-server.sh +++ b/sessionflow-server.sh @@ -611,8 +611,22 @@ do_backfill_agent_status() { fi } +xml_escape() { + local s="$1" + s="${s//&/&}" + s="${s///>}" + printf '%s' "$s" +} + 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" < @@ -623,13 +637,15 @@ write_eval_agent_plist() { ProgramArguments - ${PYTHON} - ${SCRIPT_DIR}/eval_retrieval.py + ${python} + ${script_dir}/eval_retrieval.py + --url + ${url} nightly --queries - ${SESSIONFLOW_EVAL_QUERIES} + ${queries} --out - ${SESSIONFLOW_EVAL_OUT} + ${out} StartCalendarInterval @@ -652,7 +668,7 @@ write_eval_agent_plist() { StandardErrorPath ${EVAL_AGENT_STDERR} WorkingDirectory - ${SCRIPT_DIR} + ${script_dir} PLIST @@ -663,6 +679,13 @@ do_install_eval_agent() { 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 @@ -673,6 +696,10 @@ do_install_eval_agent() { 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 diff --git a/tests/test_eval_retrieval.py b/tests/test_eval_retrieval.py index b1715f6..abfc03f 100644 --- a/tests/test_eval_retrieval.py +++ b/tests/test_eval_retrieval.py @@ -364,3 +364,72 @@ def test_canary_text_passes_the_memory_secret_check(): 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 index cd83ff7..884b79b 100644 --- a/tests/test_ranking_settings.py +++ b/tests/test_ranking_settings.py @@ -26,21 +26,28 @@ def test_out_of_range_weight_falls_back(monkeypatch, value): assert rag_engine.ranking_settings()["recency_weight"] == rag_engine.RECENCY_WEIGHT_DEFAULT -def test_hybrid_rank_uses_the_reported_weight(monkeypatch): - monkeypatch.setenv("SESSIONFLOW_RECENCY_WEIGHT", "0") +@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"}] - ranked = rag_engine._rank_results(rows, "hybrid", 2) - assert [r["doc_id"] for r in ranked] == ["old", "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): +async def test_health_reports_ranking(stub_rag_engine, monkeypatch): import importlib import json import sys - sys.modules.pop("http_server", None) + 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 index b4583d3..42e74c4 100644 --- a/tests/test_search_log.py +++ b/tests/test_search_log.py @@ -140,3 +140,20 @@ def test_client_key_names_the_mcp_client(home, monkeypatch, service): 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 5785a15..9ad1df5 100644 --- a/tools.py +++ b/tools.py @@ -600,11 +600,11 @@ async def _call_memory_tool(name: str, arguments: dict, client: str | None = Non details = getattr(exc, "details", None) return _memory_error(code, message, fmt, details if isinstance(details, dict) else None) if name == "search_memories": - search_log.record_search(name, client=client, project=project, + 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": - search_log.record_follow_up(name, client=client, item_id=args["memory_id"]) + 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)] @@ -860,8 +860,8 @@ def _client_key() -> str | None: name = getattr(getattr(params, "clientInfo", None), "name", None) or "unknown" return f"{name}:{id(session):x}" - def _log_turn_search(name: str, arguments: dict, scope: ProjectScope, results: list) -> None: - search_log.record_search( + 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]) @@ -899,7 +899,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: issue_id=issue_id_arg, db_path=db, ) - _log_turn_search(name, arguments, scope, results) + await _log_turn_search(name, arguments, scope, results) return [types.TextContent(type="text", text=format_results(results, scope))] elif name == "search_all_sessions": @@ -936,7 +936,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: issue_id=issue_id_arg, db_path=db, ) - _log_turn_search(name, arguments, scope, results) + await _log_turn_search(name, arguments, scope, results) if arguments.get("format") == "json": payload = { "scope": {"roots": scope.filter_value(), "note": scope.note}, @@ -952,7 +952,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: context=arguments.get("context", 2), db_path=db, ) - search_log.record_follow_up( + 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))] From 564245f272228ad62c12e890cd01f35134e43d02 Mon Sep 17 00:00:00 2001 From: lbruton Date: Fri, 30 Oct 2026 19:07:11 -0500 Subject: [PATCH 6/7] fix(SESF-82): install-eval-agent requires the venv interpreter --- sessionflow-server.sh | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/sessionflow-server.sh b/sessionflow-server.sh index 0fab0a8..b075fd2 100755 --- a/sessionflow-server.sh +++ b/sessionflow-server.sh @@ -690,6 +690,10 @@ do_install_eval_agent() { 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 From 462f6c4e25ec389f86c5a55469b34e221830c1ef Mon Sep 17 00:00:00 2001 From: lbruton Date: Fri, 30 Oct 2026 19:12:30 -0500 Subject: [PATCH 7/7] fix(SESF-82): escape plist paths with sed Bash 5.2 (CI) expands & in a ${s//x/y} replacement to the match, so < became /\>/g' } write_eval_agent_plist() {