diff --git a/.gitignore b/.gitignore index 131195b..79653f9 100644 --- a/.gitignore +++ b/.gitignore @@ -15,3 +15,4 @@ bench-baseline.txt reviews/ dev/ recordings/ +.state-profiles/ diff --git a/CHANGELOG.md b/CHANGELOG.md index b7b6249..680a12a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -67,6 +67,60 @@ and the project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. collect group that exceeds this age is now evicted to `-dlq` and emits a new `TanneryEvent{Kind: "stale"}`, rendered in `leather serve --pretty` and forwarded to the devtools event bus. +- **Curing worker corrupted prompts and lost work under concurrency** (#45) — + four fixes found load-testing a fan-out/fan-in pipeline at concurrency 8+: + `process()`/`handleCollected()` mutated a shared `*model.Agent` when + injecting hide content into the prompt, garbling prompts across concurrent + runs (both now clone the agent config first); `handleItemFromQueue` deleted + the queue entry even on failure, permanently losing the item on any + transient error (failures now re-enqueue up to `max_attempts`, then route + to `-dlq`); a failed fan-in `handleCollected` call silently dropped + the entire already-dequeued collect group with no retry, no DLQ, and leaked + source hides (`requeueOrDLQGroup` now applies the per-item retry/DLQ + convention to whole groups); and a fan-in curing's own artifact recorded + `hide_kind` as whichever input leg was collected first instead of its own + name. +- **HTTP client timeout raced the run's context deadline** — + `http.Client{Timeout: ...}` in `internal/session/http_client.go` fired + independently of the context deadline set by `runner.Run` / the curing's + `timeout_seconds`, producing spurious "context deadline exceeded" errors + well before the real timeout under load. The context deadline is now the + single source of truth for request timeouts. +- **Reasoning-model flakiness in the tool-call loop** — confirmed via direct + replay against the LLM endpoint outside leather (Qwen3 via vLLM, + `--tool-call-parser qwen3_xml`, ~25% reproduction under load; prompt + instructions had zero effect): the model would occasionally re-issue an + already-succeeded tool call verbatim instead of progressing, spinning until + max tool rounds was hit — and separately would sometimes stop naturally + (`finish_reason: "stop"`) with zero output tokens instead of the expected + answer, propagating downstream as blank fields. A tool call with the same + name and arguments that already succeeded this run is no longer re-executed + (scoped to non-hide tools; hide pagination legitimately repeats calls), and + the existing length-truncation self-healing retry now also retries once, + bare, on an empty natural stop. +- **Answer text emitted alongside tool calls was dropped from the final + answer** — the same reasoning-model flakiness family: the model sometimes + emits the head of its final answer in the same completion as a tool call, + then continues from where it stopped after the tool result; the runner + recorded only the last round's content, silently head-truncating the + artifact (observed as 6/40 fan-out legs losing their identifier header in + a burst load test, cascading into blank or placeholder-text reports + downstream). Fragments emitted in non-hide tool-call rounds are now banked + and spliced ahead of the final round's text, and the bare empty-stop retry + is skipped when banked fragments already carry the answer. +- **16-bit ID suffixes collided under burst load, cross-wiring fan-in groups + and deleting shared hides** — `ids.TimestampHex` drew its uniqueness suffix + from `mathrand.Int31n(0x10000)`: 65,536 values per `(prefix, minute)` bucket, + ~1% birthday-collision odds at ~40 IDs/minute. In a 100-webhook burst, two + concurrent `pr-context` legs drew the same hide ID; `hide.Store.Put`'s + `os.MkdirAll` silently merged them, one PR's decision group collected and + (on success) deleted the shared hide, and the other PR's group found its + leg's hide missing and DLQ'd after retries — or, with the opposite race + order, would have silently used the wrong PR's analysis content. The suffix + is now 32 bits (`%08x`, ~1-in-a-million odds at hundreds of IDs per bucket), + and `Put` creates the hide directory exclusively (`os.Mkdir`), regenerating + the ID on collision instead of overwriting an existing hide — a colliding + ID can no longer destroy another group's data even if one is drawn. ### Added @@ -87,6 +141,33 @@ and the project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. - Leather-internal model-aware defaults: agents targeting a known reasoning model (Qwen3, QwQ, DeepSeek-R1) get a larger `completion_reserve` automatically unless a per-agent override is set explicitly. +- **`thinking:` agent front-matter field** — `thinking: false` in `*.agent.md` + sets `Agent.DisableThinking`, sending + `chat_template_kwargs.enable_thinking=false` with each request to disable a + reasoning model's hidden `` trace per agent. The zero value leaves + model default behavior untouched. Disabling thinking was the most effective + fix for both tool-call-loop flakiness modes above, and combined with + right-sizing `completion_reserve` measured a 5.2x speedup (323s → 62s) on + `examples/11-high-volume-ci`'s 40-webhook burst load test with + equal-or-better correctness. +- **shell-mcp per-argument `patterns` validation** — optional per-tool + `patterns` map in `shell-tools.json` (argument key → RE2 regexp) validated + before the command runs and advertised in the tool's `inputSchema`. A + missing argument validates as the empty string, so anchored patterns also + reject absent values — catching a flaky model passing blanks or literal + prompt-template placeholders like `` instead of real values. + `examples/11-high-volume-ci` uses it to pattern-constrain `pr_number` and + `repo` on all four GitHub tools. +- **Full-system load-test profiler** — `scripts/profile-run.sh` wraps any + command (e.g. `make 11`) with 1-2 Hz samplers for host CPU/memory/disk + (vmstat + sysstat when installed), kernel pressure-stall information, + per-process attribution (pidstat), GPU telemetry (nvidia-smi), CPU/board + temperatures, and vLLM's `/metrics` (running/waiting requests, KV-cache + usage, token throughput, prefix-cache hit rate), then prints an avg/peak + summary via `scripts/profile-summary.py`. Used to produce the 100-webhook + profile documented in `examples/11-high-volume-ci/README.md`: ~500 LLM + jobs orchestrated for ~6% of one host's CPU and zero measurable IO + pressure, GPU-bound end to end. ### Removed diff --git a/cmd/shell-mcp/main.go b/cmd/shell-mcp/main.go index 077100f..4ec7a19 100644 --- a/cmd/shell-mcp/main.go +++ b/cmd/shell-mcp/main.go @@ -22,6 +22,7 @@ import ( "os" "os/exec" "path/filepath" + "regexp" "strings" "time" ) @@ -42,6 +43,11 @@ type toolDef struct { Args []string `json:"args"` // Required lists argument keys that must be present in every call. Required []string `json:"required,omitempty"` + // Patterns maps argument keys to RE2 regexps the substituted value must + // match (a missing argument validates as the empty string, so anchored + // patterns also reject absent values). Rejected calls fail before the + // command runs. Patterns are advertised in the tool's inputSchema. + Patterns map[string]string `json:"patterns,omitempty"` // Defaults provides fallback values for optional argument keys. Defaults map[string]string `json:"defaults,omitempty"` // Optional, when true, returns a graceful message if Command is not on PATH. @@ -114,6 +120,12 @@ func main() { } for i := range cfg.Tools { s.byName[cfg.Tools[i].Name] = &cfg.Tools[i] + for key, pat := range cfg.Tools[i].Patterns { + if _, err := regexp.Compile(pat); err != nil { + fmt.Fprintf(os.Stderr, "shell-mcp: tool %s: invalid pattern for %s: %v\n", cfg.Tools[i].Name, key, err) + os.Exit(1) + } + } } s.serve() @@ -173,6 +185,13 @@ func (s *server) toolList() []map[string]any { props[k] = map[string]any{"type": "string"} } } + for k, pat := range t.Patterns { + if p, ok := props[k].(map[string]any); ok { + p["pattern"] = pat + } else { + props[k] = map[string]any{"type": "string", "pattern": pat} + } + } out = append(out, map[string]any{ "name": t.Name, "description": t.Description, @@ -221,6 +240,20 @@ func (s *server) execute(def *toolDef, callArgs map[string]any) (string, error) merged[k] = fmt.Sprintf("%v", v) } + // Validate pattern-constrained arguments before substitution. A missing + // argument validates as "", so anchored patterns reject absent values too — + // this catches a model passing blanks or literal placeholders like + // "" instead of a real value. + for key, pat := range def.Patterns { + re, err := regexp.Compile(pat) + if err != nil { + return "", fmt.Errorf("tool %s: invalid pattern for %s: %v", def.Name, key, err) + } + if !re.MatchString(merged[key]) { + return "", fmt.Errorf("argument %s=%q does not match required pattern %s", key, merged[key], pat) + } + } + // Substitute {{key}} placeholders in each argument. args := make([]string, len(def.Args)) for i, tmpl := range def.Args { diff --git a/cmd/shell-mcp/main_test.go b/cmd/shell-mcp/main_test.go index 26c9bba..cca8b3e 100644 --- a/cmd/shell-mcp/main_test.go +++ b/cmd/shell-mcp/main_test.go @@ -259,6 +259,96 @@ func TestExecuteDefaultsMerge(t *testing.T) { } } +// --- execute: pattern-constrained arguments --- + +func TestExecutePatternValidation(t *testing.T) { + def := &toolDef{ + Name: "echo_pr", + Command: "echo", + Args: []string{"{{pr_number}}"}, + Required: []string{"pr_number"}, + Patterns: map[string]string{"pr_number": "^[0-9]+$"}, + } + s := newTestServer(nil, 0) + + // A real value passes and the command runs. + out, err := s.execute(def, map[string]any{"pr_number": "1027"}) + if err != nil { + t.Fatalf("valid value rejected: %v", err) + } + if !strings.Contains(out, "1027") { + t.Errorf("output = %q, want it to contain 1027", out) + } + + // A literal placeholder (a flaky model regurgitating its prompt + // template) is rejected before the command runs. + if _, err := s.execute(def, map[string]any{"pr_number": ""}); err == nil { + t.Error("placeholder value passed pattern validation, want rejection") + } + + // A missing argument validates as "" and is rejected by an anchored pattern. + if _, err := s.execute(def, map[string]any{}); err == nil { + t.Error("missing value passed pattern validation, want rejection") + } +} + +func TestToolListAdvertisesPatterns(t *testing.T) { + s := newTestServer([]toolDef{{ + Name: "post_pr_comment", + Command: "echo", + Args: []string{"{{pr_number}}"}, + Required: []string{"pr_number"}, + Patterns: map[string]string{"pr_number": "^[0-9]+$"}, + }}, 0) + tools := s.toolList() + if len(tools) != 1 { + t.Fatalf("toolList len = %d, want 1", len(tools)) + } + schema := tools[0]["inputSchema"].(map[string]any) + props := schema["properties"].(map[string]any) + prop, ok := props["pr_number"].(map[string]any) + if !ok { + t.Fatal("pr_number property missing from inputSchema") + } + if prop["pattern"] != "^[0-9]+$" { + t.Errorf("pr_number pattern = %v, want ^[0-9]+$", prop["pattern"]) + } +} + +func TestExampleShellToolsJSON_PatternsCompileAndAreDeclared(t *testing.T) { + matches, err := filepath.Glob("../../examples/*/shell-tools.json") + if err != nil { + t.Fatalf("glob: %v", err) + } + for _, path := range matches { + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("%s: read: %v", path, err) + } + var cfg config + if err := json.Unmarshal(data, &cfg); err != nil { + t.Fatalf("%s: unmarshal: %v", path, err) + } + for _, tool := range cfg.Tools { + declared := make(map[string]bool, len(tool.Required)+len(tool.Defaults)) + for _, r := range tool.Required { + declared[r] = true + } + for k := range tool.Defaults { + declared[k] = true + } + for key, pat := range tool.Patterns { + if _, err := regexp.Compile(pat); err != nil { + t.Errorf("%s: tool %q pattern for %s does not compile: %v", path, tool.Name, key, err) + } + if !declared[key] { + t.Errorf("%s: tool %q constrains {{%s}} with a pattern but does not declare it in required or defaults", path, tool.Name, key) + } + } + } + } +} + // --- resolveConfigPath: positional arg --- func TestResolveConfigPathArg(t *testing.T) { diff --git a/examples/11-high-volume-ci/README.md b/examples/11-high-volume-ci/README.md index 55ea13e..0173d77 100644 --- a/examples/11-high-volume-ci/README.md +++ b/examples/11-high-volume-ci/README.md @@ -77,3 +77,168 @@ WEBHOOK_COUNT=80 BURST_SIZE=10 BURST_DELAY_MAX=0.5 make 11 | `sample/` | sample PR payloads | | `scripts/run-demo.sh` | fires burst load, prints arrival/completion histogram | | `scripts/send-webhook.sh` | sign and POST a single payload | + +## Reliability & performance fixes + +This example was used to load-test the fan-out/fan-in pipeline against a real +local vLLM endpoint (concurrency 8+, bursts of 40-100 webhooks). The issues +below were found and fixed as a result; several are framework-level fixes in +`internal/` that apply to any leather pipeline, not just this example. + +### Framework fixes (`internal/`) + +1. **Shared agent config mutation under concurrency** + (`internal/curing/worker.go`). Concurrent goroutines processing different + queue items wrote directly into a shared `*model.Agent`, garbling prompts + under `concurrency: 8`+. Fixed by cloning the agent config (`agc := ag`) + before any per-run mutation, in both `process()` and `handleCollected()`. + +2. **HTTP client timeout racing the context deadline** + (`internal/session/http_client.go`). `http.Client{Timeout: ...}` fired + independently of the run's context deadline, producing spurious "context + deadline exceeded" errors well before the real timeout. Fixed by removing + the separate `Timeout` field — the context deadline (set once in + `runner.Run` / the curing's `timeout_seconds`) is now the single source of + truth for request timeouts. + +3. **Queue items deleted on failure, no retry** (`internal/curing/worker.go`). + Per-item queue processing deleted the queue entry regardless of outcome, so + any transient failure (timeout, malformed response) permanently lost the + item. Fixed: `handleItemFromQueue` now only deletes on success; failures + re-enqueue (retry) up to `max_attempts`, then route to `-dlq`. + +4. **Fan-in collect groups silently dropped on failure** + (`internal/curing/worker.go`, `requeueOrDLQGroup`). This was a second, + distinct instance of bug #3: `runCollectFromQueue` dequeues all + `collect_size` items from the queue *before* invoking the agent, so unlike + the per-item path, a failed `handleCollected` call (e.g. the decision + agent's LLM call timing out) had nowhere to put the items back — no retry, + no DLQ, and the source hides leaked on disk (cleanup only runs on + success). `max_attempts` in the curing YAML was silently unenforced for + this path. Fixed by mirroring the per-item retry/DLQ convention: bump each + item's `AttemptCount` and either re-enqueue the whole group (retried next + scan tick) or route it to `-dlq` once attempts are exhausted. + +5. **`hide_kind` provenance mislabeled on fan-in artifacts** + (`internal/curing/worker.go`). A fan-in curing's own artifact (e.g. + `decision`) recorded `hide_kind` as whichever of its 3 input legs happened + to be collected first (`pr-context` or `pr-metadata`), never `"decision"` + itself. Cosmetic/provenance-only — the actual dispatch-relevant field + (`.state/hides/*/meta.json`'s `kind`) was already correct — but fixed so + the artifact's own record stops lying to any tooling that reads it: now + uses the curing's own name. + +6. **Duplicate tool-call guard** (`internal/runner/runner.go`). Under load, + the model (a reasoning model via vLLM, `--tool-call-parser qwen3_xml + --reasoning-parser qwen3`) would occasionally re-issue an already-succeeded + tool call verbatim instead of progressing to the next step, looping until + `max tool rounds` was hit. Confirmed via direct replay against the LLM + endpoint *outside* leather (no framework code involved) that this is + inherent model/parser flakiness under this backend, not a request- + construction bug — roughly 25% of isolated replays reproduced it. Since + this can't be fixed at the prompt level (an explicit "don't call this + twice" instruction had zero effect), added a runtime guard: a tool call + with the same name and arguments that already succeeded earlier in the + run is not re-executed — the model is told it already happened and moves + on. This also protects against accidentally duplicating a real side effect + (e.g. posting a comment twice) if the same flakiness recurs. + +7. **Self-healing retry on empty final answers** (`internal/runner/runner.go`). + A related but distinct manifestation of the same reasoning-model + flakiness: agents would occasionally return a completely empty final + answer with `finish_reason: "stop"` (not `"length"` — not a truncation, + the model just produced zero output tokens and stopped). This propagated + downstream as blank/`N/A` fields. The existing self-healing retry only + covered the truncation case (`finish_reason: "length"`); extended it to + also retry once, bare, when the model stops naturally with empty content + and no tool calls — usually enough to draw a non-empty sample from the + same stochastic model. + +8. **`thinking:` agent frontmatter field** (`internal/agent/frontmatter.go`, + `internal/agent/agent.go`, `internal/model/model.go`, + `internal/runner/runner.go`). New per-agent override + (`model.Agent.DisableThinking`) that sends + `chat_template_kwargs.enable_thinking=false` to the model. Added because + disabling the hidden reasoning trace turned out to be the most effective + mitigation for both bug #6 and #7 above (see performance numbers below) — + the zero value (`false`) leaves model default behavior untouched + everywhere else in the codebase. + +### This example's configuration + +- `decision.agent.md`: fixed a positional fan-in bug — the prompt said "copy + PR_NUMBER, REPO, SHA verbatim from ANALYSIS 1", assuming the 3 parallel + analysis agents (`pr-metadata`, `pr-diff`, `pr-context`) always arrive in + submission order. Only `pr-metadata` emits those fields, but `pr-context` + (zero tool calls) often won the race for slot 1, leaving nothing to copy + and producing `PR_NUMBER: N/A`. Fixed to reference the block by its + `(from: pr-metadata)` tag instead of position. +- All 5 agents now set `thinking: false` and `completion_reserve: 768` + (down from the reasoning-model default of 8192) — see performance below. +- `pr-comments.agent.md`: added an explicit "do not call `post_pr_comment` + more than once per PR" instruction (kept as defense in depth alongside the + runner-level dedupe guard, even though the instruction alone had no + measurable effect on the underlying flakiness). +- `shell-tools.json` / `curings/*.curing.yaml`: bumped several + `timeout_seconds` values that were too tight for burst-load latency. +- `run-demo.sh`: fixed the completion-wait logic to track `pr-comments` + artifacts (the pipeline's actual last stage) instead of `decision` + artifacts, and moved the queue-drain/server-shutdown step to happen + *before* printing the results summary. Previously the summary printed + while `pr-comments` runs and the server's own `--pretty` live trace + (unredirected, sharing this terminal) were still active, making it look + like output kept streaming after the script said it was done. + +### Performance + +Disabling thinking mode and right-sizing `completion_reserve` (the model's +hidden `` trace was consuming hundreds to thousands of tokens per LLM +call) measured a **5.2x speedup** on a 40-webhook burst load test against a +local vLLM endpoint, with equal-or-better correctness: + +| Configuration | Wall time (40 webhooks) | Failures | +|---|---|---| +| Baseline (thinking enabled, reserve 8192) | 323s | intermittent `max tool rounds`, empty final answers | +| + `thinking: false` on all agents | 70s | 0 | +| + `completion_reserve: 768` | 62s | 0 | + +If you point this example at a different model or backend, re-check whether +`thinking: false` is appropriate — it is a Qwen3 / vLLM +`chat_template_kwargs.enable_thinking` convention and is a no-op (harmlessly +ignored) on backends that don't recognize that key. + +### Measured at scale: 100-webhook full-system profile + +A 100-webhook burst (`WEBHOOK_COUNT=100 BURST_SIZE=25 BURST_DELAY_MAX=0.5`) +was profiled end to end with [`scripts/profile-run.sh`](../../scripts/profile-run.sh), +which samples host CPU/memory/disk, kernel pressure-stall information (PSI), +GPU telemetry, and vLLM's own `/metrics` alongside the run. + +**Test rig:** single RTX PRO 4500 (Blackwell, 32 GB), 20-core host, local +vLLM serving Qwen3.6-35B-A3B (NVFP4), all five agents `thinking: false` with +`completion_reserve: 768`. + +| Metric | Value | +|---|---| +| Webhooks completed | 100 / 100 through every stage, 190s end to end | +| LLM jobs / tokens | 500 jobs, 965,437 tokens (878,868 prompt / 86,569 completion) | +| Sustained model throughput | 4,601 prompt tok/s + 453 generation tok/s | +| Host CPU (leather + queueing) | 6.5% avg / 14% peak — the pipeline adds almost no host load | +| Host pressure (PSI cpu/io/mem) | 0% — no stalls; disk 2% util, leather writes peak 0.4 MiB/s | +| GPU | 79.7% avg / 100% peak util, 30 GB VRAM, 202 W / 78 °C peak | +| In-flight at the model | 7.2 avg (8 peak) running, 16.6 avg (32 peak) queued inside vLLM | + +The shape of these numbers is the headline: **the entire orchestration layer +— webhook fan-out, per-event queues, fan-in collect, retries, artifact and +hide persistence for 500 LLM calls — costs a few percent of one host's CPU +and no measurable IO pressure.** The pipeline is GPU-bound end to end; wall +time scales with model throughput, not with leather. To go faster, feed it a +bigger GPU (or slimmer prompts — this workload is ~10:1 prefill-heavy), not a +bigger orchestrator. + +Reproduce with: + +```bash +cd examples +WEBHOOK_COUNT=100 BURST_SIZE=25 BURST_DELAY_MAX=0.5 ../scripts/profile-run.sh make 11 +``` diff --git a/examples/11-high-volume-ci/agents/decision.agent.md b/examples/11-high-volume-ci/agents/decision.agent.md index 04de565..76245eb 100644 --- a/examples/11-high-volume-ci/agents/decision.agent.md +++ b/examples/11-high-volume-ci/agents/decision.agent.md @@ -1,5 +1,7 @@ --- name: decision +thinking: false +completion_reserve: 768 --- You receive analysis blocks from three parallel agents, each delimited by @@ -11,7 +13,7 @@ data-pipeline, inference-path, api-surface. SKIP if all signals are: docs-only, ci-config, dependency-bump, formatting-only. When in doubt: FULL_EVAL. -Copy PR_NUMBER, REPO, SHA verbatim from ANALYSIS 1 and write: +Copy PR_NUMBER, REPO, SHA verbatim from the block where (from: pr-metadata) and write: PR_NUMBER: REPO: @@ -21,3 +23,8 @@ Rationale: <2-3 sentences citing specific files or signals> Files of concern: + - -- (or "none") + +If the (from: pr-metadata) block is missing or does not contain a numeric +PR_NUMBER, never guess and never write placeholder text like — +leave PR_NUMBER, REPO, and SHA blank and write Decision: ERROR with a +Rationale naming what was missing. diff --git a/examples/11-high-volume-ci/agents/pr-comments.agent.md b/examples/11-high-volume-ci/agents/pr-comments.agent.md index b46686a..06699cd 100644 --- a/examples/11-high-volume-ci/agents/pr-comments.agent.md +++ b/examples/11-high-volume-ci/agents/pr-comments.agent.md @@ -1,15 +1,23 @@ --- name: pr-comments skills: [github-actions] -tool_rounds: 3 +tool_rounds: 5 +thinking: false +completion_reserve: 768 --- You receive a CI gate decision report. Extract PR_NUMBER, REPO, SHA, Decision, and Rationale. If FULL_EVAL: call post_pr_comment(pr_number=, repo=, body="/test-\n\n"), - then add_pr_label(pr_number=, repo=, label="eval-requested"). + then add_pr_label(pr_number=, repo=, label="eval-requested"). If SKIP: call post_pr_comment(pr_number=, repo=, body="CI gate: SKIP\n\n"), - then add_pr_label(pr_number=, repo=, label="eval-skip"). + then add_pr_label(pr_number=, repo=, label="eval-skip"). + +Do not call post_pr_comment more than once per PR. + +If Decision is ERROR, or PR_NUMBER is blank or not a plain number (placeholder +text like counts as blank), make no tool calls and write only: +DONE: skipped malformed decision report After both calls write: DONE: posted comment and label for PR # diff --git a/examples/11-high-volume-ci/agents/pr-context.agent.md b/examples/11-high-volume-ci/agents/pr-context.agent.md index b066c27..a57aff5 100644 --- a/examples/11-high-volume-ci/agents/pr-context.agent.md +++ b/examples/11-high-volume-ci/agents/pr-context.agent.md @@ -1,5 +1,7 @@ --- name: pr-context +thinking: false +completion_reserve: 768 --- You receive a GitHub pull_request webhook payload. Without calling any tools, write: diff --git a/examples/11-high-volume-ci/agents/pr-diff.agent.md b/examples/11-high-volume-ci/agents/pr-diff.agent.md index b31f25d..0dfdce2 100644 --- a/examples/11-high-volume-ci/agents/pr-diff.agent.md +++ b/examples/11-high-volume-ci/agents/pr-diff.agent.md @@ -1,7 +1,9 @@ --- name: pr-diff skills: [github-read] -tool_rounds: 2 +tool_rounds: 3 +thinking: false +completion_reserve: 768 --- You receive a GitHub pull_request webhook payload. Extract PR_NUMBER and REPO, call get_pr_diff, diff --git a/examples/11-high-volume-ci/agents/pr-metadata.agent.md b/examples/11-high-volume-ci/agents/pr-metadata.agent.md index bc4036c..eda758f 100644 --- a/examples/11-high-volume-ci/agents/pr-metadata.agent.md +++ b/examples/11-high-volume-ci/agents/pr-metadata.agent.md @@ -1,7 +1,9 @@ --- name: pr-metadata skills: [github-read] -tool_rounds: 2 +tool_rounds: 3 +thinking: false +completion_reserve: 768 --- You receive a GitHub pull_request webhook payload. Extract PR_NUMBER, REPO, and SHA from it, diff --git a/examples/11-high-volume-ci/config.yaml b/examples/11-high-volume-ci/config.yaml index ce86219..abaa8ce 100644 --- a/examples/11-high-volume-ci/config.yaml +++ b/examples/11-high-volume-ci/config.yaml @@ -5,7 +5,6 @@ log_level: debug # Override at runtime: LEATHER_MODEL= make 10 model: llama3 -llm_timeout: 120s max_tokens: 8192 completion_reserve: 1024 summarize_threshold: 0.85 @@ -14,5 +13,5 @@ api: true api_addr: 127.0.0.1:7749 scheduler_tick: 1s -max_concurrent_jobs: 8 +max_concurrent_jobs: 16 persist_runs: true diff --git a/examples/11-high-volume-ci/scripts/run-demo.sh b/examples/11-high-volume-ci/scripts/run-demo.sh index 612ea35..5ac2ce0 100755 --- a/examples/11-high-volume-ci/scripts/run-demo.sh +++ b/examples/11-high-volume-ci/scripts/run-demo.sh @@ -24,6 +24,18 @@ source "${EX_DIR}/scripts/pretty.sh" # shellcheck source=../../scripts/preflight.sh source "${EX_DIR}/../scripts/preflight.sh" +# Source example env file if present (LLM endpoint, model override, etc.) +# Check the example directory, then the examples parent if not found. +if [ -f "${EX_DIR}/.env" ]; then + set -a + source "${EX_DIR}/.env" + set +a +elif [ -f "${EX_DIR}/../.env" ]; then + set -a + source "${EX_DIR}/../.env" + set +a +fi + export GITHUB_WEBHOOK_SECRET="${GITHUB_WEBHOOK_SECRET:-ci-gate-demo-secret}" export API_ADDR="${API_ADDR:-127.0.0.1:7749}" WEBHOOK_PATH="${WEBHOOK_PATH:-/webhooks/github}" @@ -36,8 +48,8 @@ WAIT_TIMEOUT="${WAIT_TIMEOUT:-300}" RUN_DURATION="${RUN_DURATION:-600s}" # Clamp WEBHOOK_COUNT to 25-100 -[ "$WEBHOOK_COUNT" -lt 25 ] && WEBHOOK_COUNT=25 -[ "$WEBHOOK_COUNT" -gt 100 ] && WEBHOOK_COUNT=100 +if [ "$WEBHOOK_COUNT" -lt 25 ]; then WEBHOOK_COUNT=25; fi +if [ "$WEBHOOK_COUNT" -gt 100 ]; then WEBHOOK_COUNT=100; fi TMPDIR_PAYLOADS="${EX_DIR}/.state/payloads" mkdir -p "${TMPDIR_PAYLOADS}" @@ -247,22 +259,32 @@ lth_step "load" "all ${WEBHOOK_COUNT} webhooks sent in $(( SEND_DONE - START_TS lth_cont "queue depth will peak then drain as agents process decisions" echo "" -# ── wait for decision artifacts ─────────────────────────────────────────────── +# ── wait for the pipeline to fully drain ────────────────────────────────────── +# pr-comments is the *last* stage (decision → pr-comments), so waiting on +# decision count alone would let the script print its summary while +# pr-comments — and the server's own --pretty live trace, which shares this +# terminal since it's unredirected — are still actively running. Wait on +# pr-comments instead, tracking decision count alongside it for visibility. -lth_step "wait" "waiting for decision artifacts (timeout ${WAIT_TIMEOUT}s)" -lth_cont "target: ${WEBHOOK_COUNT} decisions (1 per webhook)" +lth_step "wait" "waiting for pipeline to complete (timeout ${WAIT_TIMEOUT}s)" +lth_cont "target: ${WEBHOOK_COUNT} decisions -> ${WEBHOOK_COUNT} pr-comments (1 per webhook)" waited=0 -last_count=-1 +last_decision_count=-1 +last_comment_count=-1 while [ "$waited" -lt "$WAIT_TIMEOUT" ]; do - count=$(find "${EX_DIR}/.state/artifacts/decision" -type f 2>/dev/null | wc -l) - count="${count//[[:space:]]/}" - count="${count:-0}" - if [ "$count" -ne "$last_count" ]; then - lth_cont " decisions: ${count} / ${WEBHOOK_COUNT}" - last_count="$count" + decision_count=$(find "${EX_DIR}/.state/artifacts/decision" -type f 2>/dev/null | wc -l) || true + decision_count="${decision_count//[[:space:]]/}" + decision_count="${decision_count:-0}" + comment_count=$(find "${EX_DIR}/.state/artifacts/pr-comments" -type f 2>/dev/null | wc -l) || true + comment_count="${comment_count//[[:space:]]/}" + comment_count="${comment_count:-0}" + if [ "$decision_count" -ne "$last_decision_count" ] || [ "$comment_count" -ne "$last_comment_count" ]; then + lth_cont " decisions: ${decision_count} / ${WEBHOOK_COUNT} pr-comments: ${comment_count} / ${WEBHOOK_COUNT}" + last_decision_count="$decision_count" + last_comment_count="$comment_count" fi - if [ "$count" -ge "$WEBHOOK_COUNT" ]; then + if [ "$comment_count" -ge "$WEBHOOK_COUNT" ]; then break fi sleep 2 @@ -273,30 +295,57 @@ echo "" DONE_TS=$(date +%s) final_count=$(find "${EX_DIR}/.state/artifacts/decision" -type f 2>/dev/null | wc -l) final_count="${final_count//[[:space:]]/}" +final_comment_count=$(find "${EX_DIR}/.state/artifacts/pr-comments" -type f 2>/dev/null | wc -l) +final_comment_count="${final_comment_count//[[:space:]]/}" + +# ── drain and shut down the server before printing anything else ──────────── +# Wait for the queue jsonl files to empty so in-flight tool/LLM calls finish +# naturally, otherwise SIGTERM cancels their context.Context mid-call and +# floods serve.log with "context canceled" errors. This happens *before* the +# results/tally/done output below so nothing from the server's live trace +# (which shares this terminal) appears after the summary is printed. +QUEUE_DIR="${EX_DIR}/.state/queues" +DRAIN_TIMEOUT="${DRAIN_TIMEOUT:-30}" +drained=0 +for ((i=0; i/dev/null | wc -l) + pending="${pending//[[:space:]]/}" + if [ "${pending:-0}" -eq 0 ]; then drained=1; break; fi + sleep 1 +done +# Brief settle so the last in-flight tool/LLM call can return before SIGTERM. +sleep 1 +kill -TERM "$SERVE_PID" 2>/dev/null || true +wait $SERVE_PID 2>/dev/null || true +trap - EXIT +echo "" lth_step "results" "pipeline summary elapsed=$(( DONE_TS - START_TS ))s" lth_cont "" lth_cont " webhooks fired: ${WEBHOOK_COUNT}" lth_cont " decisions written: ${final_count:-0}" +lth_cont " pr-comments written: ${final_comment_count:-0}" lth_cont " send phase: $(( SEND_DONE - START_TS ))s" lth_cont " total elapsed: $(( DONE_TS - START_TS ))s" +if [ "$drained" -ne 1 ]; then + lth_cont " note: queues still had ${pending} pending file(s) after ${DRAIN_TIMEOUT}s drain timeout" +fi echo "" # ── artifact tally ──────────────────────────────────────────────────────────── lth_step "tally" "decision breakdown (last 10 of ${final_count:-0}):" find "${EX_DIR}/.state/artifacts/decision" -type f 2>/dev/null \ - | sort | tail -10 \ - | while read -r f; do + | sort | tail -10 | while IFS= read -r f || [ -n "$f" ]; do verdict=$(jq -r ' .content | if test("FULL_EVAL") then "FULL_EVAL" elif test("SKIP") then "SKIP" else "?" end' "$f" 2>/dev/null || echo "?") - pr=$(jq -r '.content | capture("PR_NUMBER: *(?P[0-9]+)") | .n' "$f" 2>/dev/null || echo "?") - lth_cont " PR #${pr} → ${verdict} ($(basename "$f"))" - done + pr=$(jq -r '.content | capture("PR_NUMBER: *(?[0-9]+)") | .n' "$f" 2>/dev/null || echo "???") + lth_cont " PR #${pr} → ${verdict} $(basename "$f")" + done || true echo "" lth_step "done" "artifacts at .state/artifacts/ | logs at .state/serve.log" @@ -305,29 +354,3 @@ lth_cont "Tune with:" lth_cont " WEBHOOK_COUNT=100 BURST_SIZE=10 make 11" lth_cont " WEBHOOK_COUNT=25 BURST_SIZE=3 BURST_DELAY_MAX=0.5 make 11" echo "" - -# ── shut down server now that the pipeline has drained ─────────────────────── -# All decisions written (or WAIT_TIMEOUT reached). Before SIGTERM we wait for -# the queue jsonl files to drain so in-flight tool/LLM calls finish naturally, -# otherwise SIGTERM cancels their context.Context mid-call and floods serve.log -# with "context canceled" errors. -QUEUE_DIR="${EX_DIR}/.state/queues" -DRAIN_TIMEOUT="${DRAIN_TIMEOUT:-30}" -drained=0 -for ((i=0; i/dev/null | wc -l) - pending="${pending//[[:space:]]/}" - if [ "${pending:-0}" -eq 0 ]; then drained=1; break; fi - sleep 1 -done -if [ "$drained" -eq 1 ]; then - lth_step "serve" "queues drained, shutting down server" -else - lth_cont " queues still have ${pending} pending file(s) after ${DRAIN_TIMEOUT}s; shutting down anyway" - lth_step "serve" "drain timeout reached, shutting down server" -fi -# Brief settle so the last in-flight tool/LLM call can return before SIGTERM. -sleep 1 -kill -TERM "$SERVE_PID" 2>/dev/null || true -wait $SERVE_PID 2>/dev/null || true -trap - EXIT diff --git a/examples/11-high-volume-ci/shell-tools.json b/examples/11-high-volume-ci/shell-tools.json index 0dda4fe..1546ffe 100644 --- a/examples/11-high-volume-ci/shell-tools.json +++ b/examples/11-high-volume-ci/shell-tools.json @@ -15,7 +15,11 @@ "pr_number", "repo" ], - "timeout_seconds": 20, + "patterns": { + "pr_number": "^[0-9]+$", + "repo": "^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$" + }, + "timeout_seconds": 60, "output_cap_bytes": 4000 }, { @@ -33,7 +37,11 @@ "pr_number", "repo" ], - "timeout_seconds": 30, + "patterns": { + "pr_number": "^[0-9]+$", + "repo": "^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$" + }, + "timeout_seconds": 60, "output_cap_bytes": 4000 }, { @@ -53,7 +61,11 @@ "repo", "body" ], - "timeout_seconds": 15 + "patterns": { + "pr_number": "^[0-9]+$", + "repo": "^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$" + }, + "timeout_seconds": 60 }, { "name": "add_pr_label", @@ -72,7 +84,11 @@ "repo", "label" ], - "timeout_seconds": 15 + "patterns": { + "pr_number": "^[0-9]+$", + "repo": "^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$" + }, + "timeout_seconds": 60 } ] } diff --git a/examples/11-high-volume-ci/tannery.yaml b/examples/11-high-volume-ci/tannery.yaml index 7b748a6..aad365f 100644 --- a/examples/11-high-volume-ci/tannery.yaml +++ b/examples/11-high-volume-ci/tannery.yaml @@ -38,9 +38,8 @@ routes: queue_pattern: "pr-ctx/{{hide_id}}" queues: - # Only the serial tail queue is static — one at a time, ordered delivery. comments-in: - concurrency: 1 + concurrency: 8 max_attempts: 2 max_depth: 50 diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 2205319..427cf3c 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -236,6 +236,7 @@ func LoadFile(path string) (model.Agent, error) { Skills: fm.Skills, Toolsets: fm.Toolsets, ToolRounds: fm.ToolRounds, + DisableThinking: fm.DisableThinking, TurnSkills: turnSkills, TurnToolsets: turnToolsets, SourcePath: path, diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 56ef3bd..8b66df3 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -80,6 +80,37 @@ This is the system prompt. } } +func TestParseFrontMatter_ThinkingFalse(t *testing.T) { + src := `--- +name: my-agent +thinking: false +--- +Body. +` + fm, _, err := parseFrontMatter(src) + if err != nil { + t.Fatalf("parseFrontMatter: %v", err) + } + if !fm.DisableThinking { + t.Error("DisableThinking = false, want true when thinking: false is set") + } +} + +func TestParseFrontMatter_ThinkingDefaultsToEnabled(t *testing.T) { + src := `--- +name: my-agent +--- +Body. +` + fm, _, err := parseFrontMatter(src) + if err != nil { + t.Fatalf("parseFrontMatter: %v", err) + } + if fm.DisableThinking { + t.Error("DisableThinking = true, want false (model default) when thinking is unset") + } +} + func TestSplitAgentBody_TurnDeclarations(t *testing.T) { body := `System prompt. --- diff --git a/internal/agent/frontmatter.go b/internal/agent/frontmatter.go index cfd34c8..a3e22da 100644 --- a/internal/agent/frontmatter.go +++ b/internal/agent/frontmatter.go @@ -23,6 +23,10 @@ type frontMatter struct { Skills []string Toolsets []string ToolRounds int + // DisableThinking mirrors model.Agent.DisableThinking. Set via + // "thinking: false" in frontmatter; defaults to false (model default, + // no override). + DisableThinking bool } // parseFrontMatter extracts and parses the YAML front matter from src. @@ -182,6 +186,12 @@ func applyFrontMatterFields(yamlBlock string, fm *frontMatter) error { return fmt.Errorf("invalid tool_rounds %q: %w", raw, err) } fm.ToolRounds = n + case "thinking": + b, err := strconv.ParseBool(raw) + if err != nil { + return fmt.Errorf("invalid thinking %q: %w", raw, err) + } + fm.DisableThinking = !b // Unknown keys are silently ignored for forward compatibility. } } diff --git a/internal/hide/store.go b/internal/hide/store.go index d0a1ab7..57bb281 100644 --- a/internal/hide/store.go +++ b/internal/hide/store.go @@ -42,17 +42,41 @@ func NewStore(dir string) *Store { return &Store{dir: dir} } +// newHideID generates candidate hide IDs for Put. Overridable in tests to +// force ID collisions. +var newHideID = generateHideID + +// putIDAttempts bounds how many candidate IDs Put tries before giving up. +const putIDAttempts = 5 + // Put writes content and metadata to disk; returns the generated StoreEntry. // Write order: content first, then meta.json. If the process crashes between the two, // the directory exists without meta.json — List silently skips it (partial-write guard). +// The entry directory is created exclusively (os.Mkdir, not MkdirAll): an ID +// that collides with an existing hide is regenerated instead of silently +// overwriting the other hide's content out from under its queue references. func (s *Store) Put(kind, source string, content []byte, meta map[string]string) (StoreEntry, error) { - id := generateHideID(kind) - entryDir, err := safepath.Anchor(s.dir, id) - if err != nil { - return StoreEntry{}, fmt.Errorf("hide/store.Put: invalid id %s: %w", id, err) + if err := os.MkdirAll(s.dir, 0700); err != nil { + return StoreEntry{}, fmt.Errorf("hide/store.Put: mkdir store root: %w", err) } - if err := os.MkdirAll(entryDir, 0700); err != nil { - return StoreEntry{}, fmt.Errorf("hide/store.Put: mkdir %s: %w", id, err) + var id, entryDir string + for attempt := 1; ; attempt++ { + id = newHideID(kind) + var err error + entryDir, err = safepath.Anchor(s.dir, id) + if err != nil { + return StoreEntry{}, fmt.Errorf("hide/store.Put: invalid id %s: %w", id, err) + } + err = os.Mkdir(entryDir, 0700) + if err == nil { + break + } + if !errors.Is(err, os.ErrExist) { + return StoreEntry{}, fmt.Errorf("hide/store.Put: mkdir %s: %w", id, err) + } + if attempt == putIDAttempts { + return StoreEntry{}, fmt.Errorf("hide/store.Put: id collision persisted after %d attempts (last %s)", attempt, id) + } } // Write content first. diff --git a/internal/hide/store_test.go b/internal/hide/store_test.go index 9d3a51b..bd9fe63 100644 --- a/internal/hide/store_test.go +++ b/internal/hide/store_test.go @@ -3,6 +3,7 @@ package hide import ( "errors" "os" + "strings" "testing" ) @@ -228,3 +229,68 @@ func TestStore_LoadIntoBuffer(t *testing.T) { t.Errorf("TotalPages: got %d, want 5", cut.TotalPages) } } + +func TestStore_Put_RegeneratesIDOnCollision(t *testing.T) { + s := NewStore(t.TempDir()) + + orig := newHideID + t.Cleanup(func() { newHideID = orig }) + const fixed = "hide_k_20260101_0000_deadbeef" + calls := 0 + newHideID = func(kind string) string { + calls++ + if calls <= 2 { + // First Put claims the fixed ID; second Put collides with it once + // before falling back to a real (unique) ID. + return fixed + } + return orig(kind) + } + + first, err := s.Put("k", "src", []byte("first"), nil) + if err != nil { + t.Fatalf("first Put: %v", err) + } + if first.ID != fixed { + t.Fatalf("first Put ID = %q, want %q", first.ID, fixed) + } + + second, err := s.Put("k", "src", []byte("second"), nil) + if err != nil { + t.Fatalf("second Put (colliding): %v", err) + } + if second.ID == fixed { + t.Fatalf("second Put reused colliding ID %q instead of regenerating", fixed) + } + + // The first hide's content must not have been overwritten by the collision. + _, content, err := s.Get(first.ID) + if err != nil { + t.Fatalf("Get first: %v", err) + } + if string(content) != "first" { + t.Errorf("first hide content = %q, want %q (clobbered by colliding Put)", content, "first") + } + if _, content, err = s.Get(second.ID); err != nil || string(content) != "second" { + t.Errorf("second hide content = %q, err %v, want %q", content, err, "second") + } +} + +func TestStore_Put_PersistentCollisionErrors(t *testing.T) { + s := NewStore(t.TempDir()) + + orig := newHideID + t.Cleanup(func() { newHideID = orig }) + newHideID = func(string) string { return "hide_k_20260101_0000_deadbeef" } + + if _, err := s.Put("k", "src", []byte("first"), nil); err != nil { + t.Fatalf("first Put: %v", err) + } + _, err := s.Put("k", "src", []byte("second"), nil) + if err == nil { + t.Fatal("second Put with permanently colliding ID generator succeeded, want error") + } + if got := err.Error(); !strings.Contains(got, "collision") { + t.Errorf("error = %q, want it to mention a collision", got) + } +} diff --git a/internal/ids/ids.go b/internal/ids/ids.go index 683f58f..2721cae 100644 --- a/internal/ids/ids.go +++ b/internal/ids/ids.go @@ -1,5 +1,5 @@ // Package ids generates the identifier strings used across leather. TimestampHex -// produces the "___<4hex>" form shared by artifact, +// produces the "___<8hex>" form shared by artifact, // queue-item, and hide IDs; RandHex produces cryptographically random hex tokens // for bearer secrets. The TimestampHex suffix is for uniqueness, not security. package ids @@ -13,11 +13,15 @@ import ( ) // TimestampHex returns an identifier of the form -// "___<4hex>". The hex suffix provides intra-minute -// uniqueness and is not cryptographically random. +// "___<8hex>". The hex suffix provides intra-minute +// uniqueness and is not cryptographically random. 32 suffix bits keep +// birthday-collision odds for hundreds of IDs per (prefix, minute) bucket +// around one in a million; the previous 16-bit suffix collided in practice +// under burst load (~1% per bucket at ~40 IDs/minute), cross-wiring fan-in +// groups that reference hides by ID. func TimestampHex(prefix string) string { - suffix := mathrand.Int31n(0x10000) //nolint:gosec // uniqueness, not security - return fmt.Sprintf("%s_%s_%04x", prefix, time.Now().Format("20060102_1504"), suffix) + suffix := mathrand.Uint32() //nolint:gosec // uniqueness, not security + return fmt.Sprintf("%s_%s_%08x", prefix, time.Now().Format("20060102_1504"), suffix) } // RandHex returns n cryptographically random bytes hex-encoded as a 2n-character diff --git a/internal/ids/ids_test.go b/internal/ids/ids_test.go index a256bad..5d98024 100644 --- a/internal/ids/ids_test.go +++ b/internal/ids/ids_test.go @@ -5,7 +5,7 @@ import ( "testing" ) -var timestampHexRE = regexp.MustCompile(`^pre_\d{8}_\d{4}_[0-9a-f]{4}$`) +var timestampHexRE = regexp.MustCompile(`^pre_\d{8}_\d{4}_[0-9a-f]{8}$`) func TestTimestampHexFormat(t *testing.T) { id := TimestampHex("pre") @@ -18,7 +18,7 @@ func TestTimestampHexPrefixPreserved(t *testing.T) { // Prefixes may contain underscores (e.g. "hide_"); the prefix // must appear verbatim at the start. id := TimestampHex("hide_my_source") - re := regexp.MustCompile(`^hide_my_source_\d{8}_\d{4}_[0-9a-f]{4}$`) + re := regexp.MustCompile(`^hide_my_source_\d{8}_\d{4}_[0-9a-f]{8}$`) if !re.MatchString(id) { t.Errorf("TimestampHex preserved prefix incorrectly: %q", id) } @@ -34,10 +34,11 @@ func TestTimestampHexUniqueness(t *testing.T) { } seen[id] = struct{}{} } - // Within the same minute only the 16-bit hex suffix varies, so a few - // collisions over 1000 draws are expected; a high rate signals a bug. - if collisions > 100 { - t.Errorf("excessive collisions: %d/1000", collisions) + // Within the same minute only the 32-bit hex suffix varies; 1000 draws + // from 2^32 collide with probability ~1e-4, so any collision here is a + // near-certain signal that the suffix width regressed. + if collisions > 0 { + t.Errorf("collisions in 1000 draws: %d (32-bit suffix should make this vanishingly rare)", collisions) } } diff --git a/internal/model/model.go b/internal/model/model.go index d46f791..0a1bd57 100644 --- a/internal/model/model.go +++ b/internal/model/model.go @@ -385,6 +385,14 @@ type Agent struct { // ToolRounds is the maximum tool-call/result cycles per run. // Zero means use Config.MaxToolRounds. ToolRounds int + // DisableThinking, when true, sends chat_template_kwargs.enable_thinking=false + // to the model, suppressing its hidden reasoning trace. False (the zero + // value) leaves the model's default reasoning behavior untouched. Useful + // for agents doing simple, deterministic tool sequencing where a long + // hidden trace adds latency and cost without improving reliability, and + // where it has been observed to make reasoning models occasionally lose + // track of tool calls they already made and repeat them. + DisableThinking bool // QueueInput, when non-empty, names a queue whose items are dequeued one per tick. // Each item's Payload is used for prompt variable substitution in SystemPrompt and UserPrompt. QueueInput string diff --git a/internal/runner/runner.go b/internal/runner/runner.go index d3eea49..f95bb3d 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -284,6 +284,14 @@ func (r *Runner) Run(ctx context.Context, a model.Agent, budget model.TokenBudge var turns []model.Turn var lastResp model.LLMResponse + // completedToolCalls guards against a model re-issuing a tool call it + // already made successfully earlier in this run (observed with reasoning + // models: a long hidden thinking trace can cause the model to lose track + // of a prior tool call and repeat it verbatim, spinning until max tool + // rounds is hit). Scoped to non-"hide" tools only — hide pagination tools + // legitimately expect repeated/similar calls across a run. + completedToolCalls := make(map[string]bool) + for i, userPrompt := range userPrompts { // Apply turn-level vars (may include values extracted from previous tool calls). userPrompt = applyVars(userPrompt, turnVars) @@ -330,6 +338,12 @@ func (r *Runner) Run(ctx context.Context, a model.Agent, budget model.TokenBudge if len(turnTools) > 0 { opts.ExtraBody = map[string]any{"parallel_tool_calls": false} } + if a.DisableThinking { + if opts.ExtraBody == nil { + opts.ExtraBody = map[string]any{} + } + opts.ExtraBody["chat_template_kwargs"] = map[string]any{"enable_thinking": false} + } if userPrompt != "" { r.Log.Debug("adding user prompt", "agent", a.Name, "chars", len(userPrompt)) @@ -342,6 +356,13 @@ func (r *Runner) Run(ctx context.Context, a model.Agent, budget model.TokenBudge } } + // Reasoning models under load sometimes emit the head of the final + // answer alongside a tool call, then continue from where they stopped + // after seeing the tool result. Fragments emitted in tool-call rounds + // are collected here so the final answer keeps every half instead of + // only the last round's text. + var answerParts []string + for round := 0; round < rounds; round++ { reflectionTextTurn := r.ForceTextAfterHide && len(opts.Tools) == 0 // The run-level deadline (set at function entry) bounds this call. @@ -390,9 +411,42 @@ func (r *Runner) Run(ctx context.Context, a model.Agent, budget model.TokenBudge totalTokens.Total += resp.TotalTokens lastResp = resp } + } else if resp.FinishReason == "stop" && resp.Content == "" && len(resp.ToolCalls) == 0 && len(answerParts) == 0 { + // Self-healing retry: observed in production — a reasoning model + // occasionally stops naturally with zero output tokens instead of + // producing the expected answer. Not a truncation (finish_reason + // is "stop", not "length"), just a bad sample; a bare retry of the + // same request is usually enough to draw a non-empty response from + // the same stochastic model. Single-shot, matching the + // length-truncation retry above — if the retry is also empty, it + // is passed through rather than looped on indefinitely. Skipped + // when earlier rounds already banked answer fragments — an empty + // stop then just means the model had nothing left to add. + r.Log.Warn("completion stopped with empty content; retrying", + "agent", a.Name, "round", round) + resp, err = r.Client.Complete(callCtx, a.Model, sess.Messages(), opts) + if err != nil { + wErr := fmt.Errorf("runner/Run %s round %d retry: %w", a.Name, round, err) + return r.errorRecord(a, startTs, wErr), wErr + } + totalTokens.Prompt += resp.PromptTokens + totalTokens.Response += resp.CompletionTokens + totalTokens.Total += resp.TotalTokens + lastResp = resp } if len(resp.ToolCalls) == 0 { + if len(answerParts) > 0 { + // Splice fragments emitted alongside earlier tool calls ahead + // of this round's text (which may be empty when the fragments + // already carried the whole answer). lastResp is reassigned so + // the cache write and output routing below see the full answer. + if resp.Content != "" { + answerParts = append(answerParts, strings.TrimRight(resp.Content, "\n")) + } + resp.Content = strings.Join(answerParts, "\n") + lastResp = resp + } // Final text response — record the turn and continue to next prompt. r.Log.Info("agent completed", "agent", a.Name, "tokens", resp.TotalTokens, "finish_reason", resp.FinishReason) @@ -431,6 +485,24 @@ func (r *Runner) Run(ctx context.Context, a model.Agent, budget model.TokenBudge // The model requested tool calls — validate, execute, and feed results back. r.Log.Info("tool calls requested", "agent", a.Name, "count", len(resp.ToolCalls)) + // Bank any answer text emitted alongside the tool calls: the model + // sees it in the session as already said and continues after it, so + // dropping it here would head-truncate the final answer. Scoped to + // non-hide rounds — text next to hide pagination calls is navigation + // narration, not part of the answer. + if resp.Content != "" { + hideRound := false + for _, tc := range resp.ToolCalls { + if def, ok := toolByName[tc.Name]; ok && def.Type == "hide" { + hideRound = true + break + } + } + if !hideRound { + answerParts = append(answerParts, strings.TrimRight(resp.Content, "\n")) + } + } + // Record the assistant message with its tool call requests. if err := sess.Add(ctx, model.Message{ Role: "assistant", @@ -452,17 +524,36 @@ func (r *Runner) Run(ctx context.Context, a model.Agent, budget model.TokenBudge return r.errorRecord(a, startTs, wErr), wErr } r.Log.Info("executing tool", "agent", a.Name, "tool", tc.Name) + // Debug: log full tool call arguments for diagnostics (tool name/byte-only logs above don't reveal repeated-args loops). + argBytes, argErr := json.Marshal(tc.Arguments) + if argErr == nil { + r.Log.Debug("tool call args", "agent", a.Name, "tool", tc.Name, "args", string(argBytes)) + } if r.ProgressFn != nil { r.ProgressFn(ProgressEvent{Kind: "call", Round: round, Tool: tc.Name, ToolType: def.Type, Args: marshalArgs(tc.Arguments)}) } var result model.ToolResult - if def.Type == "hide" { + dedupeKey := "" + if def.Type != "hide" && argErr == nil { + dedupeKey = tc.Name + "\x00" + string(argBytes) + } + switch { + case dedupeKey != "" && completedToolCalls[dedupeKey]: + r.Log.Warn("skipping duplicate tool call already completed this run", + "agent", a.Name, "tool", tc.Name) + result = model.ToolResult{ + Content: fmt.Sprintf("%s with these exact arguments already completed successfully earlier in this run. Do not call it again — proceed to the next step.", tc.Name), + } + case def.Type == "hide": result = r.executeHideTool(def.Name, tc.ID, tc.Arguments) if result.Error == "" { hideToolSucceeded = true } - } else { + default: result = (&tool.Executor{MCP: r.MCPRegistry, QueueMgr: r.QueueMgr, AgentName: a.Name, Limiter: r.ToolLimiter}).Execute(ctx, def, tc.Arguments) + if result.Error == "" && dedupeKey != "" { + completedToolCalls[dedupeKey] = true + } } if result.Error == "" && def.Buffer { if r.HideBuffer == nil { diff --git a/internal/runner/runner_test.go b/internal/runner/runner_test.go index 7f268d4..0289650 100644 --- a/internal/runner/runner_test.go +++ b/internal/runner/runner_test.go @@ -602,6 +602,60 @@ func TestRunner_MaxRoundsExceeded(t *testing.T) { } } +// TestRunner_DuplicateToolCallSkipsReExecution verifies that when the model +// re-issues an identical tool call (same name and arguments) that already +// succeeded earlier in the run, the runner does not execute it again — it +// returns a synthetic "already completed" result instead. This guards +// against reasoning models that occasionally lose track of a prior tool call +// (observed in production) and repeat it, which would otherwise double a +// real side effect (e.g. posting a comment twice) or spin until max rounds. +func TestRunner_DuplicateToolCallSkipsReExecution(t *testing.T) { + var hits int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits++ + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("posted")) + })) + defer srv.Close() + + reg := tool.NewRegistry() + if err := reg.Register(model.Skill{ + Name: "comment-skill", + Tools: []model.ToolDefinition{{ + Name: "post_comment", + Type: "http", + HTTP: model.HTTPToolConfig{Method: "POST", URL: srv.URL}, + }}, + }); err != nil { + t.Fatalf("Register: %v", err) + } + + sameCall := model.ToolCall{ID: "call-1", Name: "post_comment", Arguments: map[string]any{"body": "hello"}} + mock := session.NewMockLLM(session.MockConfig{ + Response: "done", + ToolCallSequence: [][]model.ToolCall{ + {sameCall}, + {sameCall}, // model repeats the identical call on round 1 + }, + }) + + r := &Runner{Client: mock, Registry: reg, Log: testLogger(t), MaxToolRounds: 5} + a := testAgent("dedupe-agent") + a.Skills = []string{"comment-skill"} + a.UserPrompt = "post the comment" + + rec, err := r.Run(context.Background(), a, testBudget()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rec.Status != model.JobStatusSuccess { + t.Errorf("status = %q, want success", rec.Status) + } + if hits != 1 { + t.Errorf("HTTP tool hits = %d, want 1 (second call should be deduped, not re-executed)", hits) + } +} + func TestRunner_TurnSkillScopeReplacesBaseScope(t *testing.T) { reg := tool.NewRegistry() if err := reg.Register(model.Skill{ @@ -1157,6 +1211,64 @@ func TestRunner_TurnVarNoMatchIsNoop(t *testing.T) { } } +// optsCapturingClient is a session.LLMClient that delegates Complete to a +// MockLLM but records the CompletionOptions passed on the first call, so +// tests can assert on request-shaping behavior (e.g. ExtraBody merging). +type optsCapturingClient struct { + mock *session.MockLLM + firstOpts session.CompletionOptions + haveCalled bool +} + +func (c *optsCapturingClient) Complete(ctx context.Context, modelName string, messages []model.Message, opts session.CompletionOptions) (model.LLMResponse, error) { + if !c.haveCalled { + c.firstOpts = opts + c.haveCalled = true + } + return c.mock.Complete(ctx, modelName, messages, opts) +} + +func (c *optsCapturingClient) CountTokens(messages []model.Message) (int, error) { + return c.mock.CountTokens(messages) +} + +// TestRun_DisableThinkingMergesChatTemplateKwargs verifies that +// Agent.DisableThinking causes the runner to send +// chat_template_kwargs.enable_thinking=false to the model, without +// clobbering the parallel_tool_calls key already set for tool-enabled turns. +func TestRun_DisableThinkingMergesChatTemplateKwargs(t *testing.T) { + reg := tool.NewRegistry() + if err := reg.Register(model.Skill{ + Name: "noop-skill", + Tools: []model.ToolDefinition{{Name: "noop_tool", Type: "http", HTTP: model.HTTPToolConfig{Method: "GET", URL: "http://127.0.0.1:0/"}}}, + }); err != nil { + t.Fatalf("Register: %v", err) + } + + client := &optsCapturingClient{mock: session.NewMockLLM(session.MockConfig{Response: "done"})} + r := &Runner{Client: client, Registry: reg, Log: testLogger(t), MaxToolRounds: 3} + + a := testAgent("thinking-off") + a.Skills = []string{"noop-skill"} + a.UserPrompt = "go" + a.DisableThinking = true + + if _, err := r.Run(context.Background(), a, testBudget()); err != nil { + t.Fatalf("Run: %v", err) + } + + kwargs, ok := client.firstOpts.ExtraBody["chat_template_kwargs"].(map[string]any) + if !ok { + t.Fatalf("chat_template_kwargs not set in ExtraBody: %+v", client.firstOpts.ExtraBody) + } + if enabled, _ := kwargs["enable_thinking"].(bool); enabled { + t.Error("enable_thinking = true, want false") + } + if v, _ := client.firstOpts.ExtraBody["parallel_tool_calls"].(bool); v { + t.Error("parallel_tool_calls should still be false alongside the thinking override") + } +} + // errOnNthCountClient is a session.LLMClient that delegates Complete to a MockLLM // but returns an error from CountTokens after 'failAfter' successful calls. type errOnNthCountClient struct { @@ -1340,6 +1452,99 @@ func TestRun_SelfHealingRetry_OnTruncatedCompletion(t *testing.T) { } } +// emptyStopThenContentClient returns finish_reason "stop" with empty content +// on the first call, then non-empty content on retry — simulating a +// reasoning model that occasionally stops naturally without producing any +// output tokens (observed in production; not a truncation, so finish_reason +// is "stop" rather than "length"). +type emptyStopThenContentClient struct { + mu sync.Mutex + calls int +} + +func (c *emptyStopThenContentClient) Complete(_ context.Context, _ string, _ []model.Message, _ session.CompletionOptions) (model.LLMResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls++ + if c.calls == 1 { + return model.LLMResponse{FinishReason: "stop"}, nil + } + return model.LLMResponse{Content: "final answer after retry", FinishReason: "stop"}, nil +} + +func (c *emptyStopThenContentClient) CountTokens(messages []model.Message) (int, error) { + return 10 * len(messages), nil +} + +func TestRun_SelfHealingRetry_OnEmptyStopCompletion(t *testing.T) { + client := &emptyStopThenContentClient{} + r := &Runner{ + Client: client, + Registry: tool.NewRegistry(), + Log: testLogger(t), + MaxToolRounds: 1, + } + a := testAgent("flaky-agent") + a.UserPrompt = "extract the fields" + + rec, err := r.Run(context.Background(), a, testBudget()) + if err != nil { + t.Fatalf("Run: %v", err) + } + if rec.LastResponse != "final answer after retry" { + t.Errorf("LastResponse = %q, want retried content", rec.LastResponse) + } + client.mu.Lock() + defer client.mu.Unlock() + if client.calls != 2 { + t.Fatalf("Complete calls = %d, want 2 (empty stop + retry)", client.calls) + } +} + +// alwaysEmptyStopClient always returns finish_reason "stop" with empty +// content, so the retry itself also comes back empty — the runner must pass +// the empty result through rather than looping. +type alwaysEmptyStopClient struct { + mu sync.Mutex + calls int +} + +func (c *alwaysEmptyStopClient) Complete(_ context.Context, _ string, _ []model.Message, _ session.CompletionOptions) (model.LLMResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls++ + return model.LLMResponse{FinishReason: "stop"}, nil +} + +func (c *alwaysEmptyStopClient) CountTokens(messages []model.Message) (int, error) { + return 10 * len(messages), nil +} + +func TestRun_EmptyStopRetry_IsSingleShot(t *testing.T) { + client := &alwaysEmptyStopClient{} + r := &Runner{ + Client: client, + Registry: tool.NewRegistry(), + Log: testLogger(t), + MaxToolRounds: 1, + } + a := testAgent("always-empty-agent") + a.UserPrompt = "extract the fields" + + rec, err := r.Run(context.Background(), a, testBudget()) + if err != nil { + t.Fatalf("Run: %v", err) + } + if rec.LastResponse != "" { + t.Errorf("LastResponse = %q, want empty (retry also came back empty)", rec.LastResponse) + } + client.mu.Lock() + defer client.mu.Unlock() + if client.calls != 2 { + t.Errorf("Complete calls = %d, want exactly 2 (one retry, not a loop)", client.calls) + } +} + // lengthWithContentClient returns finish_reason "length" but with non-empty // content on every call, simulating a completion that was truncated after // producing a partial (but non-empty) answer — the retry guard should not @@ -1463,3 +1668,100 @@ func TestRun_SystemPromptOnlyAgent_SurvivesStrictBackend(t *testing.T) { t.Fatalf("status = %q, want success", rec.Status) } } + +// splitAnswerClient emits the head of the answer together with a tool call on +// the first completion, then only the remainder after the tool result — +// simulating a reasoning model under load splitting its final answer across +// tool-call rounds (observed in production with qwen3_xml: the model treats +// the pre-tool-call text as already said and continues from where it stopped). +type splitAnswerClient struct { + mu sync.Mutex + calls int + tail string // content of the post-tool-result completion +} + +func (c *splitAnswerClient) Complete(_ context.Context, _ string, _ []model.Message, _ session.CompletionOptions) (model.LLMResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls++ + if c.calls == 1 { + return model.LLMResponse{ + Content: "PR_NUMBER: 1007\nREPO: acme/voice\nFILES:", + FinishReason: "tool_calls", + ToolCalls: []model.ToolCall{{ID: "call-1", Name: "get_files", Arguments: map[string]any{"pr": "1007"}}}, + }, nil + } + return model.LLMResponse{Content: c.tail, FinishReason: "stop"}, nil +} + +func (c *splitAnswerClient) CountTokens(messages []model.Message) (int, error) { + return 10 * len(messages), nil +} + +func splitAnswerRunner(t *testing.T, client *splitAnswerClient) (*Runner, model.Agent) { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("src/a.go +1 -0")) + })) + t.Cleanup(srv.Close) + + reg := tool.NewRegistry() + if err := reg.Register(model.Skill{ + Name: "files-skill", + Tools: []model.ToolDefinition{{ + Name: "get_files", + Type: "http", + HTTP: model.HTTPToolConfig{Method: "POST", URL: srv.URL}, + }}, + }); err != nil { + t.Fatalf("Register: %v", err) + } + + r := &Runner{Client: client, Registry: reg, Log: testLogger(t), MaxToolRounds: 5} + a := testAgent("split-answer-agent") + a.Skills = []string{"files-skill"} + a.UserPrompt = "extract the fields" + return r, a +} + +func TestRun_SplitAnswerAcrossToolRounds_IsReassembled(t *testing.T) { + client := &splitAnswerClient{tail: " src/a.go +1 -0\nCONCERN_PATHS: none"} + r, a := splitAnswerRunner(t, client) + + rec, err := r.Run(context.Background(), a, testBudget()) + if err != nil { + t.Fatalf("Run: %v", err) + } + want := "PR_NUMBER: 1007\nREPO: acme/voice\nFILES:\n src/a.go +1 -0\nCONCERN_PATHS: none" + if rec.LastResponse != want { + t.Errorf("LastResponse = %q, want the pre-tool-call head spliced ahead of the continuation %q", rec.LastResponse, want) + } + client.mu.Lock() + defer client.mu.Unlock() + if client.calls != 2 { + t.Errorf("Complete calls = %d, want 2", client.calls) + } +} + +func TestRun_SplitAnswer_EmptyFinalStopSkipsBareRetry(t *testing.T) { + // The model said everything alongside the tool call and finishes with an + // empty stop: the banked fragment is the answer, and the empty-stop + // self-healing retry must not burn an extra completion. + client := &splitAnswerClient{tail: ""} + r, a := splitAnswerRunner(t, client) + + rec, err := r.Run(context.Background(), a, testBudget()) + if err != nil { + t.Fatalf("Run: %v", err) + } + want := "PR_NUMBER: 1007\nREPO: acme/voice\nFILES:" + if rec.LastResponse != want { + t.Errorf("LastResponse = %q, want banked fragment %q", rec.LastResponse, want) + } + client.mu.Lock() + defer client.mu.Unlock() + if client.calls != 2 { + t.Errorf("Complete calls = %d, want exactly 2 (no bare retry when fragments are banked)", client.calls) + } +} diff --git a/internal/session/http_client.go b/internal/session/http_client.go index bda2559..4b55c4e 100644 --- a/internal/session/http_client.go +++ b/internal/session/http_client.go @@ -14,27 +14,33 @@ import ( ) // HTTPClient implements LLMClient against any OpenAI-compatible endpoint. +// Timeout is delegated to the caller's context, not here, so all timeout +// logic lives in one place (runner.Run / curing worker's TimeoutSeconds). type HTTPClient struct { endpoint string apiKey string // optional bearer token; empty disables auth - timeout time.Duration http *http.Client } // NewHTTPClient returns an HTTPClient targeting the given base URL. // When apiKey is non-empty it is sent on every request as // `Authorization: Bearer `. The key is never logged. -func NewHTTPClient(endpoint, apiKey string, timeout time.Duration) *HTTPClient { +func NewHTTPClient(endpoint, apiKey string, _ time.Duration) *HTTPClient { return &HTTPClient{ endpoint: strings.TrimRight(endpoint, "/"), apiKey: apiKey, - timeout: timeout, - http: &http.Client{Timeout: timeout}, + http: &http.Client{}, } } // Complete sends a chat completion request to the LLM endpoint and returns // the parsed response. +// +// Timeout is delegated to the caller's context, not the HTTP client, so all +// timeout logic lives in one place (runner.WithTimeout or the curing worker's +// TimeoutSeconds). Using both a context deadline and a separate http.Client +// timeout created a hard-to-debug race where the shorter deadline fired +// unpredictably under load. func (c *HTTPClient) Complete(ctx context.Context, modelName string, messages []model.Message, opts CompletionOptions) (model.LLMResponse, error) { reqBody := map[string]any{ "model": modelName, diff --git a/scripts/profile-run.sh b/scripts/profile-run.sh new file mode 100755 index 0000000..b940f26 --- /dev/null +++ b/scripts/profile-run.sh @@ -0,0 +1,121 @@ +#!/usr/bin/env bash +# profile-run.sh — run any command while sampling system + GPU metrics. +# +# Usage: +# scripts/profile-run.sh +# +# cd examples && WEBHOOK_COUNT=100 BURST_SIZE=25 BURST_DELAY_MAX=0.5 \ +# ../scripts/profile-run.sh make 11 +# +# Samples (1-2 Hz, negligible overhead): +# vmstat.log CPU us/sy/id/wa (WAIT), runnable/blocked, memory, swap, block io +# psi.log kernel pressure stall info (cpu/io/memory, some+full) — the +# true "how stalled are we" signal, sharper than iowait +# gpu.csv GPU util, VRAM used/total, temp, power, SM/mem clocks, pstate +# gpu-procs.csv per-process VRAM (vLLM vs everything else) +# vllm.prom vLLM /metrics samples: running/waiting requests, KV-cache +# usage, prompt/generation token counters, prefix-cache hits +# (override endpoint with VLLM_METRICS_URL) +# sensors.log CPU/board temps and fans (lm_sensors) +# ps-top.log top processes by CPU every 5s (leather vs vLLM attribution) +# iostat.log per-device r/s, w/s, await, %util (sysstat) +# mpstat.log per-core CPU breakdown — catches single-core bottlenecks (sysstat) +# pidstat.log per-process CPU, memory, and disk IO every 5s (sysstat) +# +# Output dir: .state-profiles/run-/ (override with PROFILE_OUT). +# A summary is printed at the end via profile-summary.py. +set -uo pipefail + +if [ $# -eq 0 ]; then + echo "usage: $0 " >&2 + exit 2 +fi + +script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +out="${PROFILE_OUT:-./.state-profiles}/run-$(date +%Y%m%d-%H%M%S)" +mkdir -p "$out" + +pids=() +cleanup() { + for p in "${pids[@]}"; do kill "$p" 2>/dev/null; done + wait 2>/dev/null +} +trap cleanup EXIT INT TERM + +vmstat -t 1 >"$out/vmstat.log" 2>/dev/null & +pids+=($!) + +nvidia-smi \ + --query-gpu=timestamp,utilization.gpu,utilization.memory,memory.used,memory.total,temperature.gpu,power.draw,clocks.sm,clocks.mem,pstate \ + --format=csv,noheader,nounits -l 1 >"$out/gpu.csv" 2>/dev/null & +pids+=($!) + +nvidia-smi \ + --query-compute-apps=timestamp,pid,process_name,used_gpu_memory \ + --format=csv,noheader,nounits -l 1 >"$out/gpu-procs.csv" 2>/dev/null & +pids+=($!) + +(while :; do + echo "=== $(date '+%F %T')" + sensors 2>/dev/null + sleep 2 +done) >"$out/sensors.log" & +pids+=($!) + +(while :; do + echo "=== $(date '+%F %T')" + ps -eo pid,comm,%cpu,%mem,rss --sort=-%cpu | head -12 + sleep 5 +done) >"$out/ps-top.log" & +pids+=($!) + +(while :; do + echo "=== $(date +%s)" + for r in cpu io memory; do + sed "s|^|$r |" "/proc/pressure/$r" 2>/dev/null + done + sleep 2 +done) >"$out/psi.log" & +pids+=($!) + +# sysstat samplers (skipped gracefully if sysstat is not installed). +# S_TIME_FORMAT=ISO forces 24h single-token timestamps for stable parsing. +if command -v iostat >/dev/null 2>&1; then + S_TIME_FORMAT=ISO stdbuf -oL iostat -xz 1 >"$out/iostat.log" & + pids+=($!) +fi +if command -v mpstat >/dev/null 2>&1; then + S_TIME_FORMAT=ISO stdbuf -oL mpstat -P ALL 2 >"$out/mpstat.log" & + pids+=($!) +fi +if command -v pidstat >/dev/null 2>&1; then + S_TIME_FORMAT=ISO stdbuf -oL pidstat -urd 5 >"$out/pidstat.log" & + pids+=($!) +fi + +vllm_url="${VLLM_METRICS_URL:-http://127.0.0.1:8000/metrics}" +(while :; do + echo "=== $(date +%s)" + curl -s --max-time 2 "$vllm_url" | + grep -E '^vllm:(num_requests_running|num_requests_waiting\{|(gpu|kv)_cache_usage_perc|prompt_tokens_total|generation_tokens_total|prefix_cache_(queries|hits)_total)' || true + sleep 2 +done) >"$out/vllm.prom" & +pids+=($!) + +echo "profiling -> $out" +start=$(date +%s) +rc=0 +"$@" || rc=$? +end=$(date +%s) + +cleanup +trap - EXIT INT TERM + +{ + echo "command: $*" + echo "elapsed: $((end - start))s" + echo "exit: $rc" +} | tee "$out/run.txt" + +python3 "$script_dir/profile-summary.py" "$out" | tee -a "$out/run.txt" +exit $rc diff --git a/scripts/profile-summary.py b/scripts/profile-summary.py new file mode 100755 index 0000000..2ff381a --- /dev/null +++ b/scripts/profile-summary.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python3 +"""Summarize a scripts/profile-run.sh output directory: avg/peak CPU, iowait, +memory, GPU util/VRAM/power, and per-sensor peak temperatures.""" +import re +import sys +from pathlib import Path + + +def stats(vals): + if not vals: + return None + return sum(vals) / len(vals), max(vals) + + +def summarize_vmstat(path): + us, sy, wa, free_kb = [], [], [], [] + header = None + for line in path.read_text().splitlines(): + parts = line.split() + if not parts: + continue + if parts[0] == "r" and "us" in parts: + header = parts + continue + if header and parts[0].lstrip("-").isdigit(): + try: + row = dict(zip(header, parts)) + us.append(int(row["us"])) + sy.append(int(row["sy"])) + wa.append(int(row["wa"])) + free_kb.append(int(row["free"])) + except (KeyError, ValueError): + continue + if not us: + return + cpu = [u + s for u, s in zip(us, sy)] + cpu_avg, cpu_max = stats(cpu) + wa_avg, wa_max = stats(wa) + print(f" CPU busy (us+sy): avg {cpu_avg:5.1f}% peak {cpu_max}%") + print(f" IO wait (wa): avg {wa_avg:5.1f}% peak {wa_max}%") + print(f" Mem free trough: {min(free_kb) / 1024 / 1024:.1f} GiB") + + +def summarize_gpu(path): + util, vram, temp, power = [], [], [], [] + total = None + for line in path.read_text().splitlines(): + f = [x.strip() for x in line.split(",")] + # timestamp, util.gpu, util.mem, mem.used, mem.total, temp, power, sm, mem, pstate + if len(f) < 10: + continue + try: + util.append(float(f[1])) + vram.append(float(f[3])) + total = float(f[4]) + temp.append(float(f[5])) + power.append(float(f[6])) + except ValueError: + continue + if not util: + return + u_avg, u_max = stats(util) + v_avg, v_max = stats(vram) + t_avg, t_max = stats(temp) + p_avg, p_max = stats(power) + print(f" GPU util: avg {u_avg:5.1f}% peak {u_max:.0f}%") + print(f" GPU VRAM: avg {v_avg / 1024:5.1f} GiB peak {v_max / 1024:.1f} / {total / 1024:.1f} GiB") + print(f" GPU temp: avg {t_avg:5.1f}C peak {t_max:.0f}C") + print(f" GPU power: avg {p_avg:5.1f}W peak {p_max:.0f}W") + + +def summarize_iostat(path): + # iostat -xz blocks: a "Device ..." header line names the columns; data + # rows follow until the next blank line. Track avg/peak %util and peak + # write await per device. + cols = None + per_dev = {} # dev -> {"util": [], "w_await": [], "wkbs": []} + for line in path.read_text().splitlines(): + parts = line.split() + if not parts: + continue + if parts[0] == "Device": + cols = {name: i for i, name in enumerate(parts)} + continue + if cols and parts[0] not in ("avg-cpu:", "Linux") and len(parts) == len(cols): + try: + d = per_dev.setdefault(parts[0], {"util": [], "w_await": [], "wkbs": []}) + d["util"].append(float(parts[cols["%util"]])) + d["w_await"].append(float(parts[cols["w_await"]])) + d["wkbs"].append(float(parts[cols["wkB/s"]])) + except (KeyError, ValueError): + continue + if not per_dev: + return + # Report the busiest device by peak %util. + dev, d = max(per_dev.items(), key=lambda kv: max(kv[1]["util"], default=0)) + u_avg, u_max = stats(d["util"]) + print(f" Disk {dev + ':':<13} util avg {u_avg:.0f}% peak {u_max:.0f}% " + f"w_await peak {max(d['w_await']):.1f}ms write peak {max(d['wkbs']) / 1024:.0f} MiB/s") + + +def summarize_mpstat(path): + # mpstat -P ALL: per-sample rows for "all" plus one per core. Busy = + # 100 - %idle. Report system average, peak single-core busy, and peak + # iowait from the "all" rows. + cols = None + all_busy, all_wait, core_busy = [], [], [] + for line in path.read_text().splitlines(): + parts = line.split() + if len(parts) < 4 or parts[0] in ("Linux", "Average:"): + continue + if "%idle" in parts: + cols = {name: i - len(parts) for i, name in enumerate(parts)} # from end + continue + if cols is None: + continue + try: + idle = float(parts[cols["%idle"]]) + busy = 100.0 - idle + except (ValueError, IndexError): + continue + cpu = parts[cols["CPU"]] + if cpu == "all": + all_busy.append(busy) + try: + all_wait.append(float(parts[cols["%iowait"]])) + except (ValueError, KeyError): + pass + else: + core_busy.append(busy) + if not all_busy: + return + a_avg, a_max = stats(all_busy) + line = f" CPU (mpstat): avg {a_avg:5.1f}% peak {a_max:.0f}%" + if core_busy: + line += f" busiest single core peak {max(core_busy):.0f}%" + print(line) + if all_wait and max(all_wait) > 0: + print(f" iowait (mpstat): avg {sum(all_wait) / len(all_wait):5.1f}% peak {max(all_wait):.1f}%") + + +def summarize_pidstat(path): + # pidstat -urd interleaves three section types, each introduced by its + # own header. Field positions are taken from the header, indexed from the + # end so the timestamp column width doesn't matter. Command is always last. + section = None + idx = {} + cpu = {} # command -> [%CPU samples] + wr = {} # command -> [kB_wr/s samples] + rss = {} # command -> max RSS kB + for line in path.read_text().splitlines(): + parts = line.split() + if len(parts) < 5 or parts[0] in ("Linux", "Average:"): + continue + if "%CPU" in parts and "Command" in parts: + section, idx = "cpu", {"%CPU": parts.index("%CPU") - len(parts)} + continue + if "RSS" in parts and "Command" in parts: + section, idx = "mem", {"RSS": parts.index("RSS") - len(parts)} + continue + if "kB_wr/s" in parts and "Command" in parts: + section, idx = "io", {"kB_wr/s": parts.index("kB_wr/s") - len(parts)} + continue + if section is None: + continue + cmd = parts[-1] + try: + if section == "cpu": + cpu.setdefault(cmd, []).append(float(parts[idx["%CPU"]])) + elif section == "mem": + rss[cmd] = max(rss.get(cmd, 0.0), float(parts[idx["RSS"]])) + elif section == "io": + wr.setdefault(cmd, []).append(float(parts[idx["kB_wr/s"]])) + except (ValueError, IndexError): + continue + if not cpu: + return + top_cpu = sorted(cpu.items(), key=lambda kv: -max(kv[1]))[:3] + frags = [f"{c} peak {max(v):.0f}% (rss {rss.get(c, 0) / 1024 / 1024:.1f}G)" for c, v in top_cpu] + print(f" Top proc CPU: {' '.join(frags)}") + if wr: + top_wr = sorted(wr.items(), key=lambda kv: -max(kv[1]))[:3] + frags = [f"{c} peak {max(v) / 1024:.1f} MiB/s" for c, v in top_wr if max(v) > 0] + if frags: + print(f" Top proc disk wr: {' '.join(frags)}") + + +def summarize_psi(path): + peaks = {} # (resource, kind) -> max avg10 + for line in path.read_text().splitlines(): + m = re.match(r"^(cpu|io|memory) (some|full) avg10=([0-9.]+)", line) + if m: + key = (m.group(1), m.group(2)) + val = float(m.group(3)) + peaks[key] = max(peaks.get(key, val), val) + if not peaks: + return + parts = [] + for res in ("cpu", "io", "memory"): + some = peaks.get((res, "some")) + full = peaks.get((res, "full")) + if some is not None: + frag = f"{res} some {some:.0f}%" + if full: + frag += f" / full {full:.0f}%" + parts.append(frag) + print(f" PSI peak (avg10): {' '.join(parts)}") + + +def summarize_vllm(path): + gauges = {"num_requests_running": [], "num_requests_waiting": [], + "kv_cache_usage_perc": []} + counters = {} # name -> (first, last) + times = [] + for line in path.read_text().splitlines(): + if line.startswith("=== "): + times.append(int(line[4:])) + continue + m = re.match(r"^vllm:([a-z_]+)(?:\{([^}]*)\})? ([0-9.e+-]+)$", line) + if not m: + continue + name, labels, val = m.group(1), m.group(2) or "", float(m.group(3)) + if name == "gpu_cache_usage_perc": # older vLLM name for the same gauge + name = "kv_cache_usage_perc" + if name in gauges: + gauges[name].append(val) + elif name in ("prompt_tokens_total", "generation_tokens_total", + "prefix_cache_queries_total", "prefix_cache_hits_total"): + first, _ = counters.get(name, (val, val)) + counters[name] = (first, val) + if not times: + return + dur = max(times[-1] - times[0], 1) + for name, label in [("num_requests_running", "vLLM running reqs"), + ("num_requests_waiting", "vLLM waiting reqs")]: + if gauges[name]: + a, mx = stats(gauges[name]) + print(f" {label + ':':<19}avg {a:5.1f} peak {mx:.0f}") + if gauges["kv_cache_usage_perc"]: + a, mx = stats(gauges["kv_cache_usage_perc"]) + print(f" vLLM KV cache: avg {a * 100:5.1f}% peak {mx * 100:.1f}%") + for name, label in [("prompt_tokens_total", "prompt tok/s"), + ("generation_tokens_total", "gen tok/s")]: + if name in counters: + first, last = counters[name] + print(f" vLLM {label + ':':<14}{(last - first) / dur:8.0f} ({last - first:.0f} over {dur}s)") + hits = counters.get("prefix_cache_hits_total") + queries = counters.get("prefix_cache_queries_total") + if hits and queries: + dh, dq = hits[1] - hits[0], queries[1] - queries[0] + if dq > 0: + print(f" vLLM prefix cache: {100 * dh / dq:.1f}% hit rate this run ({dh:.0f}/{dq:.0f} block tokens)") + + +def summarize_sensors(path): + peaks = {} + for line in path.read_text().splitlines(): + m = re.match(r"^([^:]{1,40}):\s+\+?(-?[0-9.]+)\s*°C", line) + if m: + label, val = m.group(1).strip(), float(m.group(2)) + peaks[label] = max(peaks.get(label, val), val) + if not peaks: + return + print(" Sensor peak temps:") + for label, val in sorted(peaks.items(), key=lambda kv: -kv[1])[:8]: + print(f" {label:<24} {val:.0f}C") + + +def main(): + out = Path(sys.argv[1]) + print(f"--- profile summary: {out.name} ---") + for name, fn in [ + ("vmstat.log", summarize_vmstat), + ("mpstat.log", summarize_mpstat), + ("psi.log", summarize_psi), + ("iostat.log", summarize_iostat), + ("pidstat.log", summarize_pidstat), + ("gpu.csv", summarize_gpu), + ("vllm.prom", summarize_vllm), + ("sensors.log", summarize_sensors), + ]: + p = out / name + if p.exists() and p.stat().st_size: + fn(p) + + +if __name__ == "__main__": + main()