Turn-level observability — correlation_id binds a question's agent refinements

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;

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.

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'<div class="muted" style="margin:2px 0 8px">↳ {ui.esc(q["agent_query"])} '
-            f'<span class="muted">· self-reported</span></div>'
-            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'<div class="muted" style="margin:0 0 6px">↳ {ui.esc(q["agent_query"])} '
+                f'<span class="muted">· agent-reported</span></div>'
+                if q.get("agent_query")
+                else ""
+            )
+            sql = (
+                f'<pre class="code" style="white-space:pre-wrap;padding:12px;display:block;margin-top:4px">'
+                f"{ui.esc(q['sql'])}</pre>"
+                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 += (
+                '<div style="border-top:1px solid var(--line);padding:9px 0 11px">'
+                f"{sub}{sql}"
+                f'<div class="muted" style="font-size:13px;margin-top:6px">{_utc(q["ts"])} · '
+                f"{lat} {_ok_pill(q['success'])} · {rc} rows</div></div>"
+            )
+        # 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'<span class="muted">User asked</span> <strong>{ui.esc(question)}</strong> '
+            '<span class="muted" style="font-size:13px">· self-reported</span>'
+            if t.get("question")
+            else f'<strong class="muted">{ui.esc(question)}</strong>'
         )
-        sql = (
-            f'<pre class="code" style="white-space:pre-wrap;padding:12px;display:block;margin-top:6px">'
-            f"{ui.esc(q['sql'])}</pre>"
-            if q.get("sql")
-            else ""
-        )
-        lat = (str(q["execution_ms"]) + " ms") if q.get("execution_ms") is not None else ""
         cards += (
-            '<div style="border-top:1px solid var(--line);padding:14px 0">'
-            f'<div style="display:flex;justify-content:space-between"><strong>{ui.esc(question)}</strong>'
-            f'<span class="muted" style="font-size:13px">{_utc(q["ts"])} · {lat} {_ok_pill(q["success"])}</span></div>'
-            f"{sub}{sql}"
-            f'<div class="muted" style="font-size:13px;margin-top:6px">{q.get("row_count") if q.get("row_count") is not None else "—"} rows</div></div>'
+            '<div style="border-top:2px solid var(--line);padding:14px 0 2px;margin-top:8px">'
+            f'<div style="margin-bottom:4px">{asked} '
+            f'<span class="muted" style="font-size:13px">· {n} {"query" if n == 1 else "queries"}</span>'
+            f"</div>{qrows}</div>"
         )
     return f"""<input type="checkbox" id="{sid}" class="drawer-toggle">
 <div class="drawer-wrap"><label for="{sid}" class="drawer-backdrop"></label>

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

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

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"],

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),
 ]

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