fix(turn): keep eBPF-sourced turns whose opening call was missed

server/h-common/src/internal_metrics.rs

@@ -208,6 +208,14 @@ define_metrics! {
     TurnClosedByGrace        => { kind: Counter, group: Turn, short: "turns_closed_grace"            },
     TurnClosedByIdle         => { kind: Counter, group: Turn, short: "turns_closed_idle"             },
     TurnDiscardedNoUserStart => { kind: Counter, group: Turn, short: "turns_discarded_no_user_start" },
+    // A partition with no main-agent `is_user_turn_start` would normally be
+    // discarded (see `TurnDiscardedNoUserStart`), but when its calls carry a
+    // common eBPF process attribution (`call.process`), they are provably one
+    // process's contiguous agent traffic — usually a turn whose opening call
+    // was missed by the eBPF source (connection-setup / uprobe-attach timing).
+    // Counted here so the fallback's hit rate is observable; it indirectly
+    // surfaces the eBPF miss rate for opening calls.
+    TurnKeptByPidAttribution => { kind: Counter, group: Turn, short: "turns_kept_by_pid_attribution" },
     TurnHeartbeatsReceived   => { kind: Counter, group: Turn, short: "turn_heartbeats_received"      },
     TurnHeartbeatsDropped    => { kind: Counter, group: Turn, short: "turn_heartbeats_dropped"       },
     TurnActive               => { kind: Gauge,   group: Turn, short: "turn_calls_buffered"           },

server/h-llm/src/wire_apis/anthropic.rs

@@ -1464,4 +1464,87 @@ mod route_classification_tests {
         let p2 = ParsedJson::from_bytes(bytes::Bytes::copy_from_slice(with_sys.as_bytes()));
         assert!(api.matches_shape(&r, &p2));
     }
+
+    // ─── Claude Code tool-loop continuation shape ───────────────────────────
+    //
+    // Background — a production defect observed on a live Claude Code session:
+    // one PID issued 8 same-session calls (tool_use×7 → end_turn), all with
+    // turn_id=None. Root cause (traced end-to-end): the turn's OPENING call
+    // (the only call whose `messages[-1]` is user text, hence the only one
+    // that reports `is_user_turn_start=Some(true)`) was missed by the eBPF
+    // source (connection-setup / uprobe-attach timing). Every captured call
+    // is a CONTINUATION: Claude Code is stateless, so each request carries
+    // full `messages` history plus the current roundtrip's tool_result, and
+    // the last user message is a `tool_result`, not the opening text.
+    //
+    // So `is_user_turn_start` correctly returns `Some(false)` for every
+    // captured call — the logic is right, the opening call just isn't among
+    // them. The defect is that the tracker then had no `has_user_start` and
+    // dropped the whole partition. The fix lives in the tracker
+    // (`partition_has_pid_attribution` fallback in `emit_or_discard`): when
+    // no main-agent user-start survived but the calls share eBPF process
+    // attribution, the partition is kept (`TurnKeptByPidAttribution`).
+    //
+    // These tests pin the CONTINUATION shape's verdict (false) so the
+    // tracker-side fix can rely on it, plus the two control cases that must
+    // not regress. See `tracker::tests::pid_attribution_keeps_turn_*`.
+    fn cc_continuation_call_request() -> Value {
+        // A continuation call: opening user text at messages[0] (carried as
+        // history), then a tool roundtrip, ending on user:tool_result.
+        serde_json::json!({
+            "messages": [
+                {"role":"user","content":[
+                    {"type":"text","text":"update the gateway config to route model A to model B"}
+                ]},
+                {"role":"assistant","content":[
+                    {"type":"text","text":"I'll edit the config."},
+                    {"type":"tool_use","id":"toolu_1","name":"Edit","input":{}}
+                ]},
+                {"role":"user","content":[
+                    {"type":"tool_result","tool_use_id":"toolu_1","content":"ok"}
+                ]}
+            ]
+        })
+    }
+
+    #[test]
+    fn is_user_turn_start_true_for_plain_text_opening() {
+        // Control case: the simple shape the current logic was designed for —
+        // a single user text message, no tool roundtrip. This MUST stay true.
+        let req = serde_json::json!({
+            "messages":[{"role":"user","content":[{"type":"text","text":"hello"}]}]
+        });
+        assert_eq!(is_user_turn_start(&req), Some(true));
+    }
+
+    #[test]
+    fn is_user_turn_start_false_for_pure_tool_result_continuation() {
+        // Control case: a pure continuation — last user message is ONLY a
+        // tool_result (a prior turn's tool_use getting its result). This MUST
+        // stay false; it is not a turn start.
+        let req = serde_json::json!({
+            "messages":[
+                {"role":"user","content":[{"type":"text","text":"old question"}]},
+                {"role":"assistant","content":[{"type":"tool_use","id":"t_old","name":"Read","input":{}}]},
+                {"role":"user","content":[{"type":"tool_result","tool_use_id":"t_old","content":"data"}]}
+            ]
+        });
+        assert_eq!(is_user_turn_start(&req), Some(false));
+    }
+
+    #[test]
+    fn is_user_turn_start_false_for_cc_tool_loop_continuation() {
+        // A Claude Code CONTINUATION call: opening user text sits at
+        // messages[0] (carried as history), but the messages array ends on a
+        // tool_result (the current roundtrip's result). This is NOT the turn's
+        // opening call — that call was missed by the eBPF source — so
+        // `is_user_turn_start` correctly returns Some(false). The fix for the
+        // production drop is NOT here (the verdict is right); it is the
+        // tracker's `partition_has_pid_attribution` fallback in
+        // `emit_or_discard`, which keeps the partition via the shared PID when
+        // no main-agent user-start survived. This test pins the verdict the
+        // tracker fix relies on.
+        let req = cc_continuation_call_request();
+        assert_eq!(is_user_turn_start(&req), Some(false));
+    }
 }

server/h-turn/src/stage.rs

@@ -70,6 +70,7 @@ pub fn spawn_turn_stage(
                 Metric::TurnClosedByGrace,
                 Metric::TurnClosedByIdle,
                 Metric::TurnDiscardedNoUserStart,
+                Metric::TurnKeptByPidAttribution,
                 Metric::TurnHeartbeatsReceived,
             ],
         );

server/h-turn/src/test_support.rs

@@ -27,6 +27,7 @@ fn rollup_metrics() -> MetricsWorker {
             Metric::TurnClosedByGrace,
             Metric::TurnClosedByIdle,
             Metric::TurnDiscardedNoUserStart,
+            Metric::TurnKeptByPidAttribution,
             Metric::TurnHeartbeatsReceived,
         ],
     );

server/h-turn/src/tracker.rs

@@ -296,14 +296,18 @@ impl TurnTracker {
             return;
         }
         // Same discard rule as emit_or_discard — only show in-progress rows
-        // whose partition has a main-agent user_turn_start. Without this, a
-        // sub-agent dispatch's first call (which carries
-        // `is_user_turn_start=Some(true)` AND `subagent_name=Some(_)`) would
-        // produce a phantom row that finalize will eventually drop.
-        let has_user_start = partition.iter().any(|ic| {
-            ic.agent.subagent_name.is_none() && ic.agent.is_user_turn_start == Some(true)
-        });
-        if !has_user_start {
+        // whose partition has a main-agent user_turn_start, OR (PID fallback)
+        // whose calls share eBPF process attribution when the turn's opening
+        // call was missed. Without this, a sub-agent dispatch's first call
+        // (which carries `is_user_turn_start=Some(true)` AND
+        // `subagent_name=Some(_)`) would produce a phantom row that finalize
+        // will eventually drop — the sub-agent guard in both predicates still
+        // filters those out. Metric is bumped only at finalize (emit_or_discard),
+        // not here — this refresh runs on every ingest and would double-count.
+        let has_user_start = partition.iter().any(|ic| main_agent_user_turn_start(ic));
+        let kept_by_pid =
+            !has_user_start && partition_has_pid_attribution(partition.iter().copied());
+        if !(has_user_start || kept_by_pid) {
             return;
         }
 
@@ -666,6 +670,30 @@ fn finalize_session(
     }
 }
 
