diff --git a/CHANGELOG.md b/CHANGELOG.md index c0230c85..d2e2b389 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,20 @@ ### Fixed +- **`plumb hooks uptake` can now see a hint being used on an unshared + connection.** The consumed proxy matched a hint only against calls stamped + with the agent's identity. plumb stamps that identity only while several agents + share a connection, so every Codex hint, and every hint to a Claude Code + session without subagents, read as unattributed, including the first-prompt + hints that matter most. A call now also counts as the agent's when it comes + from a connection that agent alone declared through `session_start`, even when + the declaration follows the hint, and that was never observed shared (it + recorded no identity-stamped call). A connection two sessions declared, or one + with a `session_start` that named no one, still counts as nobody's. An agent + that never identifies itself on a connection is invisible to this rule, for + example a hook-less desktop chat sharing a connector. Its calls can then be + credited to the session that did declare, so the report remains a proxy. + - **The Claude Code and Codex hooks stay inside their session's namespace.** `plumb hooks run-claude` and `run-codex` now send the session's `PLUMB_NAMESPACE` with every daemon request, and the daemon answers the diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 8149f827..f9d544ec 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -408,6 +408,11 @@ uptake` reports configured (handlers on disk), trusted (Codex's own approval record), invoked (the daemon's bounded hint ledger, by outcome and reason) and consumed (a later plumb call by the same agent named what the hint named: a proxy, and hints with no attributable later call are reported as unattributed). +A call is the agent's when it carries the agent's identity, or when it comes +from a connection that agent alone declared through `session_start` and that was +never observed shared, which covers Codex and single-agent Claude Code sessions. +An agent that never identifies itself on such a connection leaves no trace, so +its calls count as the declaring agent's. Lifecycle hooks are the one place plumb can reach an agent between tool calls: at the start of a session, and at the end of a turn. Bare `plumb hooks` is diff --git a/internal/cli/hooks_uptake.go b/internal/cli/hooks_uptake.go index 52af7825..c047ac2f 100644 --- a/internal/cli/hooks_uptake.go +++ b/internal/cli/hooks_uptake.go @@ -47,7 +47,10 @@ ledger and its tool-call stats, both opened read-only. invoked hook requests the daemon recorded, by outcome and reason consumed emitted hints a later plumb call by the same agent mentioned, within ten minutes — a proxy for use, not proof of comprehension; hints whose - agent made no attributable call are counted as unattributed + agent made no attributable call are counted as unattributed. A call + is the agent's when it carries the agent's identity, or comes from a + connection only that agent declared (session_start) and that was + never observed shared Rows the ledger evicted (1000 rows / 1 MiB / 7 days per workspace) are counted, so a missing row is never read as a no-op.`, diff --git a/internal/cli/hooks_uptake_attribution_test.go b/internal/cli/hooks_uptake_attribution_test.go new file mode 100644 index 00000000..88210a71 --- /dev/null +++ b/internal/cli/hooks_uptake_attribution_test.go @@ -0,0 +1,94 @@ +package cli + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" + + "github.com/plumbkit/plumb/internal/contexthints" + "github.com/plumbkit/plumb/internal/stats" +) + +// TestHooksUptake_FirstPromptHintOnAnUnsharedConnectionIsConsumed drives the real +// `plumb hooks uptake` over both real stores, in the shape the live Codex +// dogfood recorded (PLAN-511): a hint on the session's first prompt, before its +// first plumb call, then a session_start declaring that session and a read of +// the hinted file. The connection is unshared, so none of its calls carries a +// logical_agent; the hint must still read as consumed, through the connection +// its session alone declared. The decoy hint, for a session that declared +// nothing, is the control: it stays unattributed. +func TestHooksUptake_FirstPromptHintOnAnUnsharedConnectionIsConsumed(t *testing.T) { + t.Setenv("XDG_DATA_HOME", t.TempDir()) + t.Setenv("HOME", t.TempDir()) + hint := time.Now().Add(-5 * time.Minute).Truncate(time.Millisecond) + + ledger, err := contexthints.Open() + if err != nil { + t.Fatalf("open ledger: %v", err) + } + for _, sid := range []string{"codex-conv", "silent-conv"} { + if err := ledger.Record(contexthints.Observation{ + At: hint, Workspace: "/w", SessionID: sid, Host: "codex", Event: "UserPromptSubmit", + Emitted: []string{"internal/topology/indexer_calls.go"}, EmittedBytes: 955, + Outcome: contexthints.OutcomeEmitted, Detail: "cwd-root", + }); err != nil { + t.Fatalf("record observation: %v", err) + } + } + ledger.Close() + + sdb, err := stats.Open() + if err != nil { + t.Fatalf("open stats: %v", err) + } + // The declaration is recorded the way the daemon records it, through + // statsToolData: if session_start's session_id were ever redacted there, + // every unshared connection would fall back to unattributed, silently. + declared, _ := statsToolData("session_start", json.RawMessage(`{"detail":"brief","session_id":"codex-conv"}`), "orientation") + for i, c := range []stats.Call{ + {SessionID: "conn-1", Tool: "session_start", InputJSON: declared}, + {SessionID: "conn-1", Tool: "read_symbol", InputJSON: `{"path":"internal/topology/indexer_calls.go","name":"samePackageOutcome"}`}, + {SessionID: "conn-2", Tool: "read_symbol", InputJSON: `{"path":"internal/topology/indexer_calls.go","name":"x"}`}, + } { + c.Workspace, c.Success, c.CalledAt = "/w", true, hint.Add(time.Duration(i+1)*10*time.Second) + if err := sdb.Record(c); err != nil { + t.Fatalf("record call: %v", err) + } + } + sdb.Close() + + prevJSON := hooksUptakeJSON + hooksUptakeJSON = true + t.Cleanup(func() { hooksUptakeJSON = prevJSON }) + outPath := filepath.Join(t.TempDir(), "uptake.json") + out, err := os.Create(outPath) + if err != nil { + t.Fatal(err) + } + if err := runHooksUptake(out); err != nil { + t.Fatalf("runHooksUptake: %v", err) + } + out.Close() + raw, err := os.ReadFile(outPath) + if err != nil { + t.Fatal(err) + } + var report struct { + Hosts []contexthints.HostUptake `json:"hosts"` + } + if err := json.Unmarshal(raw, &report); err != nil { + t.Fatalf("decode report: %v\n%s", err, raw) + } + if len(report.Hosts) != 1 { + t.Fatalf("want one host line, got %+v", report.Hosts) + } + h := report.Hosts[0] + // conn-2 read the hinted file too, but no session declared it: it is + // evidence about nobody, so silent-conv's hint stays unattributed. + if h.Consumed != 1 || h.Unconsumed != 0 || h.Unattributed != 1 { + t.Errorf("codex consumption = consumed %d, not consumed %d, unattributed %d; want 1, 0, 1", + h.Consumed, h.Unconsumed, h.Unattributed) + } +} diff --git a/internal/contexthints/uptake.go b/internal/contexthints/uptake.go index c9c2f7de..22d601d0 100644 --- a/internal/contexthints/uptake.go +++ b/internal/contexthints/uptake.go @@ -7,8 +7,10 @@ package contexthints // consumed when a tool call by the SAME logical agent, soon after, mentions one // of the selectors the hint named. That is evidence of use, not of // comprehension, and it can only see plumb's own tool calls: a native read of -// the named file leaves no trace here. A hint whose agent made no attributable -// call in the window is counted as unattributed, never as unconsumed. +// the named file leaves no trace here. Which calls are the agent's is the call +// source's answer (stats.InputsForAgent: its stamped calls, and those of a +// connection it alone declared). A hint whose agent made no attributable call +// in the window is counted as unattributed, never as unconsumed. import ( "sort" @@ -29,8 +31,9 @@ type HostUptake struct { EmittedBytes int // Consumed is emitted hints a later call by the same agent mentioned; // Unconsumed had attributable calls that did not; Unattributed had no - // attributable call in the window at all (an unstamped host, or an agent - // that went quiet), so it is no evidence either way. + // attributable call in the window at all (a session that never declared its + // id or declared another one, a connection several sessions declared or + // shared, or an agent that went quiet), so it is no evidence either way. Consumed int Unconsumed int Unattributed int diff --git a/internal/stats/db_agent_inputs.go b/internal/stats/db_agent_inputs.go new file mode 100644 index 00000000..6989e51f --- /dev/null +++ b/internal/stats/db_agent_inputs.go @@ -0,0 +1,101 @@ +package stats + +// db_agent_inputs.go — the context-hint consumption proxy's read: which calls a +// logical agent made soon after a hint was sent to it. +// +// A call carries the agent in logical_agent only while plumb has seen more than +// one identity on its connection (cli connSession.attributedAgent); on a +// connection one agent holds, the connection IS the agent and the column stays +// "". A Codex connection usually serves one conversation, so before this join +// every Codex hint read as unattributed, and so did a Claude Code session +// without subagents. The connection's own session_start row says which agent +// holds it: its session_id argument is the same id the hook sends with the hint. + +import ( + "fmt" + "time" +) + +// agentInputsSQL selects the calls in (?1, ?2] that belong to agent ?3: its +// stamped calls on any connection, plus the unstamped calls of a connection +// the agent alone declared. "Alone" is what plumb can observe, and it fails +// closed on three counts: +// +// - every session_start on the connection declared a session_id (one that +// did not could be another agent arriving unannounced), +// - all of them declared the same one, and it is ?3, +// - the connection never recorded a stamped call (it was never observed +// shared). +// +// Each count looks at the connection's whole recorded life, not just the +// window: a second declaration or a stamped call AFTER the window withdraws the +// connection too, because plumb cannot tell when the other agent arrived. +// +// The limit, by construction: an agent that never identifies itself, with no +// session_start and no stamp, leaves no trace on the connection, so its calls +// are credited to the agent that did declare. That is a hook-less co-tenant on a +// shared desktop connector, or a conversation a Codex /clear started on a kept +// connection before it calls session_start. The proxy can over-credit there; it +// never credits a connection two declared agents held. Namespaces are not told +// apart either: the hint ledger records none, and the stamped path never did. +// +// The declaration may come after the hint: a session's first prompt reaches the +// hook before its first plumb call, and that first turn is the hint that +// matters most. Only connections with an unstamped call in the window are +// examined (idx_tc_called_at); the CROSS JOIN keeps win as the outer loop so +// each is probed through idx_tc_session rather than scanning every +// session_start, and each declared one is probed once more for a stamp. +const agentInputsSQL = ` +WITH win AS ( + SELECT DISTINCT session_id FROM tool_calls + WHERE called_at > ?1 AND called_at <= ?2 AND logical_agent = '' +), +starts AS ( + SELECT d.session_id, + CASE WHEN json_valid(d.input_json) THEN json_extract(d.input_json, '$.session_id') END AS declared + FROM win CROSS JOIN tool_calls d ON d.session_id = win.session_id + WHERE d.tool = 'session_start' +), +declared_by AS ( + SELECT session_id FROM starts + GROUP BY session_id + HAVING COUNT(*) = COUNT(declared) + AND COUNT(DISTINCT declared) = 1 + AND MAX(declared) = ?3 +), +held AS ( + SELECT b.session_id FROM declared_by b + WHERE NOT EXISTS (SELECT 1 FROM tool_calls s + WHERE s.session_id = b.session_id AND s.logical_agent <> '') +) +SELECT c.input_json FROM tool_calls c + WHERE c.called_at > ?1 AND c.called_at <= ?2 + AND (c.logical_agent = ?3 OR c.session_id IN (SELECT session_id FROM held)) + ORDER BY c.called_at, c.id + LIMIT ?4` + +// InputsForAgent returns the input_json of the first limit calls the logical +// agent made in (from, to], oldest first: its stamped calls, and the unstamped +// calls of a connection it alone declared (agentInputsSQL). It is the +// context-hint consumption proxy's read: did a call soon after a hint mention +// what the hint named. An agent of "" matches nothing — an unattributed row is +// no evidence about any one agent. nil-safe. +func (d *DB) InputsForAgent(agent string, from, to time.Time, limit int) ([]string, error) { + if d == nil || agent == "" || limit <= 0 { + return nil, nil + } + rows, err := d.db.Query(agentInputsSQL, from.UnixMilli(), to.UnixMilli(), agent, limit) + if err != nil { + return nil, fmt.Errorf("stats: inputs for agent: %w", err) + } + defer rows.Close() + var out []string + for rows.Next() { + var in string + if err := rows.Scan(&in); err != nil { + return nil, fmt.Errorf("stats: inputs for agent: %w", err) + } + out = append(out, in) + } + return out, rows.Err() +} diff --git a/internal/stats/db_agent_inputs_test.go b/internal/stats/db_agent_inputs_test.go new file mode 100644 index 00000000..9879fa67 --- /dev/null +++ b/internal/stats/db_agent_inputs_test.go @@ -0,0 +1,165 @@ +package stats + +import ( + "slices" + "strings" + "testing" + "time" +) + +// agentInputsFixture records one hint window's worth of calls across every +// connection shape the consumption proxy must tell apart. Each call's input +// names its connection, so a result list says exactly which rows were credited. +func agentInputsFixture(t *testing.T) (*DB, time.Time) { + t.Helper() + db := openFailuresDB(t) + hint := time.Date(2026, 10, 12, 10, 44, 0, 0, time.UTC) + rec := func(sid, agent, tool, in string, at time.Duration) { + t.Helper() + c := Call{Workspace: "/w", SessionID: sid, LogicalAgent: agent, Tool: tool, InputJSON: in, Success: true, CalledAt: hint.Add(at)} + if err := db.Record(c); err != nil { + t.Fatalf("Record: %v", err) + } + } + declare := func(sid, conv string, at time.Duration) { + rec(sid, "", "session_start", `{"detail":"brief","session_id":"`+conv+`"}`, at) + } + // solo: the Codex shape. The first prompt's hint comes BEFORE the agent's + // first plumb call, so its declaration lands inside the window, after the hint. + declare("solo", "conv-A", 5*time.Second) + rec("solo", "", "read_symbol", `{"conn":"solo"}`, 10*time.Second) + // early: a Claude Code session without subagents that declared long before. + declare("early", "conv-A", -time.Hour) + rec("early", "", "read_file", `{"conn":"early"}`, 20*time.Second) + // stamped: a shared connection's call carrying the agent itself. + rec("stamped", "conv-A", "read_file", `{"conn":"stamped"}`, 25*time.Second) + // two: one connection declared by two conversations, whose unstamped calls + // could be either's. + declare("two", "conv-A", -time.Minute) + declare("two", "conv-B", -30*time.Second) + rec("two", "", "read_file", `{"conn":"two"}`, 30*time.Second) + // undeclared: a session_start that named no one, so somebody unannounced + // may hold the connection too. + declare("undeclared", "conv-A", -time.Minute) + rec("undeclared", "", "session_start", `{"detail":"brief"}`, -30*time.Second) + rec("undeclared", "", "read_file", `{"conn":"undeclared"}`, 35*time.Second) + // shared: declared by conv-A alone, but it carried a stamped call (outside + // the window), so its unstamped calls belong to no one in particular. + declare("shared", "conv-A", -time.Minute) + rec("shared", "conv-B", "read_file", `{"conn":"shared-stamped-B"}`, -2*time.Hour) + rec("shared", "", "read_file", `{"conn":"shared"}`, 40*time.Second) + // stamped-later: declared by conv-A alone, but a stamped call AFTER the + // window shows it was shared; plumb cannot tell when the other agent came. + declare("stamped-later", "conv-A", -time.Minute) + rec("stamped-later", "", "read_file", `{"conn":"stamped-later"}`, 42*time.Second) + rec("stamped-later", "conv-B", "read_file", `{"conn":"stamped-later-B"}`, time.Hour) + // cleared-later: a second conversation declared on the same connection AFTER + // the window (the Codex /clear shape), so the window's calls are nobody's. + declare("cleared-later", "conv-A", -time.Minute) + rec("cleared-later", "", "read_file", `{"conn":"cleared-later"}`, 43*time.Second) + declare("cleared-later", "conv-B", time.Hour) + // other: a connection another conversation holds. + declare("other", "conv-B", -time.Minute) + rec("other", "", "read_file", `{"conn":"other"}`, 45*time.Second) + // garbage: an unparseable session_start input cannot declare anyone, and it + // must not fail the whole query either. + declare("garbage", "conv-A", -time.Minute) + rec("garbage", "", "session_start", `not json`, -30*time.Second) + rec("garbage", "", "read_file", `{"conn":"garbage"}`, 50*time.Second) + // Bounds: a call AT the hint is not after it; one at the window's end is in. + rec("solo", "", "read_file", `{"conn":"solo-at-hint"}`, 0) + rec("solo", "", "read_file", `{"conn":"solo-at-end"}`, 10*time.Minute) + rec("solo", "", "read_file", `{"conn":"solo-after-end"}`, 10*time.Minute+time.Millisecond) + return db, hint +} + +func TestInputsForAgent_CreditsTheConnectionTheAgentAloneDeclared(t *testing.T) { + db, hint := agentInputsFixture(t) + got, err := db.InputsForAgent("conv-A", hint, hint.Add(10*time.Minute), 50) + if err != nil { + t.Fatalf("InputsForAgent: %v", err) + } + want := []string{ + `{"detail":"brief","session_id":"conv-A"}`, // solo's own declaration, inside the window + `{"conn":"solo"}`, + `{"conn":"early"}`, + `{"conn":"stamped"}`, + `{"conn":"solo-at-end"}`, + } + if !slices.Equal(got, want) { + t.Errorf("inputs credited to conv-A:\n got %q\nwant %q", got, want) + } +} + +func TestInputsForAgent_AnotherAgentGetsOnlyItsOwnRows(t *testing.T) { + db, hint := agentInputsFixture(t) + got, err := db.InputsForAgent("conv-B", hint, hint.Add(10*time.Minute), 50) + if err != nil { + t.Fatalf("InputsForAgent: %v", err) + } + // other is conv-B's alone. two is shared by both declarations and shared is + // stamped, so neither is credited to conv-B either. + if want := []string{`{"conn":"other"}`}; !slices.Equal(got, want) { + t.Errorf("inputs credited to conv-B:\n got %q\nwant %q", got, want) + } +} + +func TestInputsForAgent_LimitKeepsTheOldest(t *testing.T) { + db, hint := agentInputsFixture(t) + got, err := db.InputsForAgent("conv-A", hint, hint.Add(10*time.Minute), 2) + if err != nil { + t.Fatalf("InputsForAgent: %v", err) + } + if want := []string{`{"detail":"brief","session_id":"conv-A"}`, `{"conn":"solo"}`}; !slices.Equal(got, want) { + t.Errorf("limit 2:\n got %q\nwant %q", got, want) + } +} + +// The query runs once per emitted hint over the daemon's whole stats.db, so its +// plan is part of its contract: each window connection's session_start rows +// are reached through idx_tc_session, never by walking every session_start in +// the table. A reshaped join or WHERE drops the index without failing anything +// else. +func TestInputsForAgent_PlanProbesTheSessionIndex(t *testing.T) { + db, hint := agentInputsFixture(t) + rows, err := db.db.Query("EXPLAIN QUERY PLAN "+agentInputsSQL, + hint.UnixMilli(), hint.Add(10*time.Minute).UnixMilli(), "conv-A", 20) + if err != nil { + t.Fatalf("explain: %v", err) + } + defer rows.Close() + var plan []string + for rows.Next() { + var id, parent, notUsed int + var detail string + if err := rows.Scan(&id, &parent, ¬Used, &detail); err != nil { + t.Fatalf("scan plan: %v", err) + } + plan = append(plan, detail) + } + if err := rows.Err(); err != nil { + t.Fatalf("plan rows: %v", err) + } + if !slices.Contains(plan, "SEARCH d USING INDEX idx_tc_session (session_id=?)") { + t.Errorf("session_start rows are not probed through idx_tc_session; plan:\n%s", strings.Join(plan, "\n")) + } +} + +func TestInputsForAgent_NoAgentNoLimitNoDB(t *testing.T) { + db, hint := agentInputsFixture(t) + for _, tc := range []struct { + name string + db *DB + agent string + limit int + }{ + {"empty agent", db, "", 50}, + {"zero limit", db, "conv-A", 0}, + {"nil db", nil, "conv-A", 50}, + } { + got, err := tc.db.InputsForAgent(tc.agent, hint, hint.Add(10*time.Minute), tc.limit) + if err != nil || got != nil { + t.Errorf("%s: got %q, %v; want nothing", tc.name, got, err) + } + } +} diff --git a/internal/stats/db_query.go b/internal/stats/db_query.go index 6ef4eb58..a6210992 100644 --- a/internal/stats/db_query.go +++ b/internal/stats/db_query.go @@ -476,36 +476,6 @@ func (d *DB) CallDetail(workspace, sessionID string, calledAt time.Time) (inputJ return } -// InputsForAgent returns the input_json of the first limit calls the logical -// agent made in (from, to], oldest first. It is the context-hint consumption -// proxy's read: did a call soon after a hint mention what the hint named. An -// agent of "" matches nothing — an unattributed row is no evidence about any -// one agent. nil-safe. -func (d *DB) InputsForAgent(agent string, from, to time.Time, limit int) ([]string, error) { - if d == nil || agent == "" || limit <= 0 { - return nil, nil - } - rows, err := d.db.Query( - `SELECT input_json FROM tool_calls - WHERE called_at > ? AND called_at <= ? AND logical_agent = ? - ORDER BY called_at LIMIT ?`, - from.UnixMilli(), to.UnixMilli(), agent, limit, - ) - if err != nil { - return nil, fmt.Errorf("stats: inputs for agent: %w", err) - } - defer rows.Close() - var out []string - for rows.Next() { - var in string - if err := rows.Scan(&in); err != nil { - return nil, fmt.Errorf("stats: inputs for agent: %w", err) - } - out = append(out, in) - } - return out, rows.Err() -} - // TotalCalls returns the total number of recorded calls matching filter. func (d *DB) TotalCalls(filter Filter) int64 { if d == nil {