| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "slices" |
| 9 | ) |
| 10 | |
| 11 | // trajectorySummary is the harness-side digest of one run's trajectory file: |
| 12 | // where the wall clock went, split between tool execution and everything the |
| 13 | // model spent between calls (thinking, streaming, provider latency). |
| 14 | type trajectorySummary struct { |
| 15 | Path string `json:"path"` |
| 16 | Records int `json:"records"` |
| 17 | SpanMs int64 `json:"span_ms"` |
| 18 | ToolMs int64 `json:"tool_ms"` |
| 19 | ModelMs int64 `json:"model_ms"` // SpanMs − tool wall clock, floored at zero |
| 20 | // Round decomposition: a round's gap runs from the turn start (or the |
| 21 | // batch's last tool_result) to the next top-level tool_dispatch; the |
| 22 | // final segment to the last record is the answer round. |
| 23 | ModelRounds int `json:"model_rounds"` |
| 24 | ModelGapTotalMs int64 `json:"model_gap_total_ms"` |
| 25 | ModelGapP95Ms int64 `json:"model_gap_p95_ms"` |
| 26 | Retries int `json:"retries,omitempty"` |
| 27 | Compactions int `json:"compactions,omitempty"` |
| 28 | |
| 29 | // Batch decomposition: one batch is one round's top-level tool calls. |
| 30 | // ToolWallMs is the union of execution intervals (parallel calls counted |
| 31 | // once) plus durations of calls that carried no timestamps. |
| 32 | ToolWallMs int64 `json:"tool_wall_ms,omitempty"` |
| 33 | ToolBatches int `json:"tool_batches,omitempty"` |
| 34 | TopLevelCalls int `json:"top_level_calls,omitempty"` |
| 35 | MaxBatchSize int `json:"max_batch_size,omitempty"` |
| 36 | ParallelBatches int `json:"parallel_batches,omitempty"` // ≥2 calls actually overlapped |
| 37 | ParallelSavedMs int64 `json:"parallel_saved_ms,omitempty"` // Σ durations − batch wall |
| 38 | SingleReadRounds int `json:"single_read_rounds,omitempty"` // 1-call read-only batches |
| 39 | SingleReadStreak int `json:"single_read_streak,omitempty"` // longest consecutive run |
| 40 | StartDelayP95Ms int64 `json:"start_delay_p95_ms,omitempty"` // dispatch→start, in-batch queue |
| 41 | |
| 42 | // Recovery decomposition: a round whose gap contained a provider retry, |
| 43 | // missing-reasoning replay, or empty-final retry is a recovery round; clean |
| 44 | // p95 excludes them so adapter flakiness reads as adapter cost, not agent. |
| 45 | StreamRetries int `json:"stream_retries,omitempty"` |
| 46 | HeaderRetries int `json:"header_retries,omitempty"` |
| 47 | ReasoningReplays int `json:"reasoning_replays,omitempty"` // extra exact-replay requests |
| 48 | EmptyFinalRetries int `json:"empty_final_retries,omitempty"` |
| 49 | RecoveryRounds int `json:"recovery_rounds,omitempty"` |
| 50 | RecoveryGapMs int64 `json:"recovery_gap_ms,omitempty"` |
| 51 | CleanGapP95Ms int64 `json:"clean_gap_p95_ms,omitempty"` |
| 52 | |
| 53 | // Wall decomposition: disjoint buckets partitioning the span, allocated by |
| 54 | // priority (tools > retry backoff > compaction > model streaming with the |
| 55 | // planner split out > agent overhead), so overlaps are never double-booked. |
| 56 | RetryWaitMs int64 `json:"retry_wait_ms,omitempty"` // retrying → next attempt begin |
| 57 | CompactionMs int64 `json:"compaction_ms,omitempty"` // compaction_started → done |
| 58 | PlannerStreamMs int64 `json:"planner_stream_ms,omitempty"` // attempts closed by planner usage |
| 59 | ModelStreamMs int64 `json:"model_stream_ms,omitempty"` // remaining sampling attempts |
| 60 | AgentOtherMs int64 `json:"agent_other_ms,omitempty"` // span remainder: assembly, guards, idle |
| 61 | |
| 62 | // Phase-trace inputs: content-free firsts and counts for the per-task trace. |
| 63 | TTFTMs int64 `json:"ttft_ms,omitempty"` // span start → first output delta |
| 64 | FirstToolMs int64 `json:"first_tool_ms,omitempty"` // span start → first tool start |
| 65 | PlannerRequests int `json:"planner_requests,omitempty"` |
| 66 | ExecutorRequests int `json:"executor_requests,omitempty"` |
| 67 | SubagentRequests int `json:"subagent_requests,omitempty"` |
| 68 | // RequestsBySource keeps every origin honest — goal-evaluator, compaction |
| 69 | // and capability-router calls must not masquerade as executor rounds. |
| 70 | RequestsBySource map[string]int `json:"requests_by_source,omitempty"` |
| 71 | ToolQueueMs int64 `json:"tool_queue_ms,omitempty"` // Σ dispatch→start delays |
| 72 | NoProgressSignals int `json:"no_progress_signals,omitempty"` // progress_guard escalations |
| 73 | |
| 74 | // Round outcomes: each classified round's gap booked to what it produced. |
| 75 | // Productive = evidence_gain/mutation/verification/finalization; the rest |
| 76 | // accumulates into WastedGapMs — the knife-target readout. |
| 77 | RoundOutcomes map[string]int `json:"round_outcomes,omitempty"` |
| 78 | RoundOutcomeMs map[string]int64 `json:"round_outcome_ms,omitempty"` |
| 79 | UsefulRounds int `json:"useful_rounds,omitempty"` |
| 80 | WastedGapMs int64 `json:"wasted_gap_ms,omitempty"` |
| 81 | |
| 82 | // Mechanism ledger inputs: recovery gap split per taint kind, and the |
| 83 | // executor-handoff nudge count (a correctness mechanism that buys a whole |
| 84 | // extra model round each time it fires). |
| 85 | HandoffNudges int `json:"handoff_nudges,omitempty"` |
| 86 | RecoveryGapMsByKind map[string]int64 `json:"recovery_gap_ms_by_kind,omitempty"` |
| 87 | |
| 88 | // Tool surface: the schema tax every top-level request re-pays, and the |
| 89 | // surface churn (connect_tool_source calls, provider prefix resets) that |
| 90 | // trades that tax against mid-session cache invalidation. |
| 91 | SchemaTokensMax int64 `json:"schema_tokens_max,omitempty"` // largest per-request schema footprint |
| 92 | SchemaTokensTotal int64 `json:"schema_tokens_total,omitempty"` // Σ schema tokens across requests |
| 93 | PromptTokensSeen int64 `json:"prompt_tokens_seen,omitempty"` // Σ prompt tokens (schema share denominator) |
| 94 | PrefixResets int `json:"prefix_resets,omitempty"` // usage events with prefixChanged |
| 95 | ConnectCalls int `json:"connect_calls,omitempty"` // connect_tool_source dispatches |
| 96 | |
| 97 | // Cold/warm evidence: the first top-level request's cache split. A cold |
| 98 | // session pays the whole prefix as miss; a warmed one starts near-hit. |
| 99 | FirstReqCacheHitTokens int64 `json:"first_req_cache_hit_tokens,omitempty"` |
| 100 | FirstReqCacheMissTokens int64 `json:"first_req_cache_miss_tokens,omitempty"` |
| 101 | |
| 102 | // Shadow contract audit (last of the turn): what the observing contract |
| 103 | // concluded, priced against the hidden grader by the report. |
| 104 | ShadowIntent string `json:"shadow_intent,omitempty"` |
| 105 | ShadowVerdict string `json:"shadow_verdict,omitempty"` |
| 106 | ShadowComplete bool `json:"shadow_complete,omitempty"` |
| 107 | |
| 108 | // Completion report (last of the turn): the host-authored receipt's |
| 109 | // verdict and the gaps it refused to hide, priced against the grader. |
| 110 | CompletionVerdict string `json:"completion_verdict,omitempty"` |
| 111 | CompletionGaps int `json:"completion_gaps,omitempty"` |
| 112 | CompletionGapKinds []string `json:"completion_gap_kinds,omitempty"` |
| 113 | ClaimsVerified int `json:"claims_verified,omitempty"` |
| 114 | ClaimsUnbacked int `json:"claims_unbacked,omitempty"` |
| 115 | |
| 116 | // Outcome shadow: the runtime outcome scorer's per-round series condensed, |
| 117 | // or a verification-receipt backfill for recordings that predate it. |
| 118 | Outcome *outcomeSummary `json:"outcome,omitempty"` |
| 119 | |
| 120 | // Anchor-safety shadow: content-free interval-fingerprint decisions. The |
| 121 | // trajectory never contains paths, anchors, source text, or line hashes. |
| 122 | AnchorSafety *anchorSafetySummary `json:"anchor_safety,omitempty"` |
| 123 | |
| 124 | // Cognition: executor reasoning/completion joined per model round, plus a |
| 125 | // census of slow rounds — gaps that bought unusually large thinking. |
| 126 | ReasoningTokensTotal int64 `json:"reasoning_tokens_total,omitempty"` |
| 127 | CompletionTokensTotal int64 `json:"completion_tokens_total,omitempty"` |
| 128 | SlowRounds int `json:"slow_rounds,omitempty"` |
| 129 | SlowRoundGapMs int64 `json:"slow_round_gap_ms,omitempty"` |
| 130 | SlowRoundReasoningTokens int64 `json:"slow_round_reasoning_tokens,omitempty"` |
| 131 | Rounds []roundDigest `json:"rounds,omitempty"` |
| 132 | |
| 133 | // Delegation admission shadow: verdicts recorded by the runtime, and the |
| 134 | // subagent time spent by tools the shadow would have denied. |
| 135 | DelegationCalls int `json:"delegation_calls,omitempty"` |
| 136 | DelegationDenies int `json:"delegation_denies,omitempty"` |
| 137 | DeniedDelegationMs int64 `json:"denied_delegation_ms,omitempty"` |
| 138 | } |
| 139 | |
| 140 | // toolWall is the best available tool wall-clock: interval union when the |
| 141 | // recording carried timestamps, else the duration sum (older trajectories). |
| 142 | func (s *trajectorySummary) toolWall() int64 { |
| 143 | if s.ToolWallMs > 0 { |
| 144 | return s.ToolWallMs |
| 145 | } |
| 146 | return s.ToolMs |
| 147 | } |
| 148 | |
| 149 | // roundCall is one call's outcome-relevant facts for round classification. |
| 150 | type roundCall struct { |
| 151 | name, verification string |
| 152 | readOnly, errored bool |
| 153 | resolved, dup bool |
| 154 | } |
| 155 | |
| 156 | // gapInfo carries one model gap until its batch closes and can classify it. |
| 157 | // The cognition fields are the executor tokens streamed during the gap — what |
| 158 | // the round's thinking actually bought. |
| 159 | type gapInfo struct { |
| 160 | ms int64 |
| 161 | tainted, planner, compaction, handoff bool |
| 162 | reasonTok, complTok, promptTok int64 |
| 163 | } |
| 164 | |
| 165 | // toolBatch accumulates one round's top-level calls between model gaps. |
| 166 | type toolBatch struct { |
| 167 | dispatchTS map[string]int64 |
| 168 | infos map[string]*roundCall |
| 169 | names []string |
| 170 | calls int |
| 171 | results int |
| 172 | readOnly int |
| 173 | serialMs int64 |
| 174 | intervals [][2]int64 |
| 175 | } |
| 176 | |
| 177 | // renderTimeAttribution aggregates recorded runs into one report line; empty |
| 178 | // when no run in the suite carried a trajectory. |
| 179 | func renderTimeAttribution(results []result) string { |
| 180 | var toolMs, modelMs, gapMs, savedMs, delayP95, recoveryGapMs, cleanP95 int64 |
| 181 | var retryWaitMs, compactionMs, plannerMs, modelStreamMs, agentOtherMs, startupMs int64 |
| 182 | runs, rounds, batches, calls, singleReads, parallelBatches := 0, 0, 0, 0, 0, 0 |
| 183 | recoveryRounds, streamRetries, headerRetries, replays, emptyFinals := 0, 0, 0, 0, 0 |
| 184 | for _, r := range results { |
| 185 | if r.Trajectory != nil { |
| 186 | runs++ |
| 187 | toolMs += r.Trajectory.toolWall() |
| 188 | modelMs += r.Trajectory.ModelMs |
| 189 | rounds += r.Trajectory.ModelRounds |
| 190 | gapMs += r.Trajectory.ModelGapTotalMs |
| 191 | batches += r.Trajectory.ToolBatches |
| 192 | calls += r.Trajectory.TopLevelCalls |
| 193 | singleReads += r.Trajectory.SingleReadRounds |
| 194 | parallelBatches += r.Trajectory.ParallelBatches |
| 195 | savedMs += r.Trajectory.ParallelSavedMs |
| 196 | delayP95 = max(delayP95, r.Trajectory.StartDelayP95Ms) |
| 197 | recoveryRounds += r.Trajectory.RecoveryRounds |
| 198 | recoveryGapMs += r.Trajectory.RecoveryGapMs |
| 199 | cleanP95 = max(cleanP95, r.Trajectory.CleanGapP95Ms) |
| 200 | streamRetries += r.Trajectory.StreamRetries |
| 201 | headerRetries += r.Trajectory.HeaderRetries |
| 202 | replays += r.Trajectory.ReasoningReplays |
| 203 | emptyFinals += r.Trajectory.EmptyFinalRetries |
| 204 | retryWaitMs += r.Trajectory.RetryWaitMs |
| 205 | compactionMs += r.Trajectory.CompactionMs |
| 206 | plannerMs += r.Trajectory.PlannerStreamMs |
| 207 | modelStreamMs += r.Trajectory.ModelStreamMs |
| 208 | agentOtherMs += r.Trajectory.AgentOtherMs |
| 209 | if r.WallMs > r.Trajectory.SpanMs { |
| 210 | startupMs += r.WallMs - r.Trajectory.SpanMs |
| 211 | } |
| 212 | } |
| 213 | } |
| 214 | if runs == 0 { |
| 215 | return "" |
| 216 | } |
| 217 | line := fmt.Sprintf("**Time attribution** (%d recorded runs): **tools** %s (%s) · **model** %s (%s)", |
| 218 | runs, dur(toolMs), pct(int(toolMs), int(toolMs+modelMs)), |
| 219 | dur(modelMs), pct(int(modelMs), int(toolMs+modelMs))) |
| 220 | if rounds > 0 { |
| 221 | line += fmt.Sprintf(" · **model rounds** %d (avg gap %s)", rounds, dur(gapMs/int64(rounds))) |
| 222 | } |
| 223 | if batches > 0 { |
| 224 | line += fmt.Sprintf("\n\n**Batching** (%d tool rounds): **calls/round** %.1f · **single-read rounds** %d (%s) · **parallel rounds** %d (saved %s) · **start-delay p95** %s", |
| 225 | batches, float64(calls)/float64(batches), |
| 226 | singleReads, pct(singleReads, batches), |
| 227 | parallelBatches, dur(savedMs), durMs(delayP95)) |
| 228 | } |
| 229 | if recoveryRounds+streamRetries+headerRetries+replays+emptyFinals > 0 { |
| 230 | line += fmt.Sprintf("\n\n**Recovery**: recovery rounds %d (%s of rounds, %s) · stream retries %d · header retries %d · reasoning replays %d · empty-final retries %d · clean gap p95 %s", |
| 231 | recoveryRounds, pct(recoveryRounds, rounds), dur(recoveryGapMs), |
| 232 | streamRetries, headerRetries, replays, emptyFinals, durMs(cleanP95)) |
| 233 | } |
| 234 | if plannerMs+modelStreamMs > 0 { |
| 235 | line += fmt.Sprintf("\n\n**Wall decomposition**: **startup** %s · **agent** %s · **planner** %s · **model** %s · **tools** %s · **retry** %s · **compaction** %s", |
| 236 | dur(startupMs), dur(agentOtherMs), dur(plannerMs), dur(modelStreamMs), |
| 237 | dur(toolMs), dur(retryWaitMs), dur(compactionMs)) |
| 238 | } |
| 239 | line += renderRoundEfficiency(results) |
| 240 | return line + "\n\n" |
| 241 | } |
| 242 | |
| 243 | // summarizeTrajectory reads a run's JSONL trajectory. A truncated final line |
| 244 | // (killed run) is skipped, matching the recorder's durability contract. |
| 245 | func summarizeTrajectory(path string) (*trajectorySummary, error) { |
| 246 | scan, err := scanTrajectoryFile(path) |
| 247 | if err != nil { |
| 248 | return nil, err |
| 249 | } |
| 250 | return scan.finish(), nil |
| 251 | } |
| 252 | |
| 253 | // scanTrajectoryFile runs the record pass without finishing, so callers that |
| 254 | // need the raw series (the live dashboard) can read it before finish folds it. |
| 255 | func scanTrajectoryFile(path string) (*trajScan, error) { |
| 256 | f, err := os.Open(path) |
| 257 | if err != nil { |
| 258 | return nil, err |
| 259 | } |
| 260 | defer f.Close() |
| 261 | |
| 262 | scan := &trajScan{ |
| 263 | s: &trajectorySummary{Path: path}, |
| 264 | batch: newToolBatch(), |
| 265 | attemptBegin: map[string]int64{}, |
| 266 | lastAttempt: -1, |
| 267 | seen: map[string]bool{}, |
| 268 | denyDelegations: map[string]bool{}, |
| 269 | delegationToolMs: map[string]int64{}, |
| 270 | } |
| 271 | sc := bufio.NewScanner(f) |
| 272 | sc.Buffer(make([]byte, 0, 1<<20), 16<<20) |
| 273 | for sc.Scan() { |
| 274 | var rec trajectoryRecord |
| 275 | if err := json.Unmarshal(sc.Bytes(), &rec); err != nil { |
| 276 | continue |
| 277 | } |
| 278 | scan.record(rec) |
| 279 | } |
| 280 | if err := sc.Err(); err != nil { |
| 281 | return nil, err |
| 282 | } |
| 283 | return scan, nil |
| 284 | } |
| 285 | |
| 286 | func (t *trajScan) record(rec trajectoryRecord) { |
| 287 | t.s.Records++ |
| 288 | if t.firstTS == 0 { |
| 289 | t.firstTS = rec.TS |
| 290 | t.gapStart = rec.TS |
| 291 | t.inModel = true |
| 292 | } |
| 293 | t.lastTS = rec.TS |
| 294 | if rec.ProtocolRecovery == "missing_reasoning_retry_attempted" { |
| 295 | t.s.ReasoningReplays++ |
| 296 | t.taintAs("reasoning_replay") |
| 297 | } |
| 298 | if cs := rec.ContractShadow; cs != nil { |
| 299 | t.s.ShadowIntent = cs.Intent |
| 300 | t.s.ShadowVerdict = cs.Verdict |
| 301 | t.s.ShadowComplete = cs.Complete |
| 302 | } |
| 303 | if cr := rec.CompletionReport; cr != nil { |
| 304 | t.s.CompletionVerdict = cr.Verdict |
| 305 | t.s.CompletionGaps = cr.Gaps |
| 306 | t.s.CompletionGapKinds = cr.GapKinds |
| 307 | t.s.ClaimsVerified = cr.ClaimsVerified |
| 308 | t.s.ClaimsUnbacked = cr.ClaimsUnbacked |
| 309 | } |
| 310 | if op := rec.OutcomeProgress; op != nil { |
| 311 | t.outcomePoints = append(t.outcomePoints, outcomePoint{ |
| 312 | ts: rec.TS, round: op.Round, exploration: op.Exploration, verification: op.Verification, |
| 313 | objective: op.Objective, regression: op.Regression, churn: op.Churn, |
| 314 | legacyGain: op.LegacyGain, discriminating: op.Discriminating, debtAge: op.DebtAge, |
| 315 | blindMutations: op.BlindMutations, ebmEligible: op.EBMEligible, ebmFired: op.EBMFired, |
| 316 | governorEligible: op.GovernorEligible, governorEngaged: op.GovernorEngaged, |
| 317 | runway: op.Runway, runwayDry: op.RunwayDry, runwayIdle: op.RunwayIdle, runwaySpent: op.RunwaySpent, |
| 318 | }) |
| 319 | } |
| 320 | if aa := rec.AnchorSafetyAudit; aa != nil { |
| 321 | t.recordAnchorSafetyAudit(*aa) |
| 322 | } |
| 323 | if da := rec.DelegationAdmission; da != nil { |
| 324 | t.s.DelegationCalls++ |
| 325 | if da.Verdict == "deny" { |
| 326 | t.s.DelegationDenies++ |
| 327 | t.denyDelegations[da.Tool] = true |
| 328 | } |
| 329 | } |
| 330 | if rec.Event == nil { |
| 331 | return |
| 332 | } |
| 333 | switch rec.Event.Kind { |
| 334 | case "retrying": |
| 335 | t.s.Retries++ |
| 336 | t.pendingRetry = rec.TS |
| 337 | switch rec.Event.RetryScope { |
| 338 | case "stream": |
| 339 | t.s.StreamRetries++ |
| 340 | t.taintAs("stream_retry") |
| 341 | case "headers": |
| 342 | t.s.HeaderRetries++ |
| 343 | t.taintAs("header_retry") |
| 344 | default: |
| 345 | t.taintAs("provider_retry") |
| 346 | } |
| 347 | case "notice": |
| 348 | switch rec.Event.Code { |
| 349 | case "empty_final": |
| 350 | t.s.EmptyFinalRetries++ |
| 351 | t.taintAs("empty_final_retry") |
| 352 | case "executor_handoff": |
| 353 | t.s.HandoffNudges++ |
| 354 | t.gapHandoff = true |
| 355 | case "progress_guard": |
| 356 | t.s.NoProgressSignals++ |
| 357 | } |
| 358 | case "reasoning", "text": |
| 359 | if t.firstDelta == 0 { |
| 360 | t.firstDelta = rec.TS |
| 361 | } |
| 362 | case "stream_attempt", "usage": |
| 363 | t.recordModelPhase(rec) |
| 364 | case "compaction_started": |
| 365 | t.s.Compactions++ |
| 366 | t.compFrom = rec.TS |
| 367 | t.gapCompact = true |
| 368 | case "compaction_done": |
| 369 | if t.compFrom > 0 && rec.TS > t.compFrom { |
| 370 | t.compIvs = append(t.compIvs, [2]int64{t.compFrom, rec.TS}) |
| 371 | } |
| 372 | t.compFrom = 0 |
| 373 | } |
| 374 | if rec.Event.Tool == nil || rec.Event.Tool.ParentID != "" { |
| 375 | // Subagent calls overlap the parent's wall clock; counting them |
| 376 | // would double-book the span and split the parent's rounds. |
| 377 | return |
| 378 | } |
| 379 | switch rec.Event.Kind { |
| 380 | case "tool_dispatch": |
| 381 | t.recordDispatch(rec) |
| 382 | case "tool_result": |
| 383 | t.recordResult(rec) |
| 384 | } |
| 385 | } |
| 386 | |
| 387 | // closeGap ends one model round; recovery-tainted rounds are booked apart so |
| 388 | // clean latency stays comparable across providers with different flake rates. |
| 389 | // The gap is queued until its batch closes and can classify the round. |
| 390 | func (t *trajScan) closeGap(gap int64) { |
| 391 | t.gaps = append(t.gaps, gap) |
| 392 | t.pendingGaps = append(t.pendingGaps, gapInfo{ |
| 393 | ms: gap, tainted: t.taint != "", |
| 394 | planner: t.gapPlanner, compaction: t.gapCompact, handoff: t.gapHandoff, |
| 395 | reasonTok: t.gapReason, complTok: t.gapCompl, promptTok: t.gapPrompt, |
| 396 | }) |
| 397 | t.gapPlanner, t.gapCompact, t.gapHandoff = false, false, false |
| 398 | t.gapReason, t.gapCompl, t.gapPrompt = 0, 0, 0 |
| 399 | if t.taint != "" { |
| 400 | t.s.RecoveryRounds++ |
| 401 | t.s.RecoveryGapMs += gap |
| 402 | if t.s.RecoveryGapMsByKind == nil { |
| 403 | t.s.RecoveryGapMsByKind = map[string]int64{} |
| 404 | } |
| 405 | t.s.RecoveryGapMsByKind[t.taint] += gap |
| 406 | t.taint = "" |
| 407 | return |
| 408 | } |
| 409 | t.cleanGaps = append(t.cleanGaps, gap) |
| 410 | } |
| 411 | |
| 412 | // taintAs marks the current gap as recovery; the first mechanism to fire in |
| 413 | // a gap owns its time, so per-kind splits stay disjoint. |
| 414 | func (t *trajScan) taintAs(kind string) { |
| 415 | if t.taint == "" { |
| 416 | t.taint = kind |
| 417 | } |
| 418 | } |
| 419 | |
| 420 | // recordModelPhase brackets sampling attempts (begin → commit/discard) and |
| 421 | // tags the just-closed attempt with its usage source. Subagent usage is |
| 422 | // skipped: subagent attempts never reach the parent sink, so a subagent usage |
| 423 | // arriving mid-parent-round must not claim the parent's attempt. |
| 424 | func (t *trajScan) recordModelPhase(rec trajectoryRecord) { |
| 425 | if sa := rec.Event.StreamAttempt; sa != nil { |
| 426 | switch sa.Action { |
| 427 | case "begin": |
| 428 | t.attemptBegin[sa.ID] = rec.TS |
| 429 | if t.pendingRetry > 0 && rec.TS > t.pendingRetry { |
| 430 | t.retryIvs = append(t.retryIvs, [2]int64{t.pendingRetry, rec.TS}) |
| 431 | } |
| 432 | t.pendingRetry = 0 |
| 433 | case "commit", "discard": |
| 434 | if begin, ok := t.attemptBegin[sa.ID]; ok && rec.TS > begin { |
| 435 | t.attempts = append(t.attempts, modelAttempt{iv: [2]int64{begin, rec.TS}}) |
| 436 | t.lastAttempt = len(t.attempts) - 1 |
| 437 | } |
| 438 | delete(t.attemptBegin, sa.ID) |
| 439 | } |
| 440 | return |
| 441 | } |
| 442 | if u := rec.Event.Usage; u != nil { |
| 443 | source := u.Source |
| 444 | if source == "" { |
| 445 | source = "executor" |
| 446 | } |
| 447 | if t.s.RequestsBySource == nil { |
| 448 | t.s.RequestsBySource = map[string]int{} |
| 449 | } |
| 450 | t.s.RequestsBySource[source]++ |
| 451 | switch source { |
| 452 | case "planner": |
| 453 | t.s.PlannerRequests++ |
| 454 | t.gapPlanner = true |
| 455 | case "executor": |
| 456 | t.s.ExecutorRequests++ |
| 457 | t.s.ReasoningTokensTotal += u.ReasoningTokens |
| 458 | t.s.CompletionTokensTotal += u.CompletionTokens |
| 459 | t.gapReason += u.ReasoningTokens |
| 460 | t.gapCompl += u.CompletionTokens |
| 461 | t.gapPrompt = max(t.gapPrompt, u.PromptTokens) |
| 462 | case "subagent": |
| 463 | t.s.SubagentRequests++ |
| 464 | return |
| 465 | default: |
| 466 | // Sidecar calls (goal-evaluator, compaction, capability-router) |
| 467 | // have their own prompt shape; keep them out of the executor's |
| 468 | // schema-tax and first-request accounting. |
| 469 | return |
| 470 | } |
| 471 | if t.s.PromptTokensSeen == 0 && u.PromptTokens > 0 { |
| 472 | t.s.FirstReqCacheHitTokens = u.CacheHitTokens |
| 473 | t.s.FirstReqCacheMissTokens = u.CacheMissTokens |
| 474 | } |
| 475 | t.s.PromptTokensSeen += u.PromptTokens |
| 476 | if d := u.CacheDiagnostics; d != nil { |
| 477 | t.s.SchemaTokensTotal += d.ToolSchemaTokens |
| 478 | t.s.SchemaTokensMax = max(t.s.SchemaTokensMax, d.ToolSchemaTokens) |
| 479 | if d.PrefixChanged { |
| 480 | t.s.PrefixResets++ |
| 481 | } |
| 482 | } |
| 483 | if t.lastAttempt >= 0 { |
| 484 | t.attempts[t.lastAttempt].planner = u.Source == "planner" |
| 485 | t.lastAttempt = -1 |
| 486 | } |
| 487 | } |
| 488 | } |
| 489 | |
| 490 | func (t *trajScan) recordDispatch(rec trajectoryRecord) { |
| 491 | tl := rec.Event.Tool |
| 492 | if t.inModel { |
| 493 | t.closeGap(rec.TS - t.gapStart) |
| 494 | t.inModel = false |
| 495 | t.closeBatch() |
| 496 | } |
| 497 | // Refreshed dispatches re-announce a call already counted; |
| 498 | // id-less records (older recordings) cannot be deduped. |
| 499 | if tl.Refreshed { |
| 500 | return |
| 501 | } |
| 502 | if tl.ID == "" || !t.batch.seen(tl.ID) { |
| 503 | t.batch.calls++ |
| 504 | } |
| 505 | // The full dispatch re-announces a streamed partial; keeping the later TS |
| 506 | // anchors start-delay to pre-exec queueing, not the stream tail. The dup |
| 507 | // check keys on the latest (fullest) name+args announcement. |
| 508 | if tl.ID != "" { |
| 509 | t.sawCallIDs = true |
| 510 | t.batch.dispatchTS[tl.ID] = rec.TS |
| 511 | key := tl.Name + "\x00" + tl.Args |
| 512 | info := t.batch.infos[tl.ID] |
| 513 | if info == nil { |
| 514 | info = &roundCall{} |
| 515 | t.batch.infos[tl.ID] = info |
| 516 | t.batch.names = append(t.batch.names, tl.Name) |
| 517 | if tl.Name == "connect_tool_source" { |
| 518 | t.s.ConnectCalls++ |
| 519 | } |
| 520 | } |
| 521 | info.name = tl.Name |
| 522 | info.dup = t.seen[key] |
| 523 | t.seen[key] = true |
| 524 | } |
| 525 | } |
| 526 | |
| 527 | func (t *trajScan) recordResult(rec trajectoryRecord) { |
| 528 | tl := rec.Event.Tool |
| 529 | t.s.ToolMs += tl.DurationMs |
| 530 | t.batch.results++ |
| 531 | if tl.ReadOnly { |
| 532 | t.batch.readOnly++ |
| 533 | } |
| 534 | if info, ok := t.batch.infos[tl.ID]; ok { |
| 535 | info.resolved = true |
| 536 | info.readOnly = tl.ReadOnly |
| 537 | info.errored = tl.Err != "" |
| 538 | if tl.Execution != nil { |
| 539 | info.verification = tl.Execution.Verification |
| 540 | } |
| 541 | } |
| 542 | if ex := tl.Execution; ex != nil && (ex.Verification == "passed" || ex.Verification == "failed") { |
| 543 | t.observeVerification(tl.Name+"\x00"+tl.Args, ex.Verification == "passed", rec.TS) |
| 544 | } |
| 545 | if delegationTools[tl.Name] { |
| 546 | t.delegationToolMs[tl.Name] += tl.DurationMs |
| 547 | } |
| 548 | t.batch.serialMs += tl.DurationMs |
| 549 | if tl.StartedAt > 0 && tl.EndedAt >= tl.StartedAt { |
| 550 | t.batch.intervals = append(t.batch.intervals, [2]int64{tl.StartedAt, tl.EndedAt}) |
| 551 | if t.firstToolTS == 0 || tl.StartedAt < t.firstToolTS { |
| 552 | t.firstToolTS = tl.StartedAt |
| 553 | } |
| 554 | if disp, ok := t.batch.dispatchTS[tl.ID]; ok && tl.StartedAt >= disp { |
| 555 | t.delays = append(t.delays, tl.StartedAt-disp) |
| 556 | } |
| 557 | } else { |
| 558 | t.orphanMs += tl.DurationMs |
| 559 | } |
| 560 | t.gapStart = rec.TS |
| 561 | t.inModel = true |
| 562 | } |
| 563 | |
| 564 | func (t *trajScan) closeBatch() { |
| 565 | b, s := t.batch, t.s |
| 566 | if b.calls == 0 { |
| 567 | return |
| 568 | } |
| 569 | if len(t.pendingGaps) > 0 { |
| 570 | gap := t.pendingGaps[0] |
| 571 | t.pendingGaps = t.pendingGaps[1:] |
| 572 | if len(b.infos) > 0 { |
| 573 | t.recordRound(classifyRound(gap, b), gap, b) |
| 574 | } |
| 575 | } |
| 576 | s.ToolBatches++ |
| 577 | s.TopLevelCalls += b.calls |
| 578 | s.MaxBatchSize = max(s.MaxBatchSize, b.calls) |
| 579 | if b.calls == 1 && b.results == 1 && b.readOnly == 1 { |
| 580 | s.SingleReadRounds++ |
| 581 | t.streakRun++ |
| 582 | s.SingleReadStreak = max(s.SingleReadStreak, t.streakRun) |
| 583 | } else { |
| 584 | t.streakRun = 0 |
| 585 | } |
| 586 | if len(b.intervals) > 1 { |
| 587 | wall, overlapped := intervalSpan(b.intervals) |
| 588 | if overlapped { |
| 589 | s.ParallelBatches++ |
| 590 | } |
| 591 | if saved := b.serialMs - wall; saved > 0 { |
| 592 | s.ParallelSavedMs += saved |
| 593 | } |
| 594 | } |
| 595 | t.allIntervals = append(t.allIntervals, b.intervals...) |
| 596 | t.batch = newToolBatch() |
| 597 | } |
| 598 | |
| 599 | func (t *trajScan) finish() *trajectorySummary { |
| 600 | s := t.s |
| 601 | if t.inModel && t.lastTS > t.gapStart { |
| 602 | t.closeGap(t.lastTS - t.gapStart) // final answer round |
| 603 | } |
| 604 | t.closeBatch() |
| 605 | if t.sawCallIDs { |
| 606 | for _, gap := range t.pendingGaps { |
| 607 | t.recordRound(classifyRound(gap, nil), gap, nil) |
| 608 | } |
| 609 | } |
| 610 | t.pendingGaps = nil |
| 611 | s.ModelRounds = len(t.gaps) |
| 612 | for _, g := range t.gaps { |
| 613 | s.ModelGapTotalMs += g |
| 614 | } |
| 615 | s.ModelGapP95Ms = p95(t.gaps) |
| 616 | s.CleanGapP95Ms = p95(t.cleanGaps) |
| 617 | s.StartDelayP95Ms = p95(t.delays) |
| 618 | for _, d := range t.delays { |
| 619 | s.ToolQueueMs += d |
| 620 | } |
| 621 | if t.firstDelta > t.firstTS { |
| 622 | s.TTFTMs = t.firstDelta - t.firstTS |
| 623 | } |
| 624 | if t.firstToolTS > t.firstTS { |
| 625 | s.FirstToolMs = t.firstToolTS - t.firstTS |
| 626 | } |
| 627 | if len(t.allIntervals) > 0 { |
| 628 | s.ToolWallMs = intervalUnion(t.allIntervals) + t.orphanMs |
| 629 | } |
| 630 | s.SpanMs = t.lastTS - t.firstTS |
| 631 | if s.ModelMs = s.SpanMs - s.toolWall(); s.ModelMs < 0 { |
| 632 | s.ModelMs = 0 |
| 633 | } |
| 634 | s.Outcome = t.summarizeOutcome() |
| 635 | // The admission verdict lands after the tool's result in the stream, so |
| 636 | // denied time joins by tool name once the whole file is folded. |
| 637 | for name := range t.denyDelegations { |
| 638 | s.DeniedDelegationMs += t.delegationToolMs[name] |
| 639 | } |
| 640 | t.decompose() |
| 641 | return s |
| 642 | } |
| 643 | |
| 644 | // decompose partitions the span into disjoint wall buckets by priority, so a |
| 645 | // second spent in two places is booked once, to the more specific bucket. |
| 646 | func (t *trajScan) decompose() { |
| 647 | if len(t.attempts) == 0 { |
| 648 | return // old recording without stream_attempt events |
| 649 | } |
| 650 | s := t.s |
| 651 | var planIvs, execIvs [][2]int64 |
| 652 | for _, a := range t.attempts { |
| 653 | if a.planner { |
| 654 | planIvs = append(planIvs, a.iv) |
| 655 | } else { |
| 656 | execIvs = append(execIvs, a.iv) |
| 657 | } |
| 658 | } |
| 659 | covered := mergeIntervals(t.allIntervals) |
| 660 | retry := clipIntervals(t.retryIvs, covered) |
| 661 | covered = mergeIntervals(append(covered, retry...)) |
| 662 | comp := clipIntervals(t.compIvs, covered) |
| 663 | covered = mergeIntervals(append(covered, comp...)) |
| 664 | plan := clipIntervals(planIvs, covered) |
| 665 | covered = mergeIntervals(append(covered, plan...)) |
| 666 | exec := clipIntervals(execIvs, covered) |
| 667 | s.RetryWaitMs = ivsLen(retry) |
| 668 | s.CompactionMs = ivsLen(comp) |
| 669 | s.PlannerStreamMs = ivsLen(plan) |
| 670 | s.ModelStreamMs = ivsLen(exec) |
| 671 | rem := s.SpanMs - s.toolWall() - s.RetryWaitMs - s.CompactionMs - s.PlannerStreamMs - s.ModelStreamMs |
| 672 | if rem > 0 { |
| 673 | s.AgentOtherMs = rem |
| 674 | } |
| 675 | } |
| 676 | |
| 677 | func newToolBatch() *toolBatch { |
| 678 | return &toolBatch{dispatchTS: map[string]int64{}, infos: map[string]*roundCall{}} |
| 679 | } |
| 680 | |
| 681 | func (b *toolBatch) seen(id string) bool { |
| 682 | _, ok := b.dispatchTS[id] |
| 683 | return ok |
| 684 | } |
| 685 | |
| 686 | // durMs renders small durations without the sub-second floor dur applies. |
| 687 | func durMs(ms int64) string { |
| 688 | if ms <= 0 { |
| 689 | return "0ms" |
| 690 | } |
| 691 | if ms < 1000 { |
| 692 | return fmt.Sprintf("%dms", ms) |
| 693 | } |
| 694 | return dur(ms) |
| 695 | } |
| 696 | |
| 697 | func p95(values []int64) int64 { |
| 698 | return pctile(values, 95) |
| 699 | } |
| 700 | |
| 701 | func pctile(values []int64, p int) int64 { |
| 702 | if len(values) == 0 { |
| 703 | return 0 |
| 704 | } |
| 705 | sorted := append([]int64(nil), values...) |
| 706 | slices.Sort(sorted) |
| 707 | index := min((len(sorted)*p+99)/100, len(sorted)) |
| 708 | return sorted[index-1] |
| 709 | } |
| 710 |