+/// A main-agent call that opens a user turn: `is_user_turn_start == Some(true)`
+/// AND not a sub-agent. The sub-agent guard prevents a sub-agent dispatch
+/// (whose body looks like a fresh user message) from satisfying a partition's
+/// "needs a main-agent user start" requirement on its own.
+fn main_agent_user_turn_start(ic: &AgentCall) -> bool {
+    ic.agent.subagent_name.is_none() && ic.agent.is_user_turn_start == Some(true)
+}
+
+/// Whether any *main-agent* call in the partition carries eBPF process
+/// attribution (`call.process.is_some()`). Used as the `has_user_start`
+/// fallback: when no main-agent user-turn-start survived (the turn's opening
+/// call was missed by the eBPF source), a shared PID still proves these are
+/// one process's contiguous agent traffic, so the partition is kept rather
+/// than discarded. Sub-agent calls are excluded — a sub-agent-only partition
+/// must still be dropped even if the calls happen to carry a PID.
+fn partition_has_pid_attribution<'a, I>(calls: I) -> bool
+where
+    I: IntoIterator<Item = &'a AgentCall>,
+{
+    calls
+        .into_iter()
+        .any(|ic| ic.agent.subagent_name.is_none() && ic.call.process.is_some())
+}
+
 /// Apply the discard rule and emit (or count-and-drop) one turn per drained
 /// partition. The caller has already removed `calls` from the buffer and
 /// updated bookkeeping. Pass `turn_id_override` to reuse the in-progress
@@ -699,15 +727,28 @@ fn emit_or_discard(
     // `is_user_turn_start=Some(true)` *and* `subagent_name=Some(_)`; without
     // the sub-agent guard here those would slip through and produce phantom
     // "Incomplete with user_input=None" turns.
-    let has_user_start = calls.iter().any(|bc| {
-        bc.ic.agent.subagent_name.is_none() && bc.ic.agent.is_user_turn_start == Some(true)
-    });
-    if !has_user_start {
+    let refs: Vec<&AgentCall> = calls.iter().map(|bc| &bc.ic).collect();
+    let has_user_start = refs
+        .iter()
+        .any(|ic| main_agent_user_turn_start(ic));
+    // PID fallback: when no main-agent user-turn-start survived into this
+    // partition (typically because the eBPF source missed the turn's opening
+    // call — the only call whose `messages[-1]` is user text rather than a
+    // tool_result), a shared process attribution on the partition's calls is
+    // proof enough that these are one process's contiguous agent traffic. Keep
+    // the partition rather than dropping it; the opening call is gone but the
+    // rest of the turn is real and would otherwise be lost entirely. See
+    // `TurnKeptByPidAttribution` for the hit rate. Passive-tap calls carry no
+    // `process`, so this fallback is inert for them — behavior unchanged.
+    let kept_by_pid = !has_user_start && partition_has_pid_attribution(refs.iter().copied());
+    if kept_by_pid {
+        metrics.counter(Metric::TurnKeptByPidAttribution).inc();
+    }
+    if !(has_user_start || kept_by_pid) {
         metrics.counter(Metric::TurnDiscardedNoUserStart).inc();
         return false;
     }
 
-    let refs: Vec<&AgentCall> = calls.iter().map(|bc| &bc.ic).collect();
     let mut turn = build_turn(&refs, turn_id_override, None);
     // The buffer key is authoritative for source/session — call.source_id
     // can legitimately be empty in tests; identity.session_id may differ
@@ -966,6 +1007,7 @@ mod tests {
                 Metric::TurnClosedByGrace,
                 Metric::TurnClosedByIdle,
                 Metric::TurnDiscardedNoUserStart,
+                Metric::TurnKeptByPidAttribution,
                 Metric::TurnHeartbeatsReceived,
             ],
         );
@@ -1313,6 +1355,103 @@ mod tests {
         );
     }
 
