From 6978eef9db5c24386cb5b905c6ce84de64551a3c Mon Sep 17 00:00:00 2001 From: Shaun Smith <51304449+smith-xyz@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:11:18 -0400 Subject: [PATCH] refactor(metrics): delete the metrics journal and model registry (XWING-2440) traust retired `metrics history`, `metrics spend` and the journal row behind `registry models spend` (now a traust-core spend event). Nothing in ops/, sweep/ or portfolio/ reads metrics/ or registry/, so both go with the facade's .metrics and .models; the validation benchmark's journal append (a no-op since history.py lost its __main__) goes too. --- .ratchet.json | 12 - CODEOWNERS | 3 - README.md | 5 +- pyproject.toml | 1 - src/traust_engine/__init__.py | 6 +- src/traust_engine/_ops_base.py | 2 +- src/traust_engine/assets.py | 2 +- src/traust_engine/engine.py | 19 +- src/traust_engine/locations.py | 10 +- src/traust_engine/metrics/__init__.py | 5 - src/traust_engine/metrics/attribute_spend.py | 406 -------------- src/traust_engine/metrics/collect_spend.py | 171 ------ src/traust_engine/metrics/history.py | 523 ------------------- src/traust_engine/metrics/ops.py | 120 ----- src/traust_engine/metrics/spend.py | 389 -------------- src/traust_engine/registry/__init__.py | 5 - src/traust_engine/registry/models.py | 181 ------- src/traust_engine/registry/ops.py | 82 --- src/traust_engine/storage.py | 2 +- src/traust_engine/sweep/benchmark.py | 30 +- tests/test_attribute_session_spend.py | 342 ------------ tests/test_build_spend_dashboard.py | 134 ----- tests/test_metrics_history.py | 75 --- tests/test_model_pricing.py | 156 ------ tests/test_model_registry.py | 144 ----- tests/test_traust_engine.py | 45 +- 26 files changed, 15 insertions(+), 2855 deletions(-) delete mode 100644 src/traust_engine/metrics/__init__.py delete mode 100644 src/traust_engine/metrics/attribute_spend.py delete mode 100644 src/traust_engine/metrics/collect_spend.py delete mode 100644 src/traust_engine/metrics/history.py delete mode 100644 src/traust_engine/metrics/ops.py delete mode 100644 src/traust_engine/metrics/spend.py delete mode 100644 src/traust_engine/registry/__init__.py delete mode 100644 src/traust_engine/registry/models.py delete mode 100644 src/traust_engine/registry/ops.py delete mode 100644 tests/test_attribute_session_spend.py delete mode 100644 tests/test_build_spend_dashboard.py delete mode 100644 tests/test_metrics_history.py delete mode 100644 tests/test_model_pricing.py delete mode 100644 tests/test_model_registry.py diff --git a/.ratchet.json b/.ratchet.json index 110de62..6fab074 100644 --- a/.ratchet.json +++ b/.ratchet.json @@ -25,12 +25,6 @@ "contracts": 1, "state-file": 4 }, - "src/traust_engine/metrics/history.py": { - "contracts": 2 - }, - "src/traust_engine/metrics/spend.py": { - "contracts": 1 - }, "src/traust_engine/ops/admission.py": { "core-internal": 2 }, @@ -46,12 +40,6 @@ "src/traust_engine/portfolio/graph.py": { "database": 1 }, - "src/traust_engine/registry/models.py": { - "contracts": 1 - }, - "src/traust_engine/registry/ops.py": { - "contracts": 1 - }, "src/traust_engine/sweep/engine.py": { "contracts": 1 }, diff --git a/CODEOWNERS b/CODEOWNERS index 282113b..7adac9e 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -10,7 +10,4 @@ # Disposition ledger and corpus resolution /src/traust_engine/corpus/ @traust-security/traust-admins -# Metrics journal -/src/traust_engine/metrics/history.py @traust-security/traust-admins - * @traust-security/traust-admins diff --git a/README.md b/README.md index 4dffb5a..46f8c19 100644 --- a/README.md +++ b/README.md @@ -15,10 +15,9 @@ tagged `PENDING(decision: …)`); deleted when traust retires or ports it: | Module | Kept for | |---|---| -| `traust_engine.engine` | `HarnessEngine` facade: `.metrics`, `.models`, `.adapters`, `.sweep`, `.portfolio` | +| `traust_engine.engine` | `HarnessEngine` facade: `.adapters`, `.sweep`, `.portfolio` | | `traust_engine.sweep` | sweep stages: collect, draft, sweep, emit, benchmark, rule lane, mining, calibration | | `traust_engine.portfolio` | portfolio graph layers | -| `traust_engine.metrics`, `traust_engine.registry` | metrics journal and spend pricing | | `traust_engine.adapters` | opengrep (sweep stages) and checkov (portfolio graph) | | `traust_engine.corpus` | the report population the sweep stages read | @@ -26,7 +25,7 @@ The opengrep rule pack ships under `data/secure-code-audit/opengrep-rules/`. Thi config itself — it is a pure consumer.** A caller builds the context once (`traust_contracts.load_context()`, done for you by `HarnessEngine.load(config_home=…)`) and the engine receives it *injected*: the -ops namespaces (`h.metrics`, `h.adapters`, …) hold the frozen `HarnessContext` and +ops namespaces (`h.sweep`, `h.adapters`, …) hold the frozen `HarnessContext` and derive paths from it; stateless functions take already-resolved arguments. No environment overrides, no cwd or workspace-shaped guessing; a required-but-absent file fails closed with `DeploymentConfigMissing` at the entry point. The config diff --git a/pyproject.toml b/pyproject.toml index ace4049..4eb895c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,4 +63,3 @@ ignore = ["RUF001", "RUF002", "RUF003"] [tool.ruff.lint.per-file-ignores] "tests/*" = ["S101", "E741", "E402", "E501", "RUF012"] -"src/traust_engine/metrics/history.py" = ["E501"] diff --git a/src/traust_engine/__init__.py b/src/traust_engine/__init__.py index 5c8b76e..a899e12 100644 --- a/src/traust_engine/__init__.py +++ b/src/traust_engine/__init__.py @@ -4,12 +4,10 @@ ops — continuous ops: pack spec, controller (tick), admission, reaper Kept for traust until its PENDING decisions (sweep stages, portfolio graph -layers, metrics journal) retire or port them: - engine — the ``HarnessEngine`` facade (metrics, models, adapters, sweep, portfolio) +layers) retire or port them: + engine — the ``HarnessEngine`` facade (adapters, sweep, portfolio) sweep — class-generalization scan engine portfolio — source-code dependency graph analysis - metrics — hash-chained metrics recording + spend tracking - registry — model resolution and spend pricing adapters — opengrep (sweep stages) and checkov (portfolio graph) corpus — population resolver the sweep stages read """ diff --git a/src/traust_engine/_ops_base.py b/src/traust_engine/_ops_base.py index e41a83e..fbc2385 100644 --- a/src/traust_engine/_ops_base.py +++ b/src/traust_engine/_ops_base.py @@ -4,7 +4,7 @@ needs. The resolution logic itself lives in :mod:`traust_engine.locations`; this just binds it to ``self._ctx.locations`` so no namespace re-derives it. -PENDING(decision: sweep stages | portfolio graph layers | metrics journal) — kept for traust; +PENDING(decision: sweep stages | portfolio graph layers) — kept for traust; deleted when traust retires or ports it. """ diff --git a/src/traust_engine/assets.py b/src/traust_engine/assets.py index 48b8d70..07f388f 100644 --- a/src/traust_engine/assets.py +++ b/src/traust_engine/assets.py @@ -9,7 +9,7 @@ arithmetic or a repo-layout assumption — the latter breaks the moment the package is installed as a wheel (the original HARNESS_ROOT bug). -PENDING(decision: sweep stages | metrics journal) — kept for traust; +PENDING(decision: sweep stages) — kept for traust; deleted when traust retires or ports it. """ diff --git a/src/traust_engine/engine.py b/src/traust_engine/engine.py index 51ded2b..8f07db6 100644 --- a/src/traust_engine/engine.py +++ b/src/traust_engine/engine.py @@ -8,13 +8,12 @@ from traust_engine import HarnessEngine h = HarnessEngine(ctx) # explicit context (tests, app) - h.metrics.append("census", {"repos": 812}) - ok, issues = h.metrics.verify() + h.sweep.collect(...) -Namespaces mirror the engine's submodules: `.metrics`, `.models`, `.adapters` (the rule-lane -stage's opengrep config), `.sweep`, `.portfolio`. +Namespaces mirror the engine's submodules: `.adapters` (the rule-lane stage's opengrep +config), `.sweep`, `.portfolio`. -PENDING(decision: sweep stages | portfolio graph layers | metrics journal) — kept for traust; +PENDING(decision: sweep stages | portfolio graph layers) — kept for traust; deleted when traust retires or ports it. """ @@ -26,9 +25,7 @@ from traust_contracts import HarnessContext, load_context from traust_engine.adapters.ops import AdaptersOps -from traust_engine.metrics.ops import MetricsOps from traust_engine.portfolio.ops import PortfolioOps -from traust_engine.registry.ops import ModelsOps from traust_engine.sweep.ops import SweepOps @@ -45,14 +42,6 @@ def load(cls, *, config_home: Path | None = None) -> HarnessEngine: def ctx(self) -> HarnessContext: return self._ctx - @cached_property - def metrics(self) -> MetricsOps: - return MetricsOps(self._ctx) - - @cached_property - def models(self) -> ModelsOps: - return ModelsOps(self._ctx) - @cached_property def adapters(self) -> AdaptersOps: return AdaptersOps(self._ctx) diff --git a/src/traust_engine/locations.py b/src/traust_engine/locations.py index d3ca445..5bb99e7 100644 --- a/src/traust_engine/locations.py +++ b/src/traust_engine/locations.py @@ -11,7 +11,7 @@ below the ops layer. The resolution logic lives here, once; ops pass ``self._loc`` and never re-derive it. -PENDING(decision: sweep stages | portfolio graph layers | metrics journal) — kept for traust; +PENDING(decision: sweep stages | portfolio graph layers) — kept for traust; deleted when traust retires or ports it. """ @@ -141,8 +141,6 @@ def sarif_tool_uri(loc: Locations | None) -> str | None: PHASE0_REL = SCAN_TESTING_REL / "phase-0" SWEEPS_REL = SCAN_TESTING_REL / "sweeps" VALIDATION_BENCHMARK_REL = SCAN_TESTING_REL / "validation-benchmark" -#: The metrics-history journal lives under the configured metrics root, not results. -METRICS_HISTORY_REL = Path("metrics") / "metrics-history.jsonl" #: Sweep-engine and regression-rule draft staging under the metrics root. RULE_DRAFTS_REL = Path("metrics") / "rule-mining" / "rule-drafts" @@ -197,9 +195,3 @@ def scan_testing_dir(loc: Locations | None) -> Path | None: def benchmark_dir(loc: Locations | None) -> Path | None: """``/scan-testing/validation-benchmark``, or None.""" return _under_results(VALIDATION_BENCHMARK_REL, loc) - - -def metrics_history(loc: Locations | None) -> Path | None: - """``/metrics/metrics-history.jsonl``, or None.""" - pt = progress_tracker_dir(loc) - return pt / METRICS_HISTORY_REL if pt else None diff --git a/src/traust_engine/metrics/__init__.py b/src/traust_engine/metrics/__init__.py deleted file mode 100644 index 3895f88..0000000 --- a/src/traust_engine/metrics/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Hash-chained metrics recording + spend tracking. - -PENDING(decision: metrics journal) — kept for traust; -deleted when traust retires or ports it. -""" diff --git a/src/traust_engine/metrics/attribute_spend.py b/src/traust_engine/metrics/attribute_spend.py deleted file mode 100644 index 6da4522..0000000 --- a/src/traust_engine/metrics/attribute_spend.py +++ /dev/null @@ -1,406 +0,0 @@ -"""Per-lane spend attribution — Phase 0 of the budget-guard widening plan. - -WHY THIS EXISTS. Spend is tracked from two sources (docs/model-routing.md): -drivers *declare* per-skill spend via `model_registry.py spend`, and -`collect_session_spend.py` records session *actuals* from Claude Code -transcripts. In practice the declared side is often empty: most ledger spend -rows carry tokens_in: 0, tokens_out: 0, so the spend dashboard's per-skill -table reports $0.00 for governed lanes while the actuals side shows a large -monthly total with no per-lane breakdown. - -That is NOT a bug in the emitters. It is the contract working as -written: 17 SKILL.md files instruct the agent to run -`--tokens-in --tokens-out `, and **an agent cannot -observe its own token usage mid-run** — there is no such API — so every -agent takes the `0` the contract offers. Asking the emitter to try -harder cannot fix it; the number does not exist on that side of the -boundary. - -WHAT THIS DOES INSTEAD. The transcripts hold both halves already: real -per-message token usage AND evidence of which skill the session was -running. This collector walks each session in order, tracks which skill -is currently active, and attributes each message's usage to it — -turning the one trustworthy total into a per-lane breakdown without -asking any agent to self-report. - -Attribution is INFERRED, so it is reported as such. A session's usage -before any skill signal, or in a session that never invokes one, lands -in an explicit `unattributed` bucket that is always printed. The -reconcile view proves the split is lossless against -collect_session_spend.py's own totals — an attribution that quietly -dropped tokens would be worse than no attribution at all. - -Signals, strongest first (all must name a skill that exists in -skills/, so unrelated prose cannot mint a lane): - 1. a `Skill` tool call -> input.skill - 2. a `` block -> the slash command's name - 3. a read of //SKILL.md (path segment `skills` or legacy `harnessing`) - -Usage: - traust metrics attribute-spend # month to date - traust metrics attribute-spend --month 2026-07 - traust metrics attribute-spend --date 2026-07-27 - traust metrics attribute-spend --reconcile # lossless proof - traust metrics attribute-spend --json -""" - -from __future__ import annotations - -import datetime -import json -import re -from collections import defaultdict -from pathlib import Path - -from traust_engine.metrics import collect_spend as css -from traust_engine.metrics import history as metrics_ledger - -TOKEN_KEYS = css.TOKEN_KEYS -UNATTRIBUTED = "unattributed" - -_CMD_RX = re.compile(r"\s*/?([A-Za-z0-9_-]+)\s*") -# The optional leading segment is a workflow stage directory -# (skills/4-triage/triage/SKILL.md or legacy harnessing/...); the capture is -# always the skill itself. Both spellings must match — older transcripts -# recorded the flat path and are never rewritten. -_SKILL_MD_RX = re.compile(r"(?:harnessing|skills)/(?:[A-Za-z0-9_-]+/)?([A-Za-z0-9_-]+)/SKILL\.md") -# Cheap pre-filter: only these lines can carry a signal or usage, so the -# expensive json.loads is skipped for the bulk of a transcript. -_INTERESTING = ('"usage"', "command-name", "SKILL.md", '"Skill"') - - -def known_skills(harnessing_dir: Path | None = None) -> set[str]: - """Skill names that actually exist. Gating on this keeps a prose - mention of some other tool from inventing a lane.""" - raw = harnessing_dir - if raw is None or not raw.is_dir(): - return set() - # Two levels: skills sit at the skills/ root or under a stage - # directory. Descend only into directories that are not themselves - # skills, so a SKILL.md bundled inside a skill is not counted as one. - found = set() - for child in raw.iterdir(): - if not child.is_dir(): - continue - if (child / "SKILL.md").is_file(): - found.add(child.name) - continue - found.update(g.name for g in child.iterdir() if g.is_dir() and (g / "SKILL.md").is_file()) - return found - - -def _iter_content(msg: dict): - content = msg.get("content") - if isinstance(content, list): - for block in content: - if isinstance(block, dict): - yield block - - -def detect_skill(rec: dict, valid: set[str]) -> str | None: - """Strongest available skill signal in one transcript record.""" - msg = rec.get("message") or {} - # 1. an explicit Skill tool call - for block in _iter_content(msg): - if block.get("type") == "tool_use" and block.get("name") == "Skill": - name = (block.get("input") or {}).get("skill") - if isinstance(name, str): - name = name.split(":")[-1].strip() - if name in valid: - return name - # 2/3. a slash command, or a read of the skill prompt. Both can sit - # in user text or in tool input, so match over the record's text - # form rather than guessing which field carries it. - raw = json.dumps(rec) - m = _CMD_RX.search(raw) - if m and m.group(1) in valid: - return m.group(1) - m = _SKILL_MD_RX.search(raw) - if m and m.group(1) in valid: - return m.group(1) - return None - - -def _blank() -> dict: - return {k: 0 for k in TOKEN_KEYS} | {"messages": 0, "sessions": set()} - - -def collect_attributed( - dirs: list[Path], - day: str | None = None, - month: str | None = None, - valid: set[str] | None = None, - thresholds: dict[str, int] | None = None, -) -> dict: - """-> {date: {skill: {model: {tokens..., messages, sessions[, long_prompt]}}}}. - - ``thresholds``: model -> long-prompt threshold; see collect_spend.add_usage. - - Walks each session IN ORDER so a mid-session skill invocation - attributes the messages that follow it, not the whole file. Usage - before any signal is `unattributed` rather than guessed at. - """ - valid = known_skills() if valid is None else valid - agg: dict = defaultdict(lambda: defaultdict(lambda: defaultdict(_blank))) - for d in dirs: - for f in sorted(d.glob("*.jsonl")): - sid = f.stem - current = UNATTRIBUTED - try: - fh = f.open(encoding="utf-8", errors="replace") - except OSError: - continue - with fh: - for line in fh: - if not any(tok in line for tok in _INTERESTING): - continue - try: - rec = json.loads(line) - except json.JSONDecodeError: - continue - found = detect_skill(rec, valid) - if found: - current = found - msg = rec.get("message") or {} - usage = msg.get("usage") - model = msg.get("model") - if not usage or not model or model.startswith("<"): - continue - ts = (rec.get("timestamp") or "")[:10] - if not ts: - continue - if day and ts != day: - continue - if month and not ts.startswith(month): - continue - a = agg[ts][current][model] - css.add_usage(a, usage, (thresholds or {}).get(model)) - a["messages"] += 1 - a["sessions"].add(sid) - return agg - - -def totals_by_skill(agg: dict) -> dict: - """Collapse dates/models -> {skill: {tokens..., messages}}.""" - out: dict = defaultdict(lambda: {k: 0 for k in TOKEN_KEYS} | {"messages": 0, "sessions": set()}) - for _date, skills in agg.items(): - for skill, models in skills.items(): - for _model, a in models.items(): - t = out[skill] - for k in TOKEN_KEYS: - t[k] += a[k] - t["messages"] += a["messages"] - t["sessions"] |= a["sessions"] - return out - - -def grand_total(agg: dict) -> dict: - tot = {k: 0 for k in TOKEN_KEYS} | {"messages": 0} - for skills in agg.values(): - for models in skills.values(): - for a in models.values(): - for k in TOKEN_KEYS: - tot[k] += a[k] - tot["messages"] += a["messages"] - return tot - - -def reconcile(dirs: list[Path], day: str | None, month: str | None) -> dict: - """Prove the attribution split is lossless against the existing - collector. Any mismatch means tokens were dropped on the floor — - reported as a failure, never rounded away.""" - attributed = grand_total(collect_attributed(dirs, day, month)) - baseline = {k: 0 for k in TOKEN_KEYS} | {"messages": 0} - for date, models in css.collect(dirs, day).items(): - if month and not date.startswith(month): - continue - for a in models.values(): - for k in TOKEN_KEYS: - baseline[k] += a[k] - baseline["messages"] += a["messages"] - deltas = {k: attributed[k] - baseline[k] for k in [*list(TOKEN_KEYS), "messages"]} - return { - "attributed": attributed, - "baseline": baseline, - "deltas": deltas, - "lossless": all(v == 0 for v in deltas.values()), - } - - -LEDGER_SOURCE_PREFIX = "spend-attribution:" - - -def already_recorded(ws: Path, journal: Path | None = None) -> set[tuple[str, str, str]]: - """(date, skill, model) triples already in the ledger.""" - done = set() - for r in metrics_ledger.rows(journal): - src = str(r.get("source") or "") - if not src.startswith(LEDGER_SOURCE_PREFIX): - continue - m = r.get("metrics") or {} - if m.get("date") and m.get("skill") and m.get("model"): - done.add((m["date"], m["skill"], m["model"])) - return done - - -def append_rows(ws: Path, agg: dict, reg=None, journal: Path | None = None) -> int: - """Append per (date, skill, model) rows to the hash-chained metrics - ledger under `spend-attribution:`. A DELIBERATELY separate - namespace from `model-spend:`: those are the driver-declared - routing markers whose token fields are best-effort and usually zero, - and mixing the two is what let a $0.00 table look authoritative. - - Idempotent, and only COMPLETED days are recorded — appending a - partial day would freeze that day's undercount permanently (same - rule as collect_session_spend.py).""" - from traust_engine.registry import models as model_registry - - if reg is None: - raise ValueError( - "reg is required — pass model_registry from HarnessEngine.models.registry()" - ) - today = datetime.date.today().isoformat() - done = already_recorded(ws, journal=journal) - appended = 0 - for date in sorted(agg): - if date >= today: - continue - for skill, models in sorted(agg[date].items()): - for model, a in sorted(models.items()): - if (date, skill, model) in done: - continue - try: - usd = model_registry.cost_usd( - reg, - model, - a["input_tokens"], - a["output_tokens"], - a["cache_read_input_tokens"], - a["cache_creation_input_tokens"], - a.get("long_prompt"), - ) - except Exception: - usd = None - metrics_ledger.append( - f"{LEDGER_SOURCE_PREFIX}{skill}", - { - "date": date, - "skill": skill, - "model": model, - "tokens_in": a["input_tokens"], - "tokens_out": a["output_tokens"], - "cache_read": a["cache_read_input_tokens"], - "cache_creation": a["cache_creation_input_tokens"], - **({"long_prompt": dict(a["long_prompt"])} if a.get("long_prompt") else {}), - "messages": a["messages"], - "sessions": len(a["sessions"]), - "cost_usd": usd, - "attribution": "inferred-from-transcript", - }, - journal=journal, - ) - appended += 1 - return appended - - -def _fmt(n: int) -> str: - return f"{n:,}" - - -def cost_by_skill(agg: dict, reg=None) -> tuple[dict, set]: - """{skill: usd} at registry list prices, plus the set of models with - no price. Priced per (skill, model) because rates differ by model; - unpriced models are NAMED, never silently counted as $0 — that - rounding is how the declared side came to read $0.00 everywhere.""" - try: - from traust_engine.registry import models as model_registry - - if reg is None: - return {}, set() - except Exception: - return {}, set() - out: dict = defaultdict(float) - unpriced: set = set() - for skills in agg.values(): - for skill, models in skills.items(): - for model, a in models.items(): - try: - usd = model_registry.cost_usd( - reg, - model, - a["input_tokens"], - a["output_tokens"], - a["cache_read_input_tokens"], - a["cache_creation_input_tokens"], - a.get("long_prompt"), - ) - except Exception: - usd = None - if usd is None: - unpriced.add(model) - continue - out[skill] += usd - return dict(out), unpriced - - -def render(agg: dict, month_label: str, reg=None) -> list[str]: - by_skill = totals_by_skill(agg) - costs, unpriced = cost_by_skill(agg, reg=reg) - tot_usd = sum(costs.values()) - tot_out = sum(v["output_tokens"] for v in by_skill.values()) or 1 - lines = [ - f"# Per-lane spend attribution — {month_label}", - "", - "_Inferred from session transcripts (which skill the session " - "was running), not self-reported by agents — the declared " - "side records 0 tokens by contract. `unattributed` is " - "always shown._", - "", - "| Lane | Est. USD | Share | Output tokens | Cache read | Messages | Sessions |", - "|---|---:|---:|---:|---:|---:|---:|", - ] - ordered = sorted( - by_skill.items(), key=lambda kv: -costs.get(kv[0], kv[1]["output_tokens"] / 1e9) - ) - for skill, a in ordered: - usd = costs.get(skill) - share = f"{usd / tot_usd * 100:.1f}%" if usd is not None and tot_usd else "—" - lines.append( - f"| {skill} | {f'{usd:,.2f}' if usd is not None else 'n/a'} | " - f"{share} | {_fmt(a['output_tokens'])} | " - f"{_fmt(a['cache_read_input_tokens'])} | " - f"{_fmt(a['messages'])} | {len(a['sessions'])} |" - ) - lines += ["", f"**Total priced: ${tot_usd:,.2f}**"] - lane_usd = sum(v for k, v in costs.items() if k != UNATTRIBUTED) - if tot_usd: - lines += [ - f"**Harness lanes: ${lane_usd:,.2f} " - f"({lane_usd / tot_usd * 100:.1f}%)** · " - f"unattributed ${costs.get(UNATTRIBUTED, 0):,.2f} " - f"({costs.get(UNATTRIBUTED, 0) / tot_usd * 100:.1f}%)" - ] - else: - # Nothing priceable (all models unpriced). The lane vs - # unattributed split is the headline of this report, so fall - # back to output tokens rather than dropping the line. - lane_tok = sum(v["output_tokens"] for k, v in by_skill.items() if k != UNATTRIBUTED) - un_tok = by_skill.get(UNATTRIBUTED, {}).get("output_tokens", 0) - lines += [ - f"**Harness lanes: {_fmt(lane_tok)} output tokens " - f"({lane_tok / tot_out * 100:.1f}%)** · unattributed " - f"{_fmt(un_tok)} ({un_tok / tot_out * 100:.1f}%) — " - f"no USD: no model in this window has a registry price" - ] - if unpriced: - lines += [ - "", - f"⚠ unpriced models excluded from the USD column " - f"(no registry price): {', '.join(sorted(unpriced))}", - ] - lines += [ - "", - f"_Share of output tokens is a cross-check on the USD " - f"split; total output {_fmt(tot_out)}._", - ] - return lines diff --git a/src/traust_engine/metrics/collect_spend.py b/src/traust_engine/metrics/collect_spend.py deleted file mode 100644 index 69c9a7d..0000000 --- a/src/traust_engine/metrics/collect_spend.py +++ /dev/null @@ -1,171 +0,0 @@ -"""Session-spend collector — the ACTUALS side of spend tracking. - -Claude Code writes per-message token usage into its session transcripts -(~/.claude/projects//*.jsonl → message.usage with model, -input/output/cache_read/cache_creation token counts). This collector -aggregates them so the harness can SEE its own spend: - - live view (default) today's running totals per model — the real-time - answer to "what are we burning right now" - --date YYYY-MM-DD one day's totals - --append append per-day×model rows to the hash-chained - metrics ledger (source `model-spend:sessions`), - idempotent: days already recorded are skipped - -PLANNED (operator request 2026-07-29): when the central orchestration -platform lands, teach this collector to also read centralized session -transcripts (a --transcripts-root pointing at the platform's transcript -store, same JSONL shape) so the spend-actuals leg stops being the one -dashboard input bound to operator workstations -(docs/continuous-scanning.md, standing-jobs dashboards row). - -Two-source design (docs/model-routing.md → "How spend is tracked"): -drivers *declare* per-skill spend via `model_registry.py spend`; -this collector records session *actuals*. The dashboard shows both — -drift between declared and actual is itself a signal. - -Cache-aware: cache_read tokens are tracked separately (they price ~an -order of magnitude below fresh input); cost math applies when the -registry carries prices. -""" - -from __future__ import annotations - -import datetime -import json -from collections import defaultdict -from pathlib import Path - -from traust_engine.metrics import history as metrics_ledger - -PROJECTS = Path.home() / ".claude" / "projects" -TOKEN_KEYS = ( - "input_tokens", - "output_tokens", - "cache_read_input_tokens", - "cache_creation_input_tokens", -) - - -def workspace_slugs(workspace: Path) -> list[Path]: - """Project dirs whose slug encodes a path under the workspace.""" - prefix = str(workspace).replace("/", "-") - if not PROJECTS.is_dir(): - return [] - return [d for d in PROJECTS.iterdir() if d.is_dir() and d.name.startswith(prefix)] - - -#: Usage fields that make up a request's prompt, for long-prompt pricing. -PROMPT_KEYS = ("input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens") -#: usage field -> the metrics-row name used in a long_prompt block. -_ROW_NAMES = { - "input_tokens": "tokens_in", - "output_tokens": "tokens_out", - "cache_read_input_tokens": "cache_read", - "cache_creation_input_tokens": "cache_creation", -} - - -def add_usage(a: dict, usage: dict, threshold: int | None = None) -> None: - """Add one request's usage to a bucket. - - When the model has a long-prompt threshold and this request's prompt - (input, cache reads and cache writes) is over it, the request's tokens - are also added to ``a["long_prompt"]``, so cost_usd can price them at - the long-prompt rates. Each request is judged on its own. - """ - for k in TOKEN_KEYS: - a[k] += usage.get(k) or 0 - if threshold is not None and sum(usage.get(k) or 0 for k in PROMPT_KEYS) > threshold: - lp = a.setdefault("long_prompt", {name: 0 for name in _ROW_NAMES.values()}) - for k, name in _ROW_NAMES.items(): - lp[name] += usage.get(k) or 0 - - -def collect(dirs: list[Path], day: str | None, thresholds: dict[str, int] | None = None) -> dict: - """{date: {model: {tokens..., messages, sessions:set[, long_prompt]}}} - - ``thresholds`` maps a model to its long-prompt threshold - (registry.models.long_prompt_thresholds); requests over it are recorded - in the bucket's ``long_prompt`` block as well.""" - agg: dict = defaultdict( - lambda: defaultdict(lambda: {k: 0 for k in TOKEN_KEYS} | {"messages": 0, "sessions": set()}) - ) - for d in dirs: - for f in d.glob("*.jsonl"): - sid = f.stem - try: - fh = f.open(encoding="utf-8", errors="replace") - except OSError: - continue - with fh: - for line in fh: - if '"usage"' not in line: - continue - try: - rec = json.loads(line) - except json.JSONDecodeError: - continue - msg = rec.get("message") or {} - usage = msg.get("usage") - model = msg.get("model") - if not usage or not model or model.startswith("<"): - continue - ts = (rec.get("timestamp") or "")[:10] - if not ts or (day and ts != day): - continue - a = agg[ts][model] - add_usage(a, usage, (thresholds or {}).get(model)) - a["messages"] += 1 - a["sessions"].add(sid) - return agg - - -def _row(model: str, a: dict) -> dict: - row = { - "model": model, - "tokens_in": a["input_tokens"], - "tokens_out": a["output_tokens"], - "cache_read": a["cache_read_input_tokens"], - "cache_creation": a["cache_creation_input_tokens"], - "messages": a["messages"], - "sessions": len(a["sessions"]), - } - if a.get("long_prompt"): - row["long_prompt"] = dict(a["long_prompt"]) - return row - - -def already_recorded(ws: Path, journal: Path | None = None) -> set[tuple[str, str]]: - done = set() - for r in metrics_ledger.rows(journal): - if r.get("source") == "model-spend:sessions": - m = r.get("metrics", {}) - if m.get("date") and m.get("model"): - done.add((m["date"], m["model"])) - return done - - -def append_rows( - ws: Path, - dirs: list[Path], - journal: Path | None = None, - thresholds: dict[str, int] | None = None, -) -> int: - today = datetime.date.today().isoformat() - agg = collect(dirs, None, thresholds) - done = already_recorded(ws, journal=journal) - appended = 0 - for date in sorted(agg): - if date >= today: - continue - for model, a in sorted(agg[date].items()): - if (date, model) in done: - continue - metrics_ledger.append( - "model-spend:sessions", - {"date": date, **_row(model, a)}, - journal=journal, - ) - appended += 1 - return appended diff --git a/src/traust_engine/metrics/history.py b/src/traust_engine/metrics/history.py deleted file mode 100644 index e0448c2..0000000 --- a/src/traust_engine/metrics/history.py +++ /dev/null @@ -1,523 +0,0 @@ -"""Central append-only campaign metrics ledger, shared by every dashboard. - -This module used to be called "ledger" but it really isn't the ledger. Due to that confusion -renamed to history since its an additional tamper-evidence history tracker. - -One JSONL file — `metrics/metrics-history.jsonl` — holds one -immutable row per dashboard snapshot, namespaced by `source` (the dashboard -slug). Rows are hash-chained for tamper evidence: - - row_sha = sha256(canonical-json of the row minus row_sha) - prev_sha = row_sha of the previous row in the file ("" for genesis) - -`verify` recomputes every hash and walks the chain, so any edit, deletion, or -reordering of a committed row is detectable (git history provides the second, -independent record). Rows are NEVER revised: methodology changes appear as new -snapshots with a `note`, keeping every as-reported number forever. - -The journal location is config-owned (locations.progress_tracker); no function -takes a workspace — the operator dictates where it lives. - -Library API (dashboards import this module): - previous(source) -> latest row for a source, or None - append(source, metrics, note="", hv=None) -> row - rows() -> all rows - series(source, key) -> [(date, value), ...] - trend_line(cur, prev_row, keys) -> "Δ vs : findings 1,234 (↓ 56) · ..." - verify() -> (ok: bool, issues: [str]) - -CLI: - python3 -m traust_engine.metrics.history append \\ - --source SLUG --metrics-json FILE|- [--note TEXT] - python3 -m traust_engine.metrics.history latest --source SLUG - python3 -m traust_engine.metrics.history verify - python3 -m traust_engine.metrics.history render-exec - -`render-exec` writes the executive trends dashboard -(`metrics/Executive-Trends.{md,html}`) — the -leadership-facing view of every tracked series. -""" - -import datetime -import hashlib -import json -from pathlib import Path - -from traust_contracts import DeploymentConfigMissing - -from traust_engine.assets import harness_version as _engine_version - -try: - from traust_contracts.models import MetricsRecord - - HAS_CONTRACTS = True -except ImportError: - HAS_CONTRACTS = False - - -# Metric key -> human label, per source. Drives the executive dashboard and -# each dashboard's trend line; unknown keys still land in the ledger. -KEY_LABELS = { - "repos_audited": "Reports (repos/branches audited)", - "unique_repos": "Unique repositories", - "findings_total": "Findings", - "sev_critical": "Critical", - "sev_high": "High", - "unique_critical": "Unique Critical", - "unique_high": "Unique High", - "repos_with_critical": "Repos with >=1 Critical", - "repos_with_high": "Repos with >=1 High", - "credential_findings": "Credential findings (unique)", - "hardening_backlog": "Hardening backlog", - "dispositioned_repos": "Repos with disposition ledger", - "ledger_coverage_pct": "Ledger coverage %", - "resolved_findings": "Findings resolved", - "open_ledger_view": "Open findings (ledger replay)", - "mean_cvss_open": "Mean CVSS of open findings (0-10 — legacy)", - "owasp_risk_critical": "OWASP risk rating: Critical (open)", - "owasp_risk_high": "OWASP risk rating: High (open)", - "owasp_risk_medium": "OWASP risk rating: Medium (open)", - "owasp_risk_low": "OWASP risk rating: Low/Note (open)", - "owasp_risk_high_plus_pct": "OWASP High+Critical share of open (%)", - "risk_index": "Risk index (CVSS sum over open — unnormalized, legacy)", - "risk_index_combined": "Combined risk index (CVSS sum + hardening — unnormalized, legacy)", - "hardening_risk_index": "Hardening risk index (lambda-weighted CVSS sum — legacy)", - "confirmed_exploitable": "Confirmed exploitable (live)", - "refuted": "False positives refuted (live)", - "validation_reports": "Live-validation reports", - "checks_attempted": "Finding-checks attempted", - "fuzz_bugs": "Fuzz-confirmed bugs", - "threat_models": "Threat models", - "threats_open": "Open threats", - "team_packages": "Team packages delivered", - "jira_tickets": "Jira defects filed", - "total_loc": "Lines of code scanned", - "repos_with_loc": "Repos with LoC data", - "languages": "Languages", - "code_findings": "Source-code findings (patterns)", - "distinct_cwes": "Distinct CWEs", - "pattern_repos": "Repos with >=1 pattern", - "chains_confirmed": "Attack chains with confirmed step", - "threats_unmitigated": "Unmitigated threats", - "threats_partial": "Partially mitigated threats", - "quick_wins": "Threat quick wins available", - "pqc_repos_assessed": "PQC repos assessed (Phase 1 sweep)", - "pqc_sweep_coverage_pct": "PQC sweep coverage %", - "pqc_ready": "PQC ready (repos)", - "pqc_partial": "PQC partial (repos)", - "pqc_not_ready": "PQC not-ready (repos)", - "pqc_clock_items": "PQC 2030/2035 clock items (burndown)", - "pqc_hndl_repos": "PQC HNDL-priority repos", - "pqc_tagged_findings": "PQC-tagged findings (general audits)", - "pqc_regressions": "PQC regressions (verify-remediation)", - "pqc_2030_clock_items": "PQC 2030-clock items (urgent burndown)", - "pqc_hybrid_blockers": "PQC hybrid-TLS group/KEX blockers (burndown)", - "pqc_toolchain_quickwins": "PQC go-directive quick wins (burndown)", - "attack_techniques_covered": "ATT&CK techniques covered (any evidence)", - "attack_techniques_observed": "ATT&CK techniques observed (confirmed chains)", - "attack_techniques_modeled": "ATT&CK techniques modeled (threat models)", - "attack_tactics_covered": "ATT&CK tactics covered", -} - -# Metrics where DOWN is good (risk/exposure); everything else up-is-good or -# neutral. Used only for coloring/labeling, never for the numbers themselves. -DOWN_IS_GOOD = { - "findings_total", - "sev_critical", - "sev_high", - "unique_critical", - "unique_high", - "repos_with_critical", - "repos_with_high", - "credential_findings", - "open_ledger_view", - "mean_cvss_open", - "risk_index", - "risk_index_combined", - "hardening_risk_index", - "threats_open", - "owasp_risk_critical", - "owasp_risk_high", - "owasp_risk_medium", - "owasp_risk_low", - "owasp_risk_high_plus_pct", - "pqc_not_ready", - "pqc_clock_items", - "pqc_hndl_repos", - "pqc_regressions", - "pqc_2030_clock_items", - "pqc_hybrid_blockers", - "pqc_toolchain_quickwins", - "hardening_backlog", - "code_findings", - "threats_unmitigated", - "threats_partial", -} - - -def ledger_path(journal: Path | None = None) -> Path: - """The metrics-history journal path. - - An injected ``journal`` (passed by HarnessEngine from the loaded context) - wins; otherwise it resolves from the config-owned metrics journal root. The - operator dictates where this lives — never a workspace-relative assumption. - """ - if journal is None: - raise DeploymentConfigMissing( - "metrics-history journal not provided — load via HarnessEngine.metrics " - "(the journal is resolved from the injected context, not the environment)." - ) - return journal - - -def _canonical(row: dict) -> str: - return json.dumps( - {k: v for k, v in row.items() if k != "row_sha"}, sort_keys=True, separators=(",", ":") - ) - - -def _sha(row: dict) -> str: - return hashlib.sha256(_canonical(row).encode("utf-8")).hexdigest() - - -def rows(journal: Path | None = None) -> list[dict]: - p = ledger_path(journal) - out = [] - if not p.is_file(): - return out - for line in p.read_text(encoding="utf-8").splitlines(): - line = line.strip() - if not line: - continue - try: - out.append(json.loads(line)) - except json.JSONDecodeError: - continue - return out - - -def previous(source: str, before: str | None = None, journal: Path | None = None) -> dict | None: - """Latest row for a source (optionally strictly before an ISO timestamp).""" - cand = [ - r - for r in rows(journal) - if r.get("source") == source and (before is None or r.get("snapshot_at", "") < before) - ] - return cand[-1] if cand else None - - -def series(source: str, key: str, journal: Path | None = None) -> list[tuple[str, object]]: - return [ - (r["snapshot_at"][:10], r["metrics"].get(key)) - for r in rows(journal) - if r.get("source") == source and r.get("metrics", {}).get(key) is not None - ] - - -METRICS_SOURCE = "traust-engine-metrics" - - -def append( - source: str, - metrics: dict, - note: str = "", - hv: str | None = None, - journal: Path | None = None, -) -> dict: - """Append one immutable, hash-chained snapshot row.""" - existing = rows(journal) - prev_sha = existing[-1].get("row_sha", "") if existing else "" - row = { - "snapshot_at": datetime.datetime.now(datetime.UTC).isoformat(timespec="seconds"), - "source": source, - "harness_version": hv or _engine_version(), - "note": note or "", - "metrics": {k: v for k, v in metrics.items() if v is not None}, - "prev_sha": prev_sha, - } - row["row_sha"] = _sha(row) - p = ledger_path(journal) - p.parent.mkdir(parents=True, exist_ok=True) - with p.open("a", encoding="utf-8") as fh: - fh.write(json.dumps(row) + "\n") - return row - - -def append_if_changed( - source: str, - metrics: dict, - note: str = "", - hv: str | None = None, - journal: Path | None = None, -) -> dict | None: - """Append only when the metrics differ from the source's previous row - (or a note is supplied). Dashboards rebuild frequently and often - idempotently — identical re-runs must not spam the ledger.""" - prev = previous(source, journal=journal) - clean = {k: v for k, v in metrics.items() if v is not None} - if prev and not note and prev.get("metrics") == clean: - return None - return append(source, clean, note=note, hv=hv, journal=journal) - - -def verify(journal: Path | None = None) -> tuple[bool, list[str]]: - """Recompute every row hash and walk the chain.""" - issues = [] - prev = "" - for i, r in enumerate(rows(journal), 1): - if "row_sha" not in r: - issues.append(f"row {i}: missing row_sha (pre-chain legacy row)") - prev = r.get("row_sha", "") - continue - if _sha(r) != r["row_sha"]: - issues.append( - f"row {i} ({r.get('snapshot_at')}, " - f"{r.get('source')}): row_sha mismatch — row was " - f"altered after being written" - ) - if r.get("prev_sha", "") != prev: - issues.append( - f"row {i} ({r.get('snapshot_at')}, " - f"{r.get('source')}): chain break — prev_sha does " - f"not match preceding row (insertion/deletion/" - f"reordering)" - ) - prev = r["row_sha"] - return (not issues), issues - - -# --------------------------------------------------------------------------- -# trend rendering helpers (used by the dashboards) -# --------------------------------------------------------------------------- - - -def _fmt(v): - if isinstance(v, int): - return f"{v:,}" - if isinstance(v, float): - return f"{v:,.1f}" - return str(v) - - -def _to_num(v): - try: - return float(str(v).replace(",", "")) - except (TypeError, ValueError): - return None - - -def delta(key: str, cur, prev) -> str | None: - """'1,234 (↓ 56 · improving)' style cell, or None if not comparable.""" - c, p = _to_num(cur), _to_num(prev) - if c is None or p is None: - return None - d = c - p - if d == 0: - return "→ unchanged" - arrow = "↑" if d > 0 else "↓" - good = (d < 0) if key in DOWN_IS_GOOD else (d > 0) - tone = "improving" if good else "worsening" - mag = _fmt(int(abs(d)) if float(abs(d)).is_integer() else abs(d)) - return f"{arrow} {mag} ({tone})" - - -def trend_line(current: dict, prev_row: dict | None, keys: list[str]) -> str: - """One-line trend summary vs the previous snapshot, for embedding in a - dashboard header. Empty string when there is no prior snapshot.""" - if not prev_row: - return "" - parts = [] - for k in keys: - d = delta(k, current.get(k), prev_row.get("metrics", {}).get(k)) - if d: - parts.append(f"{KEY_LABELS.get(k, k)}: {d}") - if not parts: - return "" - return f"Trend vs {prev_row['snapshot_at'][:10]} — " + " · ".join(parts) - - -# --------------------------------------------------------------------------- -# executive trends dashboard -# --------------------------------------------------------------------------- - -_EXEC_SERIES = [ - # (source, key) pairs charted for leadership, in display order. - (METRICS_SOURCE, "findings_total"), - (METRICS_SOURCE, "sev_critical"), - (METRICS_SOURCE, "sev_high"), - (METRICS_SOURCE, "open_ledger_view"), - # OWASP Risk Rating Methodology (likelihood x impact, factors derived - # from each finding's CVSS vector) — % of open findings rated High or - # Critical risk. Replaces the earlier mean-CVSS and unnormalized - # CVSS-sum series, which stay in the ledger as history only. See - # docs/risk-rating-methodology.md. - (METRICS_SOURCE, "owasp_risk_high_plus_pct"), - (METRICS_SOURCE, "hardening_backlog"), - (METRICS_SOURCE, "ledger_coverage_pct"), - (METRICS_SOURCE, "resolved_findings"), - (METRICS_SOURCE, "confirmed_exploitable"), - (METRICS_SOURCE, "threats_open"), - ("executive-summary", "credential_findings"), - ("loc-dashboard", "total_loc"), - ("insecure-patterns", "code_findings"), - ("validation-fuzz", "checks_attempted"), - ("threat-register", "threats_unmitigated"), - ("threat-register", "quick_wins"), - # Coverage metrics (up is good — not in DOWN_IS_GOOD): fleet ATT&CK - # technique coverage from the deterministic /attack-coverage roll-up. - ("attack-coverage", "attack_techniques_observed"), - ("attack-coverage", "attack_techniques_modeled"), - # PQC Phase-1 sweep: coverage climbs (up-good), clock items burn down. - ("pqc-readiness", "pqc_sweep_coverage_pct"), - ("pqc-readiness", "pqc_clock_items"), - # PQC Phase-2 roll-up views: ready count climbs; the - # 2030-clock, not-ready, HNDL, blocker and quick-win series burn down. - ("pqc-readiness", "pqc_ready"), - ("pqc-readiness", "pqc_not_ready"), - ("pqc-readiness", "pqc_hndl_repos"), - ("pqc-readiness", "pqc_2030_clock_items"), - ("pqc-readiness", "pqc_hybrid_blockers"), - ("pqc-readiness", "pqc_toolchain_quickwins"), -] - - -def render_exec(journal: Path | None = None) -> tuple[Path, Path]: - all_rows = rows(journal) - bysrc: dict[str, list[dict]] = {} - for r in all_rows: - bysrc.setdefault(r.get("source", "?"), []).append(r) - - today = datetime.date.today().isoformat() - md = ["# Executive Trends", ""] - md.append( - f"**Generated:** {today} · from the append-only, hash-chained " - f"metrics ledger (`metrics/metrics-history.jsonl`, " - f"{len(all_rows)} snapshots across {len(bysrc)} sources). " - f"As-reported numbers are never revised — every movement below " - f"is a real change or a *noted* methodology change, and the " - f"chain is verifiable with " - f"`python3 -m traust_engine.metrics.history verify`." - ) - md.append("") - md.append("| Metric | Earliest | Latest | Trend | Snapshots |") - md.append("|---|---:|---:|---|---:|") - - chart_payload = [] - seen_keys = set() - for source, key in _EXEC_SERIES: - pts = [(d, v) for d, v in series(source, key, journal)] - if len(pts) == 0 or (source, key) in seen_keys: - continue - seen_keys.add((source, key)) - label = KEY_LABELS.get(key, key) - first_d, first_v = pts[0] - last_d, last_v = pts[-1] - tr = delta(key, last_v, first_v) or "—" - md.append( - f"| {label} | {_fmt(first_v)} ({first_d}) | " - f"{_fmt(last_v)} ({last_d}) | {tr} | {len(pts)} |" - ) - nums = [(d, _to_num(v)) for d, v in pts] - if all(n is not None for _, n in nums): - chart_payload.append( - { - "label": label, - "down_good": key in DOWN_IS_GOOD, - "points": [{"x": d, "y": n} for d, n in nums], - } - ) - - notes = [(r["snapshot_at"][:10], r.get("source"), r["note"]) for r in all_rows if r.get("note")] - if notes: - md += ["", "## Methodology / snapshot notes", ""] - md += [f"- **{d}** (`{s}`) — {n}" for d, s, n in notes] - - ok, issues = verify(journal) - md += [ - "", - "**Ledger integrity:** " - + ( - "✓ chain verified" - if ok - else f"⚠ {len(issues)} issue(s) — run `python3 -m traust_engine.metrics.history verify` for details" - ), - ] - - out_dir = ledger_path(journal).parent - md_path = out_dir / "Executive-Trends.md" - md_path.write_text("\n".join(md) + "\n", encoding="utf-8") - - charts_html = "".join( - f'

