fix(hooks): uptake credits a hint's use on a connection its session alone declared

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

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

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.`,

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)
+	}
+}

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

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()
+}