Agent trace recipe for broken call chains + tpk eval harness (#17 step 4, #35)

README.md

@@ -754,6 +754,25 @@ tools, what does the docs repo say about Quickstart?").
 
 stdio MCP is local and unrestricted: no login, no role scope.
 
+## Evaluate the agent
+
+`tpk eval` runs a small fixed question set (trace, relationship, architecture,
+docs — `src/tpk/eval_questions.toml`) through the **real** chat agent against
+the live graph, and reports *how* it answered: which tools it called, whether
+it used the graph's edges (`neighbors` / `path_between`) or only keyword search,
+how many tool calls it spent, and a sanity check that the answer is about the
+right thing. Use it to measure a prompt, tool or extraction change instead of
+eyeballing chat traces:
+
+    docker compose exec app tpk eval --yes -o /tmp/before.json
+    # ...change something, rebuild...
+    docker compose exec app tpk eval --yes -o /tmp/after.json --compare /tmp/before.json
+
+`--only trace --only relationship` and `--limit N` narrow the run;
+`--questions my.toml` uses your own set. Every question is a full agent run
+against the configured LLM, so it costs tokens (12 questions by default) and
+takes a few minutes. It is a behaviour probe, not a correctness benchmark.
+
 ## Tests
 
     uv run pytest                    # integration tests skip without TIMEPLUS_HOST

docs/FEATURES.md

@@ -132,7 +132,7 @@ Login-based access control, with roles that scope what each user can query.
 | `8123` | Timeplusd SQL over HTTP (ClickHouse-compatible; override with `TIMEPLUS_PORT`) |
 | `3218` | Timeplus REST ingest API |
 | `tpk-mcp` | stdio MCP server (via `docker compose exec -T app tpk-mcp`) |
-| `tpk` CLI | `ingest`, `status`, `serve`, `export` / `import`, and `auth reset-admin` (break-glass admin recovery) |
+| `tpk` CLI | `ingest`, `status`, `serve`, `export` / `import`, `auth reset-admin` (break-glass admin recovery), and `eval` (run a fixed question set through the agent and report its tool usage) |
 
 **Images** (multi-arch: `linux/amd64` + `linux/arm64`, published per release tag by GitHub Actions): `timeplus/tpk-app` — the app alone, for a separate timeplusd; `timeplus/tpk` — all-in-one (OSS proton + tpk).
 

src/tpk/agent.py

@@ -155,7 +155,30 @@ def system_prompt(repos) -> str:
    answer text itself — never end the conversation on a tool call or with
    an empty message, and never leave the answer only in your private
    reasoning.
-7. PROTECT SOURCE CODE. Read and search the code freely to ground your
+7. TRACE RECIPE — for "trace the call path", "how does X reach Y", "what
+   happens when ..." questions. The graph's call edges are real but the
+   chain BREAKS wherever the code calls through an interface or a pointer
+   (`storage->write(...)`, `interpreter->execute()`): those calls are not
+   in the graph. Do not give up at a break — get past it:
+   a. Find both endpoints with search_entities (kinds=["function"];
+      methods are named `Class::method`, so search "InterpreterInsertQuery
+      execute", not just "execute").
+   b. path_between(id_a, id_b). mode "calls" → you have the chain, go to e.
+   c. Otherwise walk it yourself, one hop at a time: neighbors(id,
+      rels=["calls"], direction="out") from the start (and
+      direction="in" from the end), following the callee that leads
+      toward the other endpoint.
+   d. At a dead end, read_source the function you are stuck in and look at
+      what it actually calls. For a call through an interface such as
+      `x->write(...)`, find the implementations by method name —
+      search_entities("::write", kinds=["function"]) — pick the one that
+      fits the context (the class named in the code, or the subsystem you
+      are heading for), and continue from it with step c.
+   e. Answer with the chain in order, marking every hop [graph] (an edge
+      you saw) or [code] (you read the call in the source), and say where
+      the chain is still unconfirmed.
+   A trace may use up to 18 tool calls instead of the usual 12.
+8. PROTECT SOURCE CODE. Read and search the code freely to ground your
    answer, and quote only the SHORT snippets needed to explain a point —
    but never reproduce complete or near-complete files, and never
    reconstruct a whole file across several quotes. If the user asks you to

src/tpk/cli.py

@@ -179,6 +179,66 @@ def import_bundle(
     typer.echo("import complete")
 
 
+@app.command(name="eval")
+def eval_agent(
+    out: Path = typer.Option(Path("eval-report.json"), "--out", "-o", help="Where to write the JSON report"),
+    only: list[str] = typer.Option(None, "--only", help="Run only these categories (repeatable)"),
+    limit: int = typer.Option(0, "--limit", help="Run at most N questions (0 = all)"),
+    questions: Path = typer.Option(None, "--questions", help="Custom question TOML (default: the bundled set)"),
+    compare: Path = typer.Option(None, "--compare", help="A previous report to diff against"),
+    yes: bool = typer.Option(False, "--yes", "-y", help="Skip the cost confirmation"),
+):
+    """Run the fixed eval questions through the REAL chat agent and report how
+    it answered: tools called, graph-edge usage, tool-call count, basic checks.
+
+    Every question is a full agent run against the configured LLM (real token
+    cost) and the live knowledge graph. Runs unscoped, like an admin."""
+    import asyncio
+    import json as _json
+
+    from tpk import evals
+    from tpk.agent import RECURSION_LIMIT, build_agent
+    from tpk.config import AgentConfig
+    from tpk.server import _build_kg_and_repos
+    from tpk.version import version_string
+
+    qs = evals.load_questions(questions)
+    if only:
+        qs = [q for q in qs if q.category in only]
+    if limit:
+        qs = qs[:limit]
+    if not qs:
+        raise typer.BadParameter("no questions selected")
+    cfg = AgentConfig.from_env()
+    typer.echo(f"{len(qs)} questions -> {cfg.provider}/{cfg.model} (each is a full agent run; LLM cost applies)")
+    if not yes:
+        typer.confirm("Run the eval?", abort=True)
+
+    prefix = setting("TPK_STREAM_PREFIX", "db", "stream_prefix", "")
+    kg, repos = _build_kg_and_repos(prefix)
+    agent = build_agent(kg, cfg, repos)
+
+    def _progress(rec):
+        mark = "ok  " if rec["passed"] else "FAIL"
+        typer.echo(f"  {mark} {rec['id']:32s} {len(rec['tools']):>2} calls  {rec['latency_s']:>5}s  "
+                   + (rec["error"] or " ".join(rec["tools"]))[:90])
+
+    records = asyncio.run(evals.run_eval(agent, qs, recursion_limit=RECURSION_LIMIT, on_result=_progress))
+    report = evals.write_report(out, records, meta={
+        "tpk": version_string(), "provider": cfg.provider, "model": cfg.model,
+        "questions": len(qs), "at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+    })
+    typer.echo("")
+    typer.echo(evals.render_markdown(report))
+    typer.echo(f"\nreport written to {out}")
+    if compare:
+        diff = evals.compare(_json.loads(Path(compare).read_text()), report)
+        typer.echo(f"\nvs {compare}:")
+        for k in ("passed", "graph_tool_rate", "avg_tool_calls", "errors"):
+            typer.echo(f"  {k:16s} {diff[k][0]} -> {diff[k][1]}")
+        typer.echo(f"  fixed: {diff['fixed'] or '-'}   regressed: {diff['regressed'] or '-'}")
+
+
 auth_app = typer.Typer(help="Auth store maintenance (run where the DB credentials are)")
 app.add_typer(auth_app, name="auth")
 

src/tpk/eval_questions.toml

@@ -0,0 +1,96 @@
+# The bundled eval set for `tpk eval` (#35). Written against the Timeplus
+# corpus (proton / proton-enterprise, docs, helm-charts). Keep it SMALL: every
+# question is a real agent run (LLM cost). `expect_answer_any` is a sanity
+# check that the answer is about the right thing, not a correctness oracle.
+#
+# trace / relationship questions must be answered from graph EDGES, so they
+# require one of: neighbors, path_between.
+
+# -- trace: how does A reach B ------------------------------------------------
+[[question]]
+id = "trace-http-insert-nativelog"
+category = "trace"
+question = "Trace the call path from HTTP insert to nativelog write"
+expect_tools_any = ["neighbors", "path_between"]
+expect_answer_any = ["InterpreterInsertQuery", "executeQuery", "NativeLog"]
+# The longest chain in the set (HTTP -> interpreter -> storage -> sink ->
+# NativeLog, across ~4 interface breaks): measured at 21-22 calls with a
+# complete answer. 24 keeps it under the agent's hard recursion limit.
+max_tool_calls = 24
+
+[[question]]
+id = "trace-insert-build-chain"
+category = "trace"
+question = "Trace how InterpreterInsertQuery::execute ends up building the insert chain: which functions does it go through?"
+expect_tools_any = ["neighbors", "path_between"]
+expect_answer_any = ["buildChainImpl", "buildChain"]
+
+[[question]]
+id = "trace-mv-checkpoint"
+category = "trace"
+question = "Trace how a materialized view checkpoint gets triggered and written to storage."
+expect_tools_any = ["neighbors", "path_between"]
+expect_answer_any = ["checkpoint"]
+
+# -- relationship: one hop ----------------------------------------------------
+[[question]]
+id = "rel-execute-callees"
+category = "relationship"
+question = "What does InterpreterInsertQuery::execute call?"
+expect_tools_any = ["neighbors", "path_between"]
+expect_answer_any = ["buildChainImpl", "getTable", "buildQueryPipeline"]
+
+[[question]]
+id = "rel-produce-callers"
+category = "relationship"
+question = "What calls NativeLog::processProduceRequest?"
+expect_tools_any = ["neighbors", "path_between"]
+expect_answer_any = ["doProcessRequest"]
+
+[[question]]
+id = "rel-iinterpreter-subclasses"
+category = "relationship"
+question = "Which classes inherit from IInterpreter? Name a few."
+expect_tools_any = ["neighbors", "path_between"]
+expect_answer_any = ["InterpreterInsertQuery", "InterpreterSelectQuery", "InterpreterCreateQuery"]
+
+# -- architecture: graph tools help but are not required ----------------------
+[[question]]
+id = "arch-mv-checkpoints"
+category = "architecture"
+question = "How do materialized view checkpoints work?"
+expect_answer_any = ["checkpoint"]
+
+[[question]]
+id = "arch-nativelog"
+category = "architecture"
+question = "What are the main components of the NativeLog subsystem and how do they fit together?"
+expect_answer_any = ["NativeLog"]
+
+[[question]]
+id = "arch-insert-responsibility"
+category = "architecture"
+question = "Where is InterpreterInsertQuery defined and what is it responsible for?"
+expect_answer_any = ["src/Interpreters", "InterpreterInsertQuery.h", "InterpreterInsertQuery.cpp"]
+
+# -- docs / deployment: should come from documents, cheaply -------------------
+[[question]]
+id = "docs-kafka-external-stream"
+category = "docs"
+question = "How do I create an external stream to read from Kafka?"
+expect_answer_any = ["external stream"]
+max_tool_calls = 12
+
+[[question]]
+id = "docs-mutable-stream"
+category = "docs"
+question = "What is a mutable stream and when should I use one?"
+expect_answer_any = ["mutable"]
+max_tool_calls = 12
+
+[[question]]
+id = "deploy-helm-resources"
+category = "docs"
+question = "What does the Helm chart set for timeplusd resources?"
+expect_answer_any = ["resources", "memory", "cpu"]
+max_tool_calls = 12

src/tpk/evals.py

@@ -0,0 +1,161 @@
+"""Offline agent eval (#35): run a fixed question set through the real agent
+and measure HOW it answers -- which tools it called, whether it touched the
+graph's edges, how many calls it spent -- plus cheap sanity checks on the
+answer. It exists to replace eyeballing chat traces: change the prompt, the
+tools or the extraction, run it again, `--compare` the two reports.
+
+It is not a correctness benchmark: `expect_answer_any` only catches an answer
+that never mentions the thing asked about.
+"""
+
+import json
+import time
+import tomllib
+from dataclasses import dataclass, field
+from importlib import resources
+from pathlib import Path
+
+# The tools that read graph EDGES (as opposed to keyword search / raw source).
+GRAPH_TOOLS = ("neighbors", "path_between")
+DEFAULT_MAX_TOOL_CALLS = 20
+
+
+@dataclass
+class Question:
+    id: str
+    category: str
+    question: str
+    # pass if ANY of these tools was called (empty = no requirement)
+    expect_tools_any: list[str] = field(default_factory=list)
+    # pass if the answer mentions ANY of these, case-insensitively (empty = any non-empty answer)
+    expect_answer_any: list[str] = field(default_factory=list)
+    max_tool_calls: int = DEFAULT_MAX_TOOL_CALLS
+
+
+def load_questions(path: Path | None = None) -> list[Question]:
+    """The bundled set (tpk/eval_questions.toml), or a custom TOML file with the
+    same `[[question]]` tables."""
+    if path is None:
+        raw = resources.files("tpk").joinpath("eval_questions.toml").read_text()
+    else:
+        raw = Path(path).read_text()
+    return [Question(**q) for q in tomllib.loads(raw)["question"]]
+
+
+def _text(content) -> str:
+    # OpenAI yields str content; Anthropic may yield a list of content blocks.
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        return "".join(b.get("text", "") for b in content
+                       if isinstance(b, dict) and b.get("type") == "text")
+    return ""
+
+
+def extract_run(result: dict) -> dict:
+    """Tool-call sequence and final answer from a LangGraph agent result."""
+    tools: list[str] = []
+    answer = ""
+    for msg in result.get("messages", []):
+        calls = getattr(msg, "tool_calls", None)
+        if calls:
+            tools.extend(c["name"] for c in calls)
+        elif getattr(msg, "type", "") == "ai":
+            answer = _text(msg.content)   # the last AI message without tool calls
+    return {"tools": tools, "answer": answer.strip()}
+
+
+def score(q: Question, run: dict) -> dict:
+    tools, answer = run["tools"], run["answer"]
+    checks = {
+        "tools": not q.expect_tools_any or any(t in tools for t in q.expect_tools_any),
+        "answer": bool(answer) and (not q.expect_answer_any or any(
+            k.lower() in answer.lower() for k in q.expect_answer_any)),
+        "budget": len(tools) <= q.max_tool_calls,
+    }
+    return {**run, "checks": checks, "passed": all(checks.values()),
+            "used_graph_tool": any(t in GRAPH_TOOLS for t in tools)}
+
+
+async def run_eval(agent, questions: list[Question], recursion_limit: int = 40,
+                   on_result=None) -> list[dict]:
+    """One fresh conversation per question. An agent/gateway error fails that
+    question and is recorded -- it must not abort the run."""
+    records = []
+    for q in questions:
+        started = time.monotonic()
+        try:
+            result = await agent.ainvoke(
+                {"messages": [{"role": "user", "content": q.question}]},
+                config={"recursion_limit": recursion_limit},
+            )
+            rec = {**score(q, extract_run(result)), "error": ""}
+        except Exception as exc:  # noqa: BLE001 -- recorded, not swallowed
+            rec = {"tools": [], "answer": "", "checks": {}, "passed": False,
+                   "used_graph_tool": False, "error": f"{type(exc).__name__}: {exc}"[:300]}
+        rec = {"id": q.id, "category": q.category, "question": q.question,
+               "latency_s": round(time.monotonic() - started, 1), **rec}
+        records.append(rec)
+        if on_result:
+            on_result(rec)
+    return records
+
+
+def _stats(records: list[dict]) -> dict:
+    n = len(records)
+    return {
+        "questions": n,
+        "passed": sum(1 for r in records if r["passed"]),
+        "graph_tool_rate": round(sum(1 for r in records if r["used_graph_tool"]) / n, 2) if n else 0.0,
+        "avg_tool_calls": round(sum(len(r["tools"]) for r in records) / n, 2) if n else 0.0,
+        "errors": sum(1 for r in records if r.get("error")),
+    }
+
+
+def summarize(records: list[dict]) -> dict:
+    by_cat: dict[str, list[dict]] = {}
+    tool_counts: dict[str, int] = {}
+    for r in records:
+        by_cat.setdefault(r["category"], []).append(r)
+        for t in r["tools"]:
+            tool_counts[t] = tool_counts.get(t, 0) + 1
+    return {
+        "overall": _stats(records),
+        "by_category": {c: _stats(rs) for c, rs in sorted(by_cat.items())},
+        "tool_counts": dict(sorted(tool_counts.items(), key=lambda kv: -kv[1])),
+    }
+
+
+def compare(old: dict, new: dict) -> dict:
+    """(old, new) for the headline metrics, plus which questions flipped."""
+    o, n = old["summary"]["overall"], new["summary"]["overall"]
+    was = {r["id"]: r["passed"] for r in old["records"]}
+    now = {r["id"]: r["passed"] for r in new["records"]}
+    shared = sorted(set(was) & set(now))
+    return {
+        **{k: (o[k], n[k]) for k in ("passed", "graph_tool_rate", "avg_tool_calls", "errors")},
+        "fixed": [i for i in shared if now[i] and not was[i]],
+        "regressed": [i for i in shared if was[i] and not now[i]],
+    }
+
+
+def write_report(path: Path, records: list[dict], meta: dict) -> dict:
+    report = {"meta": meta, "summary": summarize(records), "records": records}
+    Path(path).write_text(json.dumps(report, indent=2))
+    return report
+
+
+def render_markdown(report: dict) -> str:
+    s = report["summary"]
+    lines = ["| category | questions | passed | used a graph tool | avg tool calls | errors |",
+             "|---|---|---|---|---|---|"]
+    for name, st in [*s["by_category"].items(), ("**overall**", s["overall"])]:
+        lines.append(f"| {name} | {st['questions']} | {st['passed']} | "
+                     f"{round(st['graph_tool_rate'] * 100)}% | {st['avg_tool_calls']} | {st['errors']} |")
+    lines += ["", "tool calls: " + ", ".join(f"{t} {c}" for t, c in s["tool_counts"].items())]
+    failed = [r for r in report["records"] if not r["passed"]]
+    if failed:
+        lines += ["", "failed:"] + [
+            f"- {r['id']}: " + (r["error"] or "failed checks: " + ", ".join(
+                k for k, ok in r["checks"].items() if not ok)) for r in failed]
+    return "\n".join(lines)