diff --git a/migrations/core/009_tool_calls_correlation.sql b/migrations/core/009_tool_calls_correlation.sql new file mode 100644 index 00000000..38f44b1a --- /dev/null +++ b/migrations/core/009_tool_calls_correlation.sql @@ -0,0 +1,7 @@ +-- Add the **turn** level to the tool-call log (ACE-015). ACE-008 captured `thread_id` (the whole +-- conversation) and the per-call `user_question`/`agent_query`; this adds `correlation_id` — the turn, +-- i.e. the ONE user question whose answer fanned out into several agent sub-queries. Self-reported by +-- Claude (the MCP protocol carries no turn boundary, so the server can't mint it), nullable and +-- best-effort like the other self-report columns; the Sessions view groups a session's calls by it and +-- degrades to ungrouped when absent. Portable (runs on SQLite + Postgres unchanged). +ALTER TABLE tool_calls ADD COLUMN correlation_id TEXT; diff --git a/packages/agami-core/README.md b/packages/agami-core/README.md index 6970f0c5..ef31a259 100644 --- a/packages/agami-core/README.md +++ b/packages/agami-core/README.md @@ -105,12 +105,14 @@ The admin console also has two read-only activity tabs: - **Tool calls** — *every* MCP tool call, newest first: who (the authenticated user), the tool, datasource, and for a query the SQL, row count, latency, and status. This is **audit-grade** — the server observes it directly, so it's always accurate. -- **Sessions** — those queries grouped into a conversation, each opening to its queries with the - natural-language **question**. This is **best-effort**: the MCP protocol carries neither the user's - question nor a conversation id, so Claude self-reports them (a `user_question` param + a `thread_id` - it reuses per conversation). When it does, you get grouping + the question; when it doesn't, the view - degrades to ungrouped and the question shows as not reported. Treat those two fields as a hint, not a - record. +- **Sessions** — those queries grouped into a conversation, and *within* it into **turns**: each turn + is one user question and the **N agent queries** Claude ran to answer it (*"User asked X → 2 + queries"*). This is **best-effort** — the MCP protocol carries neither the user's question, a + conversation id, nor a turn boundary, so Claude self-reports them: a `user_question` (kept verbatim), + a `thread_id` (per conversation), and a `correlation_id` (per turn). The turn's question is taken from + the **first** call in the turn (the model sometimes drifts it on later refinements). When Claude + doesn't supply a `correlation_id`, each query simply shows as its own turn — the view degrades, never + errors. Treat the self-reported fields as a hint, not a record. The `tool_calls` log grows one row per call and has **no automatic retention** — it's your local store, so prune it on your own schedule if it gets large. diff --git a/packages/agami-core/src/admin.py b/packages/agami-core/src/admin.py index b82fafbf..56a8f2ca 100644 --- a/packages/agami-core/src/admin.py +++ b/packages/agami-core/src/admin.py @@ -284,28 +284,49 @@ def calls_tab_html( def _session_drawer(s: dict[str, Any], idx: int) -> str: sid = f"sess-{idx}" # DOM id is the row index, never the (self-reported, attacker-influenceable) key + # Render the session as **turns** (one user question -> N agent queries), grouped on correlation_id + # by list_sessions. The turn header shows the verbatim question once; each agent refinement (its + # `agent_query` + SQL) lists beneath it. Degrades cleanly: a call with no correlation_id is its own + # one-query turn, so this reads like the old flat list when Claude doesn't self-report. cards = "" - for q in s["queries"]: - question = q.get("user_question") or "(no question reported)" - sub = ( - f'
↳ {ui.esc(q["agent_query"])} ' - f'· self-reported
' - if q.get("agent_query") - else "" + for t in s["turns"]: + question = t.get("question") or "(no question reported)" + n = len(t["queries"]) + qrows = "" + for q in t["queries"]: + sub = ( + f'
↳ {ui.esc(q["agent_query"])} ' + f'· agent-reported
' + if q.get("agent_query") + else "" + ) + sql = ( + f'
'
+                f"{ui.esc(q['sql'])}
" + if q.get("sql") + else "" + ) + lat = (str(q["execution_ms"]) + " ms") if q.get("execution_ms") is not None else "" + rc = q.get("row_count") if q.get("row_count") is not None else "—" + qrows += ( + '
' + f"{sub}{sql}" + f'
{_utc(q["ts"])} · ' + f"{lat} {_ok_pill(q['success'])} · {rc} rows
" + ) + # The question is Claude-self-reported (best-effort, attacker-influenceable) — mark it so, like + # the rest of the activity log; "User asked" is the framing, "· self-reported" the provenance. + asked = ( + f'User asked {ui.esc(question)} ' + '· self-reported' + if t.get("question") + else f'{ui.esc(question)}' ) - sql = ( - f'
'
-            f"{ui.esc(q['sql'])}
" - if q.get("sql") - else "" - ) - lat = (str(q["execution_ms"]) + " ms") if q.get("execution_ms") is not None else "" cards += ( - '
' - f'
{ui.esc(question)}' - f'{_utc(q["ts"])} · {lat} {_ok_pill(q["success"])}
' - f"{sub}{sql}" - f'
{q.get("row_count") if q.get("row_count") is not None else "—"} rows
' + '
' + f'
{asked} ' + f'· {n} {"query" if n == 1 else "queries"}' + f"
{qrows}
" ) return f"""
diff --git a/packages/agami-core/src/contracts.py b/packages/agami-core/src/contracts.py index 25db6379..306b2f8c 100644 --- a/packages/agami-core/src/contracts.py +++ b/packages/agami-core/src/contracts.py @@ -236,3 +236,4 @@ class ToolCallRecord(_Contract): user_question: str | None = None agent_query: str | None = None thread_id: str | None = None + correlation_id: str | None = None # the turn: one user question -> N agent sub-queries diff --git a/packages/agami-core/src/model_store.py b/packages/agami-core/src/model_store.py index 40b12dcc..cb01f6d1 100644 --- a/packages/agami-core/src/model_store.py +++ b/packages/agami-core/src/model_store.py @@ -334,8 +334,9 @@ def record_tool_call(self, record: Any) -> None: # `success` is a portable 0/1 (no boolean literal across SQLite/Postgres). self._store.execute( "INSERT INTO tool_calls (id, ts, actor, tool_name, datasource, sql, row_count, " - "execution_ms, success, error_kind, source, user_question, agent_query, thread_id) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "execution_ms, success, error_kind, source, user_question, agent_query, thread_id, " + "correlation_id) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( uuid4().hex, record.ts, @@ -351,6 +352,7 @@ def record_tool_call(self, record: Any) -> None: record.user_question, record.agent_query, record.thread_id, + record.correlation_id, ), ) self._store.commit() @@ -362,7 +364,7 @@ def record_tool_call(self, record: Any) -> None: _TOOL_CALL_COLS = ( "id, ts, actor, tool_name, datasource, sql, row_count, execution_ms, success, error_kind, " - "user_question, agent_query, thread_id" + "user_question, agent_query, thread_id, correlation_id" ) @@ -373,6 +375,31 @@ def list_tool_calls(store: Store, *, limit: int = 200) -> list[dict[str, Any]]: ) +def _group_turns(queries: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Group a session's query calls into **turns** by the self-reported `correlation_id` (one user + question -> N agent sub-queries). A query with no `correlation_id` is its own singleton turn (so + the view degrades to the per-call list). The turn's `question` is the **earliest** call's + `user_question` — Claude drifts `user_question` onto later refinements, so the first is the reliable + one. Turns keep newest-first order (queries arrive ts-DESC); each turn's queries read chronologically.""" + turns_map: dict[Any, list[dict[str, Any]]] = {} + turn_order: list[Any] = [] + for q in queries: + ck = q["correlation_id"] or q["id"] # no correlation_id -> singleton keyed on the row id + if ck not in turns_map: + turns_map[ck] = [] + turn_order.append(ck) + turns_map[ck].append(q) + turns: list[dict[str, Any]] = [] + for ck in turn_order: + tq = sorted(turns_map[ck], key=lambda x: x["ts"]) # chronological within the turn + turns.append({ + "question": tq[0].get("user_question"), # earliest call's question (drift-proof) + "started": tq[0]["ts"], + "queries": tq, + }) + return turns + + def list_sessions(store: Store, *, limit: int = 500) -> list[dict[str, Any]]: """Group the **query** calls (execute_sql) into sessions for the best-effort Sessions view: same `thread_id` = one session; a call with no `thread_id` (Claude didn't self-report) becomes its own @@ -408,5 +435,6 @@ def list_sessions(store: Store, *, limit: int = 500) -> list[dict[str, Any]]: s["query_count"] = len(qs) s["error_count"] = sum(1 for q in qs if not q["success"]) s["avg_ms"] = round(sum(ms) / len(ms)) if ms else None + s["turns"] = _group_turns(qs) # the within-session turn level (ACE-015) out.append(s) return out diff --git a/packages/agami-core/src/tools.py b/packages/agami-core/src/tools.py index 88be7a27..54e83078 100644 --- a/packages/agami-core/src/tools.py +++ b/packages/agami-core/src/tools.py @@ -77,9 +77,12 @@ def server_version() -> str: "COUNT/COUNT(DISTINCT)/filter/GROUP BY/JOIN on it, but never SELECT its raw per-row values. " "'unique emails' → COUNT(DISTINCT email). To disambiguate identical labels, project the " "non-sensitive id. (execute_sql enforces this and errors on a raw sensitive projection.)\n" - "Activity log: on execute_sql, pass `user_question` (the user's verbatim question) and a " - "`thread_id` you generate once per conversation and reuse on every call — so a deployment admin " - "can see what was asked and group a conversation's queries. Best-effort; omit if unknown." + "Activity log: on execute_sql, pass `user_question` (the user's question VERBATIM — keep it the " + "SAME across every query answering that one question; put your own refinement in `raw_query`, never " + "in `user_question`), a `thread_id` (one per conversation), and a `correlation_id` (one per user " + "question/turn, reused across the queries answering it, fresh when they ask something new) — so a " + "deployment admin sees the conversation, and within it 'user asked X → agent ran N queries'. " + "Best-effort; omit if unknown." ) @@ -1012,6 +1015,7 @@ def record_tool_call( "user_question": args.get("user_question"), "agent_query": args.get("raw_query"), # the existing arg is the agent's framing of the query "thread_id": args.get("thread_id"), + "correlation_id": args.get("correlation_id"), # the turn (one user question) } ) @@ -1140,18 +1144,26 @@ def _record_tool_call(rec: dict[str, Any]) -> None: }, "raw_query": { "type": "string", - "description": "Your (the agent's) framing of this query — recorded for the admin activity log.", + "description": "Your (the agent's) framing of THIS sub-query — recorded for the " + "admin activity log. Your refinement goes here, NOT in user_question.", }, "user_question": { "type": "string", - "description": "The user's ORIGINAL question, verbatim, that led to this query — " - "recorded so an admin can see what was asked. Pass it whenever you have it.", + "description": "The user's ORIGINAL question, VERBATIM. Keep it the SAME across every " + "query you run to answer one question — do not replace it with your refinement (that " + "goes in raw_query). Recorded so an admin sees what was actually asked.", }, "thread_id": { "type": "string", "description": "A short id you generate ONCE per conversation and reuse on every " "tool call in it — lets the admin group a conversation's queries into one session.", }, + "correlation_id": { + "type": "string", + "description": "A short id you generate ONCE per USER QUESTION (a turn) and reuse on " + "every query you run to answer THAT question — lets the admin see 'user asked X → " + "agent ran N queries'. Start a fresh one when the user asks something new.", + }, "max_rows": {"type": "integer", "description": "Row cap (clamped 1–10000)."}, }, "required": ["sql"], diff --git a/render_previews.py b/render_previews.py index 15d25345..47103ed6 100644 --- a/render_previews.py +++ b/render_previews.py @@ -84,19 +84,22 @@ def write(name: str, html: str) -> None: _s.run_migrations() _sink = DbActivitySink(_s) _SAMPLE_CALLS = [ - dict(ts="2026-06-27T10:39:02Z", tool_name="execute_sql", source="mcp_server", actor="jordan@example.com", - datasource="SALES_DATA", sql="SELECT id, customer_id, amount\nFROM orders\nORDER BY created_at DESC\nLIMIT 10", - row_count=10, execution_ms=73, success=True, user_question="Show me the 10 most recent orders", - agent_query="recent orders", thread_id="t1"), - dict(ts="2026-06-27T10:40:55Z", tool_name="get_datasource_schema", source="mcp_server", - actor="jordan@example.com", datasource="SALES_DATA", execution_ms=12, success=True), + # One turn (correlation c1): a single user question that the agent answered with TWO queries. dict(ts="2026-06-27T10:42:17Z", tool_name="execute_sql", source="mcp_server", actor="jordan@example.com", datasource="SALES_DATA", sql="SELECT region, SUM(amount) AS revenue\nFROM orders\nGROUP BY region\nORDER BY revenue DESC", row_count=5, execution_ms=84, success=True, user_question="What's our revenue by region this quarter?", - agent_query="revenue by region", thread_id="t1"), + agent_query="revenue by region", thread_id="t1", correlation_id="c1"), + dict(ts="2026-06-27T10:42:41Z", tool_name="execute_sql", source="mcp_server", actor="jordan@example.com", + datasource="SALES_DATA", sql="SELECT date_trunc('month', placed_at) AS month, SUM(amount)\nFROM orders\nWHERE region = 'West'\nGROUP BY 1\nORDER BY 1", + row_count=3, execution_ms=61, success=True, user_question="What's our revenue by region this quarter?", + agent_query="monthly trend for the top region (West)", thread_id="t1", correlation_id="c1"), + dict(ts="2026-06-27T10:40:55Z", tool_name="get_datasource_schema", source="mcp_server", + actor="jordan@example.com", datasource="SALES_DATA", execution_ms=12, success=True), + # A separate one-query turn that errored (correlation c2). dict(ts="2026-06-27T10:41:50Z", tool_name="execute_sql", source="mcp_server", actor="sam@example.com", datasource="SALES_DATA", sql="SELECT * FROM ordrs", execution_ms=31, success=False, - error_kind="syntax", thread_id="t2"), + error_kind="syntax", user_question="how many orders today?", agent_query="count today's orders", + thread_id="t2", correlation_id="c2"), dict(ts="2026-06-27T10:40:03Z", tool_name="list_datasources", source="mcp_server", actor="jordan@example.com", execution_ms=3, success=True), ] diff --git a/tests/test_admin_activity.py b/tests/test_admin_activity.py index f92ba6b8..800e4c75 100644 --- a/tests/test_admin_activity.py +++ b/tests/test_admin_activity.py @@ -171,3 +171,135 @@ def test_activity_tabs_drop_the_redundant_helper_text(client, env): _login(client) assert "Every tool call, newest first" not in client.get("/admin?tab=calls").text assert "Queries grouped into conversations" not in client.get("/admin?tab=sessions").text + + +# --- ACE-015: the turn level (correlation_id) -------------------------------- + + +def test_correlation_id_round_trips(env): + s = Store.connect(env) + _call( + s, + ts="2026-06-28T10:00:00Z", + actor="a", + sql="SELECT 1", + success=True, + thread_id="t1", + correlation_id="turn-1", + ) + rows = model_store.list_tool_calls(s) + s.close() + assert rows[0]["correlation_id"] == "turn-1" + + +def test_execute_sql_inputschema_exposes_correlation_id(): + import tools + + props = tools.TOOLS["execute_sql"]["inputSchema"]["properties"] + assert "correlation_id" in props + assert "turn" in props["correlation_id"]["description"].lower() + + +def test_server_instructions_ask_for_verbatim_question_and_per_turn_correlation(): + import tools + + instr = tools.SERVER_INSTRUCTIONS + assert "correlation_id" in instr and "VERBATIM" in instr + + +def test_turns_use_earliest_question_and_group_refinements(env): + # Two calls of ONE turn: the 2nd drifts user_question; the turn must keep the FIRST (verbatim) one. + s = Store.connect(env) + _call( + s, + ts="2026-06-28T10:00:00Z", + actor="a", + sql="Q1", + success=True, + thread_id="t", + correlation_id="c1", + user_question="feature adoption vs churn", + agent_query="band", + ) + _call( + s, + ts="2026-06-28T10:01:00Z", + actor="a", + sql="Q2", + success=True, + thread_id="t", + correlation_id="c1", + user_question="churn by cancel_reason (drift)", + agent_query="lost MRR", + ) + sessions = model_store.list_sessions(s) + s.close() + turns = sessions[0]["turns"] + assert len(turns) == 1 + assert turns[0]["question"] == "feature adoption vs churn" # earliest, not the drift + assert [q["sql"] for q in turns[0]["queries"]] == [ + "Q1", + "Q2", + ] # chronological, both refinements + + +def test_turns_degrade_to_singletons_without_correlation_id(env): + s = Store.connect(env) + _call( + s, + ts="2026-06-28T10:00:00Z", + actor="a", + sql="A", + success=True, + thread_id="t", + user_question="q-a", + ) # no correlation_id + _call( + s, + ts="2026-06-28T10:01:00Z", + actor="a", + sql="B", + success=True, + thread_id="t", + user_question="q-b", + ) # no correlation_id + sessions = model_store.list_sessions(s) + s.close() + turns = sessions[0]["turns"] + assert len(turns) == 2 # each bare call is its own turn (flat behaviour preserved) + assert {t["question"] for t in turns} == {"q-a", "q-b"} + + +def test_sessions_drawer_renders_turn_with_user_asked_and_agent_queries(client, env): + s = Store.connect(env) + _call( + s, + ts="2026-06-28T10:00:00Z", + actor="jordan@example.com", + sql="Q1", + success=True, + thread_id="t", + correlation_id="c1", + user_question="feature adoption vs churn", + agent_query="churn by adoption band", + ) + _call( + s, + ts="2026-06-28T10:01:00Z", + actor="jordan@example.com", + sql="Q2", + success=True, + thread_id="t", + correlation_id="c1", + user_question="drifted", + agent_query="lost MRR", + ) + s.close() + _login(client) + html = client.get("/admin?tab=sessions").text + assert ( + "User asked" in html and "feature adoption vs churn" in html + ) # the turn question (earliest) + assert "drifted" not in html # the drifted user_question is NOT shown + assert "churn by adoption band" in html and "lost MRR" in html # both agent refinements + assert "2 queries" in html