+    // Regression: an eBPF-captured agent turn whose OPENING call was missed.
+    //
+    // Claude Code is stateless — every request carries full `messages`
+    // history. The turn's opening call (the one whose `messages[-1]` is user
+    // text, hence `is_user_turn_start=true`) is the ONLY call in the turn
+    // that reports a user-start. When the eBPF source misses it (connection-
+    // setup / uprobe-attach timing — observed in production: one PID issued
+    // 8 same-session calls tool_use×7 → end_turn, all turn_id=None), every
+    // captured call is a continuation whose `messages[-1]` is a tool_result,
+    // so `is_user_turn_start=false` for all of them. The partition has no
+    // main-agent user-start → `emit_or_discard` would discard the whole turn.
+    //
+    // But the calls DO carry a shared `process` (eBPF attribution), which is
+    // proof they are one process's contiguous agent traffic. The PID fallback
+    // in `has_user_start` keeps the partition; `TurnKeptByPidAttribution` is
+    // bumped. Turn boundaries are unaffected (still driven by `is_turn_terminal`).
+    fn anthropic_continuation_call(
+        session: &str,
+        request_time_us: i64,
+        finish: &str,
+        with_pid: bool,
+    ) -> LlmCall {
+        let mut c = anthropic_call(session, request_time_us, "tool_result", finish);
+        if with_pid {
+            c.process = Some(h_common::process::ProcessInfo::new(12345, "agent-cli"));
+        }
+        c
+    }
+
+    #[test]
+    fn pid_attribution_keeps_turn_when_opening_call_missed() {
+        let metrics = test_metrics();
+        let mut t =
+            TurnTracker::new(TrackerConfig::default(), metrics.clone());
+
+        // Two continuation calls (tool_result bodies → is_user_turn_start=false),
+        // both carrying the same eBPF PID. The second closes the turn (end_turn).
+        let c1 = anthropic_continuation_call("S", 1_000_000, "tool_use", true);
+        let c2 = anthropic_continuation_call("S", 2_000_000, "end_turn", true);
+        let id1 = call_info_for_anthropic(&c1);
+        let id2 = call_info_for_anthropic(&c2);
+        // Sanity: neither call reports a user-turn-start — the precondition
+        // for the discard that the fallback must override.
+        assert_eq!(id1.is_user_turn_start, Some(false));
+        assert_eq!(id2.is_user_turn_start, Some(false));
+
+        let mut events = t.ingest(ic(c1, id1));
+        events.extend(t.ingest(ic(c2, id2)));
+        events.extend(t.flush_all());
+        let turns = drain_completed(events);
+
+        assert_eq!(turns.len(), 1, "PID fallback must keep the partition");
+        assert_eq!(turns[0].call_count, 2);
+        assert_eq!(
+            metrics.counter(Metric::TurnKeptByPidAttribution).get(),
+            1,
+            "fallback path must bump the attribution counter"
+        );
+        assert_eq!(
+            metrics.counter(Metric::TurnDiscardedNoUserStart).get(),
+            0,
+            "partition must not be counted as discarded"
+        );
+    }
+
+    #[test]
+    fn no_pid_no_user_start_still_discarded() {
+        // Control: identical continuation-only partition, but no eBPF PID
+        // (passive-tap calls carry `process=None`). The fallback must NOT
+        // fire — a partition with no user-start and no process attribution
+        // is still discarded, exactly as before.
+        let metrics = test_metrics();
+        let mut t =
+            TurnTracker::new(TrackerConfig::default(), metrics.clone());
+
+        let c1 = anthropic_continuation_call("S", 1_000_000, "tool_use", false);
+        let c2 = anthropic_continuation_call("S", 2_000_000, "end_turn", false);
+        let id1 = call_info_for_anthropic(&c1);
+        let id2 = call_info_for_anthropic(&c2);
+
+        let mut events = t.ingest(ic(c1, id1));
+        events.extend(t.ingest(ic(c2, id2)));
+        events.extend(t.flush_all());
+        let turns = drain_completed(events);
+
+        assert!(turns.is_empty(), "passive-tap partition must be discarded");
+        assert_eq!(
+            metrics.counter(Metric::TurnKeptByPidAttribution).get(),
+            0,
+            "fallback must not fire without PID"
+        );
+        assert_eq!(
+            metrics.counter(Metric::TurnDiscardedNoUserStart).get(),
+            1,
+        );
+    }
+
     #[test]
     fn subagent_assistant_text_does_not_leak_to_parent_final_answer() {
         let mut t = mk_tracker_no_grace();

server/h-turn/tests/integration.rs

@@ -881,6 +881,7 @@ async fn generic_profile_anthropic_two_call_session() {
             Metric::TurnClosedByGrace,
             Metric::TurnClosedByIdle,
             Metric::TurnDiscardedNoUserStart,
+            Metric::TurnKeptByPidAttribution,
         ],
     );
     let _turn_svc = turn_sys.start();

server/h-turn/tests/reorder.rs

@@ -34,6 +34,7 @@ fn test_metrics() -> MetricsWorker {
             Metric::TurnClosedByGrace,
             Metric::TurnClosedByIdle,
             Metric::TurnDiscardedNoUserStart,
+            Metric::TurnKeptByPidAttribution,
             Metric::TurnHeartbeatsReceived,
         ],
     );