{c["label"]}' - f"{' lower is better' if c['down_good'] else ''}" - f'

' - for i, c in enumerate(chart_payload) - ) - html = f""" - -Executive Trends - - -

Executive Trends

-
Generated {today} · append-only hash-chained metrics ledger · -{len(all_rows)} snapshots · integrity {"✓ verified" if ok else "⚠ check failed"}
-
{charts_html} -
Source: metrics/metrics-history.jsonl — -as-reported snapshots, never revised; methodology changes carry notes. -Verify the chain: python3 -m traust_engine.metrics.history verify. -Rebuild: traust-engine-metrics.
-""" - html_path = out_dir / "Executive-Trends.html" - html_path.write_text(html, encoding="utf-8") - return md_path, html_path - - -def append_typed(record: MetricsRecord, journal: Path | None = None) -> dict: - """Append a typed metrics record to the ledger.""" - if not HAS_CONTRACTS: - raise ImportError("traust_contracts required") - return append( - record.source, - record.metrics, - note=record.note, - hv=record.harness_version or None, - journal=journal, - ) - - -def read_typed(journal: Path | None = None) -> list[MetricsRecord]: - """Read ledger and return typed records.""" - if not HAS_CONTRACTS: - raise ImportError("traust_contracts required") - return [ - MetricsRecord( - timestamp=r.get("snapshot_at", ""), - source=r.get("source", ""), - harness_version=r.get("harness_version", ""), - metrics=r.get("metrics", {}), - note=r.get("note", ""), - ) - for r in rows(journal) - ] - - -# --------------------------------------------------------------------------- -# CLI -# --------------------------------------------------------------------------- diff --git a/src/traust_engine/metrics/ops.py b/src/traust_engine/metrics/ops.py deleted file mode 100644 index 234f407..0000000 --- a/src/traust_engine/metrics/ops.py +++ /dev/null @@ -1,120 +0,0 @@ -"""``HarnessEngine.metrics`` — metrics journal and spend, bound to context.""" - -from __future__ import annotations - -from pathlib import Path - -from traust_engine._ops_base import ContextOps -from traust_engine.locations import METRICS_HISTORY_REL -from traust_engine.metrics import attribute_spend, collect_spend, history, spend - - -class MetricsOps(ContextOps): - """Metrics operations bound to one loaded context. - - Paths come from ``ctx.locations``; stateless functions in sibling modules - receive injected journal/registry/roots — the engine never self-fetches. - """ - - def _journal(self) -> Path: - return self._progress_tracker() / METRICS_HISTORY_REL - - def _model_registry(self): - return self._ctx.model_registry - - def workspace(self) -> Path: - return self._workspace() - - # --- metrics history ----------------------------------------------------- - - def append(self, source: str, metrics: dict, note: str = "", hv: str | None = None) -> dict: - return history.append(source, metrics, note=note, hv=hv, journal=self._journal()) - - def append_if_changed( - self, source: str, metrics: dict, note: str = "", hv: str | None = None - ) -> dict | None: - return history.append_if_changed(source, metrics, note=note, hv=hv, journal=self._journal()) - - def rows(self) -> list[dict]: - return history.rows(self._journal()) - - def previous(self, source: str, before: str | None = None) -> dict | None: - return history.previous(source, before=before, journal=self._journal()) - - def verify(self) -> tuple[bool, list[str]]: - return history.verify(self._journal()) - - def render(self) -> tuple[Path, Path]: - return history.render_exec(self._journal()) - - # --- spend dashboard ----------------------------------------------------- - - def spend_dashboard_dir(self) -> Path: - return self._progress_tracker() / "metrics" / "dashboards" / "spend" - - def build_spend_dashboard( - self, out_dir: Path | None = None, budget_path: Path | None = None, budget_section=None - ) -> Path: - return spend.build( - self._workspace(), - out_dir or self.spend_dashboard_dir(), - budget_path=budget_path, - journal=self._journal(), - reg=self._model_registry(), - budget_section=( - budget_section if budget_section is not None else self._ctx.budget_policy - ), - ) - - # --- session / attribution spend ----------------------------------------- - - def session_project_dirs(self, workspace: Path | None = None) -> list[Path]: - return collect_spend.workspace_slugs(workspace or self._workspace()) - - def _long_prompt_thresholds(self) -> dict[str, int]: - from traust_engine.registry.models import long_prompt_thresholds - - return long_prompt_thresholds(self._model_registry()) - - def collect_session_spend(self, dirs: list[Path], day: str | None = None) -> dict: - return collect_spend.collect(dirs, day, self._long_prompt_thresholds()) - - def append_session_spend(self, dirs: list[Path] | None = None) -> int: - project_dirs = dirs if dirs is not None else self.session_project_dirs() - return collect_spend.append_rows( - self._workspace(), - project_dirs, - journal=self._journal(), - thresholds=self._long_prompt_thresholds(), - ) - - def collect_attributed( - self, - dirs: list[Path], - day: str | None = None, - month: str | None = None, - valid: set[str] | None = None, - ) -> dict: - return attribute_spend.collect_attributed( - dirs, day, month, valid, self._long_prompt_thresholds() - ) - - def attribute_reconcile( - self, dirs: list[Path], day: str | None = None, month: str | None = None - ) -> dict: - return attribute_spend.reconcile(dirs, day, month) - - def append_attribution_rows(self, agg: dict, reg=None) -> int: - return attribute_spend.append_rows( - self._workspace(), - agg, - reg=reg if reg is not None else self._model_registry(), - journal=self._journal(), - ) - - def render_attribution(self, agg: dict, month_label: str, reg=None) -> list[str]: - return attribute_spend.render( - agg, - month_label, - reg=reg if reg is not None else self._model_registry(), - ) diff --git a/src/traust_engine/metrics/spend.py b/src/traust_engine/metrics/spend.py deleted file mode 100644 index 2314769..0000000 --- a/src/traust_engine/metrics/spend.py +++ /dev/null @@ -1,389 +0,0 @@ -"""Model-spend dashboard — multi-model-strategy-plan M1. - -Aggregates the hash-chained metrics ledger's `model-spend:` rows -(appended by `traust registry models spend`) into -`metrics/dashboards/spend/spend-dashboard.md`: -spend by skill x model, token totals, cost where the registry carries -prices, and the MTTA-per-dollar headline when MTTA rows exist. - -Deterministic full rebuild; rerun after any batch that records spend. -""" - -from __future__ import annotations - -import calendar -import datetime -from collections import defaultdict -from pathlib import Path - -import yaml -from traust_contracts import BudgetPolicy - -from traust_engine.metrics import history as metrics_ledger -from traust_engine.registry import models as model_registry - - -def load_monthly_budget( - path: Path | None = None, *, budget_section: BudgetPolicy | None = None -) -> dict | None: - """The monthly_budget block of the budget policy (visibility layer - only — this module tracks, it never gates). None when absent.""" - if path is not None: - try: - bp = (yaml.safe_load(path.read_text()) or {}).get("budget_policy") or {} - except (OSError, yaml.YAMLError): - return None - elif budget_section is not None: - bp = budget_section.budget_policy or {} - else: - return None - mb = bp.get("monthly_budget") or {} - if not {"usd_central", "usd_band_low", "usd_band_high"} <= set(mb): - return None - return mb - - -def monthly_budget_section( - session_rows: list[dict], reg: dict, budget: dict, today: datetime.date -) -> list[str]: - """Budget-vs-actual per calendar month from session-actuals rows. - Months are keyed on the row's usage date (metrics.date), priced at - registry list rates; models without a price are counted as coverage - gaps per month, never silently as $0.""" - months: dict[str, dict] = defaultdict(lambda: {"usd": 0.0, "days": set(), "unpriced": set()}) - for r in session_rows: - m = r["metrics"] - date = str(m.get("date") or "") - if len(date) < 7: - continue - mo = months[date[:7]] - c = model_registry.cost_usd( - reg, - m.get("model", ""), - int(m.get("tokens_in") or 0), - int(m.get("tokens_out") or 0), - int(m.get("cache_read") or 0), - int(m.get("cache_creation") or 0), - long_prompt=m.get("long_prompt"), - ) - if c is None: - mo["unpriced"].add(m.get("model", "?")) - else: - mo["usd"] += c - mo["days"].add(date) - if not months: - return [] - - low, high = budget["usd_band_low"], budget["usd_band_high"] - central = budget["usd_central"] - cur = today.strftime("%Y-%m") - L = [ - "## Monthly budget tracking", - "", - f"_Plan: **${central:,}/month central** (band ${low:,}–${high:,})," - f" list prices — `config/budget-policy.yaml` `monthly_budget`" - f" (example policy; set your own monthly band). Tracking only; nothing" - f" is gated._", - "", - "_Caveats: actuals are list-priced from the registry at " - "CURRENT rates, recomputed from stored token counts — a rate " - "correction therefore restates history when registry rates change. " - "Models with no registry price are named per month, " - "never counted as $0. Months dominated by one-time experiment " - "bursts are expected to breach the band — the plan prices " - "steady-state scanning only._", - "", - "| Month | Actual (USD) | vs band | Status | Coverage |", - "|---|---:|---|---|---|", - ] - for month in sorted(months, reverse=True): - mo = months[month] - usd = mo["usd"] - if month == cur: - days_in = calendar.monthrange(today.year, today.month)[1] - rate = usd / today.day * days_in if today.day else 0.0 - if rate > high: - status = f"⚠ projecting ${rate:,.0f} — ABOVE band" - elif rate < low: - status = f"✓ projecting ${rate:,.0f} — below band" - else: - status = f"✓ projecting ${rate:,.0f} — within band" - vs = f"month-to-date, day {today.day}/{days_in}" - else: - if usd > high: - status, vs = "⚠ OVER band", f"+${usd - high:,.0f} over" - elif usd < low: - status, vs = "✓ under band", f"${low - usd:,.0f} headroom" - else: - status, vs = "✓ within band", f"${high - usd:,.0f} to top" - gaps = ( - f"{len(mo['days'])} days recorded; unpriced: {', '.join(sorted(mo['unpriced']))}" - if mo["unpriced"] - else f"{len(mo['days'])} days recorded" - ) - L.append(f"| {month} | {usd:,.2f} | {vs} | {status} | {gaps} |") - L.append("") - return L - - -LANE_SOURCE_PREFIX = "spend-attribution:" - - -def lane_attribution_section( - ws: Path, *, journal: Path | None = None, reg: dict | None = None -) -> list[str]: - """Per-lane cost from `spend-attribution:` ledger rows — - inferred from session transcripts by attribute_session_spend.py, - which is the ONLY trustworthy per-lane cost source (the declared - rows below carry zero tokens by contract). - - The lane vs `unattributed` split is the headline: most workstation - spend is interactive work that never invoked a skill, and judging a - scanning budget against that total overstates lane cost ~3.6x.""" - rows = [ - r - for r in metrics_ledger.rows(journal) - if str(r.get("source") or "").startswith(LANE_SOURCE_PREFIX) - ] - if not rows: - return [ - "## Per-lane attribution", - "", - "_Not yet collected. Run `python3 " - "traust metrics attribute-spend --append` (reads " - "Claude Code transcripts; completed days only) to " - "populate. Until then no per-lane cost is known — the " - "declared rows below are routing markers, not costs._", - "", - ] - by_month: dict = defaultdict( - lambda: defaultdict(lambda: {"usd": 0.0, "out": 0, "unpriced": False}) - ) - # Reprice from stored TOKENS at current registry rates rather than - # trusting the `cost_usd` frozen into each row. Without this the - # monthly-budget section above (which prices at build time) and this - # table can disagree after a rate correction — same rule as - # build_financial_report.ledger_series. - if reg is None: - raise ValueError( - "reg is required — pass model_registry from HarnessEngine.models.registry()" - ) - for r in rows: - m = r.get("metrics") or {} - date, skill = str(m.get("date") or ""), str(m.get("skill") or "?") - if len(date) < 7: - continue - cell = by_month[date[:7]][skill] - usd = m.get("cost_usd") - if m.get("model"): - try: - derived = model_registry.cost_usd( - reg, - m["model"], - int(m.get("tokens_in") or 0), - int(m.get("tokens_out") or 0), - int(m.get("cache_read") or 0), - int(m.get("cache_creation") or 0), - long_prompt=m.get("long_prompt"), - ) - except Exception: - derived = None - if derived is not None: - usd = derived - if usd is None: - cell["unpriced"] = True - else: - cell["usd"] += float(usd) - cell["out"] += int(m.get("tokens_out") or 0) - L = [ - "## Per-lane attribution (cost source)", - "", - "_Inferred from session transcripts — which skill each session " - "was running — not self-reported by agents. `unattributed` is " - "workstation sessions that never invoked a skill and is always " - "shown; a scanning budget should be judged against the lane " - "subtotal, not the workstation total._", - "", - ] - for month in sorted(by_month, reverse=True)[:2]: - skills = by_month[month] - tot = sum(c["usd"] for c in skills.values()) - lane = sum(c["usd"] for s, c in skills.items() if s != "unattributed") - un = skills.get("unattributed", {}).get("usd", 0.0) - L += [ - f"### {month}", - "", - f"**Harness lanes ${lane:,.2f}** · unattributed " - f"${un:,.2f} · total ${tot:,.2f}" - + (f" (lanes = {lane / tot * 100:.1f}%)" if tot else ""), - "", - "| Lane | Cost (USD) | Output tokens |", - "|---|---:|---:|", - ] - for skill, c in sorted(skills.items(), key=lambda kv: -kv[1]["usd"]): - note = " ⚠ unpriced model(s)" if c["unpriced"] else "" - L.append(f"| {skill} | {c['usd']:,.2f}{note} | {c['out']:,} |") - L.append("") - return L - - -def build( - ws: Path, - out_dir: Path, - budget_path: Path | None = None, - *, - journal: Path | None = None, - reg: dict | None = None, - budget_section: BudgetPolicy | None = None, -) -> Path: - all_rows = [ - r - for r in metrics_ledger.rows(journal) - if str(r.get("source", "")).startswith("model-spend:") - ] - rows = [r for r in all_rows if r["source"] != "model-spend:sessions"] - session_rows = [r for r in all_rows if r["source"] == "model-spend:sessions"] - agg: dict[tuple[str, str], dict] = defaultdict( - lambda: {"runs": 0, "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "cost_known": True} - ) - for r in rows: - skill = r["source"].split(":", 1)[1] - m = r.get("metrics", {}) - a = agg[(skill, m.get("model", "?"))] - a["runs"] += 1 - a["tokens_in"] += int(m.get("tokens_in") or 0) - a["tokens_out"] += int(m.get("tokens_out") or 0) - if m.get("cost_usd") is None: - a["cost_known"] = False - else: - a["cost_usd"] += float(m["cost_usd"]) - - L = [ - "# Model Spend Dashboard", - "", - f"**Generated:** {datetime.date.today().isoformat()} · " - f"source: hash-chained metrics ledger (`model-spend:*` rows)", - "", - "_How tracking works: drivers **declare** per-skill spend " - "(`model_registry.py spend`); the session collector records " - "**actuals** from Claude Code transcripts " - "(`collect_session_spend.py --append`, daily). Real-time view: " - "`traust metrics collect-spend`. Full chain: " - "docs/model-routing.md → How spend is tracked._", - "", - ] - - if reg is None: - raise ValueError( - "reg is required — pass model_registry from HarnessEngine.models.registry()" - ) - budget = load_monthly_budget(budget_path, budget_section=budget_section) - if budget and session_rows: - L += monthly_budget_section(session_rows, reg, budget, datetime.date.today()) - - L += lane_attribution_section(ws, journal=journal, reg=reg) - - if not agg: - L += [ - "_No spend rows recorded yet. Drivers begin appending rows as " - "the M1 rollout lands; this dashboard renders them on the next " - "rebuild._", - "", - ] - else: - L += [ - "## Declared rows (routing markers — NOT a cost source)", - "", - "_These are what drivers declare via `model_registry.py " - "spend`: which skill/model/repo/size ran. **Their token and " - "cost fields are best-effort and usually zero** — an agent " - "cannot observe its own usage mid-run, so most declared rows " - "record 0/0 tokens. Do not read cost from " - "this table; the per-lane section above is the cost " - "source._", - "", - "| Skill | Model | Runs | Tokens in | Tokens out | Cost (USD) |", - "|---|---|---:|---:|---:|---:|", - ] - for (skill, model), a in sorted(agg.items(), key=lambda kv: -(kv[1]["cost_usd"] or 0)): - cost = f"{a['cost_usd']:.2f}" if a["cost_known"] else "n/a (no price in registry)" - L.append( - f"| {skill} | {model} | {a['runs']} | " - f"{a['tokens_in']:,} | {a['tokens_out']:,} | {cost} |" - ) - L.append("") - - # session actuals (day x model), costed from registry list prices - if session_rows: - priced_total = 0.0 - unpriced: set = set() - for r in session_rows: - m = r["metrics"] - c = model_registry.cost_usd( - reg, - m.get("model", ""), - int(m.get("tokens_in") or 0), - int(m.get("tokens_out") or 0), - int(m.get("cache_read") or 0), - int(m.get("cache_creation") or 0), - long_prompt=m.get("long_prompt"), - ) - if c is None: - unpriced.add(m.get("model", "?")) - else: - priced_total += c - L += [ - "## Session actuals (Claude Code transcripts)", - "", - "_Cost = registry LIST prices (cache-aware: read×0.1, " - "write×1.25 of input) — preliminary until the rate card " - "lands; unpriced models show n/a._", - "", - "| Date | Model | Output tokens | Fresh input | Cache read | " - "Cache new | Sessions | Est. cost (USD) |", - "|---|---|---:|---:|---:|---:|---:|---:|", - ] - for r in sorted( - session_rows, - key=lambda r: (r["metrics"].get("date", ""), r["metrics"].get("model", "")), - reverse=True, - )[:30]: - m = r["metrics"] - c = model_registry.cost_usd( - reg, - m.get("model", ""), - int(m.get("tokens_in") or 0), - int(m.get("tokens_out") or 0), - int(m.get("cache_read") or 0), - int(m.get("cache_creation") or 0), - long_prompt=m.get("long_prompt"), - ) - L.append( - f"| {m.get('date')} | {m.get('model')} | " - f"{m.get('tokens_out', 0):,} | {m.get('tokens_in', 0):,} | " - f"{m.get('cache_read', 0):,} | {m.get('cache_creation', 0):,} | " - f"{m.get('sessions', 0)} | " - f"{f'{c:,.2f}' if c is not None else 'n/a'} |" - ) - L.append("") - note = f" (excludes unpriced: {', '.join(sorted(unpriced))})" if unpriced else "" - L.append( - f"**Estimated session-actuals total (all recorded days): ${priced_total:,.2f}**{note}" - ) - L.append("") - - # MTTA-per-dollar headline when both series exist - mtta = [r for r in metrics_ledger.rows(journal) if str(r.get("source", "")).startswith("mtta")] - total_cost = sum(a["cost_usd"] for a in agg.values() if a["cost_known"]) - if mtta and total_cost: - L += [ - f"**MTTA-per-dollar inputs available** — latest MTTA rows: " - f"{len(mtta)}; recorded spend ${total_cost:.2f}. Join lands " - f"with the Executive-Trends chart.", - "", - ] - - out_dir.mkdir(parents=True, exist_ok=True) - out = out_dir / "spend-dashboard.md" - out.write_text("\n".join(L) + "\n") - return out diff --git a/src/traust_engine/registry/__init__.py b/src/traust_engine/registry/__init__.py deleted file mode 100644 index 840b036..0000000 --- a/src/traust_engine/registry/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Model resolution — AI model tiers and spend pricing. - -PENDING(decision: metrics journal) — kept for traust; -deleted when traust retires or ports it. -""" diff --git a/src/traust_engine/registry/models.py b/src/traust_engine/registry/models.py deleted file mode 100644 index d825164..0000000 --- a/src/traust_engine/registry/models.py +++ /dev/null @@ -1,181 +0,0 @@ -"""Model registry loader/resolver — multi-model-strategy-plan M0/M1. - -config/model-registry.yaml is the ONLY place vendor/model identifiers -live (alignment rule A11 enforces this). This tool is how everything -else consults it: - - validate schema-validate the registry (CI/pre-commit) - list roles with floor/approved models - resolve ROLE [--tier T] print the model to use for a role — - first approved model at the role floor - (or an explicit escalation --tier) - stamp ROLE MODEL emit the metadata.additional.model_routing - JSON block (role, model, registry sha) - spend --skill S --model M --tokens-in N --tokens-out N [--batch B] - append a spend row to the hash-chained - metrics ledger (source model-spend:) - with cost computed when the registry has - prices - -Exit 0 ok; 1 validation/resolution failure; 2 usage. -""" - -from __future__ import annotations - -from traust_contracts import ModelRegistry - -_TIER_ORDER = ["haiku-class", "sonnet-class", "opus-class", "mythos-class"] - - -def validate(reg: ModelRegistry) -> list[str]: - """Semantic cross-checks beyond the JSON schema (which load_context - already applied): tier membership and floor/approved consistency.""" - errors: list[str] = [] - models = reg.all_models() - tiers = set(reg.tiers) - for mid, m in models.items(): - if m.tier not in tiers: - errors.append(f"model {mid}: unknown tier {m.tier}") - for role, r in reg.roles.items(): - if r.floor not in tiers: - errors.append(f"role {role}: unknown floor {r.floor}") - for mid in r.approved + r.candidates: - if mid not in models: - errors.append(f"role {role}: model {mid} not in any provider") - for mid in r.approved: - if mid in models and _TIER_ORDER.index(models[mid].tier) < _TIER_ORDER.index(r.floor): - errors.append( - f"role {role}: approved model {mid} " - f"({models[mid].tier}) is below the floor {r.floor}" - ) - return errors - - -def resolve(reg: ModelRegistry, role: str, tier: str | None = None) -> str: - r = reg.roles.get(role) - if r is None: - raise KeyError(f"unknown role '{role}' — roles: {sorted(reg.roles)}") - want = tier or r.floor - if want not in _TIER_ORDER: - raise KeyError(f"unknown tier '{want}'") - if _TIER_ORDER.index(want) < _TIER_ORDER.index(r.floor): - raise ValueError( - f"tier {want} is below role '{role}' floor {r.floor} — " - f"tier-downs below the floor are never allowed" - ) - models = reg.all_models() - for mid in r.approved: - m = models.get(mid) - if m is not None and m.tier == want: - return mid - # escalation request above floor with no exact-tier approval: take the - # lowest approved tier >= requested - ranked = sorted( - (mid for mid in r.approved if mid in models), - key=lambda mid: _TIER_ORDER.index(models[mid].tier), - ) - for mid in ranked: - if _TIER_ORDER.index(models[mid].tier) >= _TIER_ORDER.index(want): - return mid - raise ValueError(f"role '{role}': no approved model at or above {want}") - - -def stamp(reg: ModelRegistry, role: str, model: str) -> dict: - """The report's ``metadata.additional.model_routing`` record. - - Unknown values are omitted, never written as null: report.schema - (traust-contracts 0.55.0) types ``registry_sha`` and ``floor`` as strings, - so a role missing from the registry simply carries no floor. - """ - r = reg.roles.get(role) - out = {"role": role, "model": model} - if reg.source_sha: - out["registry_sha"] = reg.source_sha - if r is not None and r.floor: - out["floor"] = r.floor - return out - - -CACHE_READ_MULT = 0.1 # standard cache-hit price relative to input -CACHE_WRITE_MULT = 1.25 # 5-minute cache writes, the same for every current model -#: Token fields of a long_prompt usage block (a metrics row's own names). -LONG_PROMPT_KEYS = ("tokens_in", "tokens_out", "cache_read", "cache_creation") - - -def long_prompt_thresholds(reg: ModelRegistry) -> dict[str, int]: - """{model id: above_tokens} for every model with long-prompt pricing.""" - return {mid: m.long_prompt.above_tokens for mid, m in reg.all_models().items() if m.long_prompt} - - -def cost_usd( - reg: ModelRegistry, - model: str, - tok_in: int, - tok_out: int, - cache_read: int = 0, - cache_creation: int = 0, - long_prompt: dict | None = None, -) -> float | None: - """Cache-aware cost from registry list prices; None when unpriced. - - Cache hits use the model's ``cache_read_multiplier`` (0.1 when unset). - ``long_prompt`` is the part of these tokens that came from requests over - the model's long-prompt threshold, keyed by LONG_PROMPT_KEYS; it is - priced at the model's long-prompt rates and the rest at base rates. - Without it every token is priced at base rates. - """ - m = reg.all_models().get(model) - if m is None: - return None - pin, pout = m.price_per_mtok_in, m.price_per_mtok_out - if pin is None or pout is None: - return None - read_mult = m.cache_read_multiplier if m.cache_read_multiplier is not None else CACHE_READ_MULT - - def price(p_in: float, p_out: float, i: int, o: int, r: int, w: int) -> float: - return ( - i / 1e6 * p_in - + o / 1e6 * p_out - + r / 1e6 * p_in * read_mult - + w / 1e6 * p_in * CACHE_WRITE_MULT - ) - - lp = long_prompt or {} - long_part = [max(0, int(lp.get(k) or 0)) for k in LONG_PROMPT_KEYS] - whole = [tok_in, tok_out, cache_read, cache_creation] - base_part = [max(0, w - lng) for w, lng in zip(whole, long_part, strict=True)] - total = price(pin, pout, *base_part) - if any(long_part): - tier = m.long_prompt - p_in, p_out = (tier.price_per_mtok_in, tier.price_per_mtok_out) if tier else (pin, pout) - total += price(p_in, p_out, *long_part) - return round(total, 4) - - -def spend_row( - reg: ModelRegistry, - model: str, - tok_in: int, - tok_out: int, - cache_read: int = 0, - cache_creation: int = 0, - repo: str | None = None, - loc: int | None = None, -) -> dict: - row = { - "model": model, - "tokens_in": tok_in, - "tokens_out": tok_out, - "cost_usd": cost_usd(reg, model, tok_in, tok_out, cache_read, cache_creation), - "registry_sha": reg.source_sha, - } - if cache_read or cache_creation: - row["cache_read"] = cache_read - row["cache_creation"] = cache_creation - # per-repo attribution — the calibration tuple (spend, repo, size) - # estimate_scan fits against (estimate-calibration-analysis.md F5) - if repo: - row["repo"] = repo - if loc is not None: - row["loc"] = loc - return row diff --git a/src/traust_engine/registry/ops.py b/src/traust_engine/registry/ops.py deleted file mode 100644 index 8419c2b..0000000 --- a/src/traust_engine/registry/ops.py +++ /dev/null @@ -1,82 +0,0 @@ -"""``HarnessEngine.models`` — model registry and product-catalog ops.""" - -from __future__ import annotations - -from pathlib import Path - -from traust_contracts import ModelRegistry - -from traust_engine import locations -from traust_engine._ops_base import ContextOps -from traust_engine.registry.models import ( - cost_usd, - long_prompt_thresholds, - resolve, - spend_row, - stamp, - validate, -) - - -class ModelsOps(ContextOps): - def registry(self) -> ModelRegistry: - return self._ctx.model_registry - - def validate(self) -> list[str]: - return validate(self._ctx.model_registry) - - def resolve(self, role: str, tier: str | None = None) -> str: - return resolve(self._ctx.model_registry, role, tier) - - def stamp(self, role: str, model: str) -> dict: - return stamp(self._ctx.model_registry, role, model) - - def cost_usd( - self, - model: str, - tok_in: int, - tok_out: int, - cache_read: int = 0, - cache_creation: int = 0, - long_prompt: dict | None = None, - ) -> float | None: - return cost_usd( - self._ctx.model_registry, - model, - tok_in, - tok_out, - cache_read, - cache_creation, - long_prompt, - ) - - def long_prompt_thresholds(self) -> dict[str, int]: - """{model id: above_tokens} for models with long-prompt pricing.""" - return long_prompt_thresholds(self._ctx.model_registry) - - def spend_row( - self, - model: str, - tok_in: int, - tok_out: int, - cache_read: int = 0, - cache_creation: int = 0, - repo: str | None = None, - loc: int | None = None, - ) -> dict: - return spend_row( - self._ctx.model_registry, - model, - tok_in, - tok_out, - cache_read, - cache_creation, - repo=repo, - loc=loc, - ) - - def _feeds_cache(self) -> Path: - return locations.require(locations.feeds_cache_dir(self._loc), "feeds_cache") - - def feeds_cache(self) -> Path: - return self._feeds_cache() diff --git a/src/traust_engine/storage.py b/src/traust_engine/storage.py index dbb8a94..37fe26f 100644 --- a/src/traust_engine/storage.py +++ b/src/traust_engine/storage.py @@ -35,7 +35,7 @@ db = storage.localize(cfg.portfolio_graph_uri()) # -> local Path if storage.exists(results_uri / "graph/sboms"): ... -PENDING(decision: sweep stages | portfolio graph layers | metrics journal) — kept for traust; +PENDING(decision: sweep stages | portfolio graph layers) — kept for traust; deleted when traust retires or ports it. """ diff --git a/src/traust_engine/sweep/benchmark.py b/src/traust_engine/sweep/benchmark.py index 7112a7e..d0e6c7c 100644 --- a/src/traust_engine/sweep/benchmark.py +++ b/src/traust_engine/sweep/benchmark.py @@ -28,9 +28,7 @@ run /validate-findings against benchmark-findings.json → `score --variant vuln` → redeploy safe twins → re-run → `score --variant safe`. Results land in analysis-results/scan-testing/validation-benchmark/ -(harness-QA tree — excluded from campaign metrics) and the scorecard -appends to the hash-chained metrics ledger (source -validation-benchmark). +(harness-QA tree — excluded from campaign metrics). """ from __future__ import annotations @@ -204,32 +202,6 @@ def cmd_score(args, *, results_dir: Path) -> int: ) ) - # ledger row (best effort) - payload = json.dumps( - { - metric: val, - "severity_accuracy": result["severity_accuracy"], - "variant": args.variant, - "n": result["n"], - } - ) - # Best-effort metrics append, gated on the metrics destination being - # configured (locations.progress_tracker) rather than a WORKSPACE env var. - if locations.progress_tracker_dir(locations.configured_locations()) is not None: - subprocess.run( - [ - sys.executable, - "-m", - "traust_engine.metrics.history", - "append", - "--source", - "validation-benchmark", - "--metrics-json", - "-", - ], - input=payload, - text=True, - ) if val is not None and val < floor: print( f"BELOW FLOOR — a lane change that drops {metric} under {floor} must not ship", diff --git a/tests/test_attribute_session_spend.py b/tests/test_attribute_session_spend.py deleted file mode 100644 index 227ae20..0000000 --- a/tests/test_attribute_session_spend.py +++ /dev/null @@ -1,342 +0,0 @@ -"""Tests for traust metrics attribute-spend — per-lane spend -attribution (budget-guard widening plan, Phase 0).""" - -import json - -import pytest - -from traust_engine.metrics import attribute_spend as asp -from traust_engine.metrics import collect_spend as css - -VALID = {"secure-code-audit", "vuln-scan", "threat-model"} - - -def _usage(ts, model="m1", out=10, inp=1, cache_read=0, cache_new=0): - return { - "timestamp": ts, - "message": { - "model": model, - "usage": { - "input_tokens": inp, - "output_tokens": out, - "cache_read_input_tokens": cache_read, - "cache_creation_input_tokens": cache_new, - }, - }, - } - - -def _skill_call(name): - return { - "timestamp": "2026-07-01T00:00:00Z", - "message": {"content": [{"type": "tool_use", "name": "Skill", "input": {"skill": name}}]}, - } - - -def _session(tmp_path, name, records): - d = tmp_path / "proj" - d.mkdir(exist_ok=True) - f = d / f"{name}.jsonl" - f.write_text("\n".join(json.dumps(r) for r in records) + "\n", encoding="utf-8") - return d - - -class TestDetectSkill: - def test_skill_tool_call(self): - assert asp.detect_skill(_skill_call("vuln-scan"), VALID) == "vuln-scan" - - def test_scoped_skill_name_is_stripped(self): - rec = _skill_call("traust:vuln-scan") - assert asp.detect_skill(rec, VALID) == "vuln-scan" - - def test_slash_command(self): - rec = {"message": {"content": "/threat-model"}} - assert asp.detect_skill(rec, VALID) == "threat-model" - - def test_skill_md_read(self): - rec = { - "message": { - "content": [ - { - "type": "tool_use", - "name": "Read", - "input": {"file_path": "/workspace/skills/vuln-scan/SKILL.md"}, - } - ] - } - } - assert asp.detect_skill(rec, VALID) == "vuln-scan" - - def test_unknown_skill_name_is_rejected(self): - """Gating on the real skills tree stops prose about some - other tool from inventing a lane.""" - assert asp.detect_skill(_skill_call("not-a-real-skill"), VALID) is None - rec = {"message": {"content": "/nope"}} - assert asp.detect_skill(rec, VALID) is None - - def test_plain_message_has_no_signal(self): - assert asp.detect_skill(_usage("2026-07-01T00:00:00Z"), VALID) is None - - def test_known_skills_reads_the_real_tree(self, tmp_path): - for name in ("threat-model", "vuln-scan"): - skill = tmp_path / name - skill.mkdir() - (skill / "SKILL.md").write_text(f"# {name}\n") - skills = asp.known_skills(tmp_path) - assert "threat-model" in skills and "vuln-scan" in skills - - def test_known_skills_spans_both_layouts(self, tmp_path): - """Workflow skills sit under a stage directory and the rest at the - root (harness skill-usability plan 1.2). Missing the nested half - would empty `valid`, and attribution would silently mint no lanes - at all rather than failing.""" - (tmp_path / "census").mkdir() - (tmp_path / "census" / "SKILL.md").write_text("# census\n") - nested = tmp_path / "3-audit" / "vuln-scan" - nested.mkdir(parents=True) - (nested / "SKILL.md").write_text("# vuln-scan\n") - # a SKILL.md bundled inside a skill is not a skill of its own - vendored = tmp_path / "census" / "vendored" / "impostor" - vendored.mkdir(parents=True) - (vendored / "SKILL.md").write_text("# impostor\n") - assert asp.known_skills(tmp_path) == {"census", "vuln-scan"} - - def test_skill_md_signal_reads_a_staged_path(self): - """The transcript may record either layout: newer runs read - workspace/skills///SKILL.md, older ones the flat path, and - both must attribute to the same lane.""" - for path in ( - "/workspace/skills/3-audit/vuln-scan/SKILL.md", - "/workspace/skills/vuln-scan/SKILL.md", - ): - rec = { - "message": { - "content": [ - { - "type": "tool_use", - "name": "Read", - "input": {"file_path": path}, - } - ] - } - } - assert asp.detect_skill(rec, VALID) == "vuln-scan", path - - -class TestAttribution: - def test_usage_before_any_signal_is_unattributed(self, tmp_path): - d = _session( - tmp_path, - "s1", - [ - _usage("2026-07-01T00:00:00Z", out=100), - _skill_call("vuln-scan"), - _usage("2026-07-01T00:01:00Z", out=50), - ], - ) - agg = asp.collect_attributed([d], valid=VALID) - day = agg["2026-07-01"] - assert day[asp.UNATTRIBUTED]["m1"]["output_tokens"] == 100 - assert day["vuln-scan"]["m1"]["output_tokens"] == 50 - - def test_mid_session_switch_splits_the_session(self, tmp_path): - """A session that runs two skills must not book all its tokens - to whichever one it happened to start with.""" - d = _session( - tmp_path, - "s1", - [ - _skill_call("vuln-scan"), - _usage("2026-07-01T00:00:00Z", out=10), - _skill_call("threat-model"), - _usage("2026-07-01T00:01:00Z", out=7), - ], - ) - agg = asp.collect_attributed([d], valid=VALID) - day = agg["2026-07-01"] - assert day["vuln-scan"]["m1"]["output_tokens"] == 10 - assert day["threat-model"]["m1"]["output_tokens"] == 7 - assert asp.UNATTRIBUTED not in day - - def test_attribution_does_not_leak_across_sessions(self, tmp_path): - d = _session( - tmp_path, "s1", [_skill_call("vuln-scan"), _usage("2026-07-01T00:00:00Z", out=10)] - ) - _session(tmp_path, "s2", [_usage("2026-07-01T00:02:00Z", out=99)]) - agg = asp.collect_attributed([d], valid=VALID) - day = agg["2026-07-01"] - assert day["vuln-scan"]["m1"]["output_tokens"] == 10 - assert day[asp.UNATTRIBUTED]["m1"]["output_tokens"] == 99 - - def test_day_and_month_filters(self, tmp_path): - d = _session( - tmp_path, - "s1", - [_usage("2026-07-01T00:00:00Z", out=1), _usage("2026-08-02T00:00:00Z", out=2)], - ) - assert list(asp.collect_attributed([d], day="2026-07-01", valid=VALID)) == ["2026-07-01"] - assert list(asp.collect_attributed([d], month="2026-08", valid=VALID)) == ["2026-08-02"] - - def test_synthetic_model_names_are_skipped(self, tmp_path): - d = _session(tmp_path, "s1", [_usage("2026-07-01T00:00:00Z", model="", out=500)]) - assert asp.collect_attributed([d], valid=VALID) == {} - - -class TestLossless: - def test_split_matches_the_existing_collector_exactly(self, tmp_path): - """The whole point: an attribution that drops tokens is worse - than no attribution. Totals must equal collect_session_spend.""" - d = _session( - tmp_path, - "s1", - [ - _usage("2026-07-01T00:00:00Z", out=100, inp=5, cache_read=7), - _skill_call("vuln-scan"), - _usage("2026-07-01T00:01:00Z", out=50, inp=3, cache_read=9), - _usage("2026-07-02T00:00:00Z", model="m2", out=11), - ], - ) - res = asp.reconcile([d], None, None) - assert res["lossless"], res["deltas"] - assert res["attributed"]["output_tokens"] == 161 - - def test_reconcile_reports_a_mismatch_rather_than_hiding_it(self, tmp_path, monkeypatch): - d = _session(tmp_path, "s1", [_usage("2026-07-01T00:00:00Z", out=10)]) - monkeypatch.setattr( - asp, - "collect_attributed", - lambda *a, **k: { - "2026-07-01": { - "x": { - "m1": { - **{key: 0 for key in asp.TOKEN_KEYS}, - "output_tokens": 3, - "messages": 1, - "sessions": set(), - } - } - } - }, - ) - res = asp.reconcile([d], None, None) - assert res["lossless"] is False - assert res["deltas"]["output_tokens"] == -7 - - def test_cli_reconcile_exits_nonzero_on_mismatch(self, tmp_path, monkeypatch): - from traust_engine import HarnessEngine - - d = _session(tmp_path, "s1", [_usage("2026-07-01T00:00:00Z", out=10)]) - monkeypatch.setattr(css, "workspace_slugs", lambda ws: [d]) - monkeypatch.setattr( - asp, - "reconcile", - lambda *a, **k: { - "lossless": False, - "attributed": {"output_tokens": 1}, - "baseline": {"output_tokens": 2}, - "deltas": {"output_tokens": -1}, - }, - ) - metrics = HarnessEngine.load().metrics - res = metrics.attribute_reconcile([d], None, "2026-07") - assert res["lossless"] is False - - -class TestRender: - def test_unattributed_is_always_shown(self, tmp_path): - d = _session(tmp_path, "s1", [_usage("2026-07-01T00:00:00Z", out=10)]) - agg = asp.collect_attributed([d], valid=VALID) - body = "\n".join(asp.render(agg, "2026-07")) - assert asp.UNATTRIBUTED in body - assert "Harness lanes" in body - - def test_unpriced_models_are_named_not_zeroed(self, tmp_path): - """The declared side reads $0.00 everywhere precisely because a - missing price rounded to zero. Never repeat that here.""" - d = _session( - tmp_path, "s1", [_usage("2026-07-01T00:00:00Z", model="no-such-model", out=10)] - ) - agg = asp.collect_attributed([d], valid=VALID) - _costs, unpriced = asp.cost_by_skill(agg) - if unpriced: - body = "\n".join(asp.render(agg, "2026-07")) - assert "unpriced" in body and "no-such-model" in body - - -class TestCliGuards: - def test_missing_transcripts_exits_3(self, tmp_path, monkeypatch): - monkeypatch.setattr(css, "workspace_slugs", lambda ws: []) - assert css.workspace_slugs(tmp_path) == [] - - -class TestLedgerAppend: - def _agg(self, date="2026-07-01"): - base = {k: 0 for k in asp.TOKEN_KEYS} - return { - date: { - "vuln-scan": { - "m1": base - | {"output_tokens": 10, "input_tokens": 2, "messages": 1, "sessions": {"s1"}} - } - } - } - - def test_appends_completed_days(self, tmp_path, monkeypatch): - calls = [] - - monkeypatch.setattr(asp, "already_recorded", lambda ws, journal=None: set()) - monkeypatch.setattr( - asp.metrics_ledger, - "append", - lambda source, metrics, journal=None: calls.append((source, metrics)) or {}, - ) - n = asp.append_rows(tmp_path, self._agg(), reg={}) - assert n == 1 and len(calls) == 1 - source, payload = calls[0] - assert source == "spend-attribution:vuln-scan" - assert payload["skill"] == "vuln-scan" - assert payload["attribution"] == "inferred-from-transcript" - assert payload["tokens_out"] == 10 - - def test_today_is_never_recorded(self, tmp_path, monkeypatch): - """Appending a partial day would freeze that day's undercount - permanently, since the row is then 'already recorded'.""" - import datetime as _dt - - today = _dt.date.today().isoformat() - monkeypatch.setattr(asp, "already_recorded", lambda ws, journal=None: set()) - monkeypatch.setattr( - asp.metrics_ledger, - "append", - lambda *a, **k: pytest.fail("appended today"), - ) - assert asp.append_rows(tmp_path, self._agg(today), reg={}) == 0 - - def test_idempotent(self, tmp_path, monkeypatch): - monkeypatch.setattr( - asp, - "already_recorded", - lambda ws, journal=None: {("2026-07-01", "vuln-scan", "m1")}, - ) - monkeypatch.setattr( - asp.metrics_ledger, - "append", - lambda *a, **k: pytest.fail("re-appended"), - ) - assert asp.append_rows(tmp_path, self._agg(), reg={}) == 0 - - def test_uses_a_separate_ledger_namespace(self, tmp_path, monkeypatch): - """spend-attribution: must NOT collide with the declared - model-spend: rows — mixing them is what let a $0.00 table - look authoritative.""" - seen = [] - - monkeypatch.setattr(asp, "already_recorded", lambda ws, journal=None: set()) - monkeypatch.setattr( - asp.metrics_ledger, - "append", - lambda source, metrics, journal=None: seen.append(source) or {}, - ) - asp.append_rows(tmp_path, self._agg(), reg={}) - assert seen == ["spend-attribution:vuln-scan"] - assert not seen[0].startswith("model-spend:") diff --git a/tests/test_build_spend_dashboard.py b/tests/test_build_spend_dashboard.py deleted file mode 100644 index f7f60fc..0000000 --- a/tests/test_build_spend_dashboard.py +++ /dev/null @@ -1,134 +0,0 @@ -"""Monthly budget tracking in traust metrics spend. - -The dashboard's budget-vs-actual section prices session-actuals ledger -rows at registry list rates and compares each calendar month against -the monthly_budget block of config/budget-policy.yaml (visibility -only). Verified: band classification for complete months, run-rate -projection for the current month, unpriced-model coverage gaps, and -clean omission when the budget block or session rows are absent. -""" - -import datetime -import json -import tempfile -import unittest -from pathlib import Path -from unittest import mock - -from traust_contracts import load_context - -from traust_engine.metrics import spend as bsd -from traust_engine.registry import models as model_registry - -TODAY = datetime.date.today() -CUR_MONTH = TODAY.strftime("%Y-%m") - - -def _session_row(date: str, model: str, tok_out: int, cache_read: int = 0) -> dict: - return { - "snapshot_at": f"{date}T00:00:00+00:00", - "source": "model-spend:sessions", - "metrics": { - "date": date, - "model": model, - "tokens_in": 0, - "tokens_out": tok_out, - "cache_read": cache_read, - "cache_creation": 0, - "sessions": 1, - }, - } - - -def _budget_yaml(tmp: Path) -> Path: - p = tmp / "budget-policy.yaml" - p.write_text( - "budget_policy:\n" - " monthly_budget:\n" - " usd_central: 23000\n" - " usd_band_low: 19000\n" - " usd_band_high: 27000\n" - ) - return p - - -def _write_ledger(tmp: Path, rows: list[dict]) -> Path: - led = tmp / "metrics-history.jsonl" - led.write_text("\n".join(json.dumps(r) for r in rows) + "\n") - return led - - -class TestMonthlyBudgetSection(unittest.TestCase): - # mythos list rate: $50/Mtok out → tokens_out alone prices exactly. - - def _build_md(self, rows: list[dict]) -> str: - reg = load_context().model_registry - with tempfile.TemporaryDirectory() as td: - tmp = Path(td) - led = _write_ledger(tmp, rows) - with mock.patch("traust_engine.metrics.history.ledger_path", return_value=led): - out = bsd.build(tmp, tmp / "out", budget_path=_budget_yaml(tmp), reg=reg) - return out.read_text() - - def test_complete_months_classified_against_band(self): - md = self._build_md( - [ - # 2026-05: 600M tokens out x $50/M = $30,000 → over band - _session_row("2026-05-10", "claude-mythos-5", 600_000_000), - # 2026-06: 400M x $50/M = $20,000 → within band - _session_row("2026-06-10", "claude-mythos-5", 400_000_000), - # 2026-04: 100M x $50/M = $5,000 → under band - _session_row("2026-04-10", "claude-mythos-5", 100_000_000), - ] - ) - self.assertIn("## Monthly budget tracking", md) - may = next(l for l in md.splitlines() if l.startswith("| 2026-05 ")) - jun = next(l for l in md.splitlines() if l.startswith("| 2026-06 ")) - apr = next(l for l in md.splitlines() if l.startswith("| 2026-04 ")) - self.assertIn("OVER band", may) - self.assertIn("+$3,000 over", may) - self.assertIn("within band", jun) - self.assertIn("under band", apr) - - def test_current_month_shows_runrate_projection(self): - md = self._build_md( - [ - _session_row(TODAY.isoformat(), "claude-mythos-5", 20_000_000), - ] - ) - cur = next(l for l in md.splitlines() if l.startswith(f"| {CUR_MONTH} ")) - self.assertIn("projecting $", cur) - self.assertIn("month-to-date", cur) - - def test_unpriced_models_reported_as_coverage_gap_not_zero(self): - md = self._build_md( - [ - _session_row("2026-06-10", "claude-mythos-5", 400_000_000), - _session_row("2026-06-11", "totally-unpriced-model", 9_999_999), - ] - ) - jun = next(l for l in md.splitlines() if l.startswith("| 2026-06 ")) - self.assertIn("unpriced: totally-unpriced-model", jun) - self.assertIn("20,000.00", jun) # priced part unchanged - - def test_section_omitted_without_budget_block(self): - with tempfile.TemporaryDirectory() as td: - tmp = Path(td) - led = _write_ledger(tmp, [_session_row("2026-06-10", "claude-mythos-5", 1_000)]) - empty = tmp / "no-budget.yaml" - empty.write_text("budget_policy: {}\n") - with mock.patch("traust_engine.metrics.history.ledger_path", return_value=led): - reg = load_context().model_registry - out = bsd.build(tmp, tmp / "out", budget_path=empty, reg=reg) - self.assertNotIn("Monthly budget tracking", out.read_text()) - - def test_pricing_matches_registry_cost_fn(self): - reg = load_context().model_registry - expect = model_registry.cost_usd(reg, "claude-mythos-5", 0, 400_000_000, 0, 0) - md = self._build_md([_session_row("2026-06-10", "claude-mythos-5", 400_000_000)]) - jun = next(l for l in md.splitlines() if l.startswith("| 2026-06 ")) - self.assertIn(f"{expect:,.2f}", jun) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_metrics_history.py b/tests/test_metrics_history.py deleted file mode 100644 index f3a5b87..0000000 --- a/tests/test_metrics_history.py +++ /dev/null @@ -1,75 +0,0 @@ -"""Tests for the append-only hash-chained campaign metrics-history journal.""" - -import json - -import pytest - -from traust_engine.metrics import history as ml - - -@pytest.fixture(autouse=True) -def ledger_file(tmp_path, monkeypatch): - """The journal is config-located; point it at an isolated tmp file per test.""" - p = tmp_path / "metrics-history.jsonl" - monkeypatch.setattr(ml, "ledger_path", lambda journal=None: journal or p) - return p - - -def test_append_chains_and_verifies(): - r1 = ml.append("executive-summary", {"findings_total": 100}, note="first") - r2 = ml.append("loc-dashboard", {"total_loc": 5}) - assert r1["prev_sha"] == "" - assert r2["prev_sha"] == r1["row_sha"] - ok, issues = ml.verify() - assert ok, issues - - -def test_tampering_detected(ledger_file): - ml.append("executive-summary", {"findings_total": 100}) - ml.append("executive-summary", {"findings_total": 90}) - rows = [json.loads(line) for line in ledger_file.read_text().splitlines()] - rows[0]["metrics"]["findings_total"] = 1 # rewrite history - ledger_file.write_text("".join(json.dumps(r) + "\n" for r in rows)) - ok, issues = ml.verify() - assert not ok - assert any("altered" in i for i in issues) - - -def test_deletion_breaks_chain(ledger_file): - ml.append("a", {"x": 1}) - ml.append("a", {"x": 2}) - ml.append("a", {"x": 3}) - lines = ledger_file.read_text().splitlines() - ledger_file.write_text("\n".join([lines[0], lines[2]]) + "\n") # drop the middle row - ok, issues = ml.verify() - assert not ok - assert any("chain break" in i for i in issues) - - -def test_append_if_changed_dedupes(): - assert ml.append_if_changed("s", {"x": 1}) is not None - assert ml.append_if_changed("s", {"x": 1}) is None # identical - assert ml.append_if_changed("s", {"x": 1}, note="n") is not None # note forces - assert ml.append_if_changed("s", {"x": 2}) is not None # changed - ok, _ = ml.verify() - assert ok - - -def test_previous_and_series_are_per_source(): - ml.append("a", {"x": 1}) - ml.append("b", {"x": 10}) - ml.append("a", {"x": 2}) - assert ml.previous("a")["metrics"]["x"] == 2 - assert ml.previous("b")["metrics"]["x"] == 10 - assert [v for _, v in ml.series("a", "x")] == [1, 2] - - -def test_trend_line_and_polarity(): - prev = ml.append("executive-summary", {"findings_total": 100, "resolved_findings": 5}) - line = ml.trend_line( - {"findings_total": 90, "resolved_findings": 9}, - prev, - ["findings_total", "resolved_findings"], - ) - assert "↓ 10 (improving)" in line # fewer findings = good - assert "↑ 4 (improving)" in line # more resolved = good diff --git a/tests/test_model_pricing.py b/tests/test_model_pricing.py deleted file mode 100644 index 4207f4e..0000000 --- a/tests/test_model_pricing.py +++ /dev/null @@ -1,156 +0,0 @@ -"""Per-model cache-read multipliers and long-prompt pricing (traust-contracts 0.59.0).""" - -import json - -import pytest -from traust_contracts import ModelRegistry - -from traust_engine.metrics import attribute_spend, collect_spend -from traust_engine.registry.models import cost_usd, long_prompt_thresholds - - -def _reg(): - return ModelRegistry.model_validate( - { - "version": 1, - "updated": "2026-10-16", - "tiers": {"haiku-class": {}, "opus-class": {}}, - "providers": { - "anthropic": { - "data_handling": "test", - "models": { - "std": { - "tier": "opus-class", - "price_per_mtok_in": 5.0, - "price_per_mtok_out": 25.0, - }, - "cheap-cache": { - "tier": "opus-class", - "price_per_mtok_in": 4.0, - "price_per_mtok_out": 20.0, - "cache_read_multiplier": 0.05, - }, - "tiered": { - "tier": "haiku-class", - "price_per_mtok_in": 0.1, - "price_per_mtok_out": 0.5, - "long_prompt": { - "above_tokens": 100_000, - "price_per_mtok_in": 0.5, - "price_per_mtok_out": 2.5, - }, - }, - }, - } - }, - "roles": {}, - } - ) - - -M = 1_000_000 - - -def test_the_standard_cache_rate_is_unchanged(): - # 1M in + 1M out + 1M cache read at 0.1x + 1M cache write at 1.25x - assert cost_usd(_reg(), "std", M, M, M, M) == 5 + 25 + 0.5 + 6.25 - - -def test_a_model_cache_multiplier_applies(): - assert cost_usd(_reg(), "cheap-cache", 0, 0, M, 0) == pytest.approx(4 * 0.05) - - -def test_long_prompt_tokens_take_the_long_rates(): - lp = {"tokens_in": M, "tokens_out": M, "cache_read": M, "cache_creation": 0} - # the long part at 0.5/2.5 (cache read 0.1x of 0.5); the remaining 1M in at 0.1 - assert cost_usd(_reg(), "tiered", 2 * M, M, M, 0, lp) == pytest.approx(0.1 + 0.5 + 2.5 + 0.05) - - -def test_without_a_long_prompt_block_everything_is_base(): - assert cost_usd(_reg(), "tiered", M, M) == pytest.approx(0.6) - - -def test_thresholds_list_only_tiered_models(): - assert long_prompt_thresholds(_reg()) == {"tiered": 100_000} - - -def _usage(i, o=10, r=0, w=0): - return { - "input_tokens": i, - "output_tokens": o, - "cache_read_input_tokens": r, - "cache_creation_input_tokens": w, - } - - -def test_each_request_is_judged_on_its_own(): - a = {k: 0 for k in collect_spend.TOKEN_KEYS} - collect_spend.add_usage(a, _usage(60_000, r=50_000), 100_000) # prompt 110k: long - collect_spend.add_usage(a, _usage(90_000), 100_000) # prompt 90k: base - assert a["input_tokens"] == 150_000 - assert a["long_prompt"] == { - "tokens_in": 60_000, - "tokens_out": 10, - "cache_read": 50_000, - "cache_creation": 0, - } - - -def test_no_threshold_records_no_long_prompt(): - a = {k: 0 for k in collect_spend.TOKEN_KEYS} - collect_spend.add_usage(a, _usage(500_000), None) - assert "long_prompt" not in a - - -def _transcript(tmp_path, rows): - d = tmp_path / "proj" - d.mkdir() - with (d / "s1.jsonl").open("w") as fh: - for model, usage in rows: - fh.write( - json.dumps( - { - "timestamp": "2026-10-15T10:00:00Z", - "message": {"model": model, "usage": usage}, - } - ) - + "\n" - ) - return d - - -def test_both_collectors_carry_the_split_into_rows(tmp_path): - d = _transcript(tmp_path, [("tiered", _usage(150_000)), ("tiered", _usage(1_000))]) - th = long_prompt_thresholds(_reg()) - sessions = collect_spend.collect([d], None, th)["2026-10-15"]["tiered"] - row = collect_spend._row("tiered", sessions) - assert row["long_prompt"]["tokens_in"] == 150_000 and row["tokens_in"] == 151_000 - attributed = attribute_spend.collect_attributed([d], valid=set(), thresholds=th) - a = attributed["2026-10-15"][attribute_spend.UNATTRIBUTED]["tiered"] - assert a["long_prompt"]["tokens_in"] == 150_000 - # priced from the row exactly as the reports re-derive it - usd = cost_usd( - _reg(), - "tiered", - row["tokens_in"], - row["tokens_out"], - row["cache_read"], - row["cache_creation"], - long_prompt=row.get("long_prompt"), - ) - # cost_usd rounds to 4 decimal places - assert usd == pytest.approx( - 150_000 / M * 0.5 + 10 / M * 2.5 + 1_000 / M * 0.1 + 10 / M * 0.5, abs=1e-4 - ) - - -def test_the_engine_facade_passes_long_prompt_through(): - from types import SimpleNamespace - - from traust_engine.registry.ops import ModelsOps - - ops = ModelsOps.__new__(ModelsOps) - ops._ctx = SimpleNamespace(model_registry=_reg()) - lp = {"tokens_in": M, "tokens_out": 0, "cache_read": 0, "cache_creation": 0} - assert ops.cost_usd("tiered", M, 0, long_prompt=lp) == pytest.approx(0.5) - assert ops.long_prompt_thresholds() == {"tiered": 100_000} diff --git a/tests/test_model_registry.py b/tests/test_model_registry.py deleted file mode 100644 index 521221f..0000000 --- a/tests/test_model_registry.py +++ /dev/null @@ -1,144 +0,0 @@ -"""Tests for traust registry models and the shipped registry.""" - -import unittest -from pathlib import Path - -from traust_contracts import load_context - -from traust_engine.registry.models import resolve, spend_row, stamp, validate - - -def _load_registry(): - return load_context().model_registry - - -class TestShippedRegistry(unittest.TestCase): - def setUp(self): - self.reg = _load_registry() - - def test_registry_validates(self): - self.assertEqual(validate(self.reg), []) - - def test_floor_enforced(self): - with self.assertRaises(ValueError): - resolve(self.reg, "deep-audit", "haiku-class") - - def test_resolve_at_floor(self): - self.assertEqual(resolve(self.reg, "fact-judge"), "claude-sonnet-5") - - def test_escalation_above_floor(self): - self.assertEqual(resolve(self.reg, "fact-judge", "mythos-class"), "claude-mythos-5") - - def test_unknown_role(self): - with self.assertRaises(KeyError): - resolve(self.reg, "no-such-role") - - def test_ledger_writers_are_mythos_floor(self): - for role, r in self.reg.roles.items(): - if getattr(r, "ledger_validity_writer", False): - self.assertEqual( - r.floor, "mythos-class", f"{role}: validity writers keep frontier floor" - ) - - def test_stamp_carries_registry_sha(self): - s = stamp(self.reg, "fact-judge", "claude-sonnet-5") - self.assertEqual(len(s["registry_sha"]), 12) - self.assertEqual(s["floor"], "sonnet-class") - - def test_stamp_omits_floor_for_unregistered_role(self): - s = stamp(self.reg, "no-such-role", "claude-sonnet-5") - self.assertEqual(s["role"], "no-such-role") - self.assertNotIn("floor", s) - self.assertNotIn(None, s.values()) - - def test_stamp_validates_against_report_schema(self): - import json - - from jsonschema import Draft202012Validator - from traust_contracts.paths import schema_dir - - schema = json.loads((schema_dir() / "report.schema.json").read_text(encoding="utf-8")) - routing = schema["$defs"]["metadata"]["properties"]["additional"]["properties"][ - "model_routing" - ] - for role in ("fact-judge", "no-such-role"): - errors = list( - Draft202012Validator(routing).iter_errors(stamp(self.reg, role, "claude-sonnet-5")) - ) - self.assertEqual(errors, [], role) - - def test_spend_row_mythos_priced_from_rate_card(self): - # rate card filled 2026-07-24: mythos-5 at 10/50 per MTok - # (1M in = $10) + (100k out = $5) = $15 - row = spend_row(self.reg, "claude-mythos-5", 1_000_000, 100_000) - self.assertAlmostEqual(row["cost_usd"], 15.0) - - def test_spend_row_cache_aware_cost(self): - # sonnet list price: 3/15 per MTok; cache read x0.1, write x1.25 - row = spend_row( - self.reg, - "claude-sonnet-5", - 1_000_000, - 1_000_000, - cache_read=10_000_000, - cache_creation=1_000_000, - ) - # 3 + 15 + 10*0.3 + 1*3.75 = 24.75 - self.assertAlmostEqual(row["cost_usd"], 24.75, places=2) - - def test_approved_never_below_floor(self): - # covered by validate(), but assert the invariant directly - errors = [e for e in validate(self.reg) if "below the floor" in e] - self.assertEqual(errors, []) - - -class TestSessionSpendCollector(unittest.TestCase): - def test_collect_parses_usage_lines(self): - import json - import tempfile - - from traust_engine.metrics.collect_spend import collect - - d = Path(tempfile.mkdtemp()) - (d / "abc.jsonl").write_text( - json.dumps( - { - "timestamp": "2026-07-24T10:00:00Z", - "message": { - "model": "claude-sonnet-5", - "usage": { - "input_tokens": 10, - "output_tokens": 20, - "cache_read_input_tokens": 300, - "cache_creation_input_tokens": 40, - }, - }, - } - ) - + "\n" - + json.dumps( - { - "timestamp": "2026-07-24T10:01:00Z", - "message": {"model": "", "usage": {"output_tokens": 5}}, - } - ) - + "\n" - ) - agg = collect([d], "2026-07-24") - a = agg["2026-07-24"]["claude-sonnet-5"] - self.assertEqual(a["output_tokens"], 20) - self.assertEqual(a["cache_read_input_tokens"], 300) - self.assertNotIn("", agg["2026-07-24"]) - - -if __name__ == "__main__": - unittest.main() - - -def test_spend_row_carries_repo_attribution(): - """Calibration tuple (spend, repo, size) for spend attribution.""" - reg = _load_registry() - row = spend_row(reg, "test-model", 100, 50, repo="my-repo", loc=12345) - assert row["repo"] == "my-repo" and row["loc"] == 12345 - row2 = spend_row(reg, "test-model", 100, 50) - assert "repo" not in row2 and "loc" not in row2 diff --git a/tests/test_traust_engine.py b/tests/test_traust_engine.py index 8eac1c6..216a49e 100644 --- a/tests/test_traust_engine.py +++ b/tests/test_traust_engine.py @@ -21,61 +21,20 @@ def _home_with_progress_tracker(tmp_path: Path, pt: Path) -> Path: return home -def test_metrics_journal_follows_configured_progress_tracker(tmp_path): - pt = tmp_path / "estate-pt" - h = HarnessEngine.load(config_home=_home_with_progress_tracker(tmp_path, pt)) - - h.metrics.append("census", {"repos": 812}) - row = h.metrics.append("census", {"repos": 815}) - - # the engine received its location from the context — the journal landed under - # the operator-configured progress-tracker, not a workspace-relative guess. - assert (pt / "metrics" / "metrics-history.jsonl").is_file() - assert h.metrics.rows()[-1]["metrics"]["repos"] == 815 - assert h.metrics.previous("census")["row_sha"] == row["row_sha"] - ok, issues = h.metrics.verify() - assert ok, issues - - -def test_metrics_spend_build_uses_configured_paths(tmp_path): - pt = tmp_path / "estate-pt" - ws = tmp_path / "estate-ws" - ws.mkdir() - home = _home_with_progress_tracker(tmp_path, pt) - (home / "locations.yaml").write_text( - yaml.safe_dump({"progress_tracker": str(pt), "workspace": str(ws)}), - encoding="utf-8", - ) - h = HarnessEngine.load(config_home=home) - out = h.metrics.build_spend_dashboard() - assert out == pt / "metrics" / "dashboards" / "spend" / "spend-dashboard.md" - assert out.is_file() - - def test_two_engines_are_isolated_by_their_context(tmp_path): h1 = HarnessEngine.load(config_home=_home_with_progress_tracker(tmp_path, tmp_path / "a")) h2 = HarnessEngine.load(config_home=_home_with_progress_tracker(tmp_path, tmp_path / "b")) - h1.metrics.append("s", {"x": 1}) - assert h1.metrics.rows() and not h2.metrics.rows() + assert h1.sweep._progress_tracker() == tmp_path / "a" + assert h2.sweep._progress_tracker() == tmp_path / "b" def test_namespace_ops_are_cached_per_engine(tmp_path): h = HarnessEngine.load(config_home=_home_with_progress_tracker(tmp_path, tmp_path / "pt")) - assert h.metrics is h.metrics - assert h.models is h.models assert h.adapters is h.adapters assert h.sweep is h.sweep assert h.portfolio is h.portfolio -def test_models_registry_and_validate_from_context(tmp_path): - h = HarnessEngine.load(config_home=_home_with_progress_tracker(tmp_path, tmp_path / "pt")) - reg = h.models.registry() - assert reg is h.ctx.model_registry - assert h.models.validate() == [] - assert h.models.resolve("fact-judge") == "claude-sonnet-5" - - def test_sweep_paths_follow_configured_locations(tmp_path): home = tmp_path / "cfg-sweep" shutil.copytree(_FIXTURE, home)