diff --git a/CHANGELOG.md b/CHANGELOG.md index 15e3559..0fe0959 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,23 @@ All notable changes to nexum are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] +### Added +- **Caveman prompts — telegraphic plans + dispatch prompts (`caveman_prompts_enabled`, default true).** `/nx-plan` now writes the plan's prose fields (task summary, step `title`/`objective`/`contract`/`scope`) in clipped, telegraphic English — articles, copulas, and filler dropped — and `/nx-build` builds its executor dispatch prompts the same way. The plan is re-read by every executor and the shared dispatch prefix ships on every step, so trimming function words from them is a recurring token saving. Strict carve-outs stay verbatim and unambiguous: file paths, identifiers, signatures, config keys, code, and the runnable `acceptance` command — terseness never costs precision. Set `false` for normal prose. +- **Grep narrowing — a *working* PreToolUse context-savings lever (`scripts/scan_guard.py`).** An unscoped/broad search now has its **output capped** instead of being hard-denied: the `Grep` tool gets a `head_limit` injected and an unscoped recursive Bash `grep`/`rg` gets `| head -n N` appended (via PreToolUse `updatedInput`, which current Claude Code honors — unlike PostToolUse output shrink). The model still gets a bounded answer with no retry round-trip. Searches into a `scan_deny_paths` directory, and Glob (no `head_limit`), still deny; a Bash grep that already pipes falls back to deny; an explicit caller `head_limit` is never overridden. Config: `grep_narrow_enabled` (default true), `grep_head_limit` (default 80). +- **Confidence-aware, bidirectional routing calibration (`store.calibration_advice`, `store.py calib-advice` CLI).** Replaces the raw first-try pass ratio with a **Wilson score lower bound**, so a short lucky/unlucky streak no longer flips a route. Advice is now **bidirectional**: nudge a route *up* a tier when low-confidence (lower bound < `route_calib_min_success_ratio`) and *down* a tier when a cheaper tier reliably suffices (lower bound ≥ `route_calib_downgrade_ratio`, default 0.9). Falls back to a cross-repo `_global` prior when a repo lacks `route_calib_min_samples` of its own history. `/nx-plan` consumes the new `calib-advice` JSON (action/reason/samples/lower/source). Config: `route_calib_downgrade_ratio` (default 0.9; 1.0 disables downgrades). +- **Honest savings split in `/nx-report` (`scripts/report.py`, `store.savings_by_source`).** The report now separates savings into three buckets so the headline never overclaims the inert PostToolUse lever: **Realized** (PreToolUse, measured — `predup`'s exact denied-repeat tokens), **Bounded interventions** (`read_guard`/`grep_narrow` — output capped, exact saving unknowable, counted only), and **Theoretical** (`dedup`/`truncate` — PostToolUse shrink that is inert because Claude Code ignores `updatedToolOutput` for built-in tools, tracking #65403). read-guard and grep-narrow now record a 0-token intervention row so the bounded count is visible. +- **`PreCompact` hook (`scripts/precompact.py`).** Fires at the exact compaction boundary to (a) invalidate this session's `tool_calls` rows — so predup can never deny a re-read of output the compaction just evicted — and (b) write a deterministic handoff skeleton. Never blocks compaction. Closes the predup-after-compaction correctness gap that the `predup_max_age_seconds` time guard could only approximate. +- **`SessionStart` reset/prune hook (`scripts/session_reset.py`).** On `source` `clear`/`compact`, clears `tool_calls` for the same reason as PreCompact (filtered in-script, not via matcher). Also runs the throttled retention prune. +- **`SubagentStop` hook (`scripts/subagent_usage.py`) → real per-tier usage.** Maps a nexum executor agent (`nexum-impl-{haiku,sonnet,opus}`) to its tier and records a `usage` row with token totals parsed best-effort from the subagent transcript (`store.transcript_usage_totals`), so the cost-report breakdown reflects measured spend rather than the plan-preview estimate. (The SubagentStop payload carries no usage fields, so attribution is transcript-based and best-effort — documented as such.) +- **Retention/pruning (`store.prune`, `store.maybe_prune`, `store.py prune` CLI).** Rows older than `retention_days` (default 14) are pruned from the ephemeral tables (`tool_calls`, `savings`, `outputs`, `usage`, `memo`, `file_activity`), throttled to at most once/day on session start, keeping the SQLite file and predup lookups bounded. `0` disables. (`route_calibration`/`step_ledger` are persistent and intentionally not age-pruned.) +- **Fable pricing.** `store.PRICING` gains `fable` ($10/$50 per MTok) and `cost_report._model_key` recognises it (and orders matching most-specific-first), so Fable-class usage is no longer mispriced as Sonnet. Bedrock/Vertex `anthropic.`-prefixed IDs still map correctly. +- **Wasted-context analytics + `/nx-report` (inspired by the open-source `claude-context-optimizer`).** A PostToolUse tracker (extends `dedup.py`, now also matching `Edit|Write|MultiEdit`) records per-file read/edit counts and injected-token estimates in a new `file_activity` table. The new `/nx-report` command (`scripts/report.py`) prints a deterministic, no-LLM session digest: the cost summary plus a wasted-context analysis — total tokens read, tokens spent on files read but never edited, a waste ratio, an S–F efficiency grade, a per-file useful/WASTED table, and concrete "drop X → save ~N tokens" suggestions. New `store` helpers: `record_file_read`, `record_file_edit`, `file_activity_rows`, `wasted_files`. Config `file_activity_enabled` (default true). +- **Tiered budget alerts (C).** `context_watch` now compares the session's real metered cost (the status-line `session_cost` snapshot) to `budget_usd`, and cumulative input+output tokens to `budget_tokens`, emitting an escalating, once-per-tier non-blocking `systemMessage` at `budget_alert_tiers` (default 50/70/80/90%). At ≥70% it names the biggest never-edited files to drop (from the wasted-context tracker); at ≥90% it urges `/compact`. Both budgets default to `0` (disabled). + +### Changed +- **predup recall — canonicalised signatures.** `store.tool_call_sig` now realpath-canonicalises `file_path`/`path` before hashing, so `./foo.py`, `foo.py`, and the absolute path collapse to one signature and predup catches the repeat. Distinct read ranges (`offset`/`limit`) and patterns stay distinct. New `store.clear_tool_calls`. + ## [0.4.0] - 2026-06-18 ### Added - **Estimated per-tier usage recording in `/nx-build`** so the cost report breakdown is populated with token estimates per execution tier, giving visibility into where the session's cost is incurred. diff --git a/SPEC.md b/SPEC.md index 4beeb07..35482dd 100644 --- a/SPEC.md +++ b/SPEC.md @@ -52,8 +52,14 @@ reliably reduce context today are: via `updatedInput` for files above `read_guard_min_bytes` (default 262144 bytes) that do not already carry an explicit limit. Configurable via `config.json`: `read_guard_enabled`, `read_guard_min_bytes`, `read_guard_inject_lines`. -- **Scan-guard blocking** — unscoped recursive greps, broad globs, and reads into - deny paths are blocked via `permissionDecision: deny` before the tool executes. +- **Scan-guard blocking** — reads into deny paths, recursive searches over deny + paths, and unscoped `find`/`ls -R` are blocked via `permissionDecision: deny` + before the tool executes. +- **Grep narrowing** (`grep_narrow_enabled`, default true) — instead of denying a + broad/unscoped *search*, nexum caps its output via `updatedInput`: `head_limit` + on the `Grep` tool, `| head -n grep_head_limit` on an unscoped recursive Bash + `grep`/`rg`. A bounded answer is returned without a retry. Deny-path searches, + Glob, already-piped greps, and caller-set `head_limit` fall back to deny. --- @@ -111,6 +117,25 @@ nexum/ > Net rule: dedup is the authority on the final `updatedToolOutput`; truncate is a > fallback for tools dedup doesn't handle. Keep both wired; dedup wins when it acts. +**Lifecycle hooks (beyond the wiring block above).** The plugin also wires: +`SessionStart` → `resume_nudge.py`, `audit_nudge.py`, `session_reset.py` +(the last clears `tool_calls` on a `clear`/`compact` source and runs the +throttled retention prune); `PreCompact` → `precompact.py` (clears `tool_calls` +at the compaction boundary so predup can't deny a re-read of evicted output, and +writes a handoff skeleton — never blocks compaction); `SubagentStop` → +`subagent_usage.py` (records a real per-tier `usage` row for `nexum-impl-*` +executors, token totals parsed best-effort from the subagent transcript since +the payload carries none). All fail-open: print `{}` / exit 0 on any error. + +**Wasted-context analytics + budget alerts.** The PostToolUse `dedup.py` (matcher +extended to `…|Edit|Write|MultiEdit`) also records per-file read/edit counts and +injected tokens into a `file_activity` table. `/nx-report` (`report.py`) renders +a deterministic digest: cost summary + a wasted-context view (tokens spent on +files read but never edited, a waste ratio, an S–F grade, and "drop X → save ~N +tokens" picks). `context_watch` adds tiered budget alerts keyed to the real +metered `session_cost` (`budget_usd`) and cumulative tokens (`budget_tokens`), +naming the biggest never-edited files at ≥70% and urging `/compact` at ≥90%. + --- ## 2. `scripts/store.py` — foundation (build FIRST; everything imports it) · TIER: standard (Sonnet) @@ -355,16 +380,21 @@ Prompt-driven (uses the Task/subagent mechanism), referencing the agents in 4.2. - **Grep/Glob:** `path` missing or repo-root AND `glob`/`pattern` very broad (`**/*` or `*`); OR path under a `scan_deny_paths` entry. - **Read:** `file_path` under a `scan_deny_paths` entry. -- Action: emit +- Action: by default emit `{"hookSpecificOutput":{"hookEventName":"PreToolUse","permissionDecision":"deny","permissionDecisionReason":"[nexum] — scope the search to a directory or add -maxdepth/path."}}`. - For Bash where a safe narrowing exists, prefer `updatedInput` to inject a path/ - limit instead of denying (only when unambiguous; else deny). + When `grep_narrow_enabled` (default true) and the problem is *result volume* (an + unscoped/broad search, not a deny-path or traversal problem), prefer `updatedInput` + to cap the output instead: inject `head_limit` on the `Grep` tool, or append + `| head -n grep_head_limit` to an unscoped recursive Bash grep that does not + already pipe. Deny-path searches, Glob, already-piped greps, and a caller-set + `head_limit` still deny. - Respect `scan_guard_enabled`; fail-open on any error. - **EDGE CASES:** legitimately scoped commands must pass untouched; a command that already has `-maxdepth` or an explicit non-root path → allow; never deny a plain `Read` of a normal source file. False-positive risk is high — be conservative; when unsure, ALLOW. -- **ACCEPTANCE:** `grep -r foo` → deny; `grep -r foo src/` → allow; `Read node_modules/x` +- **ACCEPTANCE:** `grep -r foo` → narrow to `grep -r foo | head -n 80` (or deny when + `grep_narrow_enabled` is false); `grep -r foo src/` → allow; `Read node_modules/x` → deny; `Read src/app.py` → allow; disabled flag → always allow. ### 5.2 `scripts/context_watch.py` (UserPromptSubmit) · TIER: standard diff --git a/commands/nx-build.md b/commands/nx-build.md index a99514c..9f4d6ef 100644 --- a/commands/nx-build.md +++ b/commands/nx-build.md @@ -1,109 +1,100 @@ --- -description: "Execute a nexum plan file by dispatching steps to the cheapest capable model tier, verifying acceptance, and escalating on failure." +description: "Execute a nexum plan: dispatch each step to the cheapest capable model tier, verify acceptance, escalate on failure, resume across restarts." --- -You are the nexum implementer. You read a plan file produced by `/nx-plan`, execute its steps by delegating to the cheapest capable model tier, and handle retries and escalation. You orchestrate; you do not write step code yourself (except inline, see §4). +You are the nexum implementer. Read a `/nx-plan` plan file, run its steps via the cheapest capable tier, handle retries/escalation. You orchestrate; you don't write step code (except inline, §4). -**Orchestration is mechanical** — parsing the plan, dispatching, reading verdicts, and branching do **not** require Opus. Do not assume this command runs on Opus; only `needs-strong` *step content* needs Opus, and that is delegated to a subagent (§4). Keeping the driver cheap is a standing saving on every run. +**Orchestration is mechanical** — parse, dispatch, read verdicts, branch. Does NOT need Opus. Don't assume Opus; only `needs-strong` *step content* needs Opus, delegated to a subagent (§4). Cheap driver = standing saving. -## 1. Locate and read the plan file +**Output: terse, minimal. No prose, no narration.** Print only: cost preview (§1b), resume-skips, escalations, failures, final cost report (§10). NO per-step success chatter. -Resolve the data directory (same logic as `store.py`): `$CLAUDE_PLUGIN_DATA`, else `${CLAUDE_PLUGIN_ROOT}/.nexum-data`, else `./.nexum-data`. The plan file is at `/plan/.md` where session id comes from `$CLAUDE_SESSION_ID` (or `_nosession`). +## 1. Locate + read plan -Read the plan file in full. Parse out every step: its index, `route`, `files`, `objective`, `contract`, `scope`, and `acceptance`. +Data dir (same as `store.py`): `$CLAUDE_PLUGIN_DATA`, else `${CLAUDE_PLUGIN_ROOT}/.nexum-data`, else `./.nexum-data`. Plan: `/plan/.md`, session id = `$CLAUDE_SESSION_ID` (else `_nosession`). -If the plan file does not exist, stop and tell the user: `[nexum] No plan found for this session. Run /nx-plan first.` +Read plan in full. Parse every step: index, `route`, `files`, `objective`, `contract`, `scope`, `acceptance`. -Read the effective config once: `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py config`. The keys that drive this command are `dispatch_granularity` (`group` | `step`), `max_same_tier_retries` (default 1), `orchestrator_resume_enabled` (default true), and the model tiers. +No plan file → stop: `[nexum] No plan found for this session. Run /nx-plan first.` -## 1a. Resume from the step ledger (skip work already done) +Read config once: `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py config`. Drivers: `dispatch_granularity` (`group`|`step`), `max_same_tier_retries` (1), `orchestrator_resume_enabled` (true), `caveman_prompts_enabled` (true, §5), tiers. -The orchestrator persists each step's outcome to a durable **step ledger** so a session that died mid-plan — context/plan limit, crash, or an interrupted background dispatch — resumes instead of redoing completed steps. This is the main anti-wastage lever across restarts. +## 1a. Resume from ledger (skip done work) -If `orchestrator_resume_enabled` is true (default): +Durable step ledger lets a dead-mid-plan session resume instead of redoing. Main anti-wastage lever across restarts. -1. Compute the **plan hash** (the ledger key that ties saved state to *this* plan's content; editing the plan changes the hash and discards stale state automatically): +If `orchestrator_resume_enabled` (default true): + +1. Plan hash (ledger key; editing plan changes hash → stale state discarded): `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py plan-hash --file ` -2. Load any prior state: - `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py step-list --session --plan-hash ` -3. For each step, treat the ledger as the source of truth for what's already finished: - - **`done`** → **skip it.** Do not re-dispatch, do not re-run acceptance. Print `[nexum] Step already done (resumed) — skipping.` - - **`failed`** → resume mid-ladder, do **not** restart the retry/escalation ladder from scratch. Read the row's `attempts` and `tier_used` and continue from there: re-dispatch as a patch-retry (§7) seeded with the saved `last_diff` and `verdict`, at the tier the ladder had already reached (`tier_used`), counting the prior `attempts` against `max_same_tier_retries` before escalating. This avoids re-spending escalations a previous session already paid for. - - **`pending` or absent** → execute normally. +2. `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py step-list --session --plan-hash ` +3. Per step, ledger = source of truth: + - **done** → skip. No re-dispatch, no re-run. Print `[nexum] Step already done (resumed) — skipping.` + - **failed** → resume mid-ladder, don't restart it. Read `attempts` + `tier_used`, continue from there: patch-retry (§7) seeded with saved `last_diff`/`verdict`, at `tier_used`, counting prior `attempts` against `max_same_tier_retries` before escalating. + - **pending / absent** → run normally. -If every step is already `done`, report that the plan is complete and skip to the cost summary (§10). Hold the `plan_hash` for the rest of the run — you record every verdict against it (§6a). +All done → report complete, skip to §10. Hold `plan_hash` for the run (every verdict records against it, §6a). -## 1b. Cost preview (show projected savings up front) +## 1b. Cost preview -If `plan_preview_enabled` is true (the default), run: +If `plan_preview_enabled` (default), run and print verbatim **before any dispatch**: ``` python3 ${CLAUDE_PLUGIN_ROOT}/scripts/plan_preview.py --plan ``` -and print its output verbatim **before dispatching any steps**, so the user sees the projected cost per tier and the savings vs an all-opus run at the start of execution. - -The preview is a per-step token heuristic (an estimate, not measured usage); the authoritative post-run numbers — capturing cache writes/reads and actual token counts — still come from the §10 cost report after all steps complete. - -## 2. Group steps by route - -Partition steps into three ordered groups and execute them in this order: +Heuristic estimate, not measured. Authoritative numbers = §10 after run. -1. **mechanical** (Haiku tier) -2. **standard** (Sonnet tier) -3. **needs-strong** (Opus tier) +## 2. Group by route -Grouping keeps each model's prompt prefix stable, maximising cache hits. +Order: 1. mechanical (Haiku) → 2. standard (Sonnet) → 3. needs-strong (Opus). Stable prefix per model = cache hits. -**Dependencies override tier order.** A correct plan routes steps so dependencies never run before their prerequisites (see the planner's dependency-vs-tier rule), but verify it yourself and adapt: never dispatch a step before a step it depends on, even if that means running a cheaper tier *after* a costlier one. Concretely — a test step that exercises code written in another step must run after that step; and a final full-suite / verification step always runs **last**, regardless of its tier. When tier order and dependency order conflict, dependency order wins. +**Deps override tier order.** Never dispatch a step before a step it depends on, even if that runs a cheaper tier after a costlier one. Test exercising another step's code runs after it. Final full-suite/verify step runs **last**, any tier. Dep order wins on conflict. -## 3. Dispatch granularity — batch by tier (default) vs per-step +## 3. Dispatch granularity (main cost lever) -**This is the main cost lever.** Read `dispatch_granularity` from config: +Read `dispatch_granularity`: -- **`group` (default):** Send the route group to **one** executor dispatch. The executor reads the shared spec and source files once and reuses them across every step in the group — one warm context, one cached prefix, instead of N cold starts that each re-derive the same context. The executor returns a per-step result list. -- **`step`:** Send one dispatch per step. More isolation (a mid-batch failure can't affect siblings), but pays a cold start and re-reads context for every step. Use only when steps in a tier are large or risky enough that isolation outweighs the re-derivation cost. +- **group** (default) → whole route group = ONE dispatch. Executor reads shared spec + sources once, reuses across steps. One warm context, one cached prefix vs N cold starts. Returns per-step result list. +- **step** → one dispatch per step. More isolation, pays cold start + re-reads context each time. Use only when a tier's steps are large/risky enough that isolation beats re-derivation. -**Cap the batch size — compute the split, don't eyeball it.** Under `group`, a single oversized dispatch can overflow the executor's own context (forcing a mid-batch compaction that re-derives everything and wipes the token savings, or — as observed — stalling the stream watchdog and losing the whole batch) and widens the blast radius if one step fails. Do not judge "too big" by eye — ask the helper for the deterministic, order-preserving partition of the tier's step indices (in execution order): +**Cap batch size — compute, don't eyeball.** Oversized dispatch overflows executor context (mid-batch compaction wipes savings, or stalls the stream watchdog and loses the batch) and widens blast radius. Get the deterministic order-preserving partition: ``` -python3 ${CLAUDE_PLUGIN_ROOT}/scripts/plan_preview.py --plan --indices "" --root +python3 ${CLAUDE_PLUGIN_ROOT}/scripts/plan_preview.py --plan --indices "" --root ``` -This is **size-aware**: it estimates each step's context load (a per-step base + the byte size of its declared `files` ÷ 4) and packs the tier into sub-batches bounded by **both** `max_dispatch_context_tokens` (config, default 50000) and the `max_steps_per_dispatch` count cap (config, default 4). So a tier of a few large-file steps splits more aggressively than a tier of small ones, while a single step over the token budget still dispatches alone (a step is never split). It returns a JSON array of sub-batches (e.g. `[[1,2,3,4],[5,7]]`). Dispatch the sub-batches **sequentially**; each reuses the same shared-context prefix, so the cache stays warm while each dispatch is bounded. Order is preserved, so a dependent step never runs before its prerequisite. +Size-aware: per-step base + file bytes ÷ 4, packed under both `max_dispatch_context_tokens` (default 50000) and `max_steps_per_dispatch` count cap (default 4). Returns JSON sub-batches (e.g. `[[1,2,3,4],[5,7]]`). Dispatch sub-batches **sequentially**; each reuses the shared prefix → cache stays warm, each dispatch bounded. Order preserved → dep never before prereq. -(The older count-only helper `store.py plan-batches --indices "…"` still exists and caps by `max_steps_per_dispatch` alone — use it only if a plan file isn't available to size against.) +(Older count-only `store.py plan-batches --indices "…"` caps by `max_steps_per_dispatch` alone — use only if no plan file to size against.) -## 4. Skip the spawn when the tier already matches the session model +## 4. Skip spawn when tier == session model -A subagent exists to run a step on a **different** model than the main session without invalidating the main conversation's prompt cache. If a step's tier is the **same** model you (the orchestrator) are already running, spawning buys nothing — it only adds a cold start. In that case, implement the step(s) **inline** in this conversation instead of dispatching. +Subagent earns its keep only by running a *different* model without trashing the main cache. Step tier == your model → implement **inline**, don't dispatch (saves a cold start). -Determine the current session model from context (e.g. the model shown in the status line). Then: - -| Step route | If session model ≠ tier | If session model = tier | +| Route | session model ≠ tier | session model = tier | |---|---|---| -| `mechanical` | dispatch `nexum-impl-haiku` | implement inline | -| `standard` | dispatch `nexum-impl-sonnet` | implement inline | -| `needs-strong` | dispatch `nexum-impl-opus` | implement inline | +| mechanical | dispatch `nexum-impl-haiku` | inline | +| standard | dispatch `nexum-impl-sonnet` | inline | +| needs-strong | dispatch `nexum-impl-opus` | inline | -When in doubt about the session model, dispatch — a redundant spawn is cheaper than a cache-trashing model switch. +Doubt about session model → dispatch (redundant spawn < cache-trashing switch). -## 5. Build each delegation (stable-prefix-first for caching) +## 5. Build delegation (stable-prefix-first) -For a dispatched group (or step), construct the prompt **shared/stable content first, variable content last**, so the longest common prefix is cacheable: +Shared/stable content first, variable last → longest cacheable common prefix: ``` [SHARED CONTEXT — identical for every step in this group] You are a nexum executor. Implement the step(s) below, in order, in this one context. -Global constraints (apply to every step): +Global constraints (every step): - -- Fail-open where the plan requires it; emit deterministic JSON where required. -- Do not touch files outside each step's declared scope. -- After each step, run guardrail.py for that step and return its verbatim JSON. Invoke it as: - `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/guardrail.py --acceptance "" --changed "" --deny-path ""` - Pass the step's scope exclusions straight through as `--deny-path` (repeatable) — do not invert them into an allow-list. Paths may be absolute or repo-relative; the guardrail normalises both. +- Fail-open where required; emit deterministic JSON where required. +- Don't touch files outside each step's declared scope. +- After each step, run guardrail.py and return its verbatim JSON: + `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/guardrail.py --acceptance "" --changed "" --deny-path ""` + Pass scope exclusions straight through as `--deny-path` (repeatable) — do NOT invert to an allow-list. Abs or repo-relative both fine. -[STEP-SPECIFIC CONTENT — all steps in this group, each copied verbatim] +[STEP-SPECIFIC — all steps in group, verbatim] ### Step : - files: <...> - objective: <...> @@ -112,93 +103,86 @@ Global constraints (apply to every step): - acceptance: <...> ``` -Copy the step fields verbatim from the plan — do not summarise or compress them. +Copy step fields verbatim — don't summarise/compress. (When `caveman_prompts_enabled`, the plan's step prose is already caveman, so verbatim copying carries it through.) -## 6. The executor runs the guardrail; you read the verdict +**Caveman dispatch prompt.** If `caveman_prompts_enabled` (default true), write the SHARED CONTEXT block and any instructions *you* author in clipped, telegraphic English (drop articles/copulas/filler) — that prefix ships on every dispatch. Keep EXACT, never caveman-ify: the `guardrail.py` invocation, all paths, identifiers, config keys, and each step's verbatim `acceptance`/`contract`. -Executors run `guardrail.py` themselves as their final action per step and return the **verbatim JSON** (`{"pass": bool, "acceptance_rc": int, "scope_violations": [...], "log": "..."}`). Do **not** re-run the guardrail from the orchestrator for passing steps — that is a redundant round-trip. `guardrail.py` is deterministic, so the returned JSON is trustworthy. +## 6. Executor runs guardrail; you read verdict -**If a dispatch ran in the background** (you received only a completion *notification*, not the executor's final message), you do not have its guardrail JSON. Retrieve the executor's result before proceeding — message the agent for its verbatim per-step JSON — or, if that is unavailable, re-run each step's `acceptance` yourself once from the repo root and treat that as the verdict. Never mark a backgrounded step done on the notification alone. +Executor runs `guardrail.py` itself, returns verbatim JSON (`{"pass": bool, "acceptance_rc": int, "scope_violations": [...], "log": "..."}`). Don't re-run guardrail for passes — redundant; it's deterministic, trust it. -For each returned step verdict: +**Background dispatch** (you got only a completion *notification*, not the final message) → no guardrail JSON. Get the result before proceeding: message the agent for its verbatim per-step JSON, or re-run each step's `acceptance` once from repo root as the verdict. Never mark a backgrounded step done on the notification alone. -- **`pass` is true:** proceed. (Spot-check by re-running the guardrail yourself only if a verdict looks implausible — e.g. claims pass on a step it also reports it could not complete.) -- **`pass` is false:** follow the retry/escalation ladder in §7. +- **pass true** → proceed. (Spot-check by re-running guardrail only if a verdict looks implausible.) +- **pass false** → §7. -## 6a. Record every verdict to the ledger (so a restart can resume) +## 6a. Record every verdict (so restart resumes) -Immediately after reading each step's verdict — **before dispatching the next step** — persist it. This is what makes resume work: if the session dies on the next step, this one is already banked. Record as soon as you know the outcome, not in a batch at the end. +Right after reading each verdict — **before the next step** — persist. This is what makes resume work. -- **On pass:** - `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py step-set --session <session_id> --plan-hash <plan_hash> --index <N> --status done --title "<title>" --route <route> --tier <inline|haiku|sonnet|opus>` -- **On fail (before retrying):** record `--status failed`, and persist the attempt's diff and guardrail verdict so a post-restart retry can patch instead of reimplementing. Write the diff and verdict JSON to temp files and pass them (avoids shell-quoting large/multiline content): +- **Pass:** `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py step-set --session <session_id> --plan-hash <plan_hash> --index <N> --status done --title "<title>" --route <route> --tier <inline|haiku|sonnet|opus>` +- **Fail (before retry):** `--status failed`, persist diff + verdict for a post-restart patch. Write diff + verdict JSON to temp files (avoids shell-quoting): `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py step-set --session <session_id> --plan-hash <plan_hash> --index <N> --status failed --title "<title>" --route <route> --tier <tier> --attempts <n> --diff-file <path> --verdict-file <path>` -When a previously-failed step later passes, overwrite it with `--status done` as above. +Previously-failed step later passes → overwrite `--status done`. -If `orchestrator_resume_enabled` is false, skip all ledger writes. +`orchestrator_resume_enabled` false → skip all ledger writes. -**Immediately after recording the ledger for each step**, also record estimated usage and (when enabled) calibration counters: +**After the ledger write**, record usage + (if enabled) calibration. -**Usage recording (always; read heuristics from `store.py config`):** +Usage (always; heuristics from `store.py config`): ``` python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py record-usage \ --session <session_id> \ - --model <model id for the tier actually used: inline → current session model; dispatched → the tier's model> \ - --input-tok <plan_preview_input_tok_per_step × number of steps in this dispatch> \ - --output-tok <plan_preview_output_tok_per_step × number of steps in this dispatch> + --model <tier actually used: inline → current session model; dispatched → tier's model> \ + --input-tok <plan_preview_input_tok_per_step × steps in this dispatch> \ + --output-tok <plan_preview_output_tok_per_step × steps in this dispatch> ``` -Read `plan_preview_input_tok_per_step` and `plan_preview_output_tok_per_step` from `store.py config`. These are per-step heuristic estimates; multiply by the number of steps in the dispatch to get the total. An escalated step counts against the higher tier (the one that actually ran), not the original dispatched tier. +Escalated step counts against the higher tier that ran, not the original. -**Calibration recording (when `route_calib_enabled` from `store.py config` is true):** +Calibration (when `route_calib_enabled`): ``` -python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py calib-record \ - --repo <repo key: git toplevel basename of the repo root> \ - --route <step route> \ - --dispatched 1 +python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py calib-record --repo <git toplevel basename of repo root> --route <route> --dispatched 1 ``` -Append `--passed-first-try 1` when the step passed on the first attempt with no escalation. Append `--escalated 1` when the step required escalation to a higher tier. Usage recording is always performed; calibration is skipped only when `route_calib_enabled` is false. - -## 7. Retry and escalation ladder - -On a FAIL: - -1. **Retry same tier, up to `max_same_tier_retries` (default 1).** Re-dispatch the *failing step only* to the same tier, appending the guardrail failure (`acceptance_rc`, `scope_violations`, `log` tail) **and the diff the previous attempt produced**, instructing the executor to *patch* that diff rather than reimplement from the spec. Patching is fewer tokens and lands first-try more often. +Add `--passed-first-try 1` if passed first attempt, no escalation. Add `--escalated 1` if escalated. Usage always recorded; calibration skipped only when `route_calib_enabled` false. -2. **Escalate one tier and retry once** (haiku → sonnet → opus). Hand the higher tier the failed diff plus the guardrail output, again instructing it to patch. For an escalated/`needs-strong` step, also invoke `nexum-reviewer` on the produced diff (see §8). +## 7. Retry + escalation ladder -3. **If the escalated attempt also fails** (or a `needs-strong` step on Opus keeps failing): stop that step, report the full guardrail output to the user, and ask whether to skip or abort. Never silently continue past a failing step. Never delegate a `needs-strong` step to a weaker model. +On FAIL: -**Update the ledger on every rung.** Each time you re-dispatch a failing step, first write the latest failed attempt to the ledger via §6a with the *cumulative* `--attempts` count and the `--tier` you are now at. That is what lets a resumed session (§1a) continue this ladder from the right rung instead of re-spending retries. When the step finally passes, record it `done`. +1. **Retry same tier**, up to `max_same_tier_retries` (1). Re-dispatch the *failing step only*, append guardrail failure (`acceptance_rc`, `scope_violations`, `log` tail) + the previous diff, instruct **patch that diff**, don't reimplement. Fewer tokens, lands first-try more often. +2. **Escalate one tier, retry once** (haiku→sonnet→opus). Hand higher tier the failed diff + guardrail output, again patch. Escalated/`needs-strong` → also run `nexum-reviewer` on the diff (§8). +3. **Escalated also fails** (or needs-strong on Opus keeps failing) → stop the step, report full guardrail, ask skip or abort. Never silently continue past a failing step. Never demote a `needs-strong` step. -## 8. Gate the reviewer +**Update ledger every rung.** Before each re-dispatch, write the latest failed attempt (§6a) with *cumulative* `--attempts` and the current `--tier` — that's what lets a resumed session continue from the right rung. Final pass → record `done`. -The guardrail (acceptance + scope) is the routine review, so a step that passes its guardrail does **not** get a separate reviewer pass. Invoke `nexum-reviewer` only for: steps that failed and were escalated, `needs-strong` steps, or steps that touched many files. This avoids doubling requests on the common path. +## 8. Reviewer gate -## 9. Progress reporting +Guardrail (acceptance + scope) = the routine review. A step passing its guardrail gets NO separate reviewer pass. Invoke `nexum-reviewer` only for: failed+escalated steps, `needs-strong` steps, many-file steps. Avoids doubling requests on the common path. -After each step (or group) succeeds, print one line: -`[nexum] Step <N> done (<route>, <inline | haiku | sonnet | opus>) — acceptance passed.` +## 9. Progress (minimal) -On escalation: -`[nexum] Step <N> escalated from <old tier> to <new tier> after <N> failures.` +No per-step success lines. Print only: +- resume-skips (§1a) +- escalations: `[nexum] Step <N> escalated <old>→<new> after <N> failures.` +- failures / abort prompts (§7.3) ## 10. Cost summary -After all steps complete (or after an abort), run: +After all steps (or abort): ``` python3 ${CLAUDE_PLUGIN_ROOT}/scripts/cost_report.py --session <session_id> ``` -Print the output so the user sees actual cost vs. all-opus baseline (tiering breakdown) and the metered, cache-accurate total captured by the status line. +Print verbatim — actual cost vs all-opus baseline + the metered status-line total. ## 11. Constraints -- Never skip the guardrail — every step must pass it (run by the executor, or inline) before the next begins. +- Never skip the guardrail — every step passes it (executor or inline) before the next. - Never modify the plan file during execution. -- Keep the shared-context prefix identical across all steps within a route group so the cache prefix is maximally stable. +- Keep the shared-context prefix identical across all steps in a route group (max cache stability). diff --git a/commands/nx-plan.md b/commands/nx-plan.md index 40dd421..6bce83c 100644 --- a/commands/nx-plan.md +++ b/commands/nx-plan.md @@ -1,80 +1,92 @@ --- -description: "Produce a step-by-step implementation plan for the current task, routing each step to the cheapest model that can execute it reliably." +description: "Decompose the task into routed, self-contained steps and write a plan file. Each step routes to the cheapest model that can run it." model: opus --- -You are the nexum planner. Your job is to decompose the user's task into a precise, ordered sequence of self-contained steps and write them to a plan file. A weaker model (Haiku or Sonnet) will later execute each step in isolation, so every step must carry ALL the context it needs inline — no assumptions, no references to "the previous step." +You are the nexum planner. Decompose task into ordered, self-contained steps. Write plan file. Do NOT implement. Weaker model runs each step later in isolation — every step carries ALL its context inline. No references to "previous step." -## 1. Locate the plan file path +Output: terse. No prose. Final output = plan path + one line per step. -Resolve the data directory using the same priority used by `store.py`: -1. `$CLAUDE_PLUGIN_DATA` if set -2. `${CLAUDE_PLUGIN_ROOT}/.nexum-data` if `CLAUDE_PLUGIN_ROOT` is set -3. `./.nexum-data` otherwise +## 1. Plan path -The plan file lives at `<data_dir>/plan/<session_id>.md`. Use the current session id (available as `$CLAUDE_SESSION_ID` in the environment, or use the string `_nosession` if absent). Create the `plan/` subdirectory if it does not exist, then write the file. +Data dir: `$CLAUDE_PLUGIN_DATA`, else `${CLAUDE_PLUGIN_ROOT}/.nexum-data`, else `./.nexum-data`. +Plan file: `<data_dir>/plan/<session_id>.md`. Session id = `$CLAUDE_SESSION_ID`, else `_nosession`. mkdir `plan/` first. -## 2. Routing rubric +## 2. Routing -Assign every step exactly one route. Use the cheapest tier that is sufficient: +Every step: one route. Cheapest sufficient tier. -- **mechanical** — dispatch to Haiku. Use only when ALL of the following are true: - - The work is boilerplate, a mechanical refactor, a test scaffold, or a well-specified single-file CRUD operation. - - The full specification fits in a short prompt with no ambiguity. - - A concrete acceptance test (a runnable command or assertion) can be stated. - - No architectural judgment is required. +- **mechanical** (Haiku) — ALL true: boilerplate / mechanical refactor / test scaffold / well-specified single-file CRUD; full spec fits short prompt, no ambiguity; runnable acceptance statable; no architectural judgment. +- **standard** (Sonnet) — default. Multi-file, some reasoning, not clearly mechanical, not deep architecture. +- **needs-strong** (Opus, `nexum-impl-opus` or inline) — architecture, cross-cutting many files, ambiguous requirements, complex cross-layer debug. -- **standard** — dispatch to Sonnet. This is the default for most implementation work: multi-file changes, logic that requires some reasoning, anything not clearly mechanical and not requiring deep architectural thought. +Doubt → standard. -- **needs-strong** — route to the Opus tier (`nexum-impl-opus`, or implemented inline when the session is already on Opus). Reserve for: architecture decisions, cross-cutting concerns that span many files, ambiguous requirements needing interpretation, complex debugging across layers. - -When in doubt, prefer **standard** over **mechanical** — a false mechanical that fails wastes more time than a conservative standard. - -**Calibration nudge (per-repo history).** Before finalising routes, consult this repo's historical outcomes when `route_calib_enabled` (from `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py config`) is true. Run: +**Calibration.** If `route_calib_enabled` (from `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py config`): ``` -python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py calib-list --repo <git toplevel basename of cwd> +python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py calib-advice --repo <git toplevel basename of cwd> ``` -For any route whose `dispatched >= route_calib_min_samples` (config, default 5) **and** `passed_first_try / dispatched < route_calib_min_success_ratio` (config, default 0.6), nudge the steps you would have assigned to that route **up one tier** (mechanical→standard, standard→needs-strong) and note the calibration reason in that step's `objective` (e.g. "routed up from mechanical: this repo's mechanical steps pass first-try only 40% of the time"). This is **advisory and one-directional**: never downgrade a route, and never move a `needs-strong` step. When calibration is disabled, the repo has no rows, or a route's sample count is below `route_calib_min_samples`, route by the static rubric above unchanged. +Returns JSON per route: `action` (`up`/`down`/`keep`), `reason`, `samples`, Wilson `lower`, `source`. Apply, noting `reason` in the step's `objective`: + +- **up** → nudge that route's steps up one tier (mechanical→standard, standard→needs-strong). Fires when Wilson lower < `route_calib_min_success_ratio` (default 0.6). +- **down** → nudge down one tier (needs-strong→standard, standard→mechanical). Fires only when lower ≥ `route_calib_downgrade_ratio` (default 0.9). +- **keep** / absent → static rubric. -**Dependency-vs-tier rule.** The implementer executes tiers in the order mechanical → standard → needs-strong, so a step must never be routed to a tier that runs *before* a step it depends on. If step B consumes step A's output, B's tier must be the same as or costlier than A's. In particular: a test step that exercises code written in a `standard` step is itself `standard` (not `mechanical`); and a final full-suite / verification step takes the highest tier of any step it validates (or, if you keep it `mechanical`, state explicitly in its `objective` that it runs last). Ordering steps so prerequisites come first is not enough — the tier assignment must also respect the dependency, or the cheaper tier will run first and fail. +Advisory + conservative: never move needs-strong up, never move mechanical down. Disabled / no evidence → rubric unchanged. -## 3. Step schema +**Dependency beats tier.** Implementer runs tiers mechanical→standard→needs-strong. Step never routed to a tier running *before* a step it depends on. B consumes A → B tier ≥ A tier. Test of a standard step is standard (not mechanical). Final full-suite/verify step = highest tier of what it validates (or keep mechanical but state "runs last" in objective). Ordering prereqs first is not enough — tier must respect the dep too. -Every step MUST include all six fields, in this exact format: +## 3. Step schema (all six fields, this format) ``` ### Step N: <short title> - route: mechanical | standard | needs-strong -- files: <explicit comma-separated list of absolute or repo-relative paths to read, create, or edit> -- objective: <what this step accomplishes, stated in one or two sentences, scoped to this step only> -- contract: <the exact signatures, interfaces, data shapes, or file outputs that later steps depend on — be explicit enough that a model with no other context can satisfy them> -- scope: do NOT touch <explicit list of files or directories that are out of bounds for this step> -- acceptance: <a single runnable shell command or assertion that returns exit 0 on success, e.g. `python3 -m pytest tests/test_store.py -q` or `python3 -c "import store; store.db()"`> +- files: <explicit comma-separated absolute/repo-relative paths to read/create/edit> +- objective: <what this step does, 1-2 sentences, this step only> +- contract: <exact signatures/interfaces/data shapes/file outputs later steps depend on — explicit enough for a model with no other context> +- scope: do NOT touch <explicit out-of-bounds files/dirs> +- acceptance: <one runnable shell command/assertion, exit 0 = pass, e.g. `python3 -m pytest tests/test_store.py -q`> ``` -No field may be omitted or left empty. If a field genuinely has nothing to list (e.g. scope has no exclusions), write `none` — do not omit the key. +No field omitted/empty. Nothing to list → write `none`. -## 4. Self-containedness requirement +## 4. Self-contained -Because steps execute independently on models that share no context with each other or with you, each step's `objective`, `contract`, and `acceptance` must be fully self-explanatory. Specifically: +Steps run independently on models sharing no context. So: -- Name every file path explicitly. Do not write "the file from step 2" — write the actual path. -- State all relevant config keys, constants, or interfaces in the `contract` field so the executor does not have to infer them. -- The `acceptance` command must be copy-pasteable and runnable from the repo root with no substitutions. -- If a step depends on output from a prior step (e.g. an import), name that dependency in `objective` and state the expected interface in `contract`. +- Name every path explicitly. Not "the file from step 2" — the actual path. +- State all config keys / constants / interfaces in `contract`. +- `acceptance` = copy-pasteable, runnable from repo root, no substitutions. +- Dep on prior step → name it in `objective`, state expected interface in `contract`. -## 5. Plan file format +## 4a. Caveman style (token-saving feature) + +If `caveman_prompts_enabled` (from `python3 ${CLAUDE_PLUGIN_ROOT}/scripts/store.py config`, default true), write the plan's **prose** in clipped, telegraphic English — the plan is re-read by every executor, so fewer function words = fewer tokens each run. + +Apply to: task summary, step `title`, `objective`, `contract` prose, `scope` prose. Drop articles (a/an/the), copulas (is/are/be), and filler. Imperative, clipped. -Write the plan file as a Markdown document with this structure: +**Keep EXACT — never caveman-ify or abbreviate:** file paths, identifiers, function/class/type names, signatures, config keys, code, the `route` value, the `files` list, and the `acceptance` command (must stay copy-pasteable + runnable from repo root). + +**Caveman ≠ vague.** Drop only function words, never information. A `contract` is read cold by a weaker executor — it must stay unambiguous. If trimming a word loses precision, keep the word. + +Example: +- normal objective: "Add a function that reads the config file and returns the merged dict." +- caveman: "Add function: read config file, return merged dict." +- normal contract: "`def get_config() -> dict` returns the defaults merged with config.json." +- caveman: "`get_config() -> dict`. Return defaults merged with config.json." (signature verbatim) + +Flag false → normal prose. + +## 5. Plan file format ```markdown # Plan: <brief task title> **Session:** <session_id> -**Generated:** <ISO date, e.g. 2026-06-14> -**Task summary:** <one sentence describing what the overall task accomplishes> +**Generated:** <ISO date> +**Task summary:** <one sentence> --- @@ -85,16 +97,13 @@ Write the plan file as a Markdown document with this structure: - contract: ... - scope: ... - acceptance: ... - -### Step 2: <title> -... ``` -After writing the file, print its path to the user and give a one-line summary of each step (step number, route, title) so the user can review the plan before running `/nx-build`. +After writing: print path + one line per step (`N · route · title`). Nothing else. ## 6. Constraints -- Do not implement anything. Write the plan file only. -- Do not invent file paths — derive them from the user's task and the existing repo structure (read relevant files if needed to confirm paths). -- Do not produce vague acceptance criteria like "it works" — every acceptance must be a concrete command with a verifiable exit code. -- Adhere to §0 global constraints: stdlib only, fail-open scripts, `json.dumps(sort_keys=True)`, no third-party libs. +- Plan only. Don't implement. +- Don't invent paths — derive from task + repo (read files to confirm). +- No vague acceptance ("it works"). Every acceptance = concrete command with verifiable exit code. +- §0 globals: stdlib only, fail-open, `json.dumps(sort_keys=True)`, no third-party. diff --git a/commands/nx-report.md b/commands/nx-report.md new file mode 100644 index 0000000..7082fd8 --- /dev/null +++ b/commands/nx-report.md @@ -0,0 +1,31 @@ +--- +description: "Show this session's nexum digest: cost (actual vs all-opus + metered) and wasted-context analysis (files read but never edited, with what to stop loading)." +model: haiku +--- + +You are the nexum reporter. Run the digest script for the current session and present it clearly. You do NOT compute anything yourself — the script does all the analysis deterministically; you summarize and surface the actionable bits. + +## 1. Run the digest + +Resolve the session id from `$CLAUDE_SESSION_ID` (fall back to `_nosession`) and run: + +``` +python3 ${CLAUDE_PLUGIN_ROOT}/scripts/report.py --session "$CLAUDE_SESSION_ID" +``` + +Capture the full output. If you want an all-sessions view instead, run it with no `--session` flag. + +## 2. Present the report + +Relay the script output, then lead with the two things the user can act on: + +- **Cost** — actual spend vs the all-opus baseline (the tier-routing saving) and the metered, cache-accurate total. If the metered section says no snapshot was captured, note that the status line (`/nx-status`) must be installed for the authoritative cost. +- **Wasted context** — the efficiency grade and the waste ratio, then the concrete "drop X → save ~N tokens" suggestions. These are files this session read into context but never edited; loading them next time is avoidable cost. + +Keep it concise — prefix the header line with `[nexum] `, no emoji. + +## 3. Constraints + +- Do not edit any files. This command is read-only reporting. +- Do not run any command other than `report.py` above. +- If the script produces no output, report: `[nexum] report.py produced no output — check that CLAUDE_PLUGIN_ROOT is set correctly.` diff --git a/docs/configuration.md b/docs/configuration.md index 4ba3ff1..438cd2b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -26,12 +26,16 @@ The SQLite state file lives at `<data_dir>/nexum.db`. | `statusline_compaction_warn_pct` | `80` | Context-usage percentage at which the status line appends a `/compact` warning. Set to `0` to disable. | | `statusline_compaction_warn_tokens` | `80000` | Absolute token count at which the status line appends a `/compact` warning, regardless of window percentage. Set to `0` to disable. | | `plan_preview_enabled` | `true` | Show the projected cost preview before `/nx-build` dispatches any steps. | +| `caveman_prompts_enabled` | `true` | `/nx-plan` writes the plan's prose fields (task summary, step title/objective/contract/scope) and `/nx-build` writes its executor dispatch prompts in clipped, telegraphic English to cut tokens in the re-read plan and per-dispatch prefix. Paths, identifiers, signatures, config keys, and the runnable `acceptance` command stay verbatim; a `contract` stays unambiguous. Set `false` for normal prose. | | `resume_nudge_enabled` | `true` | Emit a session-start hint when a recent handoff exists for the current branch. | | `resume_nudge_max_age_hours` | `24` | Maximum age (in hours) of a handoff for the resume nudge to fire. | | `audit_nudge_enabled` | `true` | Surface an audit recommendation when context-blowing patterns are detected. | -| `route_calib_enabled` | `false` | Enable per-repo route calibration (nudges routes up when a tier's first-try pass rate is low). | -| `route_calib_min_samples` | `5` | Minimum number of dispatches before calibration nudges a route. | -| `route_calib_min_success_ratio` | `0.6` | First-try pass rate below which calibration nudges the route up one tier. | +| `route_calib_enabled` | `true` | Enable per-repo route calibration (Wilson-bound, bidirectional nudges from each route's first-try pass history). | +| `route_calib_min_samples` | `5` | Minimum number of dispatches before calibration advises a route (own repo, else the cross-repo `_global` prior). | +| `route_calib_min_success_ratio` | `0.6` | Wilson lower-bound first-try pass rate below which calibration nudges the route **up** one tier. | +| `route_calib_downgrade_ratio` | `0.9` | Wilson lower-bound first-try pass rate at or above which calibration nudges the route **down** one tier (cheaper). `1.0` disables downgrades. | +| `grep_narrow_enabled` | `true` | Cap an unscoped/broad search's output (`head_limit` on the Grep tool; `\| head -n N` on an unscoped recursive Bash grep) instead of hard-denying it. Deny-path searches and Glob still deny. | +| `grep_head_limit` | `80` | Result cap injected when `grep_narrow_enabled` narrows a broad/unscoped search. | | `max_steps_per_dispatch` | `6` | Maximum number of steps sent to a single executor dispatch (count cap; `0` disables). | | `max_dispatch_context_tokens` | `50000` | Token budget per dispatch sub-batch (size cap used by `plan_preview.py`). | | `dispatch_granularity` | `"group"` | `"group"`: send a whole route group to one executor; `"step"`: one dispatch per step. | diff --git a/docs/how-it-works.md b/docs/how-it-works.md index c3a4115..976e3eb 100644 --- a/docs/how-it-works.md +++ b/docs/how-it-works.md @@ -24,7 +24,9 @@ Configure via `config.json`: } ``` -**Scan-guard** (`scan_guard.py`) — unscoped recursive greps, broad globs, and reads into deny paths are blocked via `permissionDecision: deny` before the tool executes. The blocked call never reaches the model. +**Scan-guard** (`scan_guard.py`) — reads into deny paths, recursive searches over deny paths, and unscoped `find`/`ls -R` are blocked via `permissionDecision: deny` before the tool executes. The blocked call never reaches the model. + +**Grep narrowing** (`scan_guard.py`, when `grep_narrow_enabled`) — rather than denying a broad/unscoped *search* (a result-volume problem, not a noisy-directory one), nexum caps its output: it injects `head_limit` into the `Grep` tool and appends `| head -n grep_head_limit` to an unscoped recursive Bash `grep`/`rg`, via `updatedInput`. The model still gets a bounded answer without a retry round-trip. This is a *working* PreToolUse lever (unlike PostToolUse shrink). A search that already pipes, targets a deny path, or already sets `head_limit` is left to the deny path instead. **Pre-emptive dedup** (`scripts/predup.py`) — denies an identical repeated `Read`, `Grep`, or `Glob` call (and optionally read-only `Bash`) that was already executed in the same session. For `Read` calls an mtime guard is applied first: if the file has changed since the first call, the repeat is allowed through. Because a PreToolUse `deny` is actually honored, the avoided re-injection is a real saving that moves the `saved` figure in the status line. diff --git a/hooks/hooks.json b/hooks/hooks.json index 399314d..998b9b2 100644 --- a/hooks/hooks.json +++ b/hooks/hooks.json @@ -9,7 +9,7 @@ "type": "command" } ], - "matcher": "Read|Bash|Grep|Glob" + "matcher": "Read|Bash|Grep|Glob|Edit|Write|MultiEdit" } ], "PreToolUse": [ @@ -52,6 +52,37 @@ "type": "command" } ] + }, + { + "hooks": [ + { + "command": "python3 ${CLAUDE_PLUGIN_ROOT}/scripts/session_reset.py", + "timeout": 10, + "type": "command" + } + ] + } + ], + "PreCompact": [ + { + "hooks": [ + { + "command": "python3 ${CLAUDE_PLUGIN_ROOT}/scripts/precompact.py", + "timeout": 10, + "type": "command" + } + ] + } + ], + "SubagentStop": [ + { + "hooks": [ + { + "command": "python3 ${CLAUDE_PLUGIN_ROOT}/scripts/subagent_usage.py", + "timeout": 10, + "type": "command" + } + ] } ], "UserPromptSubmit": [ diff --git a/scripts/context_watch.py b/scripts/context_watch.py index b012a5c..bef8ff3 100644 --- a/scripts/context_watch.py +++ b/scripts/context_watch.py @@ -204,6 +204,68 @@ def _accumulate_tokens(session_id: str, prompt: str) -> int: return total +# --------------------------------------------------------------------------- +# Budget alerts (tiered, non-blocking) +# --------------------------------------------------------------------------- + +def _budget_alert(cfg: dict, session_id: str) -> str | None: + """Return a tiered budget warning string, or None. + + Compares the session's real metered cost (from the status-line snapshot) to + `budget_usd`, and cumulative input+output tokens to `budget_tokens`. Fires + once per tier, escalating; at >=70% it names the biggest never-edited files + to drop (from the wasted-context tracker), and at >=90% it urges /compact. + """ + try: + budget_usd = float(cfg.get("budget_usd", 0) or 0) + budget_tokens = int(cfg.get("budget_tokens", 0) or 0) + if budget_usd <= 0 and budget_tokens <= 0: + return None + tiers = sorted(int(t) for t in cfg.get("budget_alert_tiers", [50, 70, 80, 90])) + if not tiers: + return None + + rows = store.session_cost_rows(session_id) + if not rows: + return None + last = rows[-1] + cost_usd = float(last.get("cost_usd") or 0.0) + tokens = int(last.get("input_tok") or 0) + int(last.get("output_tok") or 0) + + pct_usd = (cost_usd / budget_usd * 100) if budget_usd > 0 else -1.0 + pct_tok = (tokens / budget_tokens * 100) if budget_tokens > 0 else -1.0 + pct = max(pct_usd, pct_tok) + if pct < tiers[0]: + return None + + reached = max(t for t in tiers if pct >= t) + try: + last_warned = int(_kv_get(session_id, "budget_tier") or 0) + except (TypeError, ValueError): + last_warned = 0 + if reached <= last_warned: + return None + _kv_set(session_id, "budget_tier", str(reached)) + + head = f"[nexum] Budget {pct:.0f}% used" + if budget_usd > 0: + head += f" (${cost_usd:.2f} of ${budget_usd:.2f})" + msg = head + "." + if reached >= 70: + wf = store.wasted_files(session_id, 3) + if wf: + drops = ", ".join( + f"{os.path.basename(w['file_path'])} (~{round((w['tokens_read'] or 0)/1000,1)}k)" + for w in wf + ) + msg += f" Biggest unedited reads to drop: {drops}." + if reached >= 90: + msg += " Run /compact now." + return msg + except Exception: + return None + + # --------------------------------------------------------------------------- # Response builders # --------------------------------------------------------------------------- @@ -349,6 +411,16 @@ def _handle(data: dict) -> dict: ) _kv_set(session_id, "handoff_warned", "1") + # ------------------------------------------------------------------ # + # (a2) Tiered budget alert — independent of the context-size nudges; + # merged into the same systemMessage so we still emit a single string. + # ------------------------------------------------------------------ # + budget_message = _budget_alert(cfg, session_id) + if budget_message: + system_message = ( + f"{system_message} {budget_message}" if system_message else budget_message + ) + # ------------------------------------------------------------------ # # (b) Intent-change guard # ------------------------------------------------------------------ # diff --git a/scripts/cost_report.py b/scripts/cost_report.py index 45c01d4..5c51b9e 100644 --- a/scripts/cost_report.py +++ b/scripts/cost_report.py @@ -47,7 +47,9 @@ def _model_key(model: str) -> str: """Normalise a model string to one of the PRICING keys, or return as-is.""" m = model.lower() - for key in ("opus", "sonnet", "haiku"): + # Order matters: check the most specific names first. Bedrock/Vertex IDs + # carry an "anthropic." prefix but still contain the family substring. + for key in ("fable", "opus", "sonnet", "haiku"): if key in m: return key return m # unknown model — caller handles missing key diff --git a/scripts/dedup.py b/scripts/dedup.py index 51779b9..bfbb66c 100644 --- a/scripts/dedup.py +++ b/scripts/dedup.py @@ -40,6 +40,30 @@ _MIN_LINES = 30 _MIN_CHARS = 2000 +# Edit-family tools tracked for wasted-context analytics (file_activity). +_EDIT_TOOLS = {"Edit", "Write", "MultiEdit"} + + +def _track_file_activity(session_id, tool_name, tool_input, data): + """Record per-file read/edit counters for the /nx-report waste analytics. + + A Read accumulates the file's read count + injected-token estimate (and + flags a partial read when offset/limit is set); an edit marks the file + useful. Fail-open and independent of the dedup logic below. + """ + if tool_name == "Read": + fp = tool_input.get("file_path") + if not fp: + return + out = truncate.extract_output(data) or "" + toks = store.estimate_tokens(out) if out else 0 + partial = bool(tool_input.get("offset") or tool_input.get("limit")) + store.record_file_read(session_id, fp, toks, partial) + elif tool_name in _EDIT_TOOLS: + fp = tool_input.get("file_path") + if fp: + store.record_file_edit(session_id, fp) + # Self-test: PostToolUse `updatedToolOutput` is silently ignored for built-in # tools (Bash/Read/Grep/Glob) on current Claude Code (anthropics/claude-code # #65403, #67442, #32105). When it is ignored, the shrunk/pointer output we emit @@ -174,6 +198,23 @@ def main() -> None: print("{}") return + # ---------------------------------------------------------------- + # 1b. File-activity tracking (wasted-context analytics) — runs for + # reads AND edits, independent of the dedup size logic. Edits never + # need dedup, so record and return immediately for them. + # ---------------------------------------------------------------- + _fa_tool = data.get("tool_name") or "unknown" + _fa_session = data.get("session_id") or "_nosession" + _fa_input = data.get("tool_input") or {} + try: + if store.get_config().get("file_activity_enabled", True): + _track_file_activity(_fa_session, _fa_tool, _fa_input, data) + except Exception: + pass + if _fa_tool in _EDIT_TOOLS: + print("{}") + return + # ---------------------------------------------------------------- # 2. Extract tool output # ---------------------------------------------------------------- diff --git a/scripts/precompact.py b/scripts/precompact.py new file mode 100644 index 0000000..bf83e08 --- /dev/null +++ b/scripts/precompact.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +""" +precompact.py — Nexum PreCompact hook. + +Fires immediately before Claude Code compacts the conversation. Two jobs, both +done at the exact compaction boundary (no token estimate, no polling): + +1. Invalidate predup state — compaction evicts tool output from the live + context, but the `tool_calls` rows predup keys off persist. Clearing them + here prevents predup from later denying a legitimate re-read of content the + compaction removed. +2. Write a deterministic handoff skeleton — guaranteed to capture git + task + state at the boundary even if the post-compaction session is interrupted. + +This hook NEVER blocks compaction: it performs its side effects and emits `{}` +(a PreCompact `decision: "block"` would cancel the user's compaction). + +Hook contract: + stdin → single JSON object (Claude Code PreCompact payload: session_id, + transcript_path, cwd, trigger) + stdout → "{}" always (never block) + exit 0 always (fail-open) +""" + +import json +import os +import sys + +_SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) +if _SCRIPTS_DIR not in sys.path: + sys.path.insert(0, _SCRIPTS_DIR) + +import store # noqa: E402 + + +def main() -> None: + try: + try: + data = json.loads(sys.stdin.read()) + except Exception: + print("{}") + return + if not isinstance(data, dict): + print("{}") + return + + session_id = data.get("session_id") or "_nosession" + cwd = data.get("cwd") or None + cfg = store.get_config() + + # 1. Invalidate predup's tool_calls for this session. + if cfg.get("precompact_invalidate_predup", True): + try: + store.clear_tool_calls(session_id) + except Exception: + pass + + # 2. Write a handoff skeleton at the boundary (best-effort). + if cfg.get("precompact_handoff_enabled", True): + try: + import handoff # noqa: E402 (path already set) + token_total = 0 + tp = data.get("transcript_path") or "" + if tp: + try: + token_total = store.context_tokens_from_transcript(tp) or 0 + except Exception: + token_total = 0 + handoff.write_skeleton( + session_id=session_id, cwd=cwd, token_total=token_total + ) + except Exception: + pass + + except Exception: + pass + # Always allow the compaction to proceed. + print("{}") + + +if __name__ == "__main__": + main() diff --git a/scripts/report.py b/scripts/report.py new file mode 100644 index 0000000..7081e83 --- /dev/null +++ b/scripts/report.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +""" +report.py — Nexum session digest (/nx-report). + +Deterministic, no-LLM analytics over the data nexum already records: + + * Wasted-context analysis — per-file read/edit accounting from file_activity: + which files were read into context but never edited (wasted tokens), a waste + ratio, an efficiency grade, and concrete "drop X to save ~N tokens" picks. + * Cost summary — the actual-vs-all-opus tiering breakdown and Claude Code's + own metered, cache-accurate total (reused from cost_report). + +CLI: + python3 report.py [--session <id>] + +All computation is local and rule-based; the command body just presents this. +""" + +import argparse +import os +import sys + +_SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) +if _SCRIPTS_DIR not in sys.path: + sys.path.insert(0, _SCRIPTS_DIR) + +import store # noqa: E402 +import cost_report # noqa: E402 + +# Efficiency grade by waste ratio (wasted_tokens / total_read_tokens). +_GRADE_BANDS = [ + (0.10, "S"), (0.20, "A"), (0.35, "B"), + (0.50, "C"), (0.70, "D"), (1.01, "F"), +] + + +def _grade(waste_ratio: float) -> str: + for ceiling, letter in _GRADE_BANDS: + if waste_ratio <= ceiling: + return letter + return "F" + + +def _usefulness(row: dict) -> float: + """+3 per edit, +0.5 per re-read (beyond the first), +1 if ever partial.""" + reads = int(row.get("reads") or 0) + edits = int(row.get("edits") or 0) + partial = int(row.get("partial_reads") or 0) + return 3 * edits + 0.5 * max(0, reads - 1) + (1 if partial > 0 else 0) + + +def _fmt_tok(n: int) -> str: + n = int(n or 0) + return str(n) if n < 1000 else f"{n / 1000:.1f}k" + + +def build_waste_section(rows: list) -> str: + if not rows: + return ("Wasted-context analysis:\n" + " No file reads recorded yet " + "(file_activity is empty for this scope).") + + total_read = sum(int(r.get("tokens_read") or 0) for r in rows) + wasted_rows = [r for r in rows if int(r.get("edits") or 0) == 0 + and int(r.get("tokens_read") or 0) > 0] + wasted_tok = sum(int(r.get("tokens_read") or 0) for r in wasted_rows) + ratio = (wasted_tok / total_read) if total_read else 0.0 + + lines = [] + lines.append("Wasted-context analysis:") + lines.append(f" Tokens read into context: {total_read:>10,}") + lines.append(f" Spent on never-edited files:{wasted_tok:>10,} " + f"({ratio * 100:.0f}% wasted)") + lines.append(f" Efficiency grade: {_grade(ratio):>10}") + lines.append("") + lines.append(f" {'File':<48} {'Read':>5} {'Edit':>5} {'Tok':>8} Verdict") + lines.append(" " + "-" * 82) + for r in sorted(rows, key=lambda x: int(x.get("tokens_read") or 0), reverse=True)[:12]: + fp = str(r.get("file_path") or "?") + disp = fp if len(fp) <= 48 else "…" + fp[-47:] + verdict = "useful" if int(r.get("edits") or 0) > 0 else ( + "WASTED" if int(r.get("tokens_read") or 0) > 0 else "—") + lines.append( + f" {disp:<48} {int(r.get('reads') or 0):>5} " + f"{int(r.get('edits') or 0):>5} {_fmt_tok(r.get('tokens_read')):>8} {verdict}" + ) + + if wasted_rows: + lines.append("") + lines.append(" Suggestions — stop loading these (read, never edited):") + for r in sorted(wasted_rows, key=lambda x: int(x.get("tokens_read") or 0), + reverse=True)[:5]: + fp = str(r.get("file_path") or "?") + lines.append(f" drop {fp} → save ~{_fmt_tok(r.get('tokens_read'))} tokens") + return "\n".join(lines) + + +# Savings sources, classified by how trustworthy the number is. PreToolUse +# levers are honored by current Claude Code; PostToolUse output replacement is +# not (updatedToolOutput is ignored for built-in tools — anthropics/claude-code +# #65403), so dedup/truncate are reported separately and never folded into the +# realized headline. +_SAVINGS_BOUNDED = {"read_guard", "grep_narrow"} +_SAVINGS_THEORETICAL = {"dedup", "truncate"} +_SAVINGS_LABEL = { + "predup": "repeat tool calls denied", + "read_guard": "large reads capped", + "grep_narrow": "broad searches bounded", + "dedup": "duplicate outputs (PostToolUse)", + "truncate": "oversized outputs (PostToolUse)", +} + + +def build_savings_section(by_source: dict) -> str: + """Render savings split into realized / bounded / theoretical buckets. + + Honest by construction: only *realized* (PreToolUse, measured) tokens go in + the headline; *bounded* interventions are counted but carry no token claim; + *theoretical* PostToolUse shrink is shown with the upstream-bug caveat and + never summed into the realized total. + """ + realized, bounded, theoretical = {}, {}, {} + for src, agg in (by_source or {}).items(): + if src in _SAVINGS_THEORETICAL: + theoretical[src] = agg + elif src in _SAVINGS_BOUNDED: + bounded[src] = agg + else: + realized[src] = agg + + lines = ["Savings analysis:"] + + realized_tok = sum(int(a.get("effective_tok") or 0) for a in realized.values()) + lines.append(" Realized (PreToolUse — actually removed from context):") + if realized: + for src in sorted(realized): + a = realized[src] + lines.append( + f" {_SAVINGS_LABEL.get(src, src):<32} " + f"{a.get('count', 0):>4} calls ~{_fmt_tok(a.get('effective_tok'))} tok saved" + ) + lines.append(f" {'TOTAL realized':<32} {'':>4} ~{_fmt_tok(realized_tok)} tok") + else: + lines.append(" none recorded this scope") + + lines.append(" Bounded interventions (output capped; exact saving unknowable):") + if bounded: + for src in sorted(bounded): + a = bounded[src] + lines.append(f" {_SAVINGS_LABEL.get(src, src):<32} {a.get('count', 0):>4} ×") + else: + lines.append(" none recorded this scope") + + lines.append( + " Theoretical (PostToolUse shrink — INERT on current Claude Code;\n" + " updatedToolOutput is ignored for built-in tools, tracking #65403):" + ) + if theoretical: + for src in sorted(theoretical): + a = theoretical[src] + lines.append( + f" {_SAVINGS_LABEL.get(src, src):<32} " + f"{a.get('count', 0):>4} × ~{_fmt_tok(a.get('effective_tok'))} tok " + f"would save once the field is honored" + ) + else: + lines.append(" nothing recorded — PostToolUse shrink contributes 0 today") + return "\n".join(lines) + + +def build_digest(session_id=None) -> str: + parts = ["[nexum] Session report", "=" * 48, ""] + # Cost (reuses cost_report's deterministic builders). + parts.append(cost_report.build_report(store.usage_rows(session_id=session_id))) + parts.append("") + parts.append(cost_report.build_metered_section(store.session_cost_rows(session_id=session_id))) + parts.append("") + parts.append(build_savings_section(store.savings_by_source(session_id=session_id))) + parts.append("") + parts.append(build_waste_section(store.file_activity_rows(session_id=session_id))) + return "\n".join(parts) + + +def main() -> None: + parser = argparse.ArgumentParser( + prog="report.py", description="Nexum session digest: cost + wasted context." + ) + parser.add_argument("--session", metavar="ID", default=None, + help="Filter to a session id (omit for all sessions).") + args = parser.parse_args() + print(build_digest(session_id=args.session)) + + +if __name__ == "__main__": + main() diff --git a/scripts/scan_guard.py b/scripts/scan_guard.py index 24559a5..cfcb35c 100644 --- a/scripts/scan_guard.py +++ b/scripts/scan_guard.py @@ -76,6 +76,21 @@ def _allow() -> None: sys.exit(0) +def _record_intervention(data: dict, source: str) -> None: + """Record a *bounded* PreToolUse intervention (count only, 0 tokens). + + grep-narrow and read-guard cap a tool's output but the unbounded size — and + therefore the exact saving — is unknowable at PreToolUse time, so we record + a 0-token row purely so /nx-report can count interventions under its + "bounded" bucket without inflating the measured-savings headline. Fail-open. + """ + try: + session_id = data.get("session_id") or "_nosession" + store.record_saving(session_id, source, 0, 0) + except Exception: + pass + + # --------------------------------------------------------------------------- # Path-under-deny helper # --------------------------------------------------------------------------- @@ -396,8 +411,17 @@ def main() -> None: if tool_name == "Bash": command: str = tool_input.get("command", "") or "" - # 1. Unscoped recursive grep/rg + # 1. Unscoped recursive grep/rg — narrow output instead of denying. + # An unscoped recursive grep blows context via its result volume, not + # by reading a noisy dir, so cap the output with `| head -n N` (a + # working PreToolUse lever) rather than forcing a retry. Skip the + # rewrite when the command already pipes (can't safely append) and + # fall back to the deny. if _is_grep_like(command) and _is_unscoped_grep(command): + if cfg.get("grep_narrow_enabled", True) and "|" not in command: + limit = int(cfg.get("grep_head_limit", 80)) + _record_intervention(data, "grep_narrow") + _update_input(f"{command.rstrip()} | head -n {limit}") _deny("unscoped recursive grep/rg searches the entire repo") # 2. grep/rg targeting a deny path @@ -428,8 +452,18 @@ def main() -> None: if tool_input_path and _under_deny(tool_input_path, deny_paths): _deny(f"search path is inside a high-noise directory ({tool_input_path})") - # Broad unscoped pattern + # Broad unscoped pattern. For the Grep tool, cap the result set with + # head_limit (a working PreToolUse lever) instead of denying — the + # model still gets a bounded answer. Glob has no head_limit, so it + # keeps the deny. (Deny-path cases already returned above.) if _grep_glob_is_unscoped(tool_input, deny_paths): + if (tool_name == "Grep" + and cfg.get("grep_narrow_enabled", True) + and tool_input.get("head_limit") is None): + new_input = dict(tool_input) + new_input["head_limit"] = int(cfg.get("grep_head_limit", 80)) + _record_intervention(data, "grep_narrow") + _update_tool_input(new_input) _deny("broad pattern at repo root would scan the entire tree") _allow() @@ -445,6 +479,7 @@ def main() -> None: # for Read; PostToolUse output shrink is ignored for built-in tools). narrowed = _read_limit_input(tool_input, cfg) if narrowed is not None: + _record_intervention(data, "read_guard") _update_tool_input(narrowed) _allow() diff --git a/scripts/session_reset.py b/scripts/session_reset.py new file mode 100644 index 0000000..0edf98e --- /dev/null +++ b/scripts/session_reset.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +""" +session_reset.py — Nexum SessionStart housekeeping hook. + +Two jobs on session start: + +1. Invalidate predup state on a context reset. When `source` is `clear` (the + user ran /clear) or `compact` (the session resumed after a compaction), the + tool output predup keys off is gone — so clear this session's `tool_calls` + rows to prevent predup denying a legitimate re-read. No-op for `startup`/ + `resume`, where prior context is still present. +2. Throttled retention prune (`store.maybe_prune`, at most once/day) so the + SQLite file and predup lookups stay bounded across many sessions. + +Filtering is done in-script on the `source` field rather than via a hook +matcher, so behaviour is identical regardless of SessionStart matcher support. + +Hook contract: + stdin → single JSON object (session_id, source, ...) + stdout → "{}" always + exit 0 always (fail-open) +""" + +import json +import os +import sys + +_SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) +if _SCRIPTS_DIR not in sys.path: + sys.path.insert(0, _SCRIPTS_DIR) + +import store # noqa: E402 + +_RESET_SOURCES = {"clear", "compact"} + + +def main() -> None: + try: + try: + data = json.loads(sys.stdin.read()) + except Exception: + print("{}") + return + if not isinstance(data, dict): + print("{}") + return + + source = (data.get("source") or "").strip().lower() + session_id = data.get("session_id") or "_nosession" + + if source in _RESET_SOURCES: + cfg = store.get_config() + if cfg.get("precompact_invalidate_predup", True): + try: + store.clear_tool_calls(session_id) + except Exception: + pass + + # Bounded, throttled retention prune (internally rate-limited to 1/day). + try: + store.maybe_prune() + except Exception: + pass + + except Exception: + pass + print("{}") + + +if __name__ == "__main__": + main() diff --git a/scripts/store.py b/scripts/store.py index c8da7fe..91c13ea 100644 --- a/scripts/store.py +++ b/scripts/store.py @@ -31,6 +31,7 @@ # Cache read ≈ 0.1× input; cache write ≈ 1.25× input. # --------------------------------------------------------------------------- PRICING: Dict[str, tuple] = { + "fable": (10.0, 50.0), "opus": (5.0, 25.0), "sonnet": (3.0, 15.0), "haiku": (1.0, 5.0), @@ -152,6 +153,16 @@ "plan_preview_input_tok_per_step": 8000, # Per-step token heuristic for the plan cost preview (output side). "plan_preview_output_tok_per_step": 2000, + # Caveman prompts: /nx-plan writes the plan's PROSE fields (task summary, + # step title/objective/contract/scope) in clipped, telegraphic English — + # articles, copulas, and filler dropped — and /nx-build builds its executor + # dispatch prompts the same way. The plan is re-read by every executor and + # the dispatch prefix is sent on every step, so trimming function words from + # them is a recurring token saving. STRICT carve-outs (never caveman-ified): + # file paths, identifiers, signatures, config keys, code, and the runnable + # `acceptance` command stay verbatim, and a `contract` must stay + # unambiguous — terseness never costs precision. Set False for normal prose. + "caveman_prompts_enabled": True, # Route calibration: track per-route dispatch outcomes (pass/escalate) and # feed the data back into routing decisions over time. "route_calib_enabled": True, @@ -159,11 +170,52 @@ "route_calib_min_success_ratio": 0.6, # Minimum number of dispatches before calibration data is trusted. "route_calib_min_samples": 5, + # Downgrade threshold (bidirectional calibration): when a route's Wilson + # lower-bound first-try pass rate is at or above this, calibration may nudge + # comparable steps DOWN one tier (cheaper) — the counterpart to the up-nudge. + # Conservative by default so a downgrade needs strong evidence. 1.0 disables + # downgrades (revert to up-only calibration). + "route_calib_downgrade_ratio": 0.9, + # Grep narrowing (PreToolUse): instead of hard-denying a broad/unscoped + # search, cap its output — inject `head_limit` on the Grep tool and append + # `| head -n grep_head_limit` to an unscoped recursive Bash grep/rg. This is + # a *working* context-savings lever on current Claude Code (PreToolUse + # updatedInput is honored, unlike PostToolUse output shrink). Searches that + # target a deny-listed directory are still denied outright. False reverts to + # the deny-everything behaviour. + "grep_narrow_enabled": True, + "grep_head_limit": 80, # SessionStart audit nudge: surface /nx-audit hint when ignore config has # findings (throttled to once per repo per audit_nudge_throttle_hours). "audit_nudge_enabled": True, # Hours between successive audit nudges for the same repo. "audit_nudge_throttle_hours": 24, + # PreCompact hook: clear this session's tool_calls rows before a compaction + # (which evicts the cached output predup keys off) and write a handoff + # skeleton at the exact compaction boundary (deterministic, not estimated). + "precompact_invalidate_predup": True, + "precompact_handoff_enabled": True, + # SubagentStop hook: record a real per-tier usage row (parsed best-effort + # from the subagent transcript) for nexum executor agents, replacing the + # pure estimate in the cost report. Token attribution is best-effort — the + # SubagentStop payload carries no usage fields, so this reads the transcript. + "subagent_usage_enabled": True, + # Retention: rows in the ephemeral tables (tool_calls, savings, outputs, + # usage, file_activity, memo) older than this many days are pruned on + # session start so the SQLite file and predup lookups stay bounded. 0 + # disables pruning. + "retention_days": 14, + # Wasted-context tracking: record per-file read/edit counts so /nx-report + # can flag files read into context but never edited. + "file_activity_enabled": True, + # Budget alerts (C): tiered warnings as session spend approaches a budget. + # budget_usd is checked against the real metered cost nexum captures from + # the status line; budget_tokens against cumulative session tokens. 0 + # disables that axis. Alerts fire once per tier, escalating, via a + # non-blocking systemMessage on UserPromptSubmit. + "budget_usd": 0.0, + "budget_tokens": 0, + "budget_alert_tiers": [50, 70, 80, 90], } # --------------------------------------------------------------------------- @@ -280,6 +332,18 @@ PRIMARY KEY(repo, route) ) """, + """ + CREATE TABLE IF NOT EXISTS file_activity( + session_id TEXT, + file_path TEXT, + reads INTEGER, + partial_reads INTEGER, + edits INTEGER, + tokens_read INTEGER, + ts REAL, + PRIMARY KEY(session_id, file_path) + ) + """, ] # Column migrations for databases created by an earlier schema version. @@ -546,17 +610,76 @@ def transcript_tool_result_len(transcript_path: str, tool_use_id: str) -> Option return None +def transcript_usage_totals(transcript_path: str) -> Dict[str, int]: + """Sum token usage across all assistant turns in a transcript JSONL. + + Returns ``{"input_tok", "output_tok", "cache_read_tok"}``. Used by the + SubagentStop hook to attribute real per-tier usage: the hook payload carries + no usage fields, but the subagent's transcript records per-message ``usage`` + blocks. Best-effort — returns zeros on any error or an empty/absent file. + Note: if the path is the parent transcript rather than the subagent's, this + over-counts; callers treat the result as a best-effort signal. + """ + totals = {"input_tok": 0, "output_tok": 0, "cache_read_tok": 0} + if not transcript_path: + return totals + try: + with open(transcript_path, "r", encoding="utf-8", errors="replace") as fh: + for line in fh: + if '"usage"' not in line: + continue + try: + d = json.loads(line) + except Exception: + continue + msg = d.get("message") or {} + usage = msg.get("usage") + if not isinstance(usage, dict): + continue + totals["input_tok"] += int(usage.get("input_tokens") or 0) + totals["output_tok"] += int(usage.get("output_tokens") or 0) + totals["cache_read_tok"] += int(usage.get("cache_read_input_tokens") or 0) + except Exception: + return {"input_tok": 0, "output_tok": 0, "cache_read_tok": 0} + return totals + + # --------------------------------------------------------------------------- # Input-keyed tool-call helpers (used by PreToolUse predup) # --------------------------------------------------------------------------- +def _canonical_tool_input(tool_input: dict) -> dict: + """Return a copy of tool_input with path-like keys canonicalised. + + Resolves ``file_path``/``path`` to an absolute, symlink-free path so that + ``./foo.py``, ``foo.py``, and ``/abs/foo.py`` all produce the same + signature. Other keys (e.g. a Read ``offset``/``limit``, a Grep ``pattern``) + are preserved verbatim — a different range or pattern is genuinely different + content and must not collapse. Best-effort: on any error the original value + is kept. + """ + if not isinstance(tool_input, dict): + return tool_input + canon = dict(tool_input) + for key in ("file_path", "path"): + val = canon.get(key) + if isinstance(val, str) and val: + try: + canon[key] = os.path.realpath(val) + except Exception: + pass + return canon + + def tool_call_sig(tool_name: str, tool_input: dict) -> str: - """Return the SHA-256 of tool_name + NUL + json.dumps(tool_input, sort_keys=True). + """Return the SHA-256 of tool_name + NUL + canonicalised, sorted tool_input. - Deterministic for identical inputs regardless of dict insertion order. - Uses the existing sha256() helper. + Deterministic for equivalent inputs regardless of dict insertion order or + how a path was spelled (``./foo`` vs ``foo`` vs an absolute path). Uses the + existing sha256() helper. """ - serialised = tool_name + "\x00" + json.dumps(tool_input, sort_keys=True, default=str) + canon = _canonical_tool_input(tool_input) + serialised = tool_name + "\x00" + json.dumps(canon, sort_keys=True, default=str) return sha256(serialised) @@ -600,6 +723,139 @@ def seen_tool_call(session_id: str, input_sig: str) -> Optional[Dict[str, Any]]: return None +def clear_tool_calls(session_id: str) -> int: + """Delete all tool_calls rows for a session. Returns the row count removed. + + Called when context that predup keys off is evicted — on PreCompact (before + a compaction) and on a SessionStart whose source is ``clear``/``compact`` — + so predup can never deny a re-read of content no longer in the live context. + Fail-open: returns 0 on any error. + """ + try: + conn = db() + with conn: + cur = conn.execute( + "DELETE FROM tool_calls WHERE session_id=?", (session_id,) + ) + n = cur.rowcount + conn.close() + return int(n or 0) + except Exception: + return 0 + + +# --------------------------------------------------------------------------- +# File-activity accounting (wasted-context analytics) +# +# Per-session, per-file counters used by /nx-report to flag files that were +# read into context but never edited ("wasted"), and by the budget alert to +# name the biggest such files. file_path is realpath-canonicalised so the same +# file under different spellings aggregates into one row. +# --------------------------------------------------------------------------- + +def _realpath(file_path: str) -> str: + try: + return os.path.realpath(file_path) + except Exception: + return file_path + + +def record_file_read(session_id: str, file_path: str, tokens: int, partial: bool = False) -> None: + """Increment a file's read counters and accumulate the tokens it injected.""" + if not file_path: + return + fp = _realpath(file_path) + try: + conn = db() + with conn: + row = conn.execute( + "SELECT reads, partial_reads, edits, tokens_read FROM file_activity " + "WHERE session_id=? AND file_path=?", + (session_id, fp), + ).fetchone() + reads = (row["reads"] if row else 0) + 1 + preads = (row["partial_reads"] if row else 0) + (1 if partial else 0) + edits = row["edits"] if row else 0 + toks = (row["tokens_read"] if row else 0) + int(tokens or 0) + conn.execute( + "INSERT OR REPLACE INTO file_activity" + "(session_id, file_path, reads, partial_reads, edits, tokens_read, ts) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (session_id, fp, reads, preads, edits, toks, time.time()), + ) + conn.close() + except Exception: + pass + + +def record_file_edit(session_id: str, file_path: str) -> None: + """Increment a file's edit counter (marks the file as useful, not wasted).""" + if not file_path: + return + fp = _realpath(file_path) + try: + conn = db() + with conn: + row = conn.execute( + "SELECT reads, partial_reads, edits, tokens_read FROM file_activity " + "WHERE session_id=? AND file_path=?", + (session_id, fp), + ).fetchone() + reads = row["reads"] if row else 0 + preads = row["partial_reads"] if row else 0 + edits = (row["edits"] if row else 0) + 1 + toks = row["tokens_read"] if row else 0 + conn.execute( + "INSERT OR REPLACE INTO file_activity" + "(session_id, file_path, reads, partial_reads, edits, tokens_read, ts) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (session_id, fp, reads, preads, edits, toks, time.time()), + ) + conn.close() + except Exception: + pass + + +def file_activity_rows(session_id: Optional[str] = None) -> List[Dict[str, Any]]: + """Return file_activity rows (optionally filtered by session_id).""" + cols = "session_id, file_path, reads, partial_reads, edits, tokens_read, ts" + try: + conn = db() + if session_id is not None: + rows = conn.execute( + f"SELECT {cols} FROM file_activity WHERE session_id=? ORDER BY tokens_read DESC", + (session_id,), + ).fetchall() + else: + rows = conn.execute( + f"SELECT {cols} FROM file_activity ORDER BY tokens_read DESC" + ).fetchall() + conn.close() + return [dict(r) for r in rows] + except Exception: + return [] + + +def wasted_files(session_id: str, limit: int = 5) -> List[Dict[str, Any]]: + """Return files read into context but never edited, biggest first. + + These are the prime "drop this to save tokens" candidates. Returns dicts + with file_path / tokens_read / reads. + """ + try: + conn = db() + rows = conn.execute( + "SELECT file_path, tokens_read, reads FROM file_activity " + "WHERE session_id=? AND edits=0 AND reads>=1 AND tokens_read>0 " + "ORDER BY tokens_read DESC LIMIT ?", + (session_id, int(limit)), + ).fetchall() + conn.close() + return [dict(r) for r in rows] + except Exception: + return [] + + # --------------------------------------------------------------------------- # Dedup / memo helpers # --------------------------------------------------------------------------- @@ -851,6 +1107,101 @@ def calibration_rows(repo: str) -> List[Dict[str, Any]]: return [] +# Tier order from cheapest to strongest; an "up" nudge moves one step right, +# a "down" nudge one step left. +_ROUTE_ORDER = ["mechanical", "standard", "needs-strong"] + + +def _wilson_bounds(passed: int, n: int, z: float = 1.96) -> tuple: + """Return the (lower, upper) Wilson score interval for a pass proportion. + + Deterministic (no third-party stats lib; uses ``** 0.5`` for the root). The + lower bound discounts small samples, so a 3/3 streak does not read as a + confident 100%. Returns (0.0, 0.0) for n <= 0. + """ + if n <= 0: + return (0.0, 0.0) + phat = passed / n + z2 = z * z + denom = 1.0 + z2 / n + centre = phat + z2 / (2.0 * n) + margin = z * (((phat * (1.0 - phat)) + z2 / (4.0 * n)) / n) ** 0.5 + return ((centre - margin) / denom, (centre + margin) / denom) + + +def calibration_advice(repo: str, cfg: Optional[Dict[str, Any]] = None) -> Dict[str, Dict[str, Any]]: + """Return deterministic per-route routing advice from calibration history. + + For each route with enough evidence, returns + ``{"action": "up"|"down"|"keep", "reason", "samples", "lower", "source"}``. + Uses the repo's own rows when it has >= ``route_calib_min_samples`` + dispatches; otherwise falls back to the cross-repo ``"_global"`` aggregate. + A Wilson score lower bound replaces a raw pass ratio so small samples don't + over-trigger. Advice is **bidirectional**: nudge up one tier when confidence + the tier handles this repo is low, down one tier when a tier clears first-try + very reliably (cheaper tier may suffice). Conservative and advisory only. + Fail-open: returns ``{}`` on any error or when calibration is disabled. + """ + try: + if cfg is None: + cfg = get_config() + if not cfg.get("route_calib_enabled", True): + return {} + min_samples = int(cfg.get("route_calib_min_samples", 5)) + up_ratio = float(cfg.get("route_calib_min_success_ratio", 0.6)) + down_ratio = float(cfg.get("route_calib_downgrade_ratio", 0.9)) + + own = {r["route"]: r for r in calibration_rows(repo)} + glob = {r["route"]: r for r in calibration_rows("_global")} + + advice: Dict[str, Dict[str, Any]] = {} + for idx, route in enumerate(_ROUTE_ORDER): + row = own.get(route) + source = "repo" + if not row or int(row.get("dispatched") or 0) < min_samples: + grow = glob.get(route) + if grow and int(grow.get("dispatched") or 0) >= min_samples: + row, source = grow, "global" + else: + continue # not enough evidence in this repo or globally + n = int(row.get("dispatched") or 0) + passed = int(row.get("passed_first_try") or 0) + lower, _upper = _wilson_bounds(passed, n) + tag = "" if source == "repo" else " (cross-repo prior)" + if lower < up_ratio and idx < len(_ROUTE_ORDER) - 1: + advice[route] = { + "action": "up", + "reason": ( + f"{route} steps clear first-try with only {lower:.0%} " + f"lower-bound confidence over {n} dispatches{tag} " + f"(< {up_ratio:.0%}); route up one tier" + ), + "samples": n, "lower": round(lower, 3), "source": source, + } + elif down_ratio < 1.0 and lower >= down_ratio and idx > 0: + advice[route] = { + "action": "down", + "reason": ( + f"{route} steps clear first-try reliably ({lower:.0%} " + f"lower-bound over {n}{tag} >= {down_ratio:.0%}); a " + f"cheaper tier may suffice — route down one tier" + ), + "samples": n, "lower": round(lower, 3), "source": source, + } + else: + advice[route] = { + "action": "keep", + "reason": ( + f"{route} within calibrated band ({lower:.0%} lower-bound " + f"over {n}{tag})" + ), + "samples": n, "lower": round(lower, 3), "source": source, + } + return advice + except Exception: + return {} + + def record_saving( session_id: str, source: str, @@ -901,6 +1252,45 @@ def session_savings(session_id: str) -> int: return 0 +def savings_by_source(session_id: Optional[str] = None) -> Dict[str, Dict[str, int]]: + """Return savings aggregated by source. + + ``{source: {"count": n, "saved_tok": sum, "effective_tok": sum}}``. Lets the + report separate *realized* PreToolUse savings (``predup`` — the denied repeat's + exact token count is known) from *bounded* interventions (``read_guard``, + ``grep_narrow`` — output was capped but the unbounded size, hence the exact + saving, is unknowable, so these record a count with 0 tokens) and *theoretical* + PostToolUse shrink (``dedup``/``truncate`` — inert on Claude Code that ignores + ``updatedToolOutput`` for built-in tools). When *session_id* is None, sums + across all sessions. Fail-open: returns ``{}`` on any error. + """ + try: + conn = db() + sql = ( + "SELECT source, COUNT(*) AS cnt, " + "COALESCE(SUM(saved_tok),0) AS saved_tok, " + "COALESCE(SUM(COALESCE(effective_tok, saved_tok)),0) AS effective_tok " + "FROM savings " + ) + if session_id is None: + rows = conn.execute(sql + "GROUP BY source").fetchall() + else: + rows = conn.execute( + sql + "WHERE session_id=? GROUP BY source", (session_id,) + ).fetchall() + conn.close() + return { + r["source"]: { + "count": int(r["cnt"]), + "saved_tok": int(r["saved_tok"]), + "effective_tok": int(r["effective_tok"]), + } + for r in rows + } + except Exception: + return {} + + # --------------------------------------------------------------------------- # Session cost snapshot — Claude Code's own metered, cache-accurate totals. # @@ -965,6 +1355,76 @@ def session_cost_rows(session_id: Optional[str] = None) -> List[Dict[str, Any]]: return [] +# --------------------------------------------------------------------------- +# Retention — keep the ephemeral tables (and predup lookups) bounded +# --------------------------------------------------------------------------- + +# Tables pruned by age, and the column holding their epoch-seconds timestamp. +# session_kv (flags/task) and step_ledger (resume state) are intentionally NOT +# pruned by age — they are small and semantically session-scoped. +_PRUNE_TABLES = ( + ("tool_calls", "ts"), + ("savings", "ts"), + ("outputs", "ts"), + ("usage", "ts"), + ("memo", "ts"), + ("file_activity", "ts"), +) + + +def prune(retention_days: float) -> int: + """Delete rows older than retention_days from the ephemeral tables. + + Returns the total rows removed. retention_days <= 0 is a no-op (disabled). + Fail-open: a table that doesn't exist or any error is skipped, never raised. + """ + if not retention_days or retention_days <= 0: + return 0 + cutoff = time.time() - float(retention_days) * 86400.0 + removed = 0 + try: + conn = db() + with conn: + for table, tscol in _PRUNE_TABLES: + try: + cur = conn.execute( + f"DELETE FROM {table} WHERE {tscol} < ?", (cutoff,) + ) + removed += int(cur.rowcount or 0) + except sqlite3.OperationalError: + continue # table absent in this db — skip + conn.close() + except Exception: + return removed + return removed + + +def maybe_prune() -> int: + """Run prune() at most once per day, tracked via a global kv timestamp. + + Cheap to call on every session start: returns 0 (and does nothing) unless a + day has elapsed since the last prune. Uses session_kv under a fixed sentinel + session id so the throttle is global, not per real session. + """ + try: + cfg = get_config() + days = float(cfg.get("retention_days", 14) or 0) + if days <= 0: + return 0 + last_raw = get_flag("_nexum_global", "last_prune_ts") + now = time.time() + if last_raw: + try: + if now - float(last_raw) < 86400.0: + return 0 + except (TypeError, ValueError): + pass + set_flag("_nexum_global", "last_prune_ts", str(now)) + return prune(days) + except Exception: + return 0 + + # --------------------------------------------------------------------------- # Step ledger — durable execution state for /nx-build resume # --------------------------------------------------------------------------- @@ -1165,6 +1625,15 @@ def _cmd_config() -> None: print(json.dumps(cfg, sort_keys=True, indent=2)) +def _cmd_prune(args) -> None: + """Prune aged rows. With --days, use that; otherwise the configured value.""" + if args.days is not None: + removed = prune(args.days) + else: + removed = prune(float(get_config().get("retention_days", 14) or 0)) + print(json.dumps({"removed": removed}, sort_keys=True)) + + def _read_optional_file(path: Optional[str]) -> Optional[str]: """Read a file's text, or None. '-' means stdin. Missing → None (fail-open).""" if not path: @@ -1264,6 +1733,11 @@ def _cmd_calib_list(args) -> None: print(json.dumps(calibration_rows(args.repo), sort_keys=True)) +def _cmd_calib_advice(args) -> None: + """Print per-route routing advice (up/down/keep) for a repo as JSON.""" + print(json.dumps(calibration_advice(args.repo), sort_keys=True)) + + def main() -> None: parser = argparse.ArgumentParser( prog="store.py", @@ -1324,9 +1798,19 @@ def main() -> None: p_calib.add_argument("--passed-first-try", type=int, default=0) p_calib.add_argument("--escalated", type=int, default=0) + p_prune = sub.add_parser("prune", help="Delete aged rows from the ephemeral tables.") + p_prune.add_argument("--days", type=float, default=None, + help="Override retention_days; omit to use config.") + p_clist = sub.add_parser("calib-list", help="Print calibration rows for a repo as JSON.") p_clist.add_argument("--repo", required=True) + p_cadv = sub.add_parser( + "calib-advice", + help="Print per-route routing advice (up/down/keep, JSON) for a repo.", + ) + p_cadv.add_argument("--repo", required=True) + args = parser.parse_args() if args.command == "init": @@ -1351,6 +1835,10 @@ def main() -> None: _cmd_calib_record(args) elif args.command == "calib-list": _cmd_calib_list(args) + elif args.command == "calib-advice": + _cmd_calib_advice(args) + elif args.command == "prune": + _cmd_prune(args) else: parser.print_help() sys.exit(1) diff --git a/scripts/subagent_usage.py b/scripts/subagent_usage.py new file mode 100644 index 0000000..6d7b843 --- /dev/null +++ b/scripts/subagent_usage.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +""" +subagent_usage.py — Nexum SubagentStop hook. + +Records a real per-tier usage row when a nexum executor subagent finishes, so +`cost_report.py`'s tiering breakdown reflects measured spend rather than the +plan-preview estimate. + +Limitation (documented, not hidden): the SubagentStop payload carries NO token +usage fields — only the agent name and a transcript path. We therefore map the +agent name to its model tier and parse token totals best-effort from the +transcript (`store.transcript_usage_totals`). If the transcript is unavailable +or is the parent's rather than the subagent's, token counts may be 0 or +approximate; the row still records that the tier ran, which is strictly better +than a pure estimate. When Claude Code adds usage to the SubagentStop payload, +prefer that. + +Only nexum's own executor agents are recorded (agent names starting with +`nexum-impl-`); other subagents are ignored. + +Hook contract: + stdin → single JSON object (session_id, transcript_path, cwd, agent_type) + stdout → "{}" always + exit 0 always (fail-open) +""" + +import json +import os +import sys + +_SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) +if _SCRIPTS_DIR not in sys.path: + sys.path.insert(0, _SCRIPTS_DIR) + +import store # noqa: E402 + +# Executor agent name → pricing tier key (matches store.PRICING). +_AGENT_TIER = { + "nexum-impl-haiku": "haiku", + "nexum-impl-sonnet": "sonnet", + "nexum-impl-opus": "opus", +} + + +def _tier_for_agent(agent: str) -> str: + """Return the pricing tier for a nexum executor agent name, or '' if N/A.""" + if not agent: + return "" + a = agent.strip() + if a in _AGENT_TIER: + return _AGENT_TIER[a] + # Tolerate suffixes/namespacing (e.g. "plugin:nexum-impl-haiku"). + for name, tier in _AGENT_TIER.items(): + if a.endswith(name) or name in a: + return tier + return "" + + +def main() -> None: + try: + try: + data = json.loads(sys.stdin.read()) + except Exception: + print("{}") + return + if not isinstance(data, dict): + print("{}") + return + + cfg = store.get_config() + if not cfg.get("subagent_usage_enabled", True): + print("{}") + return + + agent = ( + data.get("agent_type") + or data.get("agent_name") + or data.get("subagent_type") + or "" + ) + tier = _tier_for_agent(agent) + if not tier: + print("{}") # not a nexum executor — ignore + return + + session_id = data.get("session_id") or "_nosession" + transcript_path = data.get("transcript_path") or "" + totals = store.transcript_usage_totals(transcript_path) + + store.add_usage( + session_id=session_id, + model=tier, + input_tok=totals.get("input_tok", 0), + output_tok=totals.get("output_tok", 0), + cache_read_tok=totals.get("cache_read_tok", 0), + ) + except Exception: + pass + print("{}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_gaps.py b/tests/test_gaps.py new file mode 100644 index 0000000..9c4ef39 --- /dev/null +++ b/tests/test_gaps.py @@ -0,0 +1,182 @@ +""" +test_gaps.py — covers the v0.4.x gap-closing work: + +#1 clear_tool_calls + precompact/session_reset invalidation (predup safety) +#2 canonicalised tool_call_sig (predup recall) +#4 transcript_usage_totals + subagent_usage hook (real per-tier usage) +#5 Fable pricing in cost_report +#6 prune / maybe_prune retention +""" + +import importlib +import json +import os +import subprocess +import sys +import tempfile +import time +import unittest + +_SCRIPTS_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts" +) +if _SCRIPTS_DIR not in sys.path: + sys.path.insert(0, _SCRIPTS_DIR) + + +def _run(script, payload, data_dir): + env = os.environ.copy() + env["CLAUDE_PLUGIN_DATA"] = data_dir + env["PYTHONPATH"] = _SCRIPTS_DIR + os.pathsep + env.get("PYTHONPATH", "") + r = subprocess.run( + [sys.executable, os.path.join(_SCRIPTS_DIR, script)], + input=json.dumps(payload).encode(), + capture_output=True, env=env, timeout=15, + ) + return r.stdout.decode().strip(), r.returncode + + +class _Base(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.mkdtemp() + os.environ["CLAUDE_PLUGIN_DATA"] = self._tmp + import store as _store + importlib.reload(_store) + import store + self.store = store + + +class TestCanonicalSig(_Base): + def test_path_spelling_collapses(self): + a = self.store.tool_call_sig("Read", {"file_path": "./foo.py"}) + b = self.store.tool_call_sig("Read", {"file_path": "foo.py"}) + c = self.store.tool_call_sig("Read", {"file_path": os.path.realpath("foo.py")}) + self.assertEqual(a, b) + self.assertEqual(a, c) + + def test_different_range_stays_distinct(self): + a = self.store.tool_call_sig("Read", {"file_path": "foo.py"}) + b = self.store.tool_call_sig("Read", {"file_path": "foo.py", "offset": 100}) + self.assertNotEqual(a, b) + + +class TestClearToolCalls(_Base): + def test_clear_removes_session_rows(self): + self.store.record_tool_call("s", "sig1", "Read", 100) + self.assertIsNotNone(self.store.seen_tool_call("s", "sig1")) + removed = self.store.clear_tool_calls("s") + self.assertEqual(removed, 1) + self.assertIsNone(self.store.seen_tool_call("s", "sig1")) + + +class TestPrune(_Base): + def _insert_old(self, sig, age_days): + conn = self.store.db() + with conn: + conn.execute( + "INSERT INTO tool_calls(session_id,input_sig,tool_name,token_count,file_path,mtime,ts) " + "VALUES (?,?,?,?,?,?,?)", + ("s", sig, "Read", 100, None, None, time.time() - age_days * 86400), + ) + conn.close() + + def test_prune_removes_aged_keeps_fresh(self): + self._insert_old("old", 100) + self.store.record_tool_call("s", "fresh", "Read", 100) + removed = self.store.prune(7) + self.assertGreaterEqual(removed, 1) + self.assertIsNone(self.store.seen_tool_call("s", "old")) + self.assertIsNotNone(self.store.seen_tool_call("s", "fresh")) + + def test_prune_zero_is_noop(self): + self._insert_old("old", 100) + self.assertEqual(self.store.prune(0), 0) + self.assertIsNotNone(self.store.seen_tool_call("s", "old")) + + def test_maybe_prune_throttles(self): + self._insert_old("old", 100) + first = self.store.maybe_prune() + self.assertGreaterEqual(first, 1) + self._insert_old("old2", 100) + # Within the same day → throttled, does nothing. + self.assertEqual(self.store.maybe_prune(), 0) + + +class TestTranscriptUsage(_Base): + def test_sums_usage_blocks(self): + tp = os.path.join(self._tmp, "t.jsonl") + with open(tp, "w") as fh: + fh.write(json.dumps({"message": {"usage": {"input_tokens": 100, "output_tokens": 20, "cache_read_input_tokens": 50}}}) + "\n") + fh.write(json.dumps({"message": {"role": "user", "content": "hi"}}) + "\n") + fh.write(json.dumps({"message": {"usage": {"input_tokens": 200, "output_tokens": 30}}}) + "\n") + t = self.store.transcript_usage_totals(tp) + self.assertEqual(t["input_tok"], 300) + self.assertEqual(t["output_tok"], 50) + self.assertEqual(t["cache_read_tok"], 50) + + def test_missing_file_zeros(self): + t = self.store.transcript_usage_totals("/nonexistent/x.jsonl") + self.assertEqual(t, {"input_tok": 0, "output_tok": 0, "cache_read_tok": 0}) + + +class TestFablePricing(_Base): + def test_pricing_has_fable(self): + self.assertIn("fable", self.store.PRICING) + self.assertEqual(self.store.PRICING["fable"], (10.0, 50.0)) + + def test_cost_report_maps_fable(self): + import cost_report + importlib.reload(cost_report) + self.assertEqual(cost_report._model_key("claude-fable-5"), "fable") + self.assertEqual(cost_report._model_key("anthropic.claude-opus-4-8"), "opus") + + +class TestPrecompactHook(_Base): + def test_clears_tool_calls(self): + self.store.record_tool_call("psess", "sig", "Read", 100) + out, rc = _run("precompact.py", { + "session_id": "psess", "cwd": self._tmp, + "transcript_path": "", "trigger": "auto", + }, self._tmp) + self.assertEqual(rc, 0) + self.assertEqual(out, "{}") # never blocks compaction + self.assertIsNone(self.store.seen_tool_call("psess", "sig")) + + +class TestSessionResetHook(_Base): + def test_clear_source_invalidates(self): + self.store.record_tool_call("rsess", "sig", "Read", 100) + _run("session_reset.py", {"session_id": "rsess", "source": "clear"}, self._tmp) + self.assertIsNone(self.store.seen_tool_call("rsess", "sig")) + + def test_startup_source_keeps(self): + self.store.record_tool_call("rsess2", "sig", "Read", 100) + _run("session_reset.py", {"session_id": "rsess2", "source": "startup"}, self._tmp) + self.assertIsNotNone(self.store.seen_tool_call("rsess2", "sig")) + + +class TestSubagentUsageHook(_Base): + def test_records_tier_usage(self): + tp = os.path.join(self._tmp, "sub.jsonl") + with open(tp, "w") as fh: + fh.write(json.dumps({"message": {"usage": {"input_tokens": 500, "output_tokens": 80}}}) + "\n") + out, rc = _run("subagent_usage.py", { + "session_id": "usess", "agent_type": "nexum-impl-haiku", + "transcript_path": tp, + }, self._tmp) + self.assertEqual(rc, 0) + rows = self.store.usage_rows("usess") + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["model"], "haiku") + self.assertEqual(rows[0]["input_tok"], 500) + self.assertEqual(rows[0]["output_tok"], 80) + + def test_ignores_non_nexum_agent(self): + _run("subagent_usage.py", { + "session_id": "usess2", "agent_type": "Explore", "transcript_path": "", + }, self._tmp) + self.assertEqual(self.store.usage_rows("usess2"), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_report_budget.py b/tests/test_report_budget.py new file mode 100644 index 0000000..e546078 --- /dev/null +++ b/tests/test_report_budget.py @@ -0,0 +1,208 @@ +""" +test_report_budget.py — B (wasted-context tracker + /nx-report) and +C (tiered budget alerts). +""" + +import importlib +import json +import os +import subprocess +import sys +import tempfile +import unittest + +_SCRIPTS_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts" +) +if _SCRIPTS_DIR not in sys.path: + sys.path.insert(0, _SCRIPTS_DIR) + + +def _run(script, payload, data_dir): + env = os.environ.copy() + env["CLAUDE_PLUGIN_DATA"] = data_dir + env["PYTHONPATH"] = _SCRIPTS_DIR + os.pathsep + env.get("PYTHONPATH", "") + r = subprocess.run( + [sys.executable, os.path.join(_SCRIPTS_DIR, script)], + input=json.dumps(payload).encode(), + capture_output=True, env=env, timeout=15, + ) + return r.stdout.decode().strip(), r.returncode + + +class _Base(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.mkdtemp() + os.environ["CLAUDE_PLUGIN_DATA"] = self._tmp + import store as _store + importlib.reload(_store) + import store + self.store = store + + def _write_config(self, cfg: dict): + with open(os.path.join(self._tmp, "config.json"), "w") as fh: + json.dump(cfg, fh) + + +class TestFileActivity(_Base): + def test_read_then_edit_marks_useful(self): + self.store.record_file_read("s", "./foo.py", 1200, partial=False) + self.store.record_file_read("s", "foo.py", 800) # same file, diff spelling + rows = self.store.file_activity_rows("s") + self.assertEqual(len(rows), 1) # canonicalised to one row + self.assertEqual(rows[0]["reads"], 2) + self.assertEqual(rows[0]["tokens_read"], 2000) + # Never edited yet → wasted. + self.assertEqual(len(self.store.wasted_files("s")), 1) + self.store.record_file_edit("s", "foo.py") + self.assertEqual(self.store.wasted_files("s"), []) # now useful + + def test_wasted_files_ranked_by_tokens(self): + self.store.record_file_read("s", "/a.py", 500) + self.store.record_file_read("s", "/b.py", 5000) + wf = self.store.wasted_files("s", limit=5) + self.assertEqual(os.path.basename(wf[0]["file_path"]), "b.py") + + def test_prune_includes_file_activity(self): + conn = self.store.db() + with conn: + conn.execute( + "INSERT INTO file_activity(session_id,file_path,reads,partial_reads,edits,tokens_read,ts) " + "VALUES (?,?,?,?,?,?,?)", + ("s", "/old.py", 1, 0, 0, 100, __import__("time").time() - 100 * 86400), + ) + conn.close() + removed = self.store.prune(7) + self.assertGreaterEqual(removed, 1) + self.assertEqual(self.store.file_activity_rows("s"), []) + + +class TestReport(_Base): + def test_waste_section_grades_and_suggests(self): + import report + importlib.reload(report) + self.store.record_file_read("s", "/wasted_big.py", 8000) + self.store.record_file_read("s", "/edited.py", 1000) + self.store.record_file_edit("s", "/edited.py") + section = report.build_waste_section(self.store.file_activity_rows("s")) + self.assertIn("Efficiency grade", section) + self.assertIn("WASTED", section) + self.assertIn("wasted_big.py", section) + self.assertIn("drop", section) + + def test_digest_runs_clean_with_no_data(self): + out, rc = _run("report.py", {}, self._tmp) + self.assertEqual(rc, 0) + self.assertIn("Session report", out) + + +class TestSavingsSplit(_Base): + """D1: savings split into realized / bounded / theoretical buckets.""" + + def test_savings_by_source_aggregates(self): + self.store.record_saving("s", "predup", 1000, 100) + self.store.record_saving("s", "predup", 500, 50) + self.store.record_saving("s", "grep_narrow", 0, 0) + agg = self.store.savings_by_source("s") + self.assertEqual(agg["predup"]["count"], 2) + self.assertEqual(agg["predup"]["saved_tok"], 1500) + self.assertEqual(agg["predup"]["effective_tok"], 150) + self.assertEqual(agg["grep_narrow"]["count"], 1) + self.assertEqual(agg["grep_narrow"]["effective_tok"], 0) + + def test_realized_only_counts_predup_tokens(self): + import report + importlib.reload(report) + # predup is realized (measured); grep_narrow/read_guard are bounded + # (count only); dedup is theoretical (PostToolUse, inert). + self.store.record_saving("s", "predup", 2000, 200) + self.store.record_saving("s", "grep_narrow", 0, 0) + self.store.record_saving("s", "read_guard", 0, 0) + section = report.build_savings_section(self.store.savings_by_source("s")) + self.assertIn("Realized", section) + self.assertIn("Bounded", section) + self.assertIn("Theoretical", section) + # The realized total reflects predup's effective tokens (~200), not the + # bounded interventions which carry 0 tokens. + self.assertIn("TOTAL realized", section) + self.assertIn("repeat tool calls denied", section) + self.assertIn("broad searches bounded", section) + + def test_theoretical_empty_states_inert(self): + import report + importlib.reload(report) + section = report.build_savings_section({}) + self.assertIn("contributes 0 today", section) + + def test_dedup_classified_theoretical(self): + import report + importlib.reload(report) + self.store.record_saving("s", "dedup", 5000, 500) + section = report.build_savings_section(self.store.savings_by_source("s")) + self.assertIn("would save once the field is honored", section) + # With only dedup recorded, the realized bucket stays empty — its tokens + # are never folded into the realized headline. + realized_idx = section.index("Realized") + bounded_idx = section.index("Bounded") + realized_block = section[realized_idx:bounded_idx] + self.assertIn("none recorded", realized_block) + self.assertNotIn("5.0k", realized_block) + + +class TestDedupActivityTracking(_Base): + def test_read_and_edit_recorded(self): + _run("dedup.py", { + "session_id": "d1", "tool_name": "Read", + "tool_input": {"file_path": "/tmp/x.py"}, + "tool_response": "line\n" * 40, + }, self._tmp) + rows = self.store.file_activity_rows("d1") + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["reads"], 1) + self.assertGreater(rows[0]["tokens_read"], 0) + self.assertEqual(rows[0]["edits"], 0) + + out, rc = _run("dedup.py", { + "session_id": "d1", "tool_name": "Edit", + "tool_input": {"file_path": "/tmp/x.py"}, + "tool_response": "ok", + }, self._tmp) + self.assertEqual(out, "{}") # edits never dedup + rows = self.store.file_activity_rows("d1") + self.assertEqual(rows[0]["edits"], 1) + self.assertEqual(self.store.wasted_files("d1"), []) + + +class TestBudgetAlert(_Base): + def test_tier_fires_with_systemMessage(self): + self._write_config({"budget_usd": 1.0, "intent_guard_enabled": False}) + self.store.upsert_session_cost("bsess", "Sonnet", 0.60, 1000, 200) + out, rc = _run("context_watch.py", + {"session_id": "bsess", "prompt": "do the thing", "transcript_path": ""}, + self._tmp) + self.assertEqual(rc, 0) + msg = json.loads(out).get("systemMessage", "") + self.assertIn("Budget", msg) + self.assertIn("60%", msg) + + def test_high_tier_names_wasted_files_and_compact(self): + self._write_config({"budget_usd": 1.0, "intent_guard_enabled": False}) + self.store.upsert_session_cost("bs2", "Sonnet", 0.95, 1000, 200) + self.store.record_file_read("bs2", "/huge_unedited.py", 9000) + out, _ = _run("context_watch.py", + {"session_id": "bs2", "prompt": "go", "transcript_path": ""}, + self._tmp) + msg = json.loads(out).get("systemMessage", "") + self.assertIn("huge_unedited.py", msg) + self.assertIn("/compact", msg) + + def test_disabled_by_default(self): + self.store.upsert_session_cost("bs3", "Sonnet", 5.0, 1000, 200) + out, _ = _run("context_watch.py", + {"session_id": "bs3", "prompt": "go", "transcript_path": ""}, + self._tmp) + self.assertNotIn("Budget", json.loads(out).get("systemMessage", "")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_scan_guard.py b/tests/test_scan_guard.py index ce89206..fe44d57 100644 --- a/tests/test_scan_guard.py +++ b/tests/test_scan_guard.py @@ -59,41 +59,95 @@ def _is_allow(out): return out == {} or not _is_deny(out) +def _updated_command(out): + """Return the narrowed Bash command if this is an updatedInput, else None.""" + try: + return out["hookSpecificOutput"]["updatedInput"]["command"] + except (KeyError, TypeError): + return None + + +def _updated_input(out): + """Return the full updatedInput dict if present, else None.""" + try: + return out["hookSpecificOutput"]["updatedInput"] + except (KeyError, TypeError): + return None + + class TestScanGuardGrepDeny(unittest.TestCase): - """grep -r without scoped path → deny.""" + """Unscoped grep -r → narrowed with `| head -n N` by default (working + PreToolUse lever); denied when grep_narrow_enabled is False.""" def setUp(self): self._tmp = tempfile.mkdtemp() - def test_grep_r_no_path_denied(self): - """grep -r foo with no path is denied.""" + def test_grep_r_no_path_narrowed(self): + """grep -r foo with no path is narrowed with `| head -n 80`.""" payload = { "tool_name": "Bash", "tool_input": {"command": "grep -r foo"}, } out, rc = _run_scan_guard(payload, self._tmp) self.assertEqual(rc, 0) - self.assertTrue(_is_deny(out), f"Expected deny, got: {out}") + self.assertEqual(_updated_command(out), "grep -r foo | head -n 80", + f"Expected narrowed command, got: {out}") - def test_grep_r_dot_path_denied(self): - """grep -r foo . (root) is denied.""" + def test_grep_r_dot_path_narrowed(self): + """grep -r foo . (root) is narrowed, not denied.""" payload = { "tool_name": "Bash", "tool_input": {"command": "grep -r foo ."}, } out, rc = _run_scan_guard(payload, self._tmp) self.assertEqual(rc, 0) - self.assertTrue(_is_deny(out), f"Expected deny for 'grep -r foo .', got: {out}") + self.assertEqual(_updated_command(out), "grep -r foo . | head -n 80", + f"Expected narrowed command, got: {out}") - def test_grep_recursive_upper_denied(self): - """grep -R (uppercase) is also denied.""" + def test_grep_recursive_upper_narrowed(self): + """grep -R (uppercase) is also narrowed.""" payload = { "tool_name": "Bash", "tool_input": {"command": "grep -R foo"}, } out, rc = _run_scan_guard(payload, self._tmp) self.assertEqual(rc, 0) - self.assertTrue(_is_deny(out)) + self.assertEqual(_updated_command(out), "grep -R foo | head -n 80") + + def test_grep_head_limit_config_respected(self): + """grep_head_limit config overrides the injected `| head -n N` count.""" + payload = { + "tool_name": "Bash", + "tool_input": {"command": "grep -r foo"}, + } + out, rc = _run_scan_guard(payload, self._tmp, + extra_config={"grep_head_limit": 25}) + self.assertEqual(rc, 0) + self.assertEqual(_updated_command(out), "grep -r foo | head -n 25") + + def test_grep_r_denied_when_narrow_disabled(self): + """With grep_narrow_enabled False, an unscoped grep -r is denied.""" + payload = { + "tool_name": "Bash", + "tool_input": {"command": "grep -r foo ."}, + } + out, rc = _run_scan_guard(payload, self._tmp, + extra_config={"grep_narrow_enabled": False}) + self.assertEqual(rc, 0) + self.assertTrue(_is_deny(out), f"Expected deny when narrow disabled, got: {out}") + + def test_grep_r_piped_falls_back_to_deny(self): + """An unscoped grep that already pipes can't be safely appended → deny.""" + payload = { + "tool_name": "Bash", + "tool_input": {"command": "grep -r foo | sort"}, + } + out, rc = _run_scan_guard(payload, self._tmp) + self.assertEqual(rc, 0) + # A piped recursive grep isn't classified as unscoped (path args include + # the pipe tokens), so it is allowed through unchanged — never narrowed + # into a double pipe. + self.assertIsNone(_updated_command(out)) class TestScanGuardGrepAllow(unittest.TestCase): @@ -273,20 +327,47 @@ def test_find_specific_dir_allowed(self): class TestScanGuardGrepToolDeny(unittest.TestCase): - """Grep tool with broad pattern at root → deny.""" + """Grep tool with broad pattern at root → narrowed with head_limit by + default; node_modules path still denied.""" def setUp(self): self._tmp = tempfile.mkdtemp() - def test_grep_tool_broad_pattern_denied(self): - """Grep tool with path='' and pattern='**/*' is denied.""" + def test_grep_tool_broad_pattern_narrowed(self): + """Grep tool with path='' and pattern='**/*' gets head_limit injected.""" + payload = { + "tool_name": "Grep", + "tool_input": {"path": "", "pattern": "**/*"}, + } + out, rc = _run_scan_guard(payload, self._tmp) + self.assertEqual(rc, 0) + ui = _updated_input(out) + self.assertIsNotNone(ui, f"Expected updatedInput for broad Grep, got: {out}") + self.assertEqual(ui.get("head_limit"), 80) + # Original fields preserved. + self.assertEqual(ui.get("pattern"), "**/*") + + def test_grep_tool_broad_pattern_denied_when_narrow_disabled(self): + """With grep_narrow_enabled False, a broad Grep is denied.""" payload = { "tool_name": "Grep", "tool_input": {"path": "", "pattern": "**/*"}, } + out, rc = _run_scan_guard(payload, self._tmp, + extra_config={"grep_narrow_enabled": False}) + self.assertEqual(rc, 0) + self.assertTrue(_is_deny(out), f"Expected deny when narrow disabled, got: {out}") + + def test_grep_tool_explicit_head_limit_untouched(self): + """A broad Grep that already sets head_limit is left alone (denied, + since we don't override the caller's cap).""" + payload = { + "tool_name": "Grep", + "tool_input": {"path": "", "pattern": "**/*", "head_limit": 5}, + } out, rc = _run_scan_guard(payload, self._tmp) self.assertEqual(rc, 0) - self.assertTrue(_is_deny(out), f"Expected deny for broad Grep, got: {out}") + self.assertTrue(_is_deny(out), f"Expected deny (no override), got: {out}") def test_grep_tool_node_modules_path_denied(self): """Grep tool targeting node_modules path → deny.""" @@ -362,24 +443,37 @@ def test_valid_json_unknown_tool_allowed(self): class TestScanGuardQuotedGrep(unittest.TestCase): - """Quoted grep patterns with spaces must still be caught by the guard (Step 3 fix).""" + """Quoted grep patterns with spaces must still be caught by the guard + (shlex tokenization); caught means narrowed by default (Step 3 fix).""" def setUp(self): self._tmp = tempfile.mkdtemp() - def test_quoted_pattern_with_space_denied(self): - """grep -r "def foo" . — quoted pattern with space → deny (unscoped).""" + def test_quoted_pattern_with_space_narrowed(self): + """grep -r "def foo" . — quoted unscoped pattern → narrowed, not retried.""" payload = { "tool_name": "Bash", "tool_input": {"command": 'grep -r "def foo" .'}, } out, rc = _run_scan_guard(payload, self._tmp) self.assertEqual(rc, 0) - self.assertTrue( - _is_deny(out), - f'Expected deny for grep -r "def foo" ., got: {out}', + self.assertEqual( + _updated_command(out), + 'grep -r "def foo" . | head -n 80', + f'Expected narrowed command for quoted grep, got: {out}', ) + def test_quoted_pattern_with_space_denied_when_narrow_disabled(self): + """With narrowing off, the quoted unscoped grep is still caught (denied).""" + payload = { + "tool_name": "Bash", + "tool_input": {"command": 'grep -r "def foo" .'}, + } + out, rc = _run_scan_guard(payload, self._tmp, + extra_config={"grep_narrow_enabled": False}) + self.assertEqual(rc, 0) + self.assertTrue(_is_deny(out), f'Expected deny, got: {out}') + class TestScanGuardReadGuard(unittest.TestCase): """Read-guard: large files get a limit injected via updatedInput; small files pass through.""" diff --git a/tests/test_store.py b/tests/test_store.py index b7b083a..bf91a4e 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -853,6 +853,76 @@ def test_calibration_rows_empty_for_unknown_repo(self): self.assertEqual(store.calibration_rows("does-not-exist"), []) +class TestCalibrationAdvice(unittest.TestCase): + """calibration_advice: Wilson-bound, bidirectional, global-prior fallback.""" + + def setUp(self): + self._tmp = tempfile.mkdtemp() + os.environ["CLAUDE_PLUGIN_DATA"] = self._tmp + + def tearDown(self): + os.environ.pop("CLAUDE_PLUGIN_DATA", None) + + def test_low_pass_rate_nudges_up(self): + import store + # 3/10 first-try → Wilson lower well below 0.6 → route up. + store.record_calibration("r", "mechanical", dispatched=10, passed_first_try=3) + adv = store.calibration_advice("r") + self.assertEqual(adv["mechanical"]["action"], "up") + self.assertEqual(adv["mechanical"]["source"], "repo") + self.assertLess(adv["mechanical"]["lower"], 0.6) + + def test_high_pass_rate_nudges_down(self): + import store + # 50/50 first-try → Wilson lower >= 0.9 → route down (and not mechanical). + store.record_calibration("r", "needs-strong", dispatched=50, passed_first_try=50) + adv = store.calibration_advice("r") + self.assertEqual(adv["needs-strong"]["action"], "down") + self.assertGreaterEqual(adv["needs-strong"]["lower"], 0.9) + + def test_mid_pass_rate_keeps(self): + import store + # 24/30 → Wilson lower in [0.6, 0.9) → keep. + store.record_calibration("r", "standard", dispatched=30, passed_first_try=24) + adv = store.calibration_advice("r") + self.assertEqual(adv["standard"]["action"], "keep") + + def test_below_min_samples_no_advice(self): + import store + store.record_calibration("r", "mechanical", dispatched=3, passed_first_try=0) + adv = store.calibration_advice("r") + self.assertNotIn("mechanical", adv) + + def test_global_prior_fallback(self): + import store + # Repo "r2" has no rows; the cross-repo "_global" aggregate does. + store.record_calibration("_global", "mechanical", dispatched=10, passed_first_try=2) + adv = store.calibration_advice("r2") + self.assertEqual(adv["mechanical"]["action"], "up") + self.assertEqual(adv["mechanical"]["source"], "global") + + def test_repo_rows_preferred_over_global(self): + import store + store.record_calibration("_global", "standard", dispatched=10, passed_first_try=2) + store.record_calibration("r", "standard", dispatched=30, passed_first_try=24) + adv = store.calibration_advice("r") + self.assertEqual(adv["standard"]["source"], "repo") + self.assertEqual(adv["standard"]["action"], "keep") + + def test_disabled_returns_empty(self): + import store + store.record_calibration("r", "mechanical", dispatched=10, passed_first_try=1) + adv = store.calibration_advice("r", cfg={"route_calib_enabled": False}) + self.assertEqual(adv, {}) + + def test_deterministic_across_calls(self): + import store + store.record_calibration("r", "mechanical", dispatched=10, passed_first_try=3) + a = json.dumps(store.calibration_advice("r"), sort_keys=True) + b = json.dumps(store.calibration_advice("r"), sort_keys=True) + self.assertEqual(a, b) + + class TestCalibrationCLI(unittest.TestCase): """calib-record then calib-list round-trip through the CLI."""