diff --git a/docs/specs/benchmark-rubric-v1.md b/docs/specs/benchmark-rubric-v1.md new file mode 100644 index 00000000..36e6dc3c --- /dev/null +++ b/docs/specs/benchmark-rubric-v1.md @@ -0,0 +1,105 @@ +# Benchmark Rubric v1 — ratified defaults + +*Companion to `docs/specs/benchmark-engine.md` (issue #99). This file is the +ratification that spec §4.6 requires before the engine may ship: the engine +proposes, this document decides. The values here are the loaded defaults in +`reports/benchmark.py`; they are surfaced in every payload (`weights`, +`rubric_version`, `ci_level`, `success_threshold`) and overridable per call, so +tuning them is data, not a code change.* + +*"Rubric v1" is the version of **this scoring rubric**, not the product/PyPI +version. It has nothing to do with the maintainer-only product version and moves +on its own cadence.* + +--- + +## Composite weights + +The per-stratum composite is a weighted blend of three axes, each normalized +*within the stratum* so the comparison is like-for-like: + +| Axis | Weight | What it measures | +|---|---:|---| +| **success** | **0.45** | Did the work land — the tiered outcome signal (§ below). | +| **cost** | **0.35** | Dollars per successful outcome (min-max inverse; cheapest wins). | +| **effort** | **0.20** | Turns / retries (min-max inverse; fewest wins). | + +Weights sum to 1.0. Rationale: outcome dominates (a cheap failure is not a win), +cost is the flagship economic axis, effort is a real but noisier signal. + +**Wall-clock duration is descriptive only** — it includes human away-from-keyboard +time, so it is never scored into the winner. + +**Reasoning efficiency is descriptive only and is never scored.** Providers that +don't report a wire reasoning-token count read as 0, so cross-provider reasoning +is not apples-to-apples. It is surfaced (`reasoning_share`) for context, never +folded into the composite. + +## Success threshold τ + +``` +τ = 7.0 +``` + +A real LLM grade (`grades.success`, `session_quality_metrics`) at or above 7.0 +counts as a success on the grade tier; below counts as a failure. Fallback +(non-LLM) grades are never persisted, so only real grades reach this tier. + +## Success signal — tiered, highest-confidence-first + +Composed per session; the first tier with a signal decides `outcome_success`, +and the deciding tier is recorded. Sessions with no signal are `NULL` — excluded +from rates, but counted in the coverage figure shown beside every verdict. + +| Tier | Source | success = 1 | success = 0 | +|---|---|---|---| +| 1 ground truth | PR / CI via commit link | PR merged & not reverted, or CI passed | PR reverted, or CI failed | +| 2 code delta | `static_analysis_findings` | net-improved | net-regressed | +| 3 LLM grade | `grades.success` (real only) | ≥ τ | < τ | +| 4 behavioral | `session_mart` proxies | one-shot | high-retry (≥ 8 turns) | + +## Confidence level + +``` +CI_LEVEL = 0.90 +``` + +All intervals (Wilson for success rates; seeded percentile bootstrap for +cost/turns; difference effects) report at 90%. The maintainer may raise this to +0.95 per call; 0.90 is the default. + +## Statistical gates (honesty contract) + +Unchanged from spec §4 and pinned in `services/benchmark_stats.py`: + +| Gate | Value | +|---|---| +| Min sessions per model per stratum | 5 | +| Min qualifying models to compare a stratum | 2 | +| Min balanced sessions for a cross-task headline | 20 | +| Min relative cost effect | 10% | +| Min success effect | 10 percentage points | +| Min grade effect | 0.5 points | +| Multiple-comparison control | Benjamini–Hochberg FDR at α = 1 − CI_LEVEL | +| Bootstrap iterations / seed | 2000 / pinned (reproducible) | + +Below the sample floor the verdict is the literal string **"insufficient +evidence"** with a `confidence` of `none`. A cross-task winner must win the +composite in ≥ 2 strata (each a real, FDR-significant separation), never clearly +lose a stratum, and clear the balanced-sample floor. + +## Canonical intent taxonomy + +Ratified as the **6-label** set (adopting `ops` over the recommender's former +5), owned by `services/task_classifier.py` and shared by the tag service, the +mode recommender, and the benchmark: + +``` +build · fix · explore · refactor · test · ops +``` + +## Scope + +Phase 3 live-replay is **deferred** (no `benchmark run` in this MVP). The MVP is +the observational engine (spec §8 Moves 0–3): a natural experiment over the +user's own history, stated as such in every payload. diff --git a/docs/specs/session-schema-v1.md b/docs/specs/session-schema-v1.md index ae133101..8ad2e78b 100644 --- a/docs/specs/session-schema-v1.md +++ b/docs/specs/session-schema-v1.md @@ -1,6 +1,6 @@ # Session Schema v1 — open exchange format for AI coding sessions -**Status:** v1 (pinned to `schema_version = 27`). +**Status:** v1 (pinned to `schema_version = 28`). **Audience:** anyone writing a tool that wants to read from, or write to, the StackUnderflow store without reverse-engineering the SQL. **Scope:** the local SQLite schema at `~/.stackunderflow/store.db`. This document is the source of truth for the on-disk shape; the migrations under `stackunderflow/store/migrations/` (`.sql` DDL and `.py` data migrations) are the reference implementation. @@ -20,7 +20,7 @@ The schema described here is **additive-only**. Any future column requires a new ## Schema version -Pin to `schema_version = 27`. The current migration set is: +Pin to `schema_version = 28`. The current migration set is: | version | file | what it adds | |---|---|---| @@ -50,6 +50,7 @@ Pin to `schema_version = 27`. The current migration set is: | 25 | `v025_command_day_mart.sql` | `command_day_mart` (per-`(day, project_id)` user-command count; windows the Overview "Commands" KPI — ui-perf #25) | | 26 | `v026_reasoning_tokens.sql` | adds `usage_events.reasoning_tokens` (reasoning/"thinking" attribution — an additive-metadata SUBSET of `output_tokens`, never priced; `DEFAULT 0`) | | 27 | `v027_worktree_of.sql` | adds `projects.worktree_of` (nullable parent-project slug for git-worktree fragment projects; NULL = normal project — campaign #8) | +| 28 | `v028_sync_identity_outbox.sql` | `sync_identity` + `sync_outbox` (opt-in multi-device sync — device identity + per-shard push watermark; #100 Phase 1) | Version 15 was reserved during planning and never created — the sequence skips from 14 to 16 by design. The migration runner keys on the leading `vNNN`, so the gap is harmless. diff --git a/pyproject.toml b/pyproject.toml index 30cdfdd9..1ea4129c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,15 @@ analysis = [ "mypy>=1.5.0", "coverage>=7.0.0", ] +# Opt-in multi-device sync (docs/specs/multi-device-sync.md). The core product +# never pulls these — they are needed only for the `sync` CLI commands, which +# are import-guarded and print a one-line install hint when the extra is absent. +# `pyrage` = the age file-encryption binding (client-side encryption); `boto3` = +# the S3-compatible bucket client (bring-your-own bucket / credentials). +sync = [ + "boto3>=1.34", + "pyrage>=1.1", +] [project.scripts] stackunderflow = "stackunderflow.cli:cli" diff --git a/stackunderflow-ui/src/components/dashboard/CompareTab.tsx b/stackunderflow-ui/src/components/dashboard/CompareTab.tsx index cf9680ba..f9abdadf 100644 --- a/stackunderflow-ui/src/components/dashboard/CompareTab.tsx +++ b/stackunderflow-ui/src/components/dashboard/CompareTab.tsx @@ -10,6 +10,7 @@ import { formatCost, formatNumber, formatModelName } from '../../services/format import { useCurrency } from '../../services/currency' import { shortenModelId } from '../../services/providerStyle' import { useFilters } from '../../services/filters' +import ModelWinsPanel from './ModelWinsPanel' // --------------------------------------------------------------------------- // CompareTab — v0.6.1 multi-provider polish. @@ -399,6 +400,9 @@ export default function CompareTab() { )} + + {/* "Which model wins" — outcome-aware benchmark beneath the cost table. */} + ) } diff --git a/stackunderflow-ui/src/components/dashboard/ModelWinsPanel.tsx b/stackunderflow-ui/src/components/dashboard/ModelWinsPanel.tsx new file mode 100644 index 00000000..5e4785da --- /dev/null +++ b/stackunderflow-ui/src/components/dashboard/ModelWinsPanel.tsx @@ -0,0 +1,296 @@ +import { useState } from 'react' +import { useQuery } from '@tanstack/react-query' +import { IconScale, IconAlertCircle } from '@tabler/icons-react' +import { getBenchmark, type BenchmarkPeriod } from '../../services/api' +import type { + BenchmarkReportData, + BenchmarkStratum, + BenchmarkModelRow, + BenchmarkConfidence, + BenchmarkCellVerdict, +} from '../../types/api' +import LoadingSpinner from '../common/LoadingSpinner' +import EmptyState from '../common/EmptyState' +import { formatCost } from '../../services/format' +import { useCurrency } from '../../services/currency' + +// --------------------------------------------------------------------------- +// ModelWinsPanel — "Which model wins" (spec 26 / issue #99). +// +// Renders `GET /api/benchmark` beneath the Compare tab's model×cost table: an +// observational benchmark over the user's own history. The point of the panel +// is honesty, made visual — every row carries n, coverage, a Wilson success CI, +// and a confidence chip; under-powered rows render greyed as "insufficient +// evidence" rather than a fake rank; and a method banner states the natural- +// experiment caveat up front. +// --------------------------------------------------------------------------- + +const PERIODS: { id: BenchmarkPeriod; label: string }[] = [ + { id: 'today', label: 'Today' }, + { id: 'week', label: '7d' }, + { id: 'month', label: '30d' }, + { id: 'all', label: 'All' }, +] + +function pct(x: number | null | undefined): string { + if (x === null || x === undefined || !Number.isFinite(x)) return '—' + return `${(x * 100).toFixed(0)}%` +} + +const CONFIDENCE_CLASS: Record = { + high: 'bg-green-500/10 text-green-600 dark:text-green-400', + medium: 'bg-blue-500/10 text-blue-600 dark:text-blue-400', + low: 'bg-yellow-500/10 text-yellow-700 dark:text-yellow-400', + none: 'bg-gray-500/10 text-gray-500 dark:text-gray-400', +} + +const CELL_VERDICT_CLASS: Record = { + clear: 'bg-green-500/10 text-green-600 dark:text-green-400', + weak: 'bg-yellow-500/10 text-yellow-700 dark:text-yellow-400', + 'insufficient evidence': 'bg-gray-500/10 text-gray-500 dark:text-gray-400', +} + +function Chip({ label, className }: { label: string; className: string }) { + return ( + + {label} + + ) +} + +function MethodBanner({ report }: { report: BenchmarkReportData }) { + const cov = report.coverage + return ( +
+ + + Based on {cov.sessions_total.toLocaleString()} sessions you already ran — + a natural experiment, not a controlled trial. Success measured on{' '} + {cov.sessions_scored.toLocaleString()}/{cov.sessions_total.toLocaleString()}{' '} + sessions · grade coverage {pct(cov.grade_coverage)}. Weights: success{' '} + {report.weights.success}, cost {report.weights.cost}, effort{' '} + {report.weights.effort} · {Math.round(report.ci_level * 100)}% CI. + +
+ ) +} + +function VerdictCard({ + report, + currency, +}: { + report: BenchmarkReportData + currency: ReturnType['currency'] +}) { + const v = report.verdict + if (!v.winning_model) { + return ( +
+
+ + + No cross-task winner yet + +
+ {v.caveats[0] && ( +

{v.caveats[0]}

+ )} +
+ ) + } + return ( +
+
+ + + {v.headline} + + +
+
+ {v.cost_per_outcome_usd !== null && ( + {formatCost(v.cost_per_outcome_usd, currency)} / successful outcome + )} + {v.runner_up && · runner-up {v.runner_up}} +
+
+ ) +} + +function ModelRow({ + row, + isWinner, + currency, +}: { + row: BenchmarkModelRow + isWinner: boolean + currency: ReturnType['currency'] +}) { + const wilson = row.success_rate.ci_wilson + const cpo = row.cost_per_outcome.point + const dim = row.qualified ? '' : 'opacity-50' + return ( + + + {isWinner && } + {row.model} + {!row.qualified && ( + insufficient evidence + )} + + + {row.n} + + + {pct(row.success_rate.point)} + {wilson && ( + + {' '} + [{pct(wilson[0])}–{pct(wilson[1])}] + + )} + + + {cpo !== null ? `${formatCost(cpo, currency)}/outcome` : '—'} + + + {pct(row.coverage)} + + + {row.composite.toFixed(2)} + + + ) +} + +function StratumCard({ + stratum, + currency, +}: { + stratum: BenchmarkStratum + currency: ReturnType['currency'] +}) { + return ( +
+
+ + {stratum.intent} × {stratum.size_band} + + +
+
+ + + + + + + + + + + + + {stratum.models.map(m => ( + + ))} + +
ModelnSuccess (90% CI)Cost / outcomeCoverageComposite
+
+
+ ) +} + +export default function ModelWinsPanel() { + const { currency } = useCurrency() + const [period, setPeriod] = useState('all') + + const { data, isLoading, error } = useQuery({ + queryKey: ['benchmark', period], + queryFn: () => getBenchmark(period), + staleTime: 60_000, + }) + + const report = data?.report + + return ( +
+
+
+ +

+ Which model wins +

+ + cost per successful outcome, per task type — from your own history + +
+
+ {PERIODS.map(p => ( + + ))} +
+
+ + {isLoading && } + + {error && ( +
+ Failed to load benchmark: {error instanceof Error ? error.message : 'Unknown error'} +
+ )} + + {!isLoading && !error && report && report.coverage.sessions_total === 0 && ( + } + title="Not enough history to compare models yet" + description="Once you've run a few sessions across more than one model, this panel compares them by task type — honestly." + /> + )} + + {!isLoading && !error && report && report.coverage.sessions_total > 0 && ( + <> + + + {report.strata.length > 0 && ( +
+

+ Per task type (intent × size) +

+ {report.strata.map(s => ( + + ))} +
+ )} + + )} +
+ ) +} diff --git a/stackunderflow-ui/src/services/api.ts b/stackunderflow-ui/src/services/api.ts index a1b4fcd1..3c85a806 100644 --- a/stackunderflow-ui/src/services/api.ts +++ b/stackunderflow-ui/src/services/api.ts @@ -17,6 +17,7 @@ import type { CostByProviderResponse, YieldResponse, ForksResponse, + BenchmarkResponse, PlanResponse, BudgetResponse, BudgetUpdate, @@ -453,6 +454,21 @@ export async function getForks(period: ForksPeriod = 'all'): Promise { + const params = new URLSearchParams({ period }) + return fetchJson(`${BASE}/benchmark?${params}`) +} + export async function getPlan(): Promise { return fetchJson(`${BASE}/plan`) } diff --git a/stackunderflow-ui/src/types/api.ts b/stackunderflow-ui/src/types/api.ts index 36b90d30..2821ca0d 100644 --- a/stackunderflow-ui/src/types/api.ts +++ b/stackunderflow-ui/src/types/api.ts @@ -677,6 +677,87 @@ export interface ForksResponse { warning: string } +// --------------------------------------------------------------------------- +// Comparative benchmark engine (`GET /api/benchmark`, spec 26 / issue #99). +// An observational "which model wins for your work" verdict over the user's own +// history, with the full statistical-honesty machinery. All cost fields are +// pre-converted to the active currency by the route. +// --------------------------------------------------------------------------- + +/** A cost point estimate with its confidence interval (currency-converted). */ +export interface BenchmarkCostBlock { + point: number | null + ci: [number, number] | null +} + +/** One model's row within a stratum. */ +export interface BenchmarkModelRow { + model: string + n: number + qualified: boolean + coverage: number + success_measured_n: number + success_rate: { point: number | null; ci_wilson: [number, number] | null } + cost_per_outcome: BenchmarkCostBlock + median_cost: BenchmarkCostBlock + median_turns: number + reasoning_share: number + composite: number +} + +export type BenchmarkCellVerdict = 'clear' | 'weak' | 'insufficient evidence' + +/** One (intent × size) stratum: like-for-like tasks. */ +export interface BenchmarkStratum { + intent: string + size_band: string + models: BenchmarkModelRow[] + assignment_balance: Record + cell_verdict: BenchmarkCellVerdict + winner: string | null + effect: { + success_risk_difference?: number + cost_relative_delta?: number + statistically_separated?: boolean + practically_separated?: boolean + } +} + +export type BenchmarkConfidence = 'none' | 'low' | 'medium' | 'high' + +export interface BenchmarkVerdict { + headline: string + winning_model: string | null + confidence: BenchmarkConfidence + cost_per_outcome_usd: number | null + runner_up: string | null + caveats: string[] +} + +export interface BenchmarkReportData { + verdict: BenchmarkVerdict + strata: BenchmarkStratum[] + coverage: { + sessions_total: number + sessions_scored: number + grade_coverage: number + } + rubric_version: number + weights: Record + ci_level: number + success_threshold: number + warning: string + method_notes: string[] +} + +export interface BenchmarkResponse { + period: string + scope: string + report: BenchmarkReportData + currency: CurrencyInfo + warning: string | null +} + /** * Plan + usage payload from `GET /api/plan`. Both fields are nullable — * when no plan is configured, the route returns `{plan: null, usage: null}` diff --git a/stackunderflow/cli.py b/stackunderflow/cli.py index 0c77b06d..05156452 100644 --- a/stackunderflow/cli.py +++ b/stackunderflow/cli.py @@ -1224,6 +1224,179 @@ def _prune_backups(keep: int) -> None: click.echo(f" Pruned old backup: {old.name}") +# ── sync ────────────────────────────────────────────────────────────────────── +# +# Opt-in, client-side-encrypted, bring-your-own-bucket backup of the analytics +# aggregates (docs/specs/multi-device-sync.md, Phase 1 MVP). Default OFF: with +# no ``sync_identity`` row there is no network, no credentials, and the optional +# ``[sync]`` dependencies need not be installed. Raw transcripts, ``usage_events`` +# and the price book NEVER leave the machine — only the derived marts, re-keyed +# from the local ``project_id`` to the machine-stable ``(provider, slug)``. + +_SYNC_INSTALL_HINT = ( + " This command needs optional dependencies that aren't installed.\n" + " Install them with: pip install 'stackunderflow[sync]'" +) + + +def _sync_missing_deps(*, need_bucket: bool) -> list[str]: + """Return the missing optional ``[sync]`` package names (empty = all present).""" + import importlib.util + missing: list[str] = [] + if importlib.util.find_spec("pyrage") is None: + missing.append("pyrage") + if need_bucket and importlib.util.find_spec("boto3") is None: + missing.append("boto3") + return missing + + +def _print_sync_init_banner(identity, device_uuid: str, bucket_url: str) -> None: + """Loud, unmissable zero-knowledge / key-loss warning shown once at ``sync init``.""" + line = " " + "─" * 64 + click.echo(line) + click.echo(" SYNC ENCRYPTION KEY — READ THIS BEFORE YOU CONTINUE") + click.echo(line) + click.echo( + " This device just generated a private encryption key. Everything\n" + " pushed to your bucket is encrypted with it, and NOTHING can decrypt\n" + " it without this key — not the bucket host, not this project, no one.\n" + ) + click.echo( + " IF YOU LOSE THIS KEY, THE OFF-SITE COPY IS UNRECOVERABLE CIPHERTEXT.\n" + " There is no reset, no recovery, no backdoor. That is the point.\n" + ) + click.echo( + " Save the key below in your password manager NOW. To read this data on\n" + " another device, copy the SAME key there — devices must share one key.\n" + ) + click.echo(" Key (store securely — shown once):") + click.echo(f" {identity.secret}") + click.echo("") + click.echo(f" Fingerprint: {identity.fingerprint}") + click.echo(f" Device: {device_uuid}") + click.echo(f" Bucket: {bucket_url}") + click.echo(f" Key file: {_STATE_DIR / 'sync-identity'} (mode 0600)") + click.echo(line) + + +@cli.group("sync") +def sync_group(): + """Encrypted, bring-your-own-bucket backup of your analytics aggregates (opt-in).""" + + +@sync_group.command("init") +@click.option("--bucket", "bucket_url", required=True, + help="Destination bucket, e.g. s3://my-bucket or s3://my-bucket/prefix") +@click.option("--endpoint", "endpoint_url", default=None, + help="Custom object-store endpoint URL (set it for non-default storage providers)") +@click.option("--force", is_flag=True, default=False, + help="Replace an existing sync key on this device (destroys access to data " + "encrypted under the old key — back it up first)") +def sync_init(bucket_url: str, endpoint_url: str | None, force: bool): + """Generate this device's encryption key and record the bucket destination. + + Prints the freshly generated key ONCE — save it, and copy it to your other + devices. Only the key's fingerprint is stored in the database; the secret + lives in a 0600 file (or the keychain / STACKUNDERFLOW_SYNC_KEY env var). + """ + if _sync_missing_deps(need_bucket=False): + click.echo(_SYNC_INSTALL_HINT) + sys.exit(1) + from stackunderflow.sync import keys, runner + + conn = _open_store() + try: + existing = runner.load_identity(conn) + if existing is not None and not force: + click.echo(" Sync is already configured on this device.") + click.echo(f" device: {existing['device_uuid']}") + click.echo(f" key fingerprint: {existing['key_fingerprint']}") + click.echo(" Re-running will NOT change the key. To replace it, back up the") + click.echo(" current key first, then re-run with --force (this destroys access") + click.echo(" to any data already encrypted under the old key).") + sys.exit(1) + + identity = keys.generate_identity() + keys.store_secret_file(identity.secret, _STATE_DIR) + device_uuid = existing["device_uuid"] if existing is not None else runner.new_device_uuid() + runner.write_identity( + conn, + device_uuid=device_uuid, + key_fingerprint=identity.fingerprint, + bucket_url=bucket_url, + endpoint_url=endpoint_url, + created_at=runner.utcnow_iso(), + ) + _print_sync_init_banner(identity, device_uuid, bucket_url) + finally: + conn.close() + + +@sync_group.command("push") +def sync_push(): + """Encrypt and upload changed aggregate shards to your bucket. + + Idempotent — an unchanged shard is skipped (zero uploads). Exits non-zero on + any failure so it is safe to script. + """ + if _sync_missing_deps(need_bucket=True): + click.echo(_SYNC_INSTALL_HINT) + sys.exit(1) + from stackunderflow.sync import runner + + conn = _open_store() + try: + if not runner.is_enabled(conn): + click.echo(" Sync is not configured. Run: stackunderflow sync init --bucket s3://your-bucket") + sys.exit(1) + try: + result = runner.run_push(conn, state_dir=_STATE_DIR) + except Exception as exc: + click.echo(f" sync push failed: {exc}") + sys.exit(1) + finally: + conn.close() + + if result.uploaded == 0: + click.echo(f" Up to date — {result.skipped} shard(s) unchanged, nothing to upload.") + else: + mb = result.bytes_uploaded / (1 << 20) + click.echo(f" Pushed {result.uploaded} shard(s) ({mb:.2f} MB); {result.skipped} unchanged.") + click.echo(f" Generation {result.generation}. Manifest committed.") + + +@sync_group.command("status") +@click.option("--json", "as_json", is_flag=True, default=False, help="Emit machine-readable JSON") +def sync_status(as_json: bool): + """Show sync configuration and how many shards are pending upload (local only).""" + from stackunderflow.sync import runner + + conn = _open_store() + try: + st = runner.status(conn) + finally: + conn.close() + + if as_json: + click.echo(json.dumps(st.as_dict(), indent=2)) + return + + if not st.enabled: + click.echo(" Sync: off (no key on this device).") + click.echo(" Enable with: stackunderflow sync init --bucket s3://your-bucket") + return + + click.echo(" Sync: on") + click.echo(f" device: {st.device_uuid}") + click.echo(f" key fingerprint: {st.fingerprint}") + click.echo(f" bucket: {st.bucket_url}") + if st.endpoint_url: + click.echo(f" endpoint: {st.endpoint_url}") + click.echo(f" shards (local): {st.shard_count}") + click.echo(f" pending upload: {len(st.pending)}") + click.echo(f" last push: {st.last_push_ts or 'never'}") + + # ── data commands ──────────────────────────────────────────────────────────── _VALID_FORMATS = ("text", "json") @@ -2095,6 +2268,223 @@ def memory_embed(batch): ) +# ── benchmark: outcome-aware "which model wins for your work" ───────────────── +# +# ``stackunderflow benchmark`` surfaces the comparative benchmark engine +# (spec 26 / issue #99): an observational, statistically-honest verdict over the +# user's own history — per-task-type winners, or an honest "insufficient +# evidence". ``show`` prints the leaderboard + per-stratum honesty; ``recommend`` +# picks a model for a described task. ``--json`` emits the same +# ``stackunderflow.memory/1`` envelope the ``memory`` namespace uses, so an agent +# can ask "which model for this refactor?" and splice a bounded, evidence- +# carrying answer straight into its context. There is deliberately no ``run`` +# subcommand — nothing is executed; the benchmark is computed from history. + +_BENCH_PERIOD_ALIASES = { + "today": "today", "week": "7days", "7days": "7days", + "month": "month", "30days": "30days", "all": "all", +} + + +def _bench_scope(period): + """Resolve a friendly period to a Scope, raising a Click error on a bad one.""" + from stackunderflow.reports.scope import parse_period + + spec = _BENCH_PERIOD_ALIASES.get(period) + if spec is None: + raise click.BadParameter( + f"Invalid period {period!r}. Valid: {', '.join(_BENCH_PERIOD_ALIASES)}", + param_hint="--period", + ) + return parse_period(spec) + + +def _bench_project_ids(conn, project): + """Resolve a ``--project`` slug/path to a project_ids list, or None for all.""" + if not project: + return None + slug = Path(project).name + try: + rows = conn.execute( + "SELECT id FROM projects WHERE slug = ?", (slug,) + ).fetchall() + except Exception: # noqa: BLE001 — advisory: a bad store scopes to nothing + return [] + return [int(r["id"]) for r in rows] + + +def _bench_pack(rows, budget): + """Greedily keep leading strata within ``budget`` estimated tokens. + + Returns ``(kept, truncated)``. A non-positive budget disables packing. + """ + from stackunderflow.cli_helpers import agent_output + + if not budget or budget <= 0: + return rows, False + kept: list = [] + for r in rows: + trial = [*kept, r] + if agent_output.estimate_tokens(trial) > budget and kept: + return kept, True + kept = trial + return kept, False + + +@cli.group("benchmark") +def benchmark_group(): + """Which model wins for the kind of work you actually do. + + An observational benchmark over your own history — a natural experiment you + already ran, not live replay. Every verdict carries n, coverage, confidence + intervals and a ``confidence`` label, and says "insufficient evidence" + rather than guess. Run any subcommand with ``--json`` for the stable, + token-bounded agent-output envelope. + """ + + +@benchmark_group.command("show") +@click.option("--period", default="all", show_default=True, help="today | week | month | all") +@click.option("--project", default=None, help="Project slug/path to scope to. Default: whole store.") +@click.option("--intent", default=None, help="Filter to one intent stratum (build/fix/explore/refactor/test/ops).") +@click.option("--context-budget", "context_budget", type=int, default=None, + help="Token budget for --json output (strata are packed to fit).") +@click.option("--json", "as_json", is_flag=True, default=False, help="Shortcut for --format json.") +@click.option("--format", "fmt", type=click.Choice(_VALID_FORMATS), default="text", show_default=True, + help="Output format. 'json' emits the stable agent-output envelope.") +def benchmark_show(period, project, intent, context_budget, as_json, fmt): + """Leaderboard + per-stratum honesty for the current scope.""" + json_mode = _memory_format(fmt, as_json) == "json" + budget = _resolve_context_budget(context_budget) + from stackunderflow.reports.benchmark import analyze_benchmark + + scope = _bench_scope(period) + conn = _open_store() + try: + project_ids = _bench_project_ids(conn, project) + report = analyze_benchmark( + conn, scope=scope, project_ids=project_ids, intent=intent, + ) + finally: + conn.close() + + if json_mode: + from stackunderflow.cli_helpers import agent_output + + rows, truncated = _bench_pack(report.get("strata") or [], budget) + envelope = agent_output.build_envelope( + command="benchmark", + query={"period": period, "project": project, "intent": intent}, + results=rows, + budget=budget, + truncated=truncated, + extra={ + "verdict": report.get("verdict"), + "coverage": report.get("coverage"), + "weights": report.get("weights"), + "rubric_version": report.get("rubric_version"), + "ci_level": report.get("ci_level"), + "warning": report.get("warning"), + }, + ) + click.echo(agent_output.render(envelope)) + return + + _emit_benchmark_text(report, period=period, scope=scope.label) + + +@benchmark_group.command("recommend") +@click.option("--intent", required=True, help="Task intent: build/fix/explore/refactor/test/ops.") +@click.option("--size", default=None, help="Task size band: tiny/small/med/large.") +@click.option("--language", default=None, help="Dominant language hint (e.g. python).") +@click.option("--project", default=None, help="Project slug/path to scope to.") +@click.option("--context-budget", "context_budget", type=int, default=None, help="Token budget for --json output.") +@click.option("--json", "as_json", is_flag=True, default=False, help="Shortcut for --format json.") +@click.option("--format", "fmt", type=click.Choice(_VALID_FORMATS), default="text", show_default=True, + help="Output format. 'json' emits the stable agent-output envelope.") +def benchmark_recommend(intent, size, language, project, context_budget, as_json, fmt): + """Outcome-aware model pick for a described task.""" + json_mode = _memory_format(fmt, as_json) == "json" + budget = _resolve_context_budget(context_budget) + from stackunderflow.reports.benchmark import recommend_from_history + + conn = _open_store() + try: + project_ids = _bench_project_ids(conn, project) + rec = recommend_from_history( + conn, intent=intent, size=size, language=language, project_ids=project_ids, + ) + finally: + conn.close() + + if json_mode: + from stackunderflow.cli_helpers import agent_output + + envelope = agent_output.build_envelope( + command="benchmark-recommend", + query={"intent": intent, "size": size, "language": language, "project": project}, + results=[rec], + budget=budget, + truncated=False, + ) + click.echo(agent_output.render(envelope)) + return + + click.echo(f"Task: intent={intent} size={size or 'any'} language={language or 'any'}") + if rec.get("recommended_model"): + click.echo( + f" → {rec['recommended_model']} " + f"(confidence: {rec.get('confidence')}, basis: {rec.get('basis')})" + ) + else: + click.echo(" → insufficient evidence") + if rec.get("rationale"): + click.echo(f" {rec['rationale']}") + + +def _emit_benchmark_text(report, *, period, scope): + """Human-readable leaderboard for ``benchmark show``.""" + v = report.get("verdict") or {} + cov = report.get("coverage") or {} + click.echo(f"Benchmark — {scope} (period: {period})") + click.echo("") + if v.get("winning_model"): + cpo = v.get("cost_per_outcome_usd") + cpo_s = f" at ${cpo:.4f}/successful outcome" if cpo is not None else "" + click.echo(f"Verdict: {v['headline']}{cpo_s}") + click.echo( + f" confidence: {v.get('confidence')} runner-up: {v.get('runner_up') or '—'}" + ) + else: + click.echo(f"Verdict: {v.get('headline', 'insufficient evidence')}") + for c in (v.get("caveats") or [])[:1]: + click.echo(f" {c}") + click.echo("") + click.echo( + f"Coverage: {cov.get('sessions_scored', 0)}/{cov.get('sessions_total', 0)} " + f"sessions scored · grade coverage {cov.get('grade_coverage', 0) * 100:.0f}%" + ) + strata = report.get("strata") or [] + if strata: + click.echo("") + click.echo("Per-stratum (intent × size):") + for s in strata: + head = f" {s['intent']} × {s['size_band']}: {s['cell_verdict']}" + if s.get("winner"): + head += f" — {s['winner']} leads" + click.echo(head) + for m in s.get("models", []): + sr = m["success_rate"]["point"] + sr_s = f"{sr * 100:.0f}%" if sr is not None else "n/a" + cpo = m["cost_per_outcome"]["point"] + cpo_s = f"${cpo:.4f}/outcome" if cpo is not None else "—" + floor = "" if m["qualified"] else " [below sample floor]" + click.echo( + f" {m['model']}: n={m['n']}, success {sr_s}, " + f"{cpo_s}, composite {m['composite']:.2f}{floor}" + ) + + # ── context replay: reconstruct what the model "saw" at a point in a session ── # # ``stackunderflow context-replay --at `` reconstructs the diff --git a/stackunderflow/hooks/proactive.py b/stackunderflow/hooks/proactive.py new file mode 100644 index 00000000..11df9e5a --- /dev/null +++ b/stackunderflow/hooks/proactive.py @@ -0,0 +1,800 @@ +"""Proactive nudge governance + the command-cluster nudge (spec 27 / #97). + +Where :mod:`stackunderflow.hooks.recall` decides *what* the memory store knows +about the thing a tool is about to touch, this module decides *whether that is +worth saying* — the anti-annoyance contract that the shipped hooks lack. It is +the single deterministic gate for every proactive/recall nudge: + +* **Governance** — a pure :func:`should_surface` ``(signal, state) -> bool`` and + a stateful :func:`admit` that records a fire. Enforces the §4 contract: + per-type allowlist, relevance floor, per-session dedupe by + ``sha1(type:target_key:signal_bucket)``, a global per-session cap, a + cross-session cooldown, and dismiss-driven adaptive quieting — all backed by a + small JSON file at ``~/.stackunderflow/proactive_state.json`` (file-locked, + bounded, corrupt/missing → treated as empty). **Never** ``store.db`` — hooks + must not contend with the ingest writer on the hot path. + +* **Phase 0 (governance retrofit)** — :func:`admit_file_risk` wraps the existing + ``recall.py`` file-risk output so a chronically risky file no longer nags + every session with no throttle. + +* **Phase 1 (command-cluster nudge)** — :func:`command_cluster_block` extracts a + pending Bash command's normalised head (via ``patterns._normalise_command``, + reused verbatim for key parity), looks it up in a precomputed O(1) signal + cache (``~/.stackunderflow/proactive_signals.json``, refreshed on ingest by + :func:`refresh_signal_cache`), applies the relevance floor + governance, and + renders one deterministic advisory line. + +Invariants (this runs inside users' live sessions — non-negotiable): + +* **Opt-in, off by default.** ``proactive_enabled`` defaults false. When it is + off the module is inert (:func:`mode` → ``"passthrough"``): ``recall.py`` + keeps its shipped, ungoverned behavior and no state file is written. The env + kill-switch ``STACKUNDERFLOW_PROACTIVE_DISABLED=1`` (:func:`mode` → ``"off"``) + silences everything and wins over ``proactive_enabled``. +* **Never blocks, never raises, always exit 0.** Nothing here ever returns a + PreToolUse deny/ask decision — only advisory ``additionalContext`` text is + ever produced by the callers. Any error, missing/corrupt state, or lock + contention degrades to "silent" (empty / ``False``), never to spam. +* **Fast + local.** No LLM, no network, no ``store.db`` write on the hook path. + The command lookup is an O(1) dict read against the precomputed cache — never + a live ``mine_patterns`` scan. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import time +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import TYPE_CHECKING, Any, Iterator + +from stackunderflow.hooks.inject import _slug_from_cwd + +if TYPE_CHECKING: # pragma: no cover - import only for the type annotation + import sqlite3 + +logger = logging.getLogger("stackunderflow.hooks") + +# ── filenames / knobs ─────────────────────────────────────────────────────── + +# Governance state + the precomputed signal cache both live in the app dir +# (derived from ``deps.store_path`` so a test that relocates the store relocates +# these too). JSON files, never the DB. +_STATE_FILENAME = "proactive_state.json" +_SIGNAL_FILENAME = "proactive_signals.json" +_LOCK_SUFFIX = ".lock" + +# Hard env kill-switch — wins over ``proactive_enabled`` and every other knob. +_KILL_SWITCH_ENV = "STACKUNDERFLOW_PROACTIVE_DISABLED" + +# The nudge type ids this module understands (mirrors ``proactive_types``). +TYPE_COMMAND_CLUSTER = "command-cluster" +TYPE_FILE_RISK = "file-risk" +_KNOWN_TYPES = frozenset({TYPE_COMMAND_CLUSTER, TYPE_FILE_RISK}) + +# Relevance floor: a cluster's last failure must be at most this many days old +# for the nudge to be "in the moment". Mirrors ``patterns.DEFAULT_SINCE_DAYS`` +# (the mining window) — a soft guard so a *stale* cache can't nudge on ancient +# failures. Kept as a local constant to keep the hot path free of a +# ``patterns`` import for non-Bash fires. +_RECENT_DAYS = 90 + +# Bounds so neither JSON file can grow without limit (LRU-style eviction). +_MAX_SESSIONS = 256 +_MAX_COOLDOWNS = 1024 +_MAX_FEEDBACK = 1024 +_MAX_PROJECTS_CACHED = 128 +_MAX_CLUSTERS_PER_PROJECT = 200 +_MAX_FILE_RISK_PER_PROJECT = 200 + +# File-lock acquisition budget. A short spin, then a stale-lock breaker — a hook +# must never wedge on a leaked lock. +_LOCK_TIMEOUT_S = 1.0 +_LOCK_SPIN_S = 0.01 +_LOCK_STALE_S = 10.0 + +# Rendered command-cluster block is one line — capped defensively. +_CMD_MAX_CHARS = 600 + +_SIGNAL_CACHE_VERSION = 1 + + +# ── config snapshot / mode ────────────────────────────────────────────────── + + +def _kill_switch() -> bool: + """True when the hard env kill-switch is set (wins over everything).""" + return os.environ.get(_KILL_SWITCH_ENV, "").strip().lower() in ("1", "true", "yes", "on") + + +@dataclass(frozen=True) +class Policy: + """Resolved governance config for one decision — env > file > default.""" + + enabled: bool + kill_switch: bool + types: frozenset[str] + max_per_session: int + cooldown_hours: float + dismiss_suppress_after: int + + @property + def mode(self) -> str: + """``"off"`` (kill-switch) · ``"passthrough"`` (disabled) · ``"governed"``.""" + if self.kill_switch: + return "off" + return "governed" if self.enabled else "passthrough" + + @classmethod + def from_settings(cls) -> Policy: + import stackunderflow.deps as deps + + cfg = deps.config + return cls( + enabled=bool(cfg.get("proactive_enabled")), + kill_switch=_kill_switch(), + types=_parse_types(cfg.get("proactive_types")), + max_per_session=_as_int(cfg.get("proactive_max_per_session"), 3), + cooldown_hours=_as_float(cfg.get("proactive_cooldown_hours"), 24.0), + dismiss_suppress_after=_as_int(cfg.get("proactive_dismiss_suppress_after"), 3), + ) + + +def mode() -> str: + """Current surfacing mode without building a full :class:`Policy`. + + ``"off"`` — kill-switch set; silence every pre-tool nudge. + ``"passthrough"`` — proactive disabled (default); ``recall.py`` keeps its + shipped ungoverned behavior, no new nudge types, no state writes. + ``"governed"`` — opt-in on; governance + the command-cluster nudge are live. + """ + if _kill_switch(): + return "off" + try: + import stackunderflow.deps as deps + + return "governed" if bool(deps.config.get("proactive_enabled")) else "passthrough" + except Exception: # noqa: BLE001 - a config read must never break the hook + return "passthrough" + + +def _parse_types(raw: Any) -> frozenset[str]: + """Parse the ``proactive_types`` allowlist leniently into known type ids.""" + if not isinstance(raw, str): + return frozenset(_KNOWN_TYPES) + out = {t.strip().lower() for t in raw.split(",") if t.strip()} + return frozenset(out & _KNOWN_TYPES) + + +def _as_int(value: Any, default: int) -> int: + try: + if isinstance(value, bool): + return default + return int(value) + except (TypeError, ValueError): + return default + + +def _as_float(value: Any, default: float) -> float: + try: + if isinstance(value, bool): + return default + return float(value) + except (TypeError, ValueError): + return default + + +# ── the signal ────────────────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class Signal: + """One would-be nudge, reduced to what governance needs to decide. + + ``counts`` are the two salient integers whose *coarse bucket* forms the + ``signal_bucket`` half of the fingerprint, so a materially worse situation + (counts crossing into a higher bucket) re-arms an already-fired nudge. + ``eligible`` carries the type-specific relevance-floor result. + """ + + type: str + target_key: str + session_id: str + counts: tuple[int, int] + eligible: bool + + @property + def bucket(self) -> str: + return f"{_coarse(self.counts[0])}.{_coarse(self.counts[1])}" + + @property + def fingerprint(self) -> str: + raw = f"{self.type}:{self.target_key}:{self.bucket}" + return hashlib.sha1(raw.encode("utf-8", "replace")).hexdigest() # noqa: S324 - dedupe key, not a security digest + + +def make_signal( + sig_type: str, + target_key: str, + session_id: str | None, + counts: tuple[int, int], + *, + eligible: bool, +) -> Signal: + return Signal( + type=sig_type, + target_key=str(target_key), + session_id=session_id or "", + counts=(int(counts[0]), int(counts[1])), + eligible=bool(eligible), + ) + + +def _coarse(n: int) -> int: + """Monotonic coarse tier for a count — 0,1,{2-4},{5-9},{10-49},{50+}.""" + n = max(0, int(n)) + if n <= 1: + return n + if n <= 4: + return 2 + if n <= 9: + return 3 + if n <= 49: + return 4 + return 5 + + +# ── the gate (pure) ───────────────────────────────────────────────────────── + + +def should_surface( + signal: Signal, + state: dict, + *, + policy: Policy | None = None, + now: datetime | None = None, +) -> bool: + """Deterministic gate — may this nudge surface, given *state*? Pure, no I/O. + + An LLM decides nothing here (spec §4.7). Order is cheapest-reject-first: + mode → type allowlist → relevance floor → adaptive quieting → per-session + dedupe → cooldown → frequency cap. Any doubt resolves to ``False``. + """ + policy = policy or Policy.from_settings() + now = now or _utcnow() + + if policy.mode != "governed": + return False + if signal.type not in policy.types: + return False + if not signal.eligible: + return False + + feedback = state.get("feedback") if isinstance(state.get("feedback"), dict) else {} + threshold = policy.dismiss_suppress_after + if threshold > 0 and ( + _dismissed(feedback, signal.type) >= threshold + or _dismissed(feedback, signal.fingerprint) >= threshold + ): + return False # adaptive quieting — the user keeps dismissing this + + sessions = state.get("sessions") if isinstance(state.get("sessions"), dict) else {} + sess = sessions.get(signal.session_id) if isinstance(sessions.get(signal.session_id), dict) else {} + + fired = sess.get("fired") + if isinstance(fired, list) and signal.fingerprint in fired: + return False # per-session dedupe + + cooldowns = state.get("cooldowns") if isinstance(state.get("cooldowns"), dict) else {} + until = _parse_iso(cooldowns.get(signal.fingerprint)) + if until is not None and until > now: + return False # cross-session cooldown + + if _as_int(sess.get("count"), 0) >= policy.max_per_session: + return False # frequency cap + + return True + + +def _dismissed(feedback: dict, key: str) -> int: + entry = feedback.get(key) + if isinstance(entry, dict): + return _as_int(entry.get("dismissed"), 0) + return 0 + + +# ── the gate (stateful) ───────────────────────────────────────────────────── + + +def admit(signal: Signal, *, now: datetime | None = None, policy: Policy | None = None) -> bool: + """Try to surface *signal*: check :func:`should_surface`, and on success + record the fire (dedupe set, count, cooldown, shown counter) to disk. + + Returns True only when the nudge should be shown *and* the fire was + recorded. Never raises; lock contention / a bad state file → ``False`` + (silent), never a duplicate or a crash. + """ + policy = policy or Policy.from_settings() + now = now or _utcnow() + if policy.mode != "governed": + return False + if signal.type not in policy.types or not signal.eligible: + return False + try: + with _locked(_state_path()) as locked: + if not locked: + return False # contended — fail to silence, never double-fire + state = _read_state() + if state is None: + return False # corrupt state → fail to silence, never spam / raise + if not should_surface(signal, state, policy=policy, now=now): + return False + _record_fire(state, signal, policy, now) + _write_json(_state_path(), state) + return True + except Exception: # noqa: BLE001 - governance must never disrupt the agent + logger.debug("proactive.admit swallowed an error", exc_info=True) + return False + + +def admit_file_risk(recalls: list[dict], payload: dict, *, now: datetime | None = None) -> bool: + """Phase 0: govern the shipped ``recall.py`` file-risk finding. + + Fingerprinted on the primary (highest-risk) path with a bucket over + ``failed``/``reverted``. Called by ``recall.py`` only in governed mode; in + passthrough mode recall never routes here and keeps its shipped behavior. + """ + if not recalls: + return False + primary = recalls[0] + target = primary.get("path") or "" + failed = sum(_as_int(r.get("failed"), 0) for r in recalls) + reverted = sum(_as_int(r.get("reverted"), 0) for r in recalls) + signal = make_signal( + TYPE_FILE_RISK, + target, + _session_id(payload), + (failed, reverted), + eligible=(failed + reverted) >= 1, + ) + return admit(signal, now=now) + + +def _record_fire(state: dict, signal: Signal, policy: Policy, now: datetime) -> None: + """Mutate *state* to reflect that *signal* just fired (in-place).""" + sessions = state.setdefault("sessions", {}) + if not isinstance(sessions, dict): + sessions = state["sessions"] = {} + sess = sessions.get(signal.session_id) + if not isinstance(sess, dict): + sess = sessions[signal.session_id] = {"fired": [], "count": 0} + fired = sess.setdefault("fired", []) + if not isinstance(fired, list): + fired = sess["fired"] = [] + if signal.fingerprint not in fired: + fired.append(signal.fingerprint) + sess["count"] = _as_int(sess.get("count"), 0) + 1 + sess["ts"] = now.isoformat() + + if policy.cooldown_hours > 0: + cooldowns = state.setdefault("cooldowns", {}) + if isinstance(cooldowns, dict): + cooldowns[signal.fingerprint] = (now + timedelta(hours=policy.cooldown_hours)).isoformat() + + feedback = state.setdefault("feedback", {}) + if isinstance(feedback, dict): + _bump(feedback, signal.type, "shown") + _bump(feedback, signal.fingerprint, "shown") + + _prune_state(state, now) + + +def record_dismissal(key: str, *, now: datetime | None = None) -> None: + """Register a dashboard 'don't show this again' for a type or a fingerprint. + + The Tier-2 dismiss primitive (the retrospective panel calls this; not wired + to a route in the MVP). Increments the ``dismissed`` counter that + :func:`should_surface` reads for adaptive quieting. Never raises. + """ + now = now or _utcnow() + try: + with _locked(_state_path()) as locked: + if not locked: + return + state = _read_state() + if state is None: + state = {} # dashboard side — a corrupt file is safe to reset here + feedback = state.setdefault("feedback", {}) + if isinstance(feedback, dict): + _bump(feedback, str(key), "dismissed") + _prune_state(state, now) + _write_json(_state_path(), state) + except Exception: # noqa: BLE001 + logger.debug("proactive.record_dismissal swallowed an error", exc_info=True) + + +def _bump(feedback: dict, key: str, field_name: str) -> None: + entry = feedback.get(key) + if not isinstance(entry, dict): + entry = feedback[key] = {"shown": 0, "dismissed": 0} + entry[field_name] = _as_int(entry.get(field_name), 0) + 1 + + +# ── the command-cluster nudge (Phase 1) ───────────────────────────────────── + + +def command_cluster_block(payload: dict, *, now: datetime | None = None) -> str: + """Advisory line for a pending Bash command in a known failure cluster, or ``""``. + + O(1): normalise the command head and look it up in the precomputed cache; + apply the relevance floor (``failure_count ≥ 2`` and ``session_count ≥ 2`` + and recent) and governance. Never runs a live ``mine_patterns`` scan, never + raises. + """ + try: + if not isinstance(payload, dict) or payload.get("tool_name") != "Bash": + return "" + tool_input = payload.get("tool_input") + if not isinstance(tool_input, dict): + return "" + command = tool_input.get("command") + if not isinstance(command, str) or not command.strip(): + return "" + slug = _slug_from_cwd(payload.get("cwd")) + if not slug: + return "" + + from stackunderflow.reports.patterns import _normalise_command + + key = _normalise_command(command) # VERBATIM reuse — cluster-key parity + cluster = _lookup_cluster(slug, key) + if cluster is None: + return "" + + now = now or _utcnow() + failure_count = _as_int(cluster.get("failure_count"), 0) + session_count = _as_int(cluster.get("session_count"), 0) + eligible = ( + failure_count >= 2 + and session_count >= 2 + and _is_recent(cluster.get("last_failure_ts"), now) + ) + signal = make_signal( + TYPE_COMMAND_CLUSTER, key, _session_id(payload), (failure_count, session_count), eligible=eligible + ) + if not admit(signal, now=now): + return "" + return _render_command_cluster(cluster, key) + except Exception: # noqa: BLE001 - a nudge must never disrupt the agent + logger.debug("proactive.command_cluster_block swallowed an error", exc_info=True) + return "" + + +def _render_command_cluster(cluster: dict, key: str) -> str: + """One deterministic advisory line for a command-cluster nudge.""" + command = cluster.get("command") if isinstance(cluster.get("command"), str) else key + command = command or key + session_count = _as_int(cluster.get("session_count"), 0) + sess_word = "session" if session_count == 1 else "sessions" + text = ( + f"[StackUnderflow memory] Heads-up before this Bash call: `{command}` has failed in " + f"{session_count} recent {sess_word} in this project" + ) + top = _top_category(cluster.get("categories")) + if top: + text += f" — mostly {top}" + text += "." + date = cluster.get("last_failure_ts") + if isinstance(date, str) and date: + text += f" Last failure {date[:10]}." + if len(text) > _CMD_MAX_CHARS: + text = text[: max(1, _CMD_MAX_CHARS - 1)].rstrip() + "…" + return text + + +def _top_category(categories: Any) -> str | None: + if not isinstance(categories, dict) or not categories: + return None + try: + return max(categories.items(), key=lambda kv: (_as_int(kv[1], 0), str(kv[0])))[0] + except (TypeError, ValueError): + return None + + +def _lookup_cluster(slug: str, key: str) -> dict | None: + """O(1) read of one cluster from the precomputed cache; ``None`` if absent.""" + cache = _read_json(_signal_path()) + projects = cache.get("projects") + if not isinstance(projects, dict): + return None + entry = projects.get(slug) + if not isinstance(entry, dict): + return None + clusters = entry.get("command_clusters") + if not isinstance(clusters, dict): + return None + cluster = clusters.get(key) + return cluster if isinstance(cluster, dict) else None + + +# ── signal cache precompute (ingest side) ─────────────────────────────────── + + +def refresh_signal_cache(conn: "sqlite3.Connection", slugs: set[str] | list[str]) -> None: + """Recompute + persist the command/file signal cache for the given slugs. + + Called additively from the ingest/reindex path. **Self-gates on + ``proactive_enabled``** so the default (opt-out) path pays nothing — no + ``mine_patterns`` scan is run unless a user turned the feature on. Fenced: + any failure is swallowed so a cache hiccup can never break ingest. + """ + policy = Policy.from_settings() + if policy.mode != "governed": + return # opt-in only — no precompute cost for users who never enabled it + slug_list = [s for s in dict.fromkeys(slugs) if s] + if not slug_list: + return + try: + from stackunderflow.reports import patterns + from stackunderflow.store import queries + + now_iso = _utcnow().isoformat() + with _locked(_signal_path()) as locked: + if not locked: + return + cache = _read_json(_signal_path()) + projects = cache.get("projects") + if not isinstance(projects, dict): + projects = {} + for slug in slug_list: + ids = [row.id for row in queries.get_projects_by_slug(conn, slug=slug)] + if not ids: + continue + report = patterns.mine_patterns(conn, project_ids=ids) + projects[slug] = { + "generated_at": now_iso, + "command_clusters": _clusters_map(report.get("command_clusters")), + "file_risk": _file_risk_map(report.get("file_risk")), + } + cache = { + "version": _SIGNAL_CACHE_VERSION, + "generated_at": now_iso, + "projects": _cap_projects(projects), + } + _write_json(_signal_path(), cache) + except Exception: # noqa: BLE001 - a signal-cache refresh must never break ingest + logger.debug("proactive.refresh_signal_cache swallowed an error", exc_info=True) + + +def _clusters_map(clusters: Any) -> dict: + """``command_clusters`` list → ``{normalised_key: trimmed cluster}`` (O(1) lookup).""" + out: dict[str, dict] = {} + if not isinstance(clusters, list): + return out + for c in clusters[:_MAX_CLUSTERS_PER_PROJECT]: + if not isinstance(c, dict): + continue + key = c.get("command") + if not isinstance(key, str) or not key: + continue + cats = c.get("categories") + out[key] = { + "command": key, + "failure_count": _as_int(c.get("failure_count"), 0), + "session_count": _as_int(c.get("session_count"), 0), + "categories": cats if isinstance(cats, dict) else {}, + "last_failure_ts": c.get("last_failure_ts") if isinstance(c.get("last_failure_ts"), str) else None, + } + return out + + +def _file_risk_map(file_risk: Any) -> dict: + """``file_risk`` list → ``{path: trimmed risk}``. Cached for Tier-2 / Phase 2; + the MVP hook path does not read it (file-risk stays on the recall CLI path).""" + out: dict[str, dict] = {} + if not isinstance(file_risk, list): + return out + for f in file_risk[:_MAX_FILE_RISK_PER_PROJECT]: + if not isinstance(f, dict): + continue + path = f.get("path") + if not isinstance(path, str) or not path: + continue + out[path] = { + "failure_count": _as_int(f.get("failure_count"), 0), + "failure_session_count": _as_int(f.get("failure_session_count"), 0), + "last_failure_ts": f.get("last_failure_ts") if isinstance(f.get("last_failure_ts"), str) else None, + } + return out + + +def _cap_projects(projects: dict) -> dict: + """Bound the cache to the most-recently-generated projects.""" + if len(projects) <= _MAX_PROJECTS_CACHED: + return projects + ordered = sorted( + projects.items(), key=lambda kv: str(kv[1].get("generated_at", "")), reverse=True + ) + return dict(ordered[:_MAX_PROJECTS_CACHED]) + + +# ── state pruning (bounded LRU) ───────────────────────────────────────────── + + +def _prune_state(state: dict, now: datetime) -> None: + """Keep the state file bounded — evict old sessions, expired cooldowns.""" + sessions = state.get("sessions") + if isinstance(sessions, dict) and len(sessions) > _MAX_SESSIONS: + ordered = sorted(sessions.items(), key=lambda kv: str(kv[1].get("ts", "")), reverse=True) + state["sessions"] = dict(ordered[:_MAX_SESSIONS]) + + cooldowns = state.get("cooldowns") + if isinstance(cooldowns, dict): + live = { + fp: ts + for fp, ts in cooldowns.items() + if (parsed := _parse_iso(ts)) is not None and parsed > now + } + if len(live) > _MAX_COOLDOWNS: + ordered = sorted(live.items(), key=lambda kv: str(kv[1]), reverse=True) + live = dict(ordered[:_MAX_COOLDOWNS]) + state["cooldowns"] = live + + feedback = state.get("feedback") + if isinstance(feedback, dict) and len(feedback) > _MAX_FEEDBACK: + ordered = sorted( + feedback.items(), + key=lambda kv: (_as_int(kv[1].get("dismissed"), 0), _as_int(kv[1].get("shown"), 0)) + if isinstance(kv[1], dict) + else (0, 0), + reverse=True, + ) + state["feedback"] = dict(ordered[:_MAX_FEEDBACK]) + + +# ── paths / JSON I/O / file lock ──────────────────────────────────────────── + + +def _app_dir() -> Path: + """The app dir — derived from ``deps.store_path`` so tests relocate cleanly.""" + import stackunderflow.deps as deps + + return deps.store_path.parent + + +def _state_path() -> Path: + return _app_dir() / _STATE_FILENAME + + +def _signal_path() -> Path: + return _app_dir() / _SIGNAL_FILENAME + + +def _read_json(path: Path) -> dict: + """Load a JSON dict; missing / corrupt / non-dict → ``{}`` (fail to empty).""" + try: + if not path.exists(): + return {} + data = json.loads(path.read_text()) + except (OSError, ValueError, TypeError): + return {} + return data if isinstance(data, dict) else {} + + +def _read_state() -> dict | None: + """Governance state, distinguishing *missing* from *corrupt*. + + * **Missing** file → ``{}`` — the normal first-fire condition; an empty + state suppresses nothing, so the nudge proceeds under the usual rules. + * **Corrupt** / non-dict → ``None`` — an error the hot path resolves by + failing to silence (never spam off unreadable throttle state), never a + raise. (Writes go through ``os.replace``, so corruption is not expected + in practice.) + """ + path = _state_path() + try: + if not path.exists(): + return {} + data = json.loads(path.read_text()) + except (OSError, ValueError, TypeError): + return None + return data if isinstance(data, dict) else None + + +def _write_json(path: Path, data: dict) -> None: + """Atomically persist *data* (temp file + ``os.replace``). Best-effort.""" + try: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + f".tmp-{os.getpid()}") + tmp.write_text(json.dumps(data)) + os.replace(tmp, path) + except OSError: + logger.debug("proactive: could not write %s", path, exc_info=True) + + +@contextmanager +def _locked(target: Path) -> Iterator[bool]: + """Best-effort cross-process advisory lock for a read-modify-write on *target*. + + Uses an ``O_CREAT|O_EXCL`` sibling lock file — portable across platforms + (no ``fcntl``). Yields True when acquired, False on timeout (the caller + then bails to silence). A lock older than ``_LOCK_STALE_S`` is treated as + leaked and stolen, so a crashed hook can never wedge the feature. + """ + lock_path = target.with_suffix(target.suffix + _LOCK_SUFFIX) + try: + lock_path.parent.mkdir(parents=True, exist_ok=True) + except OSError: + yield False + return + deadline = time.monotonic() + _LOCK_TIMEOUT_S + fd: int | None = None + while True: + try: + fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + break + except FileExistsError: + try: + if time.time() - os.path.getmtime(lock_path) > _LOCK_STALE_S: + os.unlink(lock_path) + continue + except OSError: + pass + if time.monotonic() >= deadline: + yield False + return + time.sleep(_LOCK_SPIN_S) + except OSError: + yield False + return + try: + yield True + finally: + if fd is not None: + try: + os.close(fd) + except OSError: + pass + try: + os.unlink(lock_path) + except OSError: + pass + + +# ── small utils ───────────────────────────────────────────────────────────── + + +def _session_id(payload: dict) -> str: + sid = payload.get("session_id") if isinstance(payload, dict) else None + return sid if isinstance(sid, str) and sid else "" + + +def _utcnow() -> datetime: + return datetime.now(UTC) + + +def _parse_iso(value: Any) -> datetime | None: + if not isinstance(value, str) or not value: + return None + try: + dt = datetime.fromisoformat(value.replace("Z", "+00:00")) + except (ValueError, TypeError): + return None + return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC) + + +def _is_recent(ts: Any, now: datetime) -> bool: + """True when *ts* is a parseable timestamp within ``_RECENT_DAYS`` of *now*. + + A missing / unparseable timestamp is *not* recent — a nudge without a + dateable last failure stays silent (conservative).""" + parsed = _parse_iso(ts) + if parsed is None: + return False + return (now - parsed) <= timedelta(days=_RECENT_DAYS) diff --git a/stackunderflow/hooks/recall.py b/stackunderflow/hooks/recall.py index 4ba578f1..605760bb 100644 --- a/stackunderflow/hooks/recall.py +++ b/stackunderflow/hooks/recall.py @@ -52,7 +52,7 @@ import time from typing import Any -from stackunderflow.hooks import templates +from stackunderflow.hooks import proactive, templates logger = logging.getLogger("stackunderflow.hooks") @@ -97,6 +97,15 @@ def build_recall(hook_id: str, payload: dict | None) -> str: Never raises. Any failure — unknown id, bad payload, missing CLI, timeout, garbage output — returns ``""`` so the caller emits nothing and exits 0. An empty return is also the normal "file is clean" outcome. + + Governance (spec 27 / #97) rides on top without changing the default: + + * ``proactive`` disabled (the default) → **passthrough**: the shipped + file-risk warning is emitted exactly as before, ungoverned. + * kill-switch set → **off**: every pre-tool nudge is silenced. + * ``proactive_enabled`` → **governed**: the file-risk warning passes through + the dedupe / cap / cooldown layer (Phase 0), and a command-cluster nudge + (Phase 1) may be appended on the Bash path. """ try: payload = payload if isinstance(payload, dict) else {} @@ -104,35 +113,64 @@ def build_recall(hook_id: str, payload: dict | None) -> str: if event is None or hook_id not in templates.RECALL_HOOK_IDS: return "" - paths = _candidate_paths(payload) - if not paths: - return "" + pmode = proactive.mode() + if pmode == "off": + return "" # env kill-switch — silence every pre-tool nudge + + blocks: list[str] = [] - cwd = payload.get("cwd") - cwd = cwd if isinstance(cwd, str) and os.path.isdir(cwd) else None - - deadline = time.monotonic() + _timeout_seconds() - recalls: list[dict] = [] - for path in paths: - remaining = deadline - time.monotonic() - if remaining <= 0.05: - break # deadline spent — never stretch it for more paths - envelope = _query_memory_file(path, timeout=remaining, cwd=cwd) - if envelope is None: - continue - recall = _extract_recall(envelope, path) - if recall is not None: - recalls.append(recall) - - text = _render(recalls) - if not text.strip(): + # ── file-risk (shipped in #5; #97 only retrofits governance) ────────── + recalls = _collect_recalls(payload) + file_text = _render(recalls) + if file_text.strip(): + if pmode != "governed" or proactive.admit_file_risk(recalls, payload): + blocks.append(file_text) + + # ── command-cluster nudge (Phase 1 — governed mode, Bash path only) ─── + if pmode == "governed": + cmd_text = proactive.command_cluster_block(payload) + if cmd_text.strip(): + blocks.append(cmd_text) + + if not blocks: return "" + text = "\n\n".join(blocks) return json.dumps({"hookSpecificOutput": {"hookEventName": event, "additionalContext": text}}) except Exception: # noqa: BLE001 - a recall hook must never disrupt the agent logger.debug("recall hook %s swallowed an error", hook_id, exc_info=True) return "" +def _collect_recalls(payload: dict) -> list[dict]: + """Run the file-risk lookups for a fire and return the risk findings. + + The path-extraction + shared-deadline CLI loop, factored out of + :func:`build_recall` so the findings can be handed to the governance layer + before rendering. Empty list when there is nothing to look up (no + extractable path) or nothing risky came back. + """ + paths = _candidate_paths(payload) + if not paths: + return [] + + cwd = payload.get("cwd") + cwd = cwd if isinstance(cwd, str) and os.path.isdir(cwd) else None + + deadline = time.monotonic() + _timeout_seconds() + recalls: list[dict] = [] + for path in paths: + remaining = deadline - time.monotonic() + if remaining <= 0.05: + break # deadline spent — never stretch it for more paths + envelope = _query_memory_file(path, timeout=remaining, cwd=cwd) + if envelope is None: + continue + recall = _extract_recall(envelope, path) + if recall is not None: + recalls.append(recall) + return recalls + + # ── payload → candidate paths ─────────────────────────────────────────────── diff --git a/stackunderflow/ingest/__init__.py b/stackunderflow/ingest/__init__.py index a2a3cd3e..52d6de79 100644 --- a/stackunderflow/ingest/__init__.py +++ b/stackunderflow/ingest/__init__.py @@ -155,6 +155,18 @@ def auto_reindex_touched( finally: deps.is_reindexing = prior_flag + # Proactive signal cache (spec 27 / #97): precompute the O(1) command / + # file signal snapshot that the pre-tool hook reads, so the hook never runs + # a live pattern scan. Self-gates on ``proactive_enabled`` (zero cost when + # the feature is off) and swallows its own errors — additive, never blocks + # ingest. + try: + from stackunderflow.hooks import proactive + + proactive.refresh_signal_cache(conn, slug_list) + except Exception as e: # noqa: BLE001 — signal precompute must never break ingest + _logger.debug("proactive signal-cache refresh skipped: %s", e) + def _lookup(adapters: list[SourceAdapter], name: str) -> SourceAdapter: for a in adapters: diff --git a/stackunderflow/reports/benchmark.py b/stackunderflow/reports/benchmark.py new file mode 100644 index 00000000..28d0938a --- /dev/null +++ b/stackunderflow/reports/benchmark.py @@ -0,0 +1,1033 @@ +"""Comparative benchmark engine — "which model wins for your work?" (issue #99). + +An **observational** benchmark over the user's own history: a natural +experiment they already ran, not live re-execution. Spec 26 makes the central +design call that this — not replay — is the credible, local-first, zero-cost, +always-available core. This module is the join + statistics layer over data +every other surface already computes per session; the hard part is not the +join, it is refusing to name a winner when the evidence can't support one. + +Design contract (mirrors :mod:`stackunderflow.reports.forks` / +:mod:`stackunderflow.reports.anomaly`): + +* **Advisory, never raises.** A schemaless store, an empty ``session_mart``, a + single-model store, or any arithmetic edge returns an empty-but-well-formed + verdict (``"insufficient evidence"``). Callers never wrap this for + correctness. +* **Read-only + scope-bounded.** All SQL is guarded by ``sqlite_master`` and + narrowed by the caller's :class:`Scope` + optional ``project_ids``. +* **Cost is a black box.** ``cost_usd`` is read **only** from ``session_mart`` + and never recomputed — keeps ``test_pricing_invariants`` green. +* **Honesty first (§4).** Sample floors, Wilson / bootstrap CIs, difference + effects, BH-FDR across the family, direct standardization (never pooled), and + "insufficient evidence" as a first-class verdict with a ``confidence`` label. + +The rubric weights + success threshold τ are **maintainer-owned** (ratified in +``docs/specs/benchmark-rubric-v1.md``); they are surfaced in the payload and +overridable via the ``weights`` argument — never silently hard-coded. +""" + +from __future__ import annotations + +import json +import sqlite3 +import statistics +from dataclasses import dataclass, field +from typing import Any + +from stackunderflow.reports.scope import Scope +from stackunderflow.services import benchmark_stats as bs +from stackunderflow.services import task_classifier + +__all__ = [ + "analyze_benchmark", + "RUBRIC_VERSION", + "DEFAULT_WEIGHTS", + "SUCCESS_THRESHOLD", + "recommend_from_history", +] + + +# ── rubric v1 (maintainer-owned; ratified in benchmark-rubric-v1.md) ───────── + +RUBRIC_VERSION = 1 +# Composite weights — surfaced in the payload and overridable. Sum to 1.0. +DEFAULT_WEIGHTS: dict[str, float] = {"success": 0.45, "cost": 0.35, "effort": 0.20} +# τ — grade-tier success threshold (a real LLM grade ≥ τ counts as success). +SUCCESS_THRESHOLD = 7.0 + +# Behavioural (Tier-4) proxy: a session with this many assistant turns is read +# as high-retry (a soft failure signal). One-shot sessions are a soft success. +_HIGH_RETRY_TURNS = 8 + +# The natural-experiment caveat, stated verbatim in every payload (§4.7). +NATURAL_EXPERIMENT_WARNING = ( + "This compares models over sessions you already ran — a natural " + "experiment, not a controlled trial. Models were not randomly assigned to " + "tasks, so the engine stratifies by task type and size and standardizes " + "across strata to control for the confounder it can measure (task " + "difficulty). It cannot control for the ones it can't (your skill drift " + "over time, per-project difficulty, prompt-quality differences)." +) + +_METHOD_NOTES: tuple[str, ...] = ( + "Observed history is a natural experiment, not a randomized trial.", + "Models are compared only within a stratum of comparable tasks (intent × " + "size); cross-task figures use direct standardization, never a pooled mean.", + "Success is composed from the highest-confidence signal available per " + "session (PR/CI → code-delta → LLM grade → behavioral); sessions with no " + "signal are excluded from rates but counted in coverage.", + "Tier-1 commit attribution is a coarse 24h + cwd heuristic — a signal, not " + "gospel.", + "Reasoning efficiency is descriptive only and is never scored into the " + "winner (providers that report 0 reasoning tokens aren't apples-to-apples).", + "A win must clear a practical effect floor and survive Benjamini–Hochberg " + "FDR control; below the sample floor the verdict is 'insufficient evidence'.", +) + + +# ── per-session fact ───────────────────────────────────────────────────────── + + +@dataclass(slots=True) +class _SessionFact: + session_id: str + project_id: int + primary_model: str + intent: str + size_band: str + language: str | None + cost_usd: float + num_turns: int + is_one_shot: bool + output_tokens: int + reasoning_tokens: int + first_ts: str + outcome_success: int | None # 1 / 0 / None (unmeasured) + outcome_tier: str | None # which tier decided it + + +# ── table guard ────────────────────────────────────────────────────────────── + + +def _table_exists(conn: sqlite3.Connection, name: str) -> bool: + """True when *name* is a queryable table or view (``messages`` is a view).""" + try: + row = conn.execute( + "SELECT 1 FROM sqlite_master " + "WHERE type IN ('table', 'view') AND name = ? LIMIT 1", + (name,), + ).fetchone() + except sqlite3.Error: + return False + return row is not None + + +# ── success-signal composition (tiered) ────────────────────────────────────── + + +def _outcome_from_ground_truth(outcomes: dict[str, Any]) -> int | None: + """Tier 1 — PR/CI ground truth. ``1`` merged&clean / CI pass, ``0`` reverted + / CI fail, ``None`` when neither is present.""" + prs = outcomes.get("prs") or [] + ci = outcomes.get("ci_runs") or [] + reverted = any(p.get("reverted_at") for p in prs) + merged_ok = any( + (p.get("state") == "merged") and not p.get("reverted_at") for p in prs + ) + ci_pass = any(c.get("status") == "success" for c in ci) + ci_fail = any(c.get("status") == "failure" for c in ci) + if reverted: + return 0 + if merged_ok or ci_pass: + return 1 + if ci_fail: + return 0 + return None + + +def _outcome_from_static(metric_summary: dict[str, Any]) -> int | None: + """Tier 2 — net code-delta. ``1`` net-improved, ``0`` net-regressed, + ``None`` when the session was analyzed but shows no net direction.""" + improved = sum(int(m.get("improved", 0) or 0) for m in metric_summary.values()) + regressed = sum(int(m.get("regressed", 0) or 0) for m in metric_summary.values()) + if improved > regressed: + return 1 + if regressed > improved: + return 0 + return None + + +def _outcome_from_grade(grade_success: float | None) -> int | None: + """Tier 3 — real LLM grade. ``1`` if ``grades.success ≥ τ`` else ``0``.""" + if grade_success is None: + return None + return 1 if grade_success >= SUCCESS_THRESHOLD else 0 + + +def _outcome_from_behavior(is_one_shot: bool, num_turns: int) -> int | None: + """Tier 4 — behavioral proxy. One-shot → ``1``; high-retry → ``0``; else + ``None`` (no confident behavioral read).""" + if is_one_shot: + return 1 + if num_turns >= _HIGH_RETRY_TURNS: + return 0 + return None + + +# ── data loading ───────────────────────────────────────────────────────────── + + +def _load_facts( + conn: sqlite3.Connection, + *, + scope: Scope | None, + project_ids: list[int] | None, +) -> list[_SessionFact]: + """Load one :class:`_SessionFact` per scoped session, or ``[]``. + + Cost/model/tokens come from ``session_mart`` (cost is never recomputed). + Intent + text-language are derived from the first user turn via the + canonical ``task_classifier``; size band from the session's real token + volume. Success is composed from the tiered signals in bulk-loaded side + tables. Every table touch is guarded so a partial store degrades to fewer + tiers rather than raising. + """ + if not _table_exists(conn, "session_mart") or not _table_exists(conn, "sessions"): + return [] + if project_ids is not None and len(project_ids) == 0: + return [] + + sql = ( + "SELECT sm.session_id AS session_id, sm.project_id AS project_id, " + " sm.primary_model AS primary_model, " + " COALESCE(sm.cost_usd, 0.0) AS cost_usd, sm.first_ts AS first_ts, " + " COALESCE(sm.input_tokens, 0) AS input_tokens, " + " COALESCE(sm.output_tokens, 0) AS output_tokens, " + " COALESCE(sm.assistant_message_count, 0) AS assistant_message_count, " + " COALESCE(sm.is_one_shot, 0) AS is_one_shot, " + " (SELECT m.content_text FROM messages m " + " WHERE m.session_fk = s.id AND m.role = 'user' " + " ORDER BY m.seq ASC LIMIT 1) AS first_user_text " + "FROM session_mart sm " + "JOIN sessions s ON s.session_id = sm.session_id " + "WHERE sm.primary_model IS NOT NULL AND sm.primary_model != '' " + ) + params: list[Any] = [] + if project_ids: + placeholders = ",".join("?" for _ in project_ids) + sql += f"AND sm.project_id IN ({placeholders}) " + params.extend(project_ids) + if scope is not None and scope.since is not None: + sql += "AND sm.first_ts >= ? " + params.append(scope.since) + if scope is not None and scope.until is not None: + sql += "AND sm.first_ts <= ? " + params.append(scope.until) + + try: + rows = conn.execute(sql, params).fetchall() + except sqlite3.Error: + return [] + if not rows: + return [] + + grades = _load_grades(conn) + static_lang, static_outcome = _load_static(conn, {r["session_id"] for r in rows}) + ground_truth = _load_ground_truth(conn, {r["session_id"] for r in rows}) + reasoning = _load_reasoning(conn, scope=scope, project_ids=project_ids) + + facts: list[_SessionFact] = [] + for r in rows: + sid = str(r["session_id"]) + text = str(r["first_user_text"] or "") + intent = task_classifier.classify_intent(text) + size_band = task_classifier.band_for_token_count( + int(r["input_tokens"] or 0) + int(r["output_tokens"] or 0) + ) + language = static_lang.get(sid) or task_classifier.dominant_language(text) + turns = int(r["assistant_message_count"] or 0) + one_shot = bool(r["is_one_shot"]) + + success, tier = _compose_success( + sid, + ground_truth=ground_truth, + static_outcome=static_outcome, + grade_success=grades.get(sid), + is_one_shot=one_shot, + num_turns=turns, + ) + rt, ot = reasoning.get(sid, (0, int(r["output_tokens"] or 0))) + facts.append( + _SessionFact( + session_id=sid, + project_id=int(r["project_id"] or 0), + primary_model=str(r["primary_model"]), + intent=intent, + size_band=size_band, + language=language, + cost_usd=float(r["cost_usd"] or 0.0), + num_turns=turns, + is_one_shot=one_shot, + output_tokens=int(ot or 0), + reasoning_tokens=int(rt or 0), + first_ts=str(r["first_ts"] or ""), + outcome_success=success, + outcome_tier=tier, + ) + ) + return facts + + +def _compose_success( + session_id: str, + *, + ground_truth: dict[str, dict[str, Any]], + static_outcome: dict[str, int | None], + grade_success: float | None, + is_one_shot: bool, + num_turns: int, +) -> tuple[int | None, str | None]: + """Walk the four tiers in precedence order; the first non-None wins.""" + gt = ground_truth.get(session_id) + if gt is not None: + val = _outcome_from_ground_truth(gt) + if val is not None: + return val, "ground_truth" + st = static_outcome.get(session_id) + if st is not None: + return st, "code_delta" + gr = _outcome_from_grade(grade_success) + if gr is not None: + return gr, "llm_grade" + bh = _outcome_from_behavior(is_one_shot, num_turns) + if bh is not None: + return bh, "behavioral" + return None, None + + +def _load_grades(conn: sqlite3.Connection) -> dict[str, float]: + """``session_id → grades.success`` from ``session_quality_metrics`` (real + grades only ever persist there).""" + if not _table_exists(conn, "session_quality_metrics"): + return {} + try: + rows = conn.execute( + "SELECT session_id, grades_json FROM session_quality_metrics" + ).fetchall() + except sqlite3.Error: + return {} + out: dict[str, float] = {} + for r in rows: + try: + grades = json.loads(r["grades_json"]) + if isinstance(grades, dict) and "success" in grades: + out[str(r["session_id"])] = float(grades["success"]) + except (TypeError, ValueError): + continue + return out + + +def _load_static( + conn: sqlite3.Connection, session_ids: set[str] +) -> tuple[dict[str, str], dict[str, int | None]]: + """Return ``(dominant_language, net_outcome)`` from ``static_analysis_findings``. + + Reuses the canonical ``get_session_quality`` reader per session that has + findings (bounded — most sessions have none), so the improved/regressed + polarity matches the rest of the product exactly. + """ + if not _table_exists(conn, "static_analysis_findings") or not session_ids: + return {}, {} + try: + rows = conn.execute( + "SELECT DISTINCT session_id FROM static_analysis_findings" + ).fetchall() + except sqlite3.Error: + return {}, {} + analyzed = {str(r["session_id"]) for r in rows} & session_ids + if not analyzed: + return {}, {} + + from stackunderflow.services import static_analysis + + langs: dict[str, str] = {} + outcomes: dict[str, int | None] = {} + for sid in analyzed: + try: + quality = static_analysis.get_session_quality(conn, sid) + except Exception: # noqa: BLE001,S112 — advisory: skip a bad session + continue + summary = quality.summary or {} + languages = summary.get("languages") or [] + if languages: + langs[sid] = str(languages[0]) + outcomes[sid] = _outcome_from_static(summary.get("metrics") or {}) + return langs, outcomes + + +def _load_ground_truth( + conn: sqlite3.Connection, session_ids: set[str] +) -> dict[str, dict[str, Any]]: + """Return ``session_id → outcomes`` for sessions with a commit link. + + Bounded to commit-linked sessions (most stores have few), reusing the + canonical ``outcome_attribution.get_outcomes_for_session`` so PR-matching + stays in one place. + """ + if not _table_exists(conn, "commit_session_link") or not session_ids: + return {} + try: + rows = conn.execute( + "SELECT DISTINCT session_id FROM commit_session_link" + ).fetchall() + except sqlite3.Error: + return {} + linked = {str(r["session_id"]) for r in rows} & session_ids + if not linked: + return {} + + from stackunderflow.services import outcome_attribution + + out: dict[str, dict[str, Any]] = {} + for sid in linked: + try: + out[sid] = outcome_attribution.get_outcomes_for_session(conn, sid) + except Exception: # noqa: BLE001,S112 — advisory: skip a bad session + continue + return out + + +def _load_reasoning( + conn: sqlite3.Connection, + *, + scope: Scope | None, + project_ids: list[int] | None, +) -> dict[str, tuple[int, int]]: + """``session_id → (reasoning_tokens, output_tokens)`` from ``usage_events``. + + Reasoning is descriptive-only (0 for providers with no wire count); this is + surfaced, never scored. ``reasoning_tokens`` stays a subset of output. + """ + if not _table_exists(conn, "usage_events"): + return {} + sql = ( + "SELECT session_id, " + " COALESCE(SUM(reasoning_tokens), 0) AS rt, " + " COALESCE(SUM(output_tokens), 0) AS ot " + "FROM usage_events WHERE 1=1 " + ) + params: list[Any] = [] + if project_ids: + placeholders = ",".join("?" for _ in project_ids) + sql += f"AND project_id IN ({placeholders}) " + params.extend(project_ids) + if scope is not None and scope.since is not None: + sql += "AND ts >= ? " + params.append(scope.since) + if scope is not None and scope.until is not None: + sql += "AND ts <= ? " + params.append(scope.until) + sql += "GROUP BY session_id " + try: + rows = conn.execute(sql, params).fetchall() + except sqlite3.Error: + return {} + return {str(r["session_id"]): (int(r["rt"] or 0), int(r["ot"] or 0)) for r in rows} + + +# ── per-model per-cell statistics ──────────────────────────────────────────── + + +@dataclass(slots=True) +class _ModelCell: + model: str + facts: list[_SessionFact] = field(default_factory=list) + + @property + def n(self) -> int: + return len(self.facts) + + @property + def qualified(self) -> bool: + return self.n >= bs.MIN_SESSIONS_PER_CELL + + def measured(self) -> list[_SessionFact]: + return [f for f in self.facts if f.outcome_success is not None] + + def success_count(self) -> int: + return sum(1 for f in self.measured() if f.outcome_success == 1) + + def success_rate(self) -> float | None: + m = self.measured() + return (self.success_count() / len(m)) if m else None + + def total_cost(self) -> float: + return sum(f.cost_usd for f in self.facts) + + def cost_per_outcome(self) -> float | None: + succ = self.success_count() + return (self.total_cost() / succ) if succ > 0 else None + + def median_cost(self) -> float: + return statistics.median([f.cost_usd for f in self.facts]) if self.facts else 0.0 + + def median_turns(self) -> float: + return statistics.median([f.num_turns for f in self.facts]) if self.facts else 0.0 + + def reasoning_share(self) -> float: + ot = sum(f.output_tokens for f in self.facts) + rt = sum(f.reasoning_tokens for f in self.facts) + return (rt / ot) if ot > 0 else 0.0 + + +def _cost_per_outcome_ci( + facts: list[_SessionFact], *, ci_level: float +) -> tuple[float, float] | None: + """Seeded ratio bootstrap of Σcost / Σsuccess. ``None`` when < 2 successes. + + Resamples sessions with replacement; each resample yields Σcost/Σsuccess + (resamples with no successes are skipped). Deterministic via the pinned + seed, so the CI is reproducible. + """ + pairs = [(f.cost_usd, 1 if f.outcome_success == 1 else 0) for f in facts] + total_succ = sum(s for _, s in pairs) + if total_succ < 2 or len(pairs) < 2: + return None + import random + + rng = random.Random(bs.SEED) # noqa: S311 — deterministic resampling, not crypto + n = len(pairs) + ratios: list[float] = [] + for _ in range(bs.BOOTSTRAP_ITERS): + cost_sum = 0.0 + succ_sum = 0 + for _ in range(n): + c, s = pairs[rng.randrange(n)] + cost_sum += c + succ_sum += s + if succ_sum > 0: + ratios.append(cost_sum / succ_sum) + if not ratios: + return None + ratios.sort() + alpha = (1.0 - ci_level) / 2.0 + return (bs.percentile(ratios, alpha), bs.percentile(ratios, 1.0 - alpha)) + + +def _two_proportion_pvalue(s1: int, n1: int, s2: int, n2: int) -> float: + """Two-sided two-proportion z-test p-value (normal approx, stdlib). + + Used only to feed Benjamini–Hochberg (§4.4). Degenerate inputs (zero + variance) return ``1.0`` — no evidence of a difference. + """ + if n1 == 0 or n2 == 0: + return 1.0 + p1, p2 = s1 / n1, s2 / n2 + p_pool = (s1 + s2) / (n1 + n2) + var = p_pool * (1.0 - p_pool) * (1.0 / n1 + 1.0 / n2) + if var <= 0: + return 1.0 + z = (p1 - p2) / (var ** 0.5) + return 2.0 * (1.0 - statistics.NormalDist().cdf(abs(z))) + + +# ── verdict assembly ───────────────────────────────────────────────────────── + + +def analyze_benchmark( + conn: sqlite3.Connection, + *, + scope: Scope | None = None, + project_ids: list[int] | None = None, + intent: str | None = None, + weights: dict[str, float] | None = None, + ci_level: float = bs.CI_LEVEL, +) -> dict[str, Any]: + """Compute the comparative benchmark verdict over *scope*. + + Args: + conn: Open store connection; reads only, guarded so a schemaless DB + returns an empty-but-valid verdict. + scope: Optional timestamp window (``None`` = all time). + project_ids: Optional ``projects.id`` filter (``None`` = whole store). + intent: Optional single-intent filter (only that stratum family). + weights: Composite weights; defaults to the ratified rubric v1. + ci_level: Confidence level for every interval (default 0.90). + + Returns: + The full report dict (see the module docstring / spec §6.1). Always + well-formed; ``verdict.headline`` is ``"insufficient evidence"`` on any + degenerate store. + """ + used_weights = _resolve_weights(weights) + try: + facts = _load_facts(conn, scope=scope, project_ids=project_ids) + except Exception: # noqa: BLE001 — advisory: never raise from the report + facts = [] + + if intent: + facts = [f for f in facts if f.intent == intent] + + return _assemble(facts, weights=used_weights, ci_level=ci_level) + + +def _resolve_weights(weights: dict[str, float] | None) -> dict[str, float]: + """Return normalized composite weights, defaulting to rubric v1.""" + if not weights: + return dict(DEFAULT_WEIGHTS) + picked = {k: float(weights.get(k, DEFAULT_WEIGHTS[k])) for k in DEFAULT_WEIGHTS} + total = sum(picked.values()) + if total <= 0: + return dict(DEFAULT_WEIGHTS) + return {k: v / total for k, v in picked.items()} + + +def _empty_report( + weights: dict[str, float], ci_level: float, *, sessions_total: int = 0 +) -> dict[str, Any]: + return { + "verdict": { + "headline": "insufficient evidence", + "winning_model": None, + "confidence": "none", + "cost_per_outcome_usd": None, + "runner_up": None, + "caveats": [ + "Not enough comparable evidence to name a winner yet.", + ], + }, + "strata": [], + "coverage": { + "sessions_total": sessions_total, + "sessions_scored": 0, + "grade_coverage": 0.0, + }, + "rubric_version": RUBRIC_VERSION, + "weights": weights, + "ci_level": ci_level, + "success_threshold": SUCCESS_THRESHOLD, + "warning": NATURAL_EXPERIMENT_WARNING, + "method_notes": list(_METHOD_NOTES), + } + + +def _assemble( + facts: list[_SessionFact], + *, + weights: dict[str, float], + ci_level: float, +) -> dict[str, Any]: + """Turn per-session facts into the stratified verdict payload.""" + if not facts: + return _empty_report(weights, ci_level) + + # ── coverage ────────────────────────────────────────────────────────── + sessions_total = len(facts) + sessions_scored = sum(1 for f in facts if f.outcome_success is not None) + grade_scored = sum(1 for f in facts if f.outcome_tier == "llm_grade") + coverage = { + "sessions_total": sessions_total, + "sessions_scored": sessions_scored, + "grade_coverage": round(grade_scored / sessions_total, 4) if sessions_total else 0.0, + } + + # ── stratify: (intent, size_band) → model → cell ────────────────────── + strata: dict[tuple[str, str], dict[str, _ModelCell]] = {} + for f in facts: + key = (f.intent, f.size_band) + cell = strata.setdefault(key, {}).setdefault( + f.primary_model, _ModelCell(model=f.primary_model) + ) + cell.facts.append(f) + + # ── p-value family for BH-FDR (success difference per cell-pair) ────── + pvalues: list[float] = [] + pval_index: dict[tuple[tuple[str, str], str, str], int] = {} + for key, models in strata.items(): + qualified = sorted(m for m, c in models.items() if c.qualified) + for i in range(len(qualified)): + for j in range(i + 1, len(qualified)): + a, b = models[qualified[i]], models[qualified[j]] + ma, mb = a.measured(), b.measured() + pval_index[(key, a.model, b.model)] = len(pvalues) + pvalues.append( + _two_proportion_pvalue( + a.success_count(), len(ma), b.success_count(), len(mb) + ) + ) + reject = bs.benjamini_hochberg(pvalues, alpha=1.0 - ci_level) if pvalues else [] + + def _pair_significant(key: tuple[str, str], m1: str, m2: str) -> bool: + idx = pval_index.get((key, m1, m2)) + if idx is None: + idx = pval_index.get((key, m2, m1)) + return bool(idx is not None and idx < len(reject) and reject[idx]) + + # ── per-stratum payload + cell winners ──────────────────────────────── + strata_payload: list[dict[str, Any]] = [] + # winner-tracking for the cross-task headline + clear_wins: dict[str, int] = {} + clear_losses: dict[str, int] = {} + balanced_n: dict[str, int] = {} + cell_win_widths: dict[str, list[float]] = {} + cost_accum: dict[str, tuple[float, int]] = {} # model → (Σcost, Σsucc) over clear-win cells + + for key in sorted(strata.keys()): + intent_lbl, size_lbl = key + models = strata[key] + qualified = [c for c in models.values() if c.qualified] + for c in qualified: + balanced_n[c.model] = balanced_n.get(c.model, 0) + c.n + + model_rows = [_model_row(c, ci_level=ci_level) for c in models.values()] + _fill_composites(model_rows, weights) + model_rows.sort(key=lambda r: (r["qualified"], r["composite"]), reverse=True) + + cell_verdict = "insufficient evidence" + winner: str | None = None + effect: dict[str, Any] = {} + qrows = [r for r in model_rows if r["qualified"]] + if len(qrows) >= bs.MIN_MODELS_PER_CELL: + top, second = qrows[0], qrows[1] + winner = top["model"] + sr_diff = bs.risk_difference( + top["success_rate"]["point"] or 0.0, + second["success_rate"]["point"] or 0.0, + ) + cost_rel = _cost_effect(top, second) + practical = ( + abs(sr_diff) >= bs.MIN_EFFECT_SUCCESS or cost_rel >= bs.MIN_EFFECT_COST + ) + statistical = _pair_significant(key, top["model"], second["model"]) + effect = { + "success_risk_difference": round(sr_diff, 4), + "cost_relative_delta": round(cost_rel, 4), + "statistically_separated": statistical, + "practically_separated": practical, + } + if practical and statistical: + cell_verdict = "clear" + clear_wins[winner] = clear_wins.get(winner, 0) + 1 + clear_losses[second["model"]] = clear_losses.get(second["model"], 0) + 1 + wc = models[winner] + cs, ss = cost_accum.get(winner, (0.0, 0)) + cost_accum[winner] = (cs + wc.total_cost(), ss + wc.success_count()) + w_ci = top["success_rate"].get("ci_wilson") or [0.0, 1.0] + cell_win_widths.setdefault(winner, []).append(w_ci[1] - w_ci[0]) + else: + cell_verdict = "weak" + + strata_payload.append( + { + "intent": intent_lbl, + "size_band": size_lbl, + "models": model_rows, + "assignment_balance": {c.model: c.n for c in models.values()}, + "cell_verdict": cell_verdict, + "winner": winner, + "effect": effect, + } + ) + + verdict = _headline( + intent_filter=None, + clear_wins=clear_wins, + clear_losses=clear_losses, + balanced_n=balanced_n, + cell_win_widths=cell_win_widths, + cost_accum=cost_accum, + ) + + return { + "verdict": verdict, + "strata": strata_payload, + "coverage": coverage, + "rubric_version": RUBRIC_VERSION, + "weights": weights, + "ci_level": ci_level, + "success_threshold": SUCCESS_THRESHOLD, + "warning": NATURAL_EXPERIMENT_WARNING, + "method_notes": list(_METHOD_NOTES), + } + + +def _model_row(cell: _ModelCell, *, ci_level: float) -> dict[str, Any]: + """Per-model row inside a stratum (matches spec §6.1).""" + sr = cell.success_rate() + measured = cell.measured() + wilson = ( + list(bs.wilson_interval(cell.success_count(), len(measured), ci_level=ci_level)) + if measured + else None + ) + cost_ci = bs.percentile_bootstrap_ci( + [f.cost_usd for f in cell.facts], statistic="median", ci_level=ci_level + ) + cpo = cell.cost_per_outcome() + cpo_ci = _cost_per_outcome_ci(cell.facts, ci_level=ci_level) + return { + "model": cell.model, + "n": cell.n, + "qualified": cell.qualified, + "coverage": round(len(measured) / cell.n, 4) if cell.n else 0.0, + "success_measured_n": len(measured), + "success_rate": { + "point": round(sr, 4) if sr is not None else None, + "ci_wilson": [round(x, 4) for x in wilson] if wilson else None, + }, + "cost_per_outcome": { + "point": round(cpo, 6) if cpo is not None else None, + "ci": [round(x, 6) for x in cpo_ci] if cpo_ci else None, + }, + "median_cost": { + "point": round(cell.median_cost(), 6), + "ci": [round(x, 6) for x in cost_ci], + }, + "median_turns": round(cell.median_turns(), 2), + "reasoning_share": round(cell.reasoning_share(), 4), + "composite": 0.0, # filled by _fill_composites (normalized across the cell) + } + + +def _fill_composites(rows: list[dict[str, Any]], weights: dict[str, float]) -> None: + """Fill each row's normalized composite in ``[0, 1]`` (mutates ``rows``). + + The composite blends three axes, each normalized **within the stratum** so + the comparison is like-for-like: success rate (already 0..1, higher wins), + cost (min-max inverse — cheapest gets 1.0), and effort/turns (min-max + inverse — fewest gets 1.0). Reasoning efficiency is deliberately absent — + descriptive only, never scored (§3.3). A single-model cell scores every + axis at its best (nothing to compare against). + """ + if not rows: + return + + def _cost_of(r: dict[str, Any]) -> float: + cpo = r["cost_per_outcome"]["point"] + return cpo if cpo is not None else r["median_cost"]["point"] + + costs = [_cost_of(r) for r in rows] + turns = [r["median_turns"] for r in rows] + for r in rows: + success = r["success_rate"]["point"] or 0.0 + cost_norm = _inverse_minmax(costs, _cost_of(r)) + effort_norm = _inverse_minmax(turns, r["median_turns"]) + composite = ( + weights["success"] * success + + weights["cost"] * cost_norm + + weights["effort"] * effort_norm + ) + r["composite"] = round(min(1.0, max(0.0, composite)), 4) + + +def _inverse_minmax(values: list[float], v: float) -> float: + """Min-max *inverse* normalization: the smallest value → 1.0, largest → 0.0. + + Used for cost + effort where lower is better. A flat set (all equal, or a + single model) → 1.0, so a lone model isn't penalized for having no rival. + """ + lo, hi = min(values), max(values) + if hi <= lo: + return 1.0 + return 1.0 - (v - lo) / (hi - lo) + + +def _cost_effect(top: dict[str, Any], second: dict[str, Any]) -> float: + """Relative cost advantage of the composite winner over the runner-up. + + Uses cost-per-outcome when both have it, else median cost. Positive ⇒ the + winner is cheaper by that fraction. + """ + tw = top["cost_per_outcome"]["point"] + sw = second["cost_per_outcome"]["point"] + if tw is not None and sw is not None: + return bs.relative_delta(tw, sw) + return bs.relative_delta( + top["median_cost"]["point"], second["median_cost"]["point"] + ) + + +def _headline( + *, + intent_filter: str | None, + clear_wins: dict[str, int], + clear_losses: dict[str, int], + balanced_n: dict[str, int], + cell_win_widths: dict[str, list[float]], + cost_accum: dict[str, tuple[float, int]], +) -> dict[str, Any]: + """Decide the cross-task winner (§4.4) — or refuse. + + A headline winner must win the composite in ≥2 strata (with a real, BH- + significant separation), never clearly lose a stratum, and clear the + balanced-sample floor. Otherwise the verdict is "insufficient evidence". + """ + candidates = [ + m + for m in clear_wins + if clear_wins[m] >= 2 + and clear_losses.get(m, 0) == 0 + and balanced_n.get(m, 0) >= bs.MIN_BALANCED_TOTAL + ] + if len(candidates) != 1: + return { + "headline": "insufficient evidence", + "winning_model": None, + "confidence": "none", + "cost_per_outcome_usd": None, + "runner_up": None, + "caveats": _headline_caveats(candidates, clear_wins, balanced_n), + } + + winner = candidates[0] + # runner-up = next model by clear wins (0 if none) + others = sorted( + (m for m in clear_wins if m != winner), + key=lambda m: clear_wins[m], + reverse=True, + ) + runner_up = others[0] if others else None + + cost_sum, succ_sum = cost_accum.get(winner, (0.0, 0)) + cost_per_outcome = (cost_sum / succ_sum) if succ_sum > 0 else None + + confidence = _confidence( + winner=winner, + clear_wins=clear_wins, + clear_losses=clear_losses, + balanced_n=balanced_n, + cell_win_widths=cell_win_widths, + ) + label = f"{winner} wins" + (f" for {intent_filter}" if intent_filter else "") + return { + "headline": label, + "winning_model": winner, + "confidence": confidence, + "cost_per_outcome_usd": round(cost_per_outcome, 6) if cost_per_outcome else None, + "runner_up": runner_up, + "caveats": [ + "Winner holds across " + f"{clear_wins[winner]} strata with no stratum where it clearly loses.", + NATURAL_EXPERIMENT_WARNING, + ], + } + + +def _headline_caveats( + candidates: list[str], clear_wins: dict[str, int], balanced_n: dict[str, int] +) -> list[str]: + if len(candidates) > 1: + return [ + "More than one model qualifies as a cross-task winner — no single " + "winner can be named. Compare the per-stratum table instead.", + ] + if clear_wins: + best = max(clear_wins, key=lambda m: clear_wins[m]) + return [ + f"The strongest model ({best}) wins {clear_wins[best]} stratum/strata " + f"(need ≥2) with {balanced_n.get(best, 0)} balanced sessions " + f"(need ≥{bs.MIN_BALANCED_TOTAL}) — not enough to headline a winner.", + ] + return ["Not enough comparable evidence to name a winner yet."] + + +def _confidence( + *, + winner: str, + clear_wins: dict[str, int], + clear_losses: dict[str, int], + balanced_n: dict[str, int], + cell_win_widths: dict[str, list[float]], +) -> str: + """Product-of-terms confidence (mirrors ``mode_recommender``) → bucket.""" + n = balanced_n.get(winner, 0) + sample_term = min(1.0, n / (2.0 * bs.MIN_BALANCED_TOTAL)) + wins = clear_wins.get(winner, 0) + agreement_term = min(1.0, wins / 3.0) # 3+ agreeing strata → full marks + widths = cell_win_widths.get(winner) or [1.0] + ci_term = max(0.0, 1.0 - (sum(widths) / len(widths))) + score = sample_term * agreement_term * ci_term + return bs.confidence_bucket(score) + + +# ── recommendation (outcome-aware; §1 / §6.2) ──────────────────────────────── + + +def recommend_from_history( + conn: sqlite3.Connection, + *, + intent: str, + size: str | None = None, + language: str | None = None, + scope: Scope | None = None, + project_ids: list[int] | None = None, + weights: dict[str, float] | None = None, + ci_level: float = bs.CI_LEVEL, +) -> dict[str, Any]: + """Outcome-aware model pick for a *described* task (the successor to the + cost-only ``mode_recommender``). + + Restricts the benchmark to the matching stratum family and returns the + winning model with its evidence — or an honest "insufficient evidence". + Always returns a well-formed dict; never raises. + """ + report = analyze_benchmark( + conn, + scope=scope, + project_ids=project_ids, + intent=intent, + weights=weights, + ci_level=ci_level, + ) + # Filter strata to the requested size (and language, when both known). + strata = report.get("strata") or [] + if size: + strata = [s for s in strata if s.get("size_band") == size] + + # Prefer a matching clear cell; fall back to the headline verdict. + best_cell = None + for s in strata: + if s.get("cell_verdict") == "clear" and s.get("winner"): + best_cell = s + break + + verdict = report.get("verdict") or {} + if best_cell is not None: + winner = best_cell["winner"] + row = next( + (m for m in best_cell["models"] if m["model"] == winner), None + ) + return { + "intent": intent, + "size": size, + "language": language, + "recommended_model": winner, + "confidence": "medium" if verdict.get("winning_model") != winner else verdict.get("confidence", "medium"), + "basis": "stratum", + "stratum": {"intent": best_cell["intent"], "size_band": best_cell["size_band"]}, + "evidence": row, + "rationale": ( + f"In {best_cell['intent']} × {best_cell['size_band']} tasks, " + f"{winner} wins on the composite with a real, significant " + f"separation from the runner-up." + ), + "rubric_version": RUBRIC_VERSION, + "weights": report.get("weights"), + } + + return { + "intent": intent, + "size": size, + "language": language, + "recommended_model": verdict.get("winning_model"), + "confidence": verdict.get("confidence", "none"), + "basis": "headline" if verdict.get("winning_model") else "insufficient_evidence", + "stratum": None, + "evidence": None, + "rationale": ( + verdict.get("caveats", ["Not enough comparable evidence yet."]) or [""] + )[0], + "rubric_version": RUBRIC_VERSION, + "weights": report.get("weights"), + } diff --git a/stackunderflow/routes/benchmark.py b/stackunderflow/routes/benchmark.py new file mode 100644 index 00000000..da6ba4d0 --- /dev/null +++ b/stackunderflow/routes/benchmark.py @@ -0,0 +1,243 @@ +"""``GET /api/benchmark`` — comparative "which model wins for your work" verdict. + +Thin HTTP wrapper around :func:`stackunderflow.reports.benchmark.analyze_benchmark`. +An observational benchmark over the user's own history (spec 26): per-task-type +verdicts with the full statistical-honesty machinery, or — just as often and +just as valuable — an honest "insufficient evidence". + +Mirrors ``routes/forks.py`` exactly: + +* **200 ms analytical tier** (the ``/api/yield`` / ``/api/optimize`` tier, not + the 100 ms mart tier) — it is a cross-session statistical composite, so it is + wrapped in the same read-through cache forks uses (keyed on store + scope + + project ids, self-invalidated by a sessions signature that moves on ingest). +* **Currency contract** — every dollar figure is pre-converted to the active + currency before send, applied to a deep copy outside the cache so an FX + change is picked up without recompute. +* A ``warning`` field carries the natural-experiment caveat inline. +""" + +from __future__ import annotations + +import copy +import threading +from pathlib import Path +from typing import Any + +from fastapi import APIRouter, HTTPException, Query + +import stackunderflow.deps as deps +from stackunderflow.infra.currency import active_currency_payload +from stackunderflow.reports.benchmark import analyze_benchmark, recommend_from_history +from stackunderflow.reports.scope import parse_period +from stackunderflow.store import db + +router = APIRouter() + +# Read-through cache: the analysis walks every scoped session and re-derives +# intent per session, so it isn't free. Keyed on (store, scope, ids) plus a +# sessions signature (max last_ts, summed message_count) that any ingest bumps, +# so a stale entry can't outlive a refresh — the same contract forks/cost use. +# Currency conversion stays OUTSIDE the cache (applied to a deep copy). +_BENCH_CACHE: dict[ + tuple[str, str, tuple[int, ...] | None, str | None], tuple[tuple[str | None, int], dict] +] = {} +_BENCH_CACHE_LOCK = threading.Lock() + + +def _bench_signature(conn: Any, project_ids: list[int] | None) -> tuple[str | None, int]: + """(max last_ts, summed message_count) over the scoped sessions. + + ``project_ids is None`` = whole store. Any ingest that writes a message + bumps this signature and forces a recompute. Advisory: a bad store returns + a sentinel that simply misses the cache rather than raising. + """ + try: + if project_ids is None: + row = conn.execute( + "SELECT MAX(last_ts) AS max_ts, " + "COALESCE(SUM(message_count), 0) AS n FROM sessions" + ).fetchone() + elif not project_ids: + return (None, 0) + else: + placeholders = ",".join("?" for _ in project_ids) + row = conn.execute( + "SELECT MAX(last_ts) AS max_ts, " # noqa: S608 — placeholders are only ? marks + "COALESCE(SUM(message_count), 0) AS n " + f"FROM sessions WHERE project_id IN ({placeholders})", + tuple(project_ids), + ).fetchone() + except Exception: # noqa: BLE001 — advisory: a bad store just misses cache + return (None, -1) + if row is None: + return (None, 0) + return (row["max_ts"], int(row["n"] or 0)) + + +def _analyze_benchmark_cached( + conn: Any, *, scope: Any, project_ids: list[int] | None, intent: str | None +) -> dict: + """Read-through cache around :func:`analyze_benchmark` (returns USD report).""" + sig = _bench_signature(conn, project_ids) + key = ( + str(deps.store_path), + scope.label, + tuple(sorted(project_ids)) if project_ids is not None else None, + intent, + ) + with _BENCH_CACHE_LOCK: + cached = _BENCH_CACHE.get(key) + if cached is not None and cached[0] == sig: + return copy.deepcopy(cached[1]) + report = analyze_benchmark(conn, scope=scope, project_ids=project_ids, intent=intent) + with _BENCH_CACHE_LOCK: + _BENCH_CACHE[key] = (sig, report) + return copy.deepcopy(report) + + +# Friendly period superset — ``week`` maps to ``7days`` inside ``parse_period``. +# Mirrors forks / yield so all three beta surfaces accept the same selector. +_PERIOD_ALIASES = { + "today": "today", + "week": "7days", + "7days": "7days", + "month": "month", + "30days": "30days", + "all": "all", +} + +_PERIOD_QUERY = Query("all", description="today | week | month | all") +_LOG_PATH_QUERY = Query(None, description="Project log path; omit for whole-store") +_INTENT_QUERY = Query(None, description="Filter to one intent stratum (build/fix/…)") +_SIZE_QUERY = Query(None, description="Task size band (tiny/small/med/large)") +_LANG_QUERY = Query(None, description="Dominant language hint") + + +def _project_ids_for(conn: Any, path: str) -> list[int]: + """Resolve a log path to the ``projects.id`` list for its slug (own resolver).""" + slug = Path(path).name + try: + rows = conn.execute("SELECT id FROM projects WHERE slug = ?", (slug,)).fetchall() + except Exception: # noqa: BLE001 — advisory route, never 500 on a bad store + return [] + return [int(r["id"]) for r in rows] + + +def _convert_report_costs(report: dict, rate: float) -> None: + """Convert every dollar figure in the report to the active currency in place. + + Explicit walk (never a blanket multiply) so a schema change can't silently + double-convert a non-cost field. ``cost_usd`` originates in ``session_mart`` + and is only *displayed* in another currency — the invariant is untouched. + """ + if rate == 1.0: + return + verdict = report.get("verdict") or {} + if verdict.get("cost_per_outcome_usd") is not None: + verdict["cost_per_outcome_usd"] = float(verdict["cost_per_outcome_usd"]) * rate + for stratum in report.get("strata") or []: + for m in stratum.get("models") or []: + _convert_cost_block(m.get("cost_per_outcome"), rate) + _convert_cost_block(m.get("median_cost"), rate) + + +def _convert_cost_block(block: Any, rate: float) -> None: + """Scale a ``{"point": x, "ci": [lo, hi]}`` cost block by ``rate`` in place.""" + if not isinstance(block, dict): + return + if block.get("point") is not None: + block["point"] = float(block["point"]) * rate + ci = block.get("ci") + if isinstance(ci, list): + block["ci"] = [float(x) * rate for x in ci] + + +@router.get("/api/benchmark") +async def get_benchmark( + period: str = _PERIOD_QUERY, + log_path: str | None = _LOG_PATH_QUERY, + intent: str | None = _INTENT_QUERY, +): + """Return ``{period, scope, report, currency, warning}``. + + ``report`` is the full benchmark verdict with every dollar figure already + converted to the active currency. Scoped to a project when ``log_path`` (or + the active ``deps.current_log_path``) resolves to one, else the whole store. + """ + period = period if isinstance(period, str) else "all" + spec = _PERIOD_ALIASES.get(period) + if spec is None: + raise HTTPException( + status_code=400, + detail=f"Invalid period '{period}'. Valid: {', '.join(_PERIOD_ALIASES)}", + ) + scope = parse_period(spec) + + log_path_str = log_path if isinstance(log_path, str) else None + intent_str = intent if isinstance(intent, str) else None + path = log_path_str or deps.current_log_path + + conn = db.connect(deps.store_path) + try: + project_ids = _project_ids_for(conn, path) if path else None + report = _analyze_benchmark_cached( + conn, scope=scope, project_ids=project_ids, intent=intent_str + ) + finally: + conn.close() + + currency = active_currency_payload() + _convert_report_costs(report, currency["rate_from_usd"]) + + return { + "period": period, + "scope": scope.label, + "report": report, + "currency": currency, + "warning": report.get("warning"), + } + + +@router.get("/api/benchmark/recommend") +async def get_benchmark_recommend( + intent: str = Query(..., description="Task intent (build/fix/explore/refactor/test/ops)"), + size: str | None = _SIZE_QUERY, + language: str | None = _LANG_QUERY, + log_path: str | None = _LOG_PATH_QUERY, + period: str = _PERIOD_QUERY, +): + """Return the outcome-aware model recommendation for a described task.""" + period = period if isinstance(period, str) else "all" + spec = _PERIOD_ALIASES.get(period) + if spec is None: + raise HTTPException( + status_code=400, + detail=f"Invalid period '{period}'. Valid: {', '.join(_PERIOD_ALIASES)}", + ) + scope = parse_period(spec) + if not isinstance(intent, str) or not intent.strip(): + raise HTTPException(status_code=400, detail="intent is required") + + size_str = size if isinstance(size, str) else None + lang_str = language if isinstance(language, str) else None + log_path_str = log_path if isinstance(log_path, str) else None + path = log_path_str or deps.current_log_path + + conn = db.connect(deps.store_path) + try: + project_ids = _project_ids_for(conn, path) if path else None + rec = recommend_from_history( + conn, intent=intent, size=size_str, language=lang_str, + scope=scope, project_ids=project_ids, + ) + finally: + conn.close() + + currency = active_currency_payload() + rate = currency["rate_from_usd"] + if rate != 1.0 and isinstance(rec.get("evidence"), dict): + _convert_cost_block(rec["evidence"].get("cost_per_outcome"), rate) + _convert_cost_block(rec["evidence"].get("median_cost"), rate) + + return {"period": period, "scope": scope.label, "recommendation": rec, "currency": currency} diff --git a/stackunderflow/server.py b/stackunderflow/server.py index 8777a209..35f066c1 100644 --- a/stackunderflow/server.py +++ b/stackunderflow/server.py @@ -19,6 +19,7 @@ # Route modules from stackunderflow.routes import ( agent_teams, + benchmark, bookmarks, budgets, cfg, @@ -291,6 +292,7 @@ def _watcher_conn() -> "object": app.include_router(budgets.router) app.include_router(whatif.router) app.include_router(forks.router) +app.include_router(benchmark.router) app.include_router(patterns.router) app.include_router(worktrees.router) diff --git a/stackunderflow/services/benchmark_stats.py b/stackunderflow/services/benchmark_stats.py new file mode 100644 index 00000000..2a3c8121 --- /dev/null +++ b/stackunderflow/services/benchmark_stats.py @@ -0,0 +1,312 @@ +"""Small, pure, seeded statistics for the comparative benchmark engine. + +Spec 26 §4 ("statistical honesty — the crux"). A benchmark that always names a +winner is folklore with a progress bar; the whole value is refusing to conclude +when the evidence is thin, and separating a real difference from noise when it +is not. That discipline is entirely in this module. + +Everything here is: + +* **Pure + deterministic.** No store, no clock, no network. The bootstrap draws + from a ``random.Random(_SEED)`` seeded generator so two runs over the same + numbers are byte-identical (the report's read-through cache and the tests both + rely on this). +* **stdlib only.** ``statistics`` (incl. ``NormalDist`` for the z-score) — no + numpy, consistent with the rest of the tree. ``reports/anomaly.py`` already + sets this tone with a robust-MAD ``MIN_POINTS`` floor; the engine extends it. + +The functions, mapped to the spec: + +* :func:`wilson_interval` — §4.2 success-rate CI (correct for small n where the + normal approximation breaks). +* :func:`percentile_bootstrap_ci` — §4.2 cost/grade/turns CI (skewed + continuous), seeded. +* :func:`benjamini_hochberg` — §4.4 FDR control across the family of tests. +* :func:`standardized_difference` / :func:`pooled_rate` — §3.2/§4.7 direct + standardization vs the confounded pooled mean (Simpson's-paradox defense). +* sample floors + effect thresholds + :func:`confidence_bucket` — §4.1/§4.3/§4.5 + ("insufficient evidence" as a first-class verdict). +""" + +from __future__ import annotations + +import random +import statistics +from typing import Any + +__all__ = [ + "SEED", + "CI_LEVEL", + "BOOTSTRAP_ITERS", + "MIN_SESSIONS_PER_CELL", + "MIN_MODELS_PER_CELL", + "MIN_BALANCED_TOTAL", + "MIN_EFFECT_COST", + "MIN_EFFECT_SUCCESS", + "MIN_EFFECT_GRADE", + "z_for_confidence", + "wilson_interval", + "percentile", + "percentile_bootstrap_ci", + "benjamini_hochberg", + "pooled_rate", + "standardized_rate", + "standardized_difference", + "relative_delta", + "risk_difference", + "confidence_bucket", +] + + +# ── pinned tunables ────────────────────────────────────────────────────────── + +# Pinned so the seeded bootstrap is reproducible: two runs on the same store +# agree, and a test can assert byte-identical CIs. +SEED = 1729 + +# Default confidence level for every interval. The maintainer may set 0.95; +# 0.90 is the ratified default (docs/specs/benchmark-rubric-v1.md). +CI_LEVEL = 0.90 + +# Bootstrap resamples for a continuous-metric CI. 2000 is the spec's figure — +# enough for a stable 90% interval without making the live join expensive. +BOOTSTRAP_ITERS = 2000 + +# Sample-size floors — refuse before you mislead (§4.1). 5 mirrors +# ``anomaly.MIN_POINTS`` and ``mode_recommender``'s n/5 term. +MIN_SESSIONS_PER_CELL = 5 # a model needs ≥5 sessions in a stratum to be scored +MIN_MODELS_PER_CELL = 2 # need ≥2 qualifying models to compare a stratum at all +MIN_BALANCED_TOTAL = 20 # per model, across strata, for a headline verdict + +# Practical-effect floors — statistical separation is necessary, not sufficient +# (§4.3). A win must also clear a floor a human would care about. +MIN_EFFECT_COST = 0.10 # ≥10% relative cost difference +MIN_EFFECT_SUCCESS = 0.10 # ≥10 percentage points +MIN_EFFECT_GRADE = 0.5 # ≥0.5 grade points + + +# ── z-score ────────────────────────────────────────────────────────────────── + + +def z_for_confidence(ci_level: float = CI_LEVEL) -> float: + """Two-sided z critical value for ``ci_level`` (e.g. 0.90 → 1.6449). + + Uses ``statistics.NormalDist`` — stdlib, no scipy. ``ci_level`` is clamped + to a sane open interval so a degenerate 0 or 1 can't blow up ``inv_cdf``. + """ + ci = min(max(float(ci_level), 0.5), 0.999999) + return statistics.NormalDist().inv_cdf(1.0 - (1.0 - ci) / 2.0) + + +# ── Wilson score interval (proportions) ────────────────────────────────────── + + +def wilson_interval( + successes: int, n: int, *, ci_level: float = CI_LEVEL +) -> tuple[float, float]: + """Wilson score interval for a proportion — correct at small n. + + The normal approximation (p ± z·√(p(1-p)/n)) is badly wrong for small n or + p near 0/1 (it can leave [0,1]); Wilson is the standard fix and the reason + a 3-of-4 success rate reports honest uncertainty instead of a fake 0.75±. + + ``n == 0`` → ``(0.0, 1.0)`` (no information → widest possible interval); + callers gate on the sample floor long before this, but the function stays + total. The result is always clamped to ``[0, 1]``. + """ + if n <= 0: + return (0.0, 1.0) + z = z_for_confidence(ci_level) + p = successes / n + z2 = z * z + denom = 1.0 + z2 / n + center = (p + z2 / (2.0 * n)) / denom + margin = (z / denom) * ((p * (1.0 - p) / n + z2 / (4.0 * n * n)) ** 0.5) + lo = max(0.0, center - margin) + hi = min(1.0, center + margin) + return (lo, hi) + + +# ── percentile + bootstrap (continuous, skewed) ────────────────────────────── + + +def percentile(sorted_values: list[float], q: float) -> float: + """Linear-interpolation percentile of an already-sorted list. + + ``q`` in ``[0, 1]``. Matches numpy's default ``'linear'`` method so the + bootstrap CI edges line up with what a reader would compute by hand. + """ + if not sorted_values: + return 0.0 + if len(sorted_values) == 1: + return float(sorted_values[0]) + q = min(max(float(q), 0.0), 1.0) + rank = q * (len(sorted_values) - 1) + lo_idx = int(rank) + frac = rank - lo_idx + if lo_idx + 1 >= len(sorted_values): + return float(sorted_values[-1]) + lo = sorted_values[lo_idx] + hi = sorted_values[lo_idx + 1] + return float(lo + (hi - lo) * frac) + + +def percentile_bootstrap_ci( + values: list[float], + *, + statistic: str = "median", + iters: int = BOOTSTRAP_ITERS, + ci_level: float = CI_LEVEL, + seed: int = SEED, +) -> tuple[float, float]: + """Percentile bootstrap CI for a skewed continuous metric (cost / turns). + + Resamples ``values`` with replacement ``iters`` times, recomputes + ``statistic`` (``"median"`` or ``"mean"``) on each resample, and returns the + central ``ci_level`` percentile band of that bootstrap distribution. + + Deterministic: the resampling uses ``random.Random(seed)`` with ``seed`` + pinned to :data:`SEED`, so the same input yields byte-identical bounds on + every run. Empty → ``(0.0, 0.0)``; a single value → ``(v, v)`` (a + degenerate but honest interval — one point has no spread). + """ + clean = [float(v) for v in values] + if not clean: + return (0.0, 0.0) + if len(clean) == 1: + return (clean[0], clean[0]) + + stat_fn = statistics.mean if statistic == "mean" else statistics.median + # Deterministic pseudo-random resampling — reproducibility, not crypto. + rng = random.Random(seed) # noqa: S311 + n = len(clean) + draws: list[float] = [] + for _ in range(max(1, iters)): + sample = [clean[rng.randrange(n)] for _ in range(n)] + draws.append(float(stat_fn(sample))) + draws.sort() + alpha = (1.0 - ci_level) / 2.0 + return (percentile(draws, alpha), percentile(draws, 1.0 - alpha)) + + +# ── Benjamini–Hochberg FDR ─────────────────────────────────────────────────── + + +def benjamini_hochberg(pvalues: list[float], *, alpha: float = 0.10) -> list[bool]: + """Benjamini–Hochberg step-up FDR control. + + Testing M models × S strata × K metrics inflates false positives fast + (§4.4). BH bounds the expected false-discovery rate at ``alpha``: sort the + p-values ascending, find the largest rank ``k`` with ``p_(k) ≤ (k/m)·α``, + and reject **every** hypothesis up to that rank (the step-up — a hypothesis + can be rejected even if it fails its own threshold, because a smaller p + later in the order carries it). + + Returns a list of booleans aligned to the **input** order (``True`` = + reject / a real finding). Empty input → ``[]``. + """ + m = len(pvalues) + if m == 0: + return [] + order = sorted(range(m), key=lambda i: pvalues[i]) + max_k = 0 + for rank, idx in enumerate(order, start=1): + if pvalues[idx] <= (rank / m) * alpha: + max_k = rank + reject = [False] * m + for rank, idx in enumerate(order, start=1): + if rank <= max_k: + reject[idx] = True + return reject + + +# ── standardization vs pooling (Simpson's-paradox defense) ─────────────────── + + +def pooled_rate(cells: dict[Any, tuple[int, float]]) -> float: + """Naive pooled mean: ``Σ(n·rate) / Σn`` over a model's own strata. + + This is the *confounded* number — it re-imports the selection bias because + a model's own assignment mix (mostly-easy vs mostly-hard) weights it. Kept + for disclosure/contrast only; the engine never ranks on it. + """ + num = 0.0 + den = 0 + for n, rate in cells.values(): + num += n * rate + den += n + return (num / den) if den else 0.0 + + +def standardized_rate( + cells: dict[Any, tuple[int, float]], weights: dict[Any, float] +) -> float: + """Direct-standardized rate: a model's per-stratum rates under a *common* + stratum weighting, over strata present in ``weights``.""" + num = 0.0 + den = 0.0 + for stratum, w in weights.items(): + if stratum in cells and w > 0: + num += w * cells[stratum][1] + den += w + return (num / den) if den else 0.0 + + +def standardized_difference( + a_cells: dict[Any, tuple[int, float]], + b_cells: dict[Any, tuple[int, float]], +) -> float: + """Standardized ``rate(A) - rate(B)`` over strata where **both** have data. + + Direct standardization with a common weight per shared stratum (the + combined sample there). This is the honest comparison the spec mandates: + pooling A's and B's own means separately would let a lopsided assignment + (A drew the easy tasks, B the hard ones) manufacture or reverse a winner + — the textbook Simpson's paradox (§3.2, §4.7). ``0.0`` when no stratum is + shared (→ "untested here", never imputed). + """ + shared = set(a_cells) & set(b_cells) + weights: dict[Any, float] = { + s: float(a_cells[s][0] + b_cells[s][0]) for s in shared + } + if not weights: + return 0.0 + return standardized_rate(a_cells, weights) - standardized_rate(b_cells, weights) + + +# ── effect sizes ───────────────────────────────────────────────────────────── + + +def relative_delta(new: float, base: float) -> float: + """Signed relative change ``(base - new) / base``; positive ⇒ ``new`` lower. + + Used for the cost axis: a positive value means the candidate is cheaper by + that fraction. ``0.0`` when ``base`` is 0 (no meaningful ratio).""" + if base == 0: + return 0.0 + return (base - new) / base + + +def risk_difference(p_a: float, p_b: float) -> float: + """Risk difference ``p_a - p_b`` (percentage-point gap between two rates).""" + return p_a - p_b + + +# ── confidence label ───────────────────────────────────────────────────────── + + +def confidence_bucket(score: float) -> str: + """Map a ``[0, 1]`` confidence score to ``{none, low, medium, high}``. + + ``none`` is the "insufficient evidence" verdict — the single most-surfaced + field (§4.5). Thresholds are deliberately conservative: the engine should + reach ``high`` only when the sample, balance, CI width, and cross-stratum + agreement all line up. + """ + if score >= 0.66: + return "high" + if score >= 0.40: + return "medium" + if score >= 0.15: + return "low" + return "none" diff --git a/stackunderflow/services/meta_agent.py b/stackunderflow/services/meta_agent.py index a9f6edb2..55dc9da8 100644 --- a/stackunderflow/services/meta_agent.py +++ b/stackunderflow/services/meta_agent.py @@ -595,6 +595,49 @@ def to_dict(self) -> dict[str, Any]: }, }, }, + { + "type": "function", + "function": { + "name": "recommend_model_for_task", + "description": ( + "Recommend which model to use for a described task, based on " + "the user's own OUTCOMES — not just cost. Where " + "``recommend_mode`` ranks on cost alone, this consults the " + "comparative benchmark: for the matching task stratum (intent × " + "size) it returns the model that historically won on the " + "composite of success, cost-per-successful-outcome, and effort, " + "with its evidence. Returns 'insufficient_evidence' honestly " + "when the user's history can't support a call. Use for 'which " + "model should I use for this refactor?' style routing." + ), + "parameters": { + "type": "object", + "properties": { + "intent": { + "type": "string", + "description": ( + "Task intent. One of build / fix / explore / " + "refactor / test / ops. Required." + ), + "enum": ["build", "fix", "explore", "refactor", "test", "ops"], + }, + "size": { + "type": "string", + "description": ( + "Optional task size band: tiny / small / med / " + "large. Narrows to that stratum when given." + ), + "enum": ["tiny", "small", "med", "large"], + }, + "language": { + "type": "string", + "description": "Optional dominant language hint (e.g. python).", + }, + }, + "required": ["intent"], + }, + }, + }, ] @@ -1215,6 +1258,33 @@ def _exec_get_session_quality( return quality_to_dict(quality) +def _exec_recommend_model_for_task( + conn: sqlite3.Connection, args: dict[str, Any] +) -> dict[str, Any]: + """Outcome-aware model recommendation from the comparative benchmark. + + Hands off to :func:`reports.benchmark.recommend_from_history`, which is + advisory-never-raises and returns an honest ``insufficient_evidence`` basis + when the user's history can't support a call. ``intent`` is required. + """ + from stackunderflow.reports import benchmark + + intent = str(args.get("intent") or "").strip() + if not intent: + return {"error": "intent is required"} + valid = {"build", "fix", "explore", "refactor", "test", "ops"} + if intent not in valid: + return {"error": f"intent must be one of {sorted(valid)} (got {intent!r})"} + size = args.get("size") + language = args.get("language") + return benchmark.recommend_from_history( + conn, + intent=intent, + size=str(size) if size else None, + language=str(language) if language else None, + ) + + # Dispatcher table — name → (conn, args) callable. Keeping this flat # (instead of dynamic getattr) makes the surface explicit: a new tool # has to be added in three places: catalogue, dispatcher, and tests. @@ -1233,6 +1303,7 @@ def _exec_get_session_quality( "get_pr_outcomes": _exec_get_pr_outcomes, "get_ci_runs": _exec_get_ci_runs, "get_session_quality": _exec_get_session_quality, + "recommend_model_for_task": _exec_recommend_model_for_task, } diff --git a/stackunderflow/services/mode_recommender.py b/stackunderflow/services/mode_recommender.py index 8bcc65c5..8ac300b2 100644 --- a/stackunderflow/services/mode_recommender.py +++ b/stackunderflow/services/mode_recommender.py @@ -84,6 +84,8 @@ from datetime import UTC, datetime, timedelta from typing import Any +from stackunderflow.services import task_classifier + __all__ = [ "Recommendation", "extract_features", @@ -98,47 +100,11 @@ CACHE_TTL_HOURS = 24 -# Token bands for the "same shape" filter. Edges are token-count thresholds -# (chars/4 estimate) — anything below the upper bound of a band falls in it. -# Keeping bands wide on purpose: v1 is matching shape, not token-budget -# accounting. -TOKEN_BANDS: tuple[tuple[str, int], ...] = ( - ("tiny", 200), - ("small", 800), - ("med", 3000), - ("large", 10**9), # catch-all -) - -# Intent keyword lookup. Earlier patterns win on overlap (a prompt that -# says "fix the test" maps to ``fix``, not ``test``). Keep the keyword -# lists short and obvious — heuristic v1, not NLP. -_INTENT_KEYWORDS: tuple[tuple[str, tuple[str, ...]], ...] = ( - ( - "fix", - ("fix", "bug", "broken", "regression", "debug", "patch", - "error", "fail", "crash", "stack trace"), - ), - ( - "refactor", - ("refactor", "clean up", "rename", "extract", "rewrite", - "simplify", "deduplicate", "tidy", "untangle"), - ), - ( - "test", - ("test", "tests", "unit test", "pytest", "spec", "coverage", - "fixture", "snapshot test"), - ), - ( - "build", - ("build", "add", "implement", "create", "new feature", - "ship", "write a", "scaffold"), - ), - ( - "explore", - ("explain", "what does", "how does", "summarise", "summarize", - "describe", "trace", "show me", "find", "search"), - ), -) +# Token bands for the "same shape" filter — re-exported from the canonical +# ``task_classifier`` so the recommender, the tag service, and the benchmark +# share one definition (Move 0 of spec 26). Same thresholds this module has +# always used; ``_token_band`` below still applies them to a chars/4 estimate. +TOKEN_BANDS: tuple[tuple[str, int], ...] = task_classifier.TOKEN_BANDS # Language hints. Lowercased substring match. The list is intentionally # short and high-signal — extending it is cheap (no ALTER TABLE because @@ -211,23 +177,16 @@ def to_dict(self) -> dict[str, Any]: def _intent_of(prompt: str) -> str: - """Map the prompt to a coarse intent label. - - Scans the keyword lists in declaration order and returns the **first - label whose keyword set has the most hits**. Ties broken by - declaration order (so ``fix`` beats ``test`` on a "fix the test" - prompt). Default ``"explore"`` when nothing matches — read-only - questions are the most common no-keyword case. + """Map the prompt to a single coarse intent label. + + Delegates to the canonical :func:`task_classifier.classify_intent` (Move 0 + of spec 26) so the recommender, the tag service, and the benchmark agree on + the taxonomy. The canonical classifier adopts the 6-label set (adds + ``ops``) and resolves multi-intent prompts by a fixed precedence that + keeps this module's historical single-label picks intact (e.g. "fix the + failing test" → ``fix``). Default ``"explore"`` when nothing matches. """ - if not prompt: - return "explore" - lowered = prompt.lower() - best: tuple[str, int] = ("explore", 0) - for label, keywords in _INTENT_KEYWORDS: - hits = sum(1 for kw in keywords if kw in lowered) - if hits > best[1]: - best = (label, hits) - return best[0] + return task_classifier.classify_intent(prompt or "") def _token_band(prompt: str) -> str: diff --git a/stackunderflow/services/tag_service.py b/stackunderflow/services/tag_service.py index 49359c50..9112a839 100644 --- a/stackunderflow/services/tag_service.py +++ b/stackunderflow/services/tag_service.py @@ -11,6 +11,8 @@ from collections import defaultdict from pathlib import Path +from stackunderflow.services import task_classifier + logger = logging.getLogger(__name__) # Well-known GitHub language colors @@ -127,25 +129,11 @@ "intent:ops": "#64748b", # slate — deploy / config / infra } -# Intent detection patterns. Order matters: first match wins when a session -# has evidence of multiple intents, but we keep all matches (a session CAN -# have multiple intents — e.g. a "build" that ends in "fix"). -INTENT_PATTERNS = [ - # build — adding something new - (r"\b(add|adding|added|implement|implementing|implemented|create|creating|created|build|building|built|new feature|scaffold|scaffolding|set up|setup)\b", "intent:build"), - # fix — bug or error - (r"\b(fix|fixing|fixed|bug|bugs|broken|breaks|breaking|crash|crashes|crashing|error|errors|traceback|stack trace|exception|regression|doesn't work|not working|failing|failed)\b", "intent:fix"), - # explore — reading / understanding - (r"\b(explain|explaining|explained|understand|understanding|walk me through|how does|how do|what does|what is|where is|show me|why is|why does|read|reading|review|reviewing|reviewed|look at|trace)\b", "intent:explore"), - # refactor — restructuring without behavior change - (r"\b(refactor|refactoring|refactored|clean up|cleanup|cleaning up|simplify|simplifying|simplified|restructure|restructuring|reorganize|reorganizing|rename|renaming|extract|extracting|inline|consolidate|dedup|deduplicate)\b", "intent:refactor"), - # test — writing or running tests - (r"\b(test|tests|testing|tested|unit test|integration test|pytest|jest|vitest|mocha|jasmine|rspec|assert|asserts|asserting|mock|mocking|mocked|spec|specs|coverage|tdd)\b", "intent:test"), - # ops — deployment, config, infra - # NOTE: `.env` is matched as a separate alternative with lookarounds because - # word-boundary (\b) can't anchor a pattern starting with a non-word char (.). - (r"(?:\b(?:deploy|deploying|deployed|deployment|ci/cd|ci\b|cd\b|github actions|gitlab ci|jenkins|docker|dockerfile|kubernetes|k8s|terraform|ansible|helm|env var|environment variable|nginx|caddy|systemd|pm2)\b|(? set[str]: """Return the set of intent:* tags that match the given text. A session can legitimately have multiple intents (e.g. the user started - building something, hit an error, and debugged it). We return all matches. + building something, hit an error, and debugged it). We return all + matches. Delegates to the canonical multi-label classifier and + re-applies the ``intent:`` storage prefix, so tags, the recommender, + and the benchmark never drift on what a "fix" is. """ - matches: set[str] = set() if not combined_text: - return matches - for pattern, intent in INTENT_PATTERNS: - if re.search(pattern, combined_text, re.IGNORECASE): - matches.add(intent) - return matches + return set() + return { + f"intent:{label}" + for label in task_classifier.classify_intents(combined_text) + } def auto_tag_session(self, session_id: str, messages: list[dict]) -> list[str]: """Auto-detect tags for a session from its messages. diff --git a/stackunderflow/services/task_classifier.py b/stackunderflow/services/task_classifier.py new file mode 100644 index 00000000..f0d99335 --- /dev/null +++ b/stackunderflow/services/task_classifier.py @@ -0,0 +1,258 @@ +"""Canonical task classifier — one source of truth for intent / size / language. + +Move 0 of the comparative benchmark engine (issue #99, spec 26 §8). Before +this module three surfaces each classified a session's *task* their own way: + +* ``services/tag_service.py`` — regex patterns, **6** intent labels + (``build/fix/explore/refactor/test/ops``), multi-label (a session can carry + several intents), emitted as ``intent: