From d212b8a1c9df3b9030f277180e68cc0b1e2dad14 Mon Sep 17 00:00:00 2001 From: Coden Date: Thu, 13 Aug 2026 19:41:11 +0900 Subject: [PATCH 1/2] feat: add evidence-gated cost recommendations --- .github/workflows/ci.yml | 1 + .github/workflows/release.yml | 5 + HANDOFF.md | 106 +++- README.md | 24 +- docs/cost-recommendation.md | 183 +++++++ src/weightclass/cli.py | 110 +++++ src/weightclass/cost_recommendation.py | 589 +++++++++++++++++++++++ tests/test_ci_workflow_structure.py | 1 + tests/test_cost_recommendation.py | 472 ++++++++++++++++++ tests/test_release_workflow_structure.py | 2 + 10 files changed, 1471 insertions(+), 22 deletions(-) create mode 100644 docs/cost-recommendation.md create mode 100644 src/weightclass/cost_recommendation.py create mode 100644 tests/test_cost_recommendation.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a62f09a..f821cc1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -98,5 +98,6 @@ jobs: tests.test_router tests.test_native_v2_runtime \ tests.test_executable_observation tests.test_foreground_process \ tests.test_process_context tests.test_triage tests.test_json_input \ + tests.test_cost_recommendation \ tests.test_delegation_runtime tests.test_delegation_v2_runtime \ tests.test_guarded_runtime_suite tests.test_delegation_conformance diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 21be3bd..e45979f 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -100,6 +100,8 @@ jobs: "$installed_venv/bin/python" -m pip install --no-index --no-deps "${packages[0]}" test "$("$installed_venv/bin/wclass" --version)" = "weightclass ${GITHUB_REF_NAME#v}" test "$(printf '%s' 'Fix a spelling typo.' | "$installed_venv/bin/wclass" classify)" = '{"tier": "low"}' + "$installed_venv/bin/wclass" review-cost-profile --help >/dev/null + "$installed_venv/bin/wclass" recommend --help >/dev/null "$installed_venv/bin/wclass" example-policy agy-cost-focused >/dev/null "$installed_venv/bin/wclass" example-policy claude-cost-focused >/dev/null "$installed_venv/bin/wclass" example-policy codex-cost-focused --model release-smoke-model >/dev/null @@ -149,6 +151,8 @@ jobs: "$installed_venv/bin/python" -m pip install --no-index --no-deps "${packages[0]}" test "$("$installed_venv/bin/wclass" --version)" = "weightclass ${GITHUB_REF_NAME#v}" test "$(printf '%s' 'Fix a spelling typo.' | "$installed_venv/bin/wclass" classify)" = '{"tier": "low"}' + "$installed_venv/bin/wclass" review-cost-profile --help >/dev/null + "$installed_venv/bin/wclass" recommend --help >/dev/null "$installed_venv/bin/wclass" example-policy agy-cost-focused >/dev/null "$installed_venv/bin/wclass" example-policy claude-cost-focused >/dev/null "$installed_venv/bin/wclass" example-policy codex-cost-focused --model release-smoke-model >/dev/null @@ -180,6 +184,7 @@ jobs: tests.test_router tests.test_native_v2_runtime \ tests.test_executable_observation tests.test_foreground_process \ tests.test_process_context tests.test_triage tests.test_json_input \ + tests.test_cost_recommendation \ tests.test_delegation_runtime tests.test_delegation_v2_runtime \ tests.test_guarded_runtime_suite tests.test_delegation_conformance diff --git a/HANDOFF.md b/HANDOFF.md index 88ac2e7..9cb524c 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -1,6 +1,6 @@ # Handoff -_Last updated: 2026-08-13 18:36 KST by Codex_ +_Last updated: 2026-08-13 19:35 KST by Codex_ ## Goal @@ -27,6 +27,10 @@ _Last updated: 2026-08-13 18:36 KST by Codex_ security/performance and release hardening described below. Custom selections remain explicitly unqualified; built-in routes and provider boundaries are unchanged. +- The current `main` worktree has an unreleased cost-recommendation feature: + task-free `review-cost-profile` plus non-executing `recommend`. It is not + committed, tagged, published, or installed through Homebrew yet. Built-in + routes and every execution path remain unchanged. - The source-of-truth Homebrew formula was updated on `main` by `e18fd3988a6fa6ad642f44a084b436c6507a2f6b`. The matching tap commit is [`1244ed6`](https://github.com/ictechgy/homebrew-tap/commit/1244ed6f0baf6cf2c557bec356230e3ea199961d). @@ -297,6 +301,42 @@ _Last updated: 2026-08-13 18:36 KST by Codex_ reviewed action metadata runs on Node 24, replacing the Node 20-based v5 generation that GitHub runners warned about. +## Current Unreleased Cost Recommendation + +- Added `src/weightclass/cost_recommendation.py` with strict version-1 cost + profile and qualification-card schemas. Inputs are bounded regular JSON, + duplicate-safe through the shared loader, and must assert that opaque IDs + are not task-derived. User/evaluator cost and quality assertions remain + explicitly unverified by the router. +- Added task-free `wclass review-cost-profile`, which validates the profile + and emits its canonical fingerprint without reading task stdin or starting a + vendor. +- Added `wclass recommend --preset -cost-focused --cost-profile ... + --qualification-card ...`. It loads both documents before task access, + selects the packaged candidate and same-vendor built-in baseline for one + tier, and emits `recommend` or valid `abstain` without starting a child. +- The fixed machine floors are 30 paired outcomes, 15% estimated-cost lower + bound, 20% maximum savings interval width, 5% maximum quality margin, zero + new critical failures, complete attempts, independent quality review, both + languages, all nine categories, and all three tiers. Stale, mismatched, + incomplete, or economically flat evidence abstains rather than falling back. +- The cost-profile fingerprint binds every cost and route value. A separate + canonical qualification-card fingerprint binds every aggregate evidence + field; both exact route fingerprints and both evidence fingerprints feed the + recommendation fingerprint. It does not authorize execution: ordinary + `run` acknowledgement is still required. +- Receipts expose provider differences without normalizing their semantics: + Claude/Codex have reviewed model+effort override surfaces, Grok model only, + and `agy` no model override. Claude/Codex retain stdin; `agy`/Grok retain + `task_delivery: argv` and the existing process-inspection disclosure. +- Added `docs/cost-recommendation.md` and README entry-point documentation. + Same-vendor packaged presets are the only initial scope; no runtime learning, + retry, fallback, telemetry, provider access, price inference, built-in + change, cross-vendor recommendation, or automatic execution was added. +- macOS CI/release boundary suites now include the focused recommendation + module, and clean installed-wheel release jobs smoke both new command + parsers. + ## Key Files & State - `src/weightclass/cli.py`: V2 route/run ordering and pre-spawn identity check. @@ -334,6 +374,10 @@ _Last updated: 2026-08-13 18:36 KST by Codex_ release tools and their exact hash-pinned installation closure. - `src/weightclass/entrypoint.py` and `classification_cli.py`: lightweight local-classification dispatch and lazy vendor-triage loading. +- `src/weightclass/cost_recommendation.py`, + `tests/test_cost_recommendation.py`, and + `docs/cost-recommendation.md`: unreleased advisory expected-completed-cost + contracts, CLI regressions, and user workflow. - `tests/verify_release_source.py`: redacted release-tag ancestry gate used before the release job installs networked tools. - `docs/completion-audit-v2.md`: requirement-to-test completion map. Goal g12 is leader-verified; retain this audit connection when refreshing this file. @@ -366,6 +410,28 @@ _Last updated: 2026-08-13 18:36 KST by Codex_ ## Verification +- Fresh unreleased cost-recommendation verification on 2026-08-13: + - Behavior-first RED→GREEN covered missing `recommend`, flat-cost + abstention, configuration-before-task ordering, full qualification-card + fingerprint binding, stale route binding, every fixed gate, all four + provider capability/task-delivery receipts, and task-free profile review. + - Python 3.10.20 and Python 3.14.6 full `unittest` suites with + `ResourceWarning` as an error: 808 tests passed on each interpreter. + - Ruff 0.16.2 check/format over 125 files, strict mypy over 105 source files, + compileall, and `git diff --check` passed. + - Offline wheel/sdist build and strict Twine checks passed. The artifacts + contain the new module, document, and focused test. Extracted-sdist + isolation passed 802 tests with 13 platform skips. + - A clean Python 3.10 wheel install completed the real + `route -> review-cost-profile -> recommend` flow, returned `recommend`, + preserved the exact candidate fingerprint, emitted no task text, and had + `PATH=/nonexistent` during recommendation to prove no vendor launch was + required. + - Installing the complete hash-pinned release requirements offline was not + possible because the macOS CPython 3.13 mypy 2.3.0 wheel was absent from + the local cache. The independently cached strict mypy 2.3.0 tool passed; + no network access was used. + - Fresh repository-review remediation verification on 2026-08-13: - RED→GREEN evidence covers a missing release lock, Python 3.10 native/V2 decoder recursion tracebacks, unmerged Git release commits, eager protocol @@ -602,6 +668,9 @@ _Last updated: 2026-08-13 18:36 KST by Codex_ ## Blockers & Open Questions - No known blocker or mandatory release/deployment step remains for `0.13.0`. +- The cost-recommendation worktree is deliberately unreleased and uncommitted. + Review the diff before deciding its version, commit, or deployment. Do not + republish immutable `0.13.0`. - GitHub repository settings were updated and re-read through the API on 2026-08-13. The active `Protect version tags` ruleset targets `refs/tags/v*`, has no bypass actor, and blocks updates and deletions. The `pypi` environment @@ -624,32 +693,33 @@ _Last updated: 2026-08-13 18:36 KST by Codex_ ## Next Steps -1. Keep the cost-focused Claude policy explicit opt-in only. Collect real-user +1. Review and commit the unreleased cost-recommendation diff. If publishing, + choose a new version, update release notes/HANDOFF, and run the normal + immutable release process; never overwrite `0.13.0`. +2. Keep the cost-focused Claude policy explicit opt-in only. Collect real-user compatibility feedback without task telemetry, provider-price inference, or a claim about actual bills. Do not change built-ins, standard/high routes, schema 2, or posture vocabulary from the low-target result. -2. Any future token-efficiency candidate still needs fresh task-free paired +3. Any future token-efficiency candidate still needs fresh task-free paired evidence and must pass the separate raw-token gate; the cost `go` does not satisfy it. -3. Evaluate the Codex, `agy`, and Grok cost-focused examples independently +4. Evaluate the Codex, `agy`, and Grok cost-focused examples independently before any promotion; do not combine vendor results or infer pricing. -4. Re-enable Linux Claude semantic triage only after reviewing a concrete +5. Re-enable Linux Claude semantic triage only after reviewing a concrete filesystem-containment command and its process-tree boundary. -5. Preserve Protocol 1 compatibility, explicit cross-vendor opt-in, task +6. Preserve Protocol 1 compatibility, explicit cross-vendor opt-in, task no-retention, and the single-reviewed-child boundary. ## Resume Prompt Open `/Users/jinhongan/Desktop/subscription-agent-router`, read `HANDOFF.md` -and `AGENTS.md`, then continue from: `weightclass 0.13.0 is published to PyPI, -has a GitHub Release, and is deployed through the verified Homebrew tap. It -adds tier-specific opaque Grok model routing while retaining packaged effort -values and argv task delivery. Custom Claude/Codex/Grok configurations remain -explicitly unqualified; built-ins are unchanged. The release also contains a -fully hash-pinned toolchain, tag-to-main ancestry gate, JSON recursion -redaction, lightweight local-classification entrypoint, and Node 24 GitHub -Actions. The immutable release workflow, public artifact hashes, Homebrew -style/audit/source install/test, and installed 0.13.0 smokes are green. GitHub -blocks update/deletion of v* tags and requires a pypi environment reviewer. -Do not attempt to republish immutable 0.13.0; use a new version for future -changes.` +and `AGENTS.md`, then continue from: `weightclass 0.13.0 remains the immutable +published release. The current main worktree has an uncommitted, unreleased +same-vendor advisory cost router: review-cost-profile plus recommend, strict +cost/qualification schemas, full evidence fingerprints, fixed conservative +gates, explicit abstention, provider capability/task-delivery receipts, docs, +and eight focused tests. Built-ins and execution paths are unchanged. Python +3.10/3.14 each passed 808 tests; Ruff, strict mypy, compileall, build, strict +Twine, extracted-sdist isolation, and a clean-wheel recommendation smoke are +green. Review the diff before choosing a new version/commit/release. Do not +republish 0.13.0.` diff --git a/README.md b/README.md index 2b7f69b..e0922fc 100644 --- a/README.md +++ b/README.md @@ -68,14 +68,17 @@ ownership labels remain opaque caller declarations. See the `wclass --help` lists the whole surface: ```text -wclass [-h] [--version] {classify,example-policy,review-preset,route,run,render,delegate,v2} ... +wclass [-h] [--version] {classify,example-policy,review-preset,review-cost-profile,recommend,route,run,render,delegate,v2} ... ``` -`classify`, `route`, and `run` read the task from standard input. `render` +`classify`, `recommend`, `route`, and `run` read the task from standard input. `render` prints the command of a policy route named by a workflow descriptor and never reads a task. `example-policy` emits packaged policy JSON; `review-preset` -prints every command and fingerprint in one packaged policy. Neither reads a -task or invokes a vendor. `v2` selects a declarative API route; see +prints every command and fingerprint in one packaged policy. +`review-cost-profile` validates and fingerprints task-free cost input. None of +those three review commands reads a task or invokes a vendor. `recommend` +emits evidence-bound advice or an explicit abstention and never invokes a +vendor. `v2` selects a declarative API route; see [V2 API routing](#v2-api-routing-through-an-external-runtime). `delegate route` reads only its policy and manifest and does not consume task standard input or inspect the supplied runtime path. `delegate run` reads the @@ -564,6 +567,19 @@ No preference is persisted and no router configuration file is written. Removing `--preset` or `--cost-focused` immediately restores the built-in route selection. +### Evidence-gated cost recommendation + +`wclass recommend` is a non-executing, same-vendor advisory layer over the +packaged presets. It consumes a user-reviewed opaque cost profile and a strict +qualification card, then returns either `recommend` or `abstain`. It does not +infer provider pricing, inspect billing, start a child, retry, fall back, or +change built-ins. A later `run` still requires the ordinary exact route review +and acknowledgement. + +See [Cost-aware recommendations](docs/cost-recommendation.md) for the input +schemas, fixed quality and uncertainty gates, canonical fingerprints, provider +capability differences, and end-to-end workflow. + A route's `vendor` is a containment label you choose, not a list of tools weightclass knows. Any printable identifier without whitespace, up to 64 bytes, is valid. Routing compares it as a string and the fingerprint hashes it as a diff --git a/docs/cost-recommendation.md b/docs/cost-recommendation.md new file mode 100644 index 0000000..fa1e303 --- /dev/null +++ b/docs/cost-recommendation.md @@ -0,0 +1,183 @@ +# Cost-aware recommendations + +`wclass recommend` is an offline, non-executing advisory command. It compares +one packaged same-vendor candidate with that vendor's built-in route for the +task tier, then either recommends the candidate or explicitly abstains. It +does not start a vendor, retry, fall back, change built-ins, persist a +preference, inspect credentials, fetch prices, or claim knowledge of a bill. + +The objective is **externally estimated cost per completed outcome**. A user or +evaluator supplies integer cost units under one reviewed measurement contract. +Those units should include every authorized attempt and rework needed to reach +the frozen completion rule. weightclass validates and binds the documents, but +does not verify their measurements or infer provider pricing. + +## Workflow + +1. Review the built-in baseline and packaged candidate for the same tier: + + ```sh + task='Fix a spelling typo.' + printf '%s' "$task" | wclass route --source-vendor claude --tier low + printf '%s' "$task" | wclass route --preset claude-cost-focused --tier low + ``` + +2. Put those exact route fingerprints and externally supplied cost units in a + cost profile. Validate it and obtain its canonical fingerprint: + + ```sh + wclass review-cost-profile --cost-profile cost-profile.json + ``` + + This command does not read task input or start a vendor. + +3. Create a qualification card that binds the profile fingerprint, both route + fingerprints, vendor, tier, measurement contract, and aggregate evaluation + gates. + +4. Request advice: + + ```sh + printf '%s' "$task" | wclass recommend \ + --preset claude-cost-focused \ + --cost-profile cost-profile.json \ + --qualification-card qualification-card.json \ + --tier low + ``` + + Exit `0` with `"decision": "recommend"` means the exact candidate passed + the local document checks. Exit `0` with `"decision": "abstain"` is also a + valid result: the evidence is missing, stale, weak, mismatched, or shows no + cost advantage. Malformed documents exit `2` with only + `{"error": "invalid_input"}`. + +5. If you decide to execute the recommendation, use the ordinary reviewed run + flow and acknowledge the candidate's exact `route_fingerprint`. The + recommendation fingerprint is evidence binding, not execution authority: + + ```sh + printf '%s' "$task" | wclass run \ + --preset claude-cost-focused \ + --tier low \ + --ack-route-fingerprint 'sha256:REVIEWED_CANDIDATE_FINGERPRINT' + ``` + +## Cost profile schema + +The profile is a strict version-1 JSON object. Unknown fields, duplicate keys, +non-integer costs, and more than 128 route entries are rejected. + +```json +{ + "schema_version": 1, + "profile_id": "team-cost-v1", + "measurement_contract_id": "reviewed-cost-units-v1", + "unit": "reviewed-cost-unit", + "identifiers_not_task_derived": true, + "pricing_inferred": false, + "actual_billing_claimed": false, + "routes": [ + { + "route_fingerprint": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "expected_completed_cost_units": 100 + }, + { + "route_fingerprint": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "expected_completed_cost_units": 40 + } + ] +} +``` + +`profile_id`, `measurement_contract_id`, and `unit` are bounded opaque labels; +weightclass assigns them no provider semantics. `pricing_inferred` and +`actual_billing_claimed` must be `false`. Route fingerprints must be unique. +`identifiers_not_task_derived` must be `true`; it is the author's assertion +that none of the opaque labels encodes prompt material. +The review output includes `values_verified_by_router: false` to distinguish +schema validation from economic verification. + +## Qualification card schema + +The card is also strict version-1 JSON. Replace every fingerprint with the +exact value from the corresponding review. The `cost_profile_fingerprint` +comes from `review-cost-profile`. + +```json +{ + "schema_version": 1, + "card_id": "claude-low-qualified-v1", + "identifiers_not_task_derived": true, + "cost_profile_fingerprint": "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "measurement_contract_id": "reviewed-cost-units-v1", + "vendor": "claude", + "tier": "low", + "baseline_route_fingerprint": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "candidate_route_fingerprint": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "status": "qualified", + "sample_size": 90, + "cost_savings_lower_bound_basis_points": 4628, + "cost_savings_ci_width_basis_points": 1772, + "quality_delta_lower_bound_basis_points": -402, + "quality_noninferiority_margin_basis_points": 500, + "new_critical_failures": 0, + "all_attempts_included": true, + "independent_quality_review": true, + "covered_languages": ["en", "ko"], + "covered_categories": [ + "concurrency", + "data-integrity", + "destructive-work", + "migration", + "performance", + "privacy", + "reliability", + "routine", + "security" + ], + "covered_tiers": ["low", "standard", "high"], + "valid_until": "9999-12-31" +} +``` + +A recommendation requires all of these conditions: + +- status is `qualified`, the card is not expired, and all profile, route, + vendor, tier, and measurement bindings match; +- at least 30 paired outcomes; +- estimated-cost savings lower bound at least 15%, with confidence-interval + width at most 20%; +- quality lower bound within a margin no greater than 5%; +- zero new critical failures; +- all attempts included and independent quality review asserted; and +- exact coverage of both languages, all nine fixed categories, and all three + tiers. + +These are conservative machine floors, not a claim that the evaluator's +assertions are true. The receipt says `assertions_verified_by_router: false`. +The full canonical qualification-card fingerprint is included in the receipt +and in the recommendation fingerprint, so changing any gate or binding changes +the recommendation identity. + +## Provider boundaries + +The initial command compares only a packaged preset with the built-in baseline +for the same vendor and tier. Cross-vendor selection is not supported. + +- Claude and Codex expose reviewed tier model and effort override surfaces. +- Grok exposes reviewed tier model overrides; its packaged effort values remain + opaque and are not user-overridden by this surface. +- `agy` exposes reviewed effort routing but no qualified model override. +- Claude and Codex deliver the task on stdin. `agy` and Grok retain + `task_delivery: argv`, including the documented local process-inspection + exposure if the candidate is later executed. + +Only the exact Claude low candidate currently has repository-recorded +promotion-grade estimated-cost evidence. A user-supplied qualification card can +describe another exact route, but weightclass does not transfer Claude evidence +to Codex, Grok, or `agy`, and does not verify the external assertion. + +Both input documents are read before task stdin. They must be bounded regular +files owned by the current user or root and must not be world-writable. The +receipt contains commands, fingerprints, fixed aggregate fields, and opaque +document labels, but never task text or a task hash. diff --git a/src/weightclass/cli.py b/src/weightclass/cli.py index 6cab84b..7ee21b0 100644 --- a/src/weightclass/cli.py +++ b/src/weightclass/cli.py @@ -21,6 +21,13 @@ read_task_from_standard_input, validate_task, ) +from .cost_recommendation import ( + CostRecommendationError, + build_cost_profile_review, + build_recommendation_receipt, + load_cost_profile, + load_qualification_card, +) from .delegation_compile import ( canonical_json_bytes, compile_delegation_descriptor, @@ -642,6 +649,23 @@ def build_parser() -> argparse.ArgumentParser: review_preset.add_argument("name", choices=tuple(EXAMPLE_POLICY_RESOURCES)) review_preset.add_argument("--model") _add_preset_override_arguments(review_preset) + review_cost_profile = subcommands.add_parser( + "review-cost-profile", + allow_abbrev=False, + description="Validate and fingerprint one task-free cost profile.", + ) + review_cost_profile.add_argument("--cost-profile", required=True, type=Path) + recommend = subcommands.add_parser( + "recommend", + allow_abbrev=False, + description="Recommend or abstain without starting a vendor process.", + ) + recommend.add_argument("--preset", required=True, choices=tuple(EXAMPLE_POLICY_RESOURCES)) + recommend.add_argument("--cost-profile", required=True, type=Path) + recommend.add_argument("--qualification-card", required=True, type=Path) + recommend.add_argument("--model") + recommend.add_argument("--tier", choices=("low", "standard", "high")) + _add_preset_override_arguments(recommend) for name, description in ( ("route", "Select and print a command for a task read from standard input."), ("run", "Select and start a command for a task read from standard input."), @@ -1253,6 +1277,81 @@ def review_packaged_preset( return 0 +def recommend_from_standard_input( + preset: str, + cost_profile_path: Path, + qualification_card_path: Path, + explicit_tier: Tier | None = None, + model: str | None = None, + overrides: PresetOverrides | None = None, +) -> int: + """Render one evidence-bound recommendation without starting a vendor.""" + overrides = overrides or PresetOverrides() + source_vendor = preset.removesuffix("-cost-focused") + try: + profile = load_cost_profile(cost_profile_path) + card = load_qualification_card(qualification_card_path) + candidate_policy = _automatic_cost_policy( + True, + None, + source_vendor, + None, + model, + overrides, + ) + except (CostRecommendationError, InvalidInputError): + print(json.dumps({"error": "invalid_input"}), file=sys.stderr) + return 2 + if candidate_policy is None: + raise AssertionError("packaged policy resolution returned no policy") + try: + tier, routing_reason_code, candidate_route, candidate_policy = select_task_route( + read_task_from_standard_input(), + candidate_policy, + source_vendor, + explicit_tier, + ) + baseline_route = select_tier_route(DEFAULT_ROUTES, tier, source_vendor, False) + baseline_fingerprint = native_route_fingerprint(baseline_route, False) + candidate_fingerprint = native_route_fingerprint( + candidate_route, + candidate_policy.allow_mixed_vendors, + candidate_policy.posture, + ) + receipt = build_recommendation_receipt( + profile, + card, + baseline_route=baseline_route, + baseline_route_fingerprint=baseline_fingerprint, + candidate_route=candidate_route, + candidate_route_fingerprint=candidate_fingerprint, + routing_reason_code=routing_reason_code, + candidate_configuration_status=_packaged_configuration_status(preset, model, overrides), + ) + except InvalidTaskError: + print(json.dumps({"error": "invalid_task"}), file=sys.stderr) + return 2 + except CostRecommendationError: + print(json.dumps({"error": "invalid_input"}), file=sys.stderr) + return 2 + except RouteSelectionError: + print(json.dumps({"error": "unsupported_route"}), file=sys.stderr) + return 3 + print(json.dumps(receipt)) + return 0 + + +def review_cost_profile(cost_profile_path: Path) -> int: + """Validate and fingerprint one cost profile without reading task input.""" + try: + profile = load_cost_profile(cost_profile_path) + except CostRecommendationError: + print(json.dumps({"error": "invalid_input"}), file=sys.stderr) + return 2 + print(json.dumps(build_cost_profile_review(profile))) + return 0 + + def route_from_standard_input( policy_path: Path | None, source_vendor: str | None, @@ -1633,6 +1732,17 @@ def main(argv: Sequence[str] | None = None) -> int: arguments.model, _preset_overrides_from_arguments(arguments), ) + if arguments.command == "recommend": + return recommend_from_standard_input( + arguments.preset, + arguments.cost_profile, + arguments.qualification_card, + arguments.tier, + arguments.model, + _preset_overrides_from_arguments(arguments), + ) + if arguments.command == "review-cost-profile": + return review_cost_profile(arguments.cost_profile) if arguments.command == "route": return route_from_standard_input( arguments.policy, diff --git a/src/weightclass/cost_recommendation.py b/src/weightclass/cost_recommendation.py new file mode 100644 index 0000000..1fe73b0 --- /dev/null +++ b/src/weightclass/cost_recommendation.py @@ -0,0 +1,589 @@ +"""Task-free cost evidence and advisory recommendation contracts.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from typing import Any, Final, Literal, cast + +from .classification import Tier +from .json_input import JsonInputError, load_json_object +from .router import Route, uses_argv_task_delivery + +MAX_COST_INPUT_BYTES: Final = 262_144 +MAX_COST_ROUTES: Final = 128 +MIN_QUALIFIED_PAIRS: Final = 30 +MIN_COST_SAVINGS_LOWER_BOUND_BASIS_POINTS: Final = 1_500 +MAX_COST_SAVINGS_CI_WIDTH_BASIS_POINTS: Final = 2_000 +MAX_QUALITY_NONINFERIORITY_MARGIN_BASIS_POINTS: Final = 500 +REQUIRED_LANGUAGES: Final = frozenset({"en", "ko"}) +REQUIRED_CATEGORIES: Final = frozenset( + { + "concurrency", + "data-integrity", + "destructive-work", + "migration", + "performance", + "privacy", + "reliability", + "routine", + "security", + } +) +REQUIRED_TIERS: Final = frozenset({"low", "standard", "high"}) +_FINGERPRINT_PATTERN: Final = re.compile(r"sha256:[0-9a-f]{64}") + +QualificationStatus = Literal["qualified", "provisional", "unqualified"] + + +class CostRecommendationError(ValueError): + """Raised without caller-controlled values when cost input is invalid.""" + + +@dataclass(frozen=True) +class CostRouteEstimate: + route_fingerprint: str + expected_completed_cost_units: int + + +@dataclass(frozen=True) +class CostProfile: + profile_id: str + measurement_contract_id: str + unit: str + routes: tuple[CostRouteEstimate, ...] + + def canonical_value(self) -> dict[str, object]: + return { + "schema_version": 1, + "profile_id": self.profile_id, + "measurement_contract_id": self.measurement_contract_id, + "unit": self.unit, + "identifiers_not_task_derived": True, + "pricing_inferred": False, + "actual_billing_claimed": False, + "routes": [ + { + "route_fingerprint": route.route_fingerprint, + "expected_completed_cost_units": route.expected_completed_cost_units, + } + for route in self.routes + ], + } + + def cost_for(self, route_fingerprint: str) -> int | None: + return next( + ( + route.expected_completed_cost_units + for route in self.routes + if route.route_fingerprint == route_fingerprint + ), + None, + ) + + +@dataclass(frozen=True) +class QualificationCard: + card_id: str + cost_profile_fingerprint: str + measurement_contract_id: str + vendor: str + tier: Tier + baseline_route_fingerprint: str + candidate_route_fingerprint: str + status: QualificationStatus + sample_size: int + cost_savings_lower_bound_basis_points: int + cost_savings_ci_width_basis_points: int + quality_delta_lower_bound_basis_points: int + quality_noninferiority_margin_basis_points: int + new_critical_failures: int + all_attempts_included: bool + independent_quality_review: bool + covered_languages: frozenset[str] + covered_categories: frozenset[str] + covered_tiers: frozenset[str] + valid_until: date + + def canonical_value(self) -> dict[str, object]: + return { + "schema_version": 1, + "card_id": self.card_id, + "identifiers_not_task_derived": True, + "cost_profile_fingerprint": self.cost_profile_fingerprint, + "measurement_contract_id": self.measurement_contract_id, + "vendor": self.vendor, + "tier": self.tier, + "baseline_route_fingerprint": self.baseline_route_fingerprint, + "candidate_route_fingerprint": self.candidate_route_fingerprint, + "status": self.status, + "sample_size": self.sample_size, + "cost_savings_lower_bound_basis_points": (self.cost_savings_lower_bound_basis_points), + "cost_savings_ci_width_basis_points": (self.cost_savings_ci_width_basis_points), + "quality_delta_lower_bound_basis_points": (self.quality_delta_lower_bound_basis_points), + "quality_noninferiority_margin_basis_points": ( + self.quality_noninferiority_margin_basis_points + ), + "new_critical_failures": self.new_critical_failures, + "all_attempts_included": self.all_attempts_included, + "independent_quality_review": self.independent_quality_review, + "covered_languages": sorted(self.covered_languages), + "covered_categories": sorted(self.covered_categories), + "covered_tiers": sorted(self.covered_tiers), + "valid_until": self.valid_until.isoformat(), + } + + +def _exact_mapping(value: object, expected_keys: frozenset[str]) -> dict[str, Any]: + if not isinstance(value, dict) or set(value) != expected_keys: + raise CostRecommendationError() + return cast(dict[str, Any], value) + + +def _reviewable_label(value: object) -> str: + if not isinstance(value, str): + raise CostRecommendationError() + try: + encoded = value.encode("utf-8") + except UnicodeEncodeError: + raise CostRecommendationError() from None + if not 1 <= len(encoded) <= 128 or any( + character.isspace() or not character.isprintable() for character in value + ): + raise CostRecommendationError() + return value + + +def _fingerprint(value: object) -> str: + if not isinstance(value, str) or _FINGERPRINT_PATTERN.fullmatch(value) is None: + raise CostRecommendationError() + return value + + +def _integer(value: object, *, minimum: int, maximum: int) -> int: + if isinstance(value, bool) or not isinstance(value, int) or not minimum <= value <= maximum: + raise CostRecommendationError() + return value + + +def _boolean(value: object) -> bool: + if not isinstance(value, bool): + raise CostRecommendationError() + return value + + +def _string_set(value: object, allowed: frozenset[str]) -> frozenset[str]: + if not isinstance(value, list) or not value: + raise CostRecommendationError() + parsed = tuple(_reviewable_label(item) for item in value) + if len(set(parsed)) != len(parsed) or not set(parsed) <= allowed: + raise CostRecommendationError() + return frozenset(parsed) + + +def parse_cost_profile(value: object) -> CostProfile: + profile = _exact_mapping( + value, + frozenset( + { + "schema_version", + "profile_id", + "measurement_contract_id", + "unit", + "identifiers_not_task_derived", + "pricing_inferred", + "actual_billing_claimed", + "routes", + } + ), + ) + if profile["schema_version"] != 1: + raise CostRecommendationError() + if not _boolean(profile["identifiers_not_task_derived"]): + raise CostRecommendationError() + if _boolean(profile["pricing_inferred"]) or _boolean(profile["actual_billing_claimed"]): + raise CostRecommendationError() + raw_routes = profile["routes"] + if not isinstance(raw_routes, list) or not 1 <= len(raw_routes) <= MAX_COST_ROUTES: + raise CostRecommendationError() + routes = [] + for raw_route in raw_routes: + route = _exact_mapping( + raw_route, + frozenset({"route_fingerprint", "expected_completed_cost_units"}), + ) + routes.append( + CostRouteEstimate( + route_fingerprint=_fingerprint(route["route_fingerprint"]), + expected_completed_cost_units=_integer( + route["expected_completed_cost_units"], + minimum=0, + maximum=2**63 - 1, + ), + ) + ) + if len({route.route_fingerprint for route in routes}) != len(routes): + raise CostRecommendationError() + return CostProfile( + profile_id=_reviewable_label(profile["profile_id"]), + measurement_contract_id=_reviewable_label(profile["measurement_contract_id"]), + unit=_reviewable_label(profile["unit"]), + routes=tuple(routes), + ) + + +def parse_qualification_card(value: object) -> QualificationCard: + card = _exact_mapping( + value, + frozenset( + { + "schema_version", + "card_id", + "identifiers_not_task_derived", + "cost_profile_fingerprint", + "measurement_contract_id", + "vendor", + "tier", + "baseline_route_fingerprint", + "candidate_route_fingerprint", + "status", + "sample_size", + "cost_savings_lower_bound_basis_points", + "cost_savings_ci_width_basis_points", + "quality_delta_lower_bound_basis_points", + "quality_noninferiority_margin_basis_points", + "new_critical_failures", + "all_attempts_included", + "independent_quality_review", + "covered_languages", + "covered_categories", + "covered_tiers", + "valid_until", + } + ), + ) + if card["schema_version"] != 1: + raise CostRecommendationError() + if not _boolean(card["identifiers_not_task_derived"]): + raise CostRecommendationError() + tier = _reviewable_label(card["tier"]) + status = _reviewable_label(card["status"]) + if tier not in REQUIRED_TIERS or status not in {"qualified", "provisional", "unqualified"}: + raise CostRecommendationError() + valid_until_value = card["valid_until"] + if not isinstance(valid_until_value, str): + raise CostRecommendationError() + try: + valid_until = date.fromisoformat(valid_until_value) + except ValueError: + raise CostRecommendationError() from None + if valid_until.isoformat() != valid_until_value: + raise CostRecommendationError() + return QualificationCard( + card_id=_reviewable_label(card["card_id"]), + cost_profile_fingerprint=_fingerprint(card["cost_profile_fingerprint"]), + measurement_contract_id=_reviewable_label(card["measurement_contract_id"]), + vendor=_reviewable_label(card["vendor"]), + tier=cast(Tier, tier), + baseline_route_fingerprint=_fingerprint(card["baseline_route_fingerprint"]), + candidate_route_fingerprint=_fingerprint(card["candidate_route_fingerprint"]), + status=cast(QualificationStatus, status), + sample_size=_integer(card["sample_size"], minimum=0, maximum=1_000_000), + cost_savings_lower_bound_basis_points=_integer( + card["cost_savings_lower_bound_basis_points"], + minimum=-1_000_000_000, + maximum=10_000, + ), + cost_savings_ci_width_basis_points=_integer( + card["cost_savings_ci_width_basis_points"], + minimum=0, + maximum=1_000_000_000, + ), + quality_delta_lower_bound_basis_points=_integer( + card["quality_delta_lower_bound_basis_points"], + minimum=-10_000, + maximum=10_000, + ), + quality_noninferiority_margin_basis_points=_integer( + card["quality_noninferiority_margin_basis_points"], + minimum=0, + maximum=10_000, + ), + new_critical_failures=_integer(card["new_critical_failures"], minimum=0, maximum=1_000_000), + all_attempts_included=_boolean(card["all_attempts_included"]), + independent_quality_review=_boolean(card["independent_quality_review"]), + covered_languages=_string_set(card["covered_languages"], REQUIRED_LANGUAGES), + covered_categories=_string_set(card["covered_categories"], REQUIRED_CATEGORIES), + covered_tiers=_string_set(card["covered_tiers"], REQUIRED_TIERS), + valid_until=valid_until, + ) + + +def load_cost_profile(path: Path) -> CostProfile: + try: + value = load_json_object( + path, + max_bytes=MAX_COST_INPUT_BYTES, + require_exclusive_write_owner=True, + ) + except JsonInputError: + raise CostRecommendationError() from None + return parse_cost_profile(value) + + +def load_qualification_card(path: Path) -> QualificationCard: + try: + value = load_json_object( + path, + max_bytes=MAX_COST_INPUT_BYTES, + require_exclusive_write_owner=True, + ) + except JsonInputError: + raise CostRecommendationError() from None + return parse_qualification_card(value) + + +def cost_profile_fingerprint(profile: CostProfile) -> str: + encoded = json.dumps( + profile.canonical_value(), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return f"sha256:{hashlib.sha256(encoded).hexdigest()}" + + +def qualification_card_fingerprint(card: QualificationCard) -> str: + encoded = json.dumps( + card.canonical_value(), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return f"sha256:{hashlib.sha256(encoded).hexdigest()}" + + +def build_cost_profile_review(profile: CostProfile) -> dict[str, object]: + """Return task-free binding metadata for a user-supplied cost profile.""" + return { + "schema_version": 1, + "profile_id": profile.profile_id, + "fingerprint": cost_profile_fingerprint(profile), + "measurement_contract_id": profile.measurement_contract_id, + "unit": profile.unit, + "identifiers_not_task_derived": True, + "pricing_inferred": False, + "actual_billing_claimed": False, + "values_verified_by_router": False, + "routes": [ + { + "route_fingerprint": route.route_fingerprint, + "expected_completed_cost_units": route.expected_completed_cost_units, + } + for route in profile.routes + ], + } + + +def _capability(vendor: str) -> dict[str, bool]: + return { + "reviewed_effort_routing": vendor in {"agy", "claude", "codex", "grok"}, + "tier_effort_override": vendor in {"claude", "codex"}, + "tier_model_override": vendor in {"claude", "codex", "grok"}, + } + + +def _route_receipt( + route: Route, + route_fingerprint: str, + expected_completed_cost_units: int | None, +) -> dict[str, object]: + receipt: dict[str, object] = { + "command": list(route.command), + "expected_completed_cost_units": expected_completed_cost_units, + "route": route.route_id, + "route_fingerprint": route_fingerprint, + "task_delivery": "argv" if uses_argv_task_delivery(route.command) else "stdin", + "tier": route.tier, + "vendor": route.vendor, + } + return receipt + + +def _qualification_reason( + profile: CostProfile, + card: QualificationCard, + *, + vendor: str, + tier: Tier, + baseline_route_fingerprint: str, + candidate_route_fingerprint: str, + today: date, +) -> str | None: + if card.status != "qualified": + return "qualification_not_qualified" + if card.valid_until < today: + return "qualification_expired" + if card.cost_profile_fingerprint != cost_profile_fingerprint(profile): + return "qualification_profile_mismatch" + if card.measurement_contract_id != profile.measurement_contract_id: + return "qualification_measurement_mismatch" + if card.vendor != vendor: + return "qualification_vendor_mismatch" + if card.tier != tier: + return "qualification_tier_mismatch" + if ( + card.baseline_route_fingerprint != baseline_route_fingerprint + or card.candidate_route_fingerprint != candidate_route_fingerprint + ): + return "qualification_route_mismatch" + if card.sample_size < MIN_QUALIFIED_PAIRS: + return "qualification_insufficient_samples" + if card.cost_savings_lower_bound_basis_points < MIN_COST_SAVINGS_LOWER_BOUND_BASIS_POINTS: + return "qualification_insufficient_cost_savings" + if card.cost_savings_ci_width_basis_points > MAX_COST_SAVINGS_CI_WIDTH_BASIS_POINTS: + return "qualification_cost_interval_too_wide" + if ( + card.quality_noninferiority_margin_basis_points + > MAX_QUALITY_NONINFERIORITY_MARGIN_BASIS_POINTS + or card.quality_delta_lower_bound_basis_points + < -card.quality_noninferiority_margin_basis_points + ): + return "qualification_quality_not_noninferior" + if card.new_critical_failures: + return "qualification_new_critical_failure" + if not card.all_attempts_included: + return "qualification_incomplete_attempts" + if not card.independent_quality_review: + return "qualification_missing_independent_quality" + if ( + card.covered_languages != REQUIRED_LANGUAGES + or card.covered_categories != REQUIRED_CATEGORIES + or card.covered_tiers != REQUIRED_TIERS + ): + return "qualification_incomplete_coverage" + return None + + +def build_recommendation_receipt( + profile: CostProfile, + card: QualificationCard, + *, + baseline_route: Route, + baseline_route_fingerprint: str, + candidate_route: Route, + candidate_route_fingerprint: str, + routing_reason_code: str, + candidate_configuration_status: str, + today: date | None = None, +) -> dict[str, object]: + """Return a task-free recommendation or an explicit abstention.""" + if baseline_route.tier is None or candidate_route.tier is None: + raise CostRecommendationError() + if baseline_route.tier != candidate_route.tier: + raise CostRecommendationError() + tier = baseline_route.tier + vendor = candidate_route.vendor + baseline_cost = profile.cost_for(baseline_route_fingerprint) + candidate_cost = profile.cost_for(candidate_route_fingerprint) + reason = _qualification_reason( + profile, + card, + vendor=vendor, + tier=tier, + baseline_route_fingerprint=baseline_route_fingerprint, + candidate_route_fingerprint=candidate_route_fingerprint, + today=today or date.today(), + ) + profile_savings_basis_points: int | None = None + if reason is None and (baseline_cost is None or candidate_cost is None): + reason = "profile_route_cost_missing" + if reason is None: + assert baseline_cost is not None + assert candidate_cost is not None + if baseline_cost == 0: + reason = "no_cost_advantage" + else: + profile_savings_basis_points = ( + (baseline_cost - candidate_cost) * 10_000 // baseline_cost + ) + if candidate_cost >= baseline_cost: + reason = "no_cost_advantage" + elif card.cost_savings_lower_bound_basis_points > profile_savings_basis_points: + reason = "qualification_cost_inconsistent" + + decision = "recommend" if reason is None else "abstain" + reason_code = "qualified_cost_advantage" if reason is None else reason + recommendation_semantics = { + "schema_version": 1, + "algorithm": "expected-completed-cost-v1", + "baseline_route_fingerprint": baseline_route_fingerprint, + "candidate_route_fingerprint": candidate_route_fingerprint, + "card_id": card.card_id, + "cost_profile_fingerprint": cost_profile_fingerprint(profile), + "qualification_card_fingerprint": qualification_card_fingerprint(card), + "decision": decision, + "reason_code": reason_code, + "tier": tier, + "vendor": vendor, + } + encoded = json.dumps( + recommendation_semantics, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + recommendation_fingerprint = f"sha256:{hashlib.sha256(encoded).hexdigest()}" + return { + "schema_version": 1, + "decision": decision, + "reason_code": reason_code, + "objective": "expected_completed_cost", + "fallback": "none", + "recommendation_only": True, + "recommendation_fingerprint": recommendation_fingerprint, + "routing_reason_code": routing_reason_code, + "cost_profile": { + "actual_billing_claimed": False, + "fingerprint": cost_profile_fingerprint(profile), + "id": profile.profile_id, + "measurement_contract_id": profile.measurement_contract_id, + "pricing_inferred": False, + "profile_cost_savings_basis_points": profile_savings_basis_points, + "unit": profile.unit, + }, + "qualification": { + "all_attempts_included": card.all_attempts_included, + "assertions_verified_by_router": False, + "card_id": card.card_id, + "cost_savings_ci_width_basis_points": (card.cost_savings_ci_width_basis_points), + "cost_savings_lower_bound_basis_points": (card.cost_savings_lower_bound_basis_points), + "fingerprint": qualification_card_fingerprint(card), + "independent_quality_review": card.independent_quality_review, + "identifiers_not_task_derived": True, + "new_critical_failures": card.new_critical_failures, + "quality_delta_lower_bound_basis_points": (card.quality_delta_lower_bound_basis_points), + "quality_noninferiority_margin_basis_points": ( + card.quality_noninferiority_margin_basis_points + ), + "sample_size": card.sample_size, + "status": card.status, + "valid_until": card.valid_until.isoformat(), + }, + "capability": _capability(vendor), + "candidate_configuration_status": candidate_configuration_status, + "baseline": _route_receipt( + baseline_route, + baseline_route_fingerprint, + baseline_cost, + ), + "candidate": _route_receipt( + candidate_route, + candidate_route_fingerprint, + candidate_cost, + ), + } diff --git a/tests/test_ci_workflow_structure.py b/tests/test_ci_workflow_structure.py index b565952..bcc7b9e 100644 --- a/tests/test_ci_workflow_structure.py +++ b/tests/test_ci_workflow_structure.py @@ -12,6 +12,7 @@ "tests.test_process_context", "tests.test_triage", "tests.test_json_input", + "tests.test_cost_recommendation", "tests.test_delegation_runtime", "tests.test_delegation_v2_runtime", "tests.test_guarded_runtime_suite", diff --git a/tests/test_cost_recommendation.py b/tests/test_cost_recommendation.py new file mode 100644 index 0000000..d9052c4 --- /dev/null +++ b/tests/test_cost_recommendation.py @@ -0,0 +1,472 @@ +import contextlib +import hashlib +import io +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from typing import Any, cast +from unittest import mock + +from weightclass import cli + + +def _weightclass(*arguments: str, task: str) -> subprocess.CompletedProcess[str]: + environment = os.environ.copy() + environment["PATH"] = "/nonexistent" + return subprocess.run( + [sys.executable, "-m", "weightclass", *arguments], + capture_output=True, + check=False, + env=environment, + input=task, + text=True, + ) + + +def _profile_fingerprint(profile: dict[str, Any]) -> str: + encoded = json.dumps( + profile, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return "sha256:" + hashlib.sha256(encoded).hexdigest() + + +def _route(*arguments: str) -> dict[str, Any]: + completed = _weightclass("route", *arguments, "--tier", "low", task="Fix a typo.") + if completed.returncode != 0: + raise AssertionError(completed.stderr) + parsed = json.loads(completed.stdout) + if not isinstance(parsed, dict): + raise AssertionError("route output was not an object") + return cast(dict[str, Any], parsed) + + +def _cost_profile(baseline: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]: + return { + "schema_version": 1, + "profile_id": "team-cost-v1", + "measurement_contract_id": "reviewed-cost-units-v1", + "unit": "reviewed-cost-unit", + "identifiers_not_task_derived": True, + "pricing_inferred": False, + "actual_billing_claimed": False, + "routes": [ + { + "route_fingerprint": baseline["route_fingerprint"], + "expected_completed_cost_units": 100, + }, + { + "route_fingerprint": candidate["route_fingerprint"], + "expected_completed_cost_units": 40, + }, + ], + } + + +def _qualification_card( + profile: dict[str, Any], baseline: dict[str, Any], candidate: dict[str, Any] +) -> dict[str, Any]: + return { + "schema_version": 1, + "card_id": f"{candidate['vendor']}-low-qualified-v1", + "identifiers_not_task_derived": True, + "cost_profile_fingerprint": _profile_fingerprint(profile), + "measurement_contract_id": profile["measurement_contract_id"], + "vendor": candidate["vendor"], + "tier": "low", + "baseline_route_fingerprint": baseline["route_fingerprint"], + "candidate_route_fingerprint": candidate["route_fingerprint"], + "status": "qualified", + "sample_size": 90, + "cost_savings_lower_bound_basis_points": 4_628, + "cost_savings_ci_width_basis_points": 1_772, + "quality_delta_lower_bound_basis_points": -402, + "quality_noninferiority_margin_basis_points": 500, + "new_critical_failures": 0, + "all_attempts_included": True, + "independent_quality_review": True, + "covered_languages": ["en", "ko"], + "covered_categories": [ + "concurrency", + "data-integrity", + "destructive-work", + "migration", + "performance", + "privacy", + "reliability", + "routine", + "security", + ], + "covered_tiers": ["low", "standard", "high"], + "valid_until": "9999-12-31", + } + + +def _recommend( + preset: str, + profile: dict[str, Any], + card: dict[str, Any], + *, + task: str = "Fix a typo.", + tier: str = "low", + extra_arguments: tuple[str, ...] = (), +) -> subprocess.CompletedProcess[str]: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + profile_path = root / "profile.json" + card_path = root / "card.json" + profile_path.write_text(json.dumps(profile), encoding="utf-8") + card_path.write_text(json.dumps(card), encoding="utf-8") + return _weightclass( + "recommend", + "--preset", + preset, + "--cost-profile", + str(profile_path), + "--qualification-card", + str(card_path), + "--tier", + tier, + *extra_arguments, + task=task, + ) + + +class CostRecommendationCliTests(unittest.TestCase): + def test_reviews_a_cost_profile_without_reading_a_task(self) -> None: + """Breaks if evidence binding requires an undocumented external hash tool.""" + baseline = {"route_fingerprint": "sha256:" + "1" * 64} + candidate = { + "route_fingerprint": "sha256:" + "2" * 64, + "vendor": "claude", + } + profile = _cost_profile(baseline, candidate) + with tempfile.TemporaryDirectory() as directory: + profile_path = Path(directory) / "profile.json" + profile_path.write_text(json.dumps(profile), encoding="utf-8") + output = io.StringIO() + with ( + mock.patch.object( + cli, + "read_task_from_standard_input", + side_effect=AssertionError("task input was read"), + ), + contextlib.redirect_stdout(output), + ): + exit_code = cli.main(["review-cost-profile", "--cost-profile", str(profile_path)]) + + self.assertEqual(exit_code, 0) + review = json.loads(output.getvalue()) + self.assertEqual(review["fingerprint"], _profile_fingerprint(profile)) + self.assertEqual(review["profile_id"], "team-cost-v1") + self.assertEqual(len(review["routes"]), 2) + self.assertTrue(review["identifiers_not_task_derived"]) + self.assertFalse(review["values_verified_by_router"]) + + def test_recommends_a_qualified_lower_cost_route_without_starting_a_vendor(self) -> None: + """Breaks if advisory cost routing stops binding qualified exact routes.""" + baseline = _route("--source-vendor", "claude") + candidate = _route("--preset", "claude-cost-focused") + profile = _cost_profile(baseline, candidate) + card = _qualification_card(profile, baseline, candidate) + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + profile_path = root / "profile.json" + card_path = root / "card.json" + profile_path.write_text(json.dumps(profile), encoding="utf-8") + card_path.write_text(json.dumps(card), encoding="utf-8") + + completed = _weightclass( + "recommend", + "--preset", + "claude-cost-focused", + "--cost-profile", + str(profile_path), + "--qualification-card", + str(card_path), + "--tier", + "low", + task="Fix a typo.", + ) + + self.assertEqual(completed.returncode, 0, completed.stderr) + receipt = json.loads(completed.stdout) + self.assertEqual(receipt["decision"], "recommend") + self.assertEqual(receipt["reason_code"], "qualified_cost_advantage") + self.assertEqual(receipt["objective"], "expected_completed_cost") + self.assertEqual(receipt["fallback"], "none") + self.assertTrue(receipt["recommendation_only"]) + self.assertEqual(receipt["baseline"]["route_fingerprint"], baseline["route_fingerprint"]) + self.assertEqual(receipt["candidate"]["route_fingerprint"], candidate["route_fingerprint"]) + self.assertEqual(receipt["baseline"]["expected_completed_cost_units"], 100) + self.assertEqual(receipt["candidate"]["expected_completed_cost_units"], 40) + self.assertEqual(receipt["candidate"]["task_delivery"], "stdin") + self.assertEqual(receipt["qualification"]["sample_size"], 90) + self.assertEqual( + receipt["qualification"]["cost_savings_lower_bound_basis_points"], + 4_628, + ) + self.assertEqual( + receipt["qualification"]["quality_delta_lower_bound_basis_points"], + -402, + ) + self.assertFalse(receipt["qualification"]["assertions_verified_by_router"]) + self.assertTrue(receipt["recommendation_fingerprint"].startswith("sha256:")) + self.assertNotIn("Fix a typo.", completed.stdout) + + def test_abstains_when_the_reviewed_cost_profile_has_no_cost_advantage(self) -> None: + """Breaks if a flat subscription-cost scenario is reported as a saving.""" + baseline = _route("--source-vendor", "claude") + candidate = _route("--preset", "claude-cost-focused") + profile = _cost_profile(baseline, candidate) + profile["routes"][1]["expected_completed_cost_units"] = 100 + card = _qualification_card(profile, baseline, candidate) + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + profile_path = root / "profile.json" + card_path = root / "card.json" + profile_path.write_text(json.dumps(profile), encoding="utf-8") + card_path.write_text(json.dumps(card), encoding="utf-8") + completed = _weightclass( + "recommend", + "--preset", + "claude-cost-focused", + "--cost-profile", + str(profile_path), + "--qualification-card", + str(card_path), + "--tier", + "low", + task="Fix a typo.", + ) + + self.assertEqual(completed.returncode, 0, completed.stderr) + receipt = json.loads(completed.stdout) + self.assertEqual(receipt["decision"], "abstain") + self.assertEqual(receipt["reason_code"], "no_cost_advantage") + self.assertEqual(receipt["cost_profile"]["profile_cost_savings_basis_points"], 0) + + def test_recommendation_fingerprint_binds_the_entire_qualification_card(self) -> None: + """Breaks if evidence can drift without invalidating its recommendation.""" + baseline = _route("--source-vendor", "claude") + candidate = _route("--preset", "claude-cost-focused") + profile = _cost_profile(baseline, candidate) + first_card = _qualification_card(profile, baseline, candidate) + second_card = dict(first_card) + second_card["sample_size"] = 91 + + first = _recommend("claude-cost-focused", profile, first_card) + second = _recommend("claude-cost-focused", profile, second_card) + + self.assertEqual(first.returncode, 0, first.stderr) + self.assertEqual(second.returncode, 0, second.stderr) + first_receipt = json.loads(first.stdout) + second_receipt = json.loads(second.stdout) + self.assertTrue(first_receipt["qualification"]["fingerprint"].startswith("sha256:")) + self.assertNotEqual( + first_receipt["qualification"]["fingerprint"], + second_receipt["qualification"]["fingerprint"], + ) + self.assertNotEqual( + first_receipt["recommendation_fingerprint"], + second_receipt["recommendation_fingerprint"], + ) + + def test_abstains_when_qualification_is_bound_to_another_route(self) -> None: + """Breaks if stale evidence can authorize a changed candidate command.""" + baseline = _route("--source-vendor", "claude") + candidate = _route("--preset", "claude-cost-focused") + profile = _cost_profile(baseline, candidate) + card = _qualification_card(profile, baseline, candidate) + card["candidate_route_fingerprint"] = "sha256:" + "0" * 64 + + completed = _recommend("claude-cost-focused", profile, card) + + self.assertEqual(completed.returncode, 0, completed.stderr) + receipt = json.loads(completed.stdout) + self.assertEqual(receipt["decision"], "abstain") + self.assertEqual(receipt["reason_code"], "qualification_route_mismatch") + + def test_reports_each_vendor_capability_and_task_delivery_without_execution(self) -> None: + """Breaks if provider differences are normalized into unsupported claims.""" + expected = { + "agy": { + "task_delivery": "argv", + "tier_effort_override": False, + "tier_model_override": False, + }, + "claude": { + "task_delivery": "stdin", + "tier_effort_override": True, + "tier_model_override": True, + }, + "codex": { + "task_delivery": "stdin", + "tier_effort_override": True, + "tier_model_override": True, + }, + "grok": { + "task_delivery": "argv", + "tier_effort_override": False, + "tier_model_override": True, + }, + } + + for vendor, expected_fields in expected.items(): + with self.subTest(vendor=vendor): + preset = f"{vendor}-cost-focused" + baseline = _route("--source-vendor", vendor) + candidate = _route("--preset", preset) + profile = _cost_profile(baseline, candidate) + card = _qualification_card(profile, baseline, candidate) + + completed = _recommend(preset, profile, card) + + self.assertEqual(completed.returncode, 0, completed.stderr) + receipt = json.loads(completed.stdout) + self.assertEqual(receipt["decision"], "recommend") + self.assertTrue(receipt["capability"]["reviewed_effort_routing"]) + self.assertEqual( + receipt["capability"]["tier_effort_override"], + expected_fields["tier_effort_override"], + ) + self.assertEqual( + receipt["capability"]["tier_model_override"], + expected_fields["tier_model_override"], + ) + self.assertEqual( + receipt["candidate"]["task_delivery"], + expected_fields["task_delivery"], + ) + self.assertNotIn("Fix a typo.", completed.stdout) + + def test_abstains_for_each_failed_qualification_gate(self) -> None: + """Breaks if a weak or stale qualification card can become a recommendation.""" + baseline = _route("--source-vendor", "claude") + candidate = _route("--preset", "claude-cost-focused") + profile = _cost_profile(baseline, candidate) + valid_card = _qualification_card(profile, baseline, candidate) + cases: tuple[tuple[str, str, object], ...] = ( + ("status", "qualification_not_qualified", "provisional"), + ("valid_until", "qualification_expired", "2000-01-01"), + ( + "cost_profile_fingerprint", + "qualification_profile_mismatch", + "sha256:" + "0" * 64, + ), + ( + "measurement_contract_id", + "qualification_measurement_mismatch", + "other-contract", + ), + ("vendor", "qualification_vendor_mismatch", "codex"), + ("tier", "qualification_tier_mismatch", "standard"), + ("sample_size", "qualification_insufficient_samples", 29), + ( + "cost_savings_lower_bound_basis_points", + "qualification_insufficient_cost_savings", + 1_499, + ), + ( + "cost_savings_ci_width_basis_points", + "qualification_cost_interval_too_wide", + 2_001, + ), + ( + "quality_delta_lower_bound_basis_points", + "qualification_quality_not_noninferior", + -501, + ), + ( + "quality_noninferiority_margin_basis_points", + "qualification_quality_not_noninferior", + 501, + ), + ("new_critical_failures", "qualification_new_critical_failure", 1), + ("all_attempts_included", "qualification_incomplete_attempts", False), + ( + "independent_quality_review", + "qualification_missing_independent_quality", + False, + ), + ("covered_languages", "qualification_incomplete_coverage", ["en"]), + ) + + for field, reason_code, value in cases: + with self.subTest(field=field): + card = dict(valid_card) + card[field] = value + + completed = _recommend("claude-cost-focused", profile, card) + + self.assertEqual(completed.returncode, 0, completed.stderr) + receipt = json.loads(completed.stdout) + self.assertEqual(receipt["decision"], "abstain") + self.assertEqual(receipt["reason_code"], reason_code) + + def test_rejects_an_invalid_cost_document_before_reading_task_input(self) -> None: + """Breaks if unsafe cost metadata can cause task input to be consumed first.""" + baseline = {"route_fingerprint": "sha256:" + "1" * 64} + candidate = { + "route_fingerprint": "sha256:" + "2" * 64, + "vendor": "claude", + } + valid_profile = _cost_profile(baseline, candidate) + valid_card = _qualification_card(valid_profile, baseline, candidate) + profile_with_task = dict(valid_profile) + profile_with_task["task"] = "must-not-be-accepted" + profile_with_derived_ids = dict(valid_profile) + profile_with_derived_ids["identifiers_not_task_derived"] = False + card_with_task = dict(valid_card) + card_with_task["task"] = "must-not-be-accepted" + card_with_derived_ids = dict(valid_card) + card_with_derived_ids["identifiers_not_task_derived"] = False + + for profile, card in ( + (profile_with_task, valid_card), + (profile_with_derived_ids, valid_card), + (valid_profile, card_with_task), + (valid_profile, card_with_derived_ids), + ): + with self.subTest(): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + profile_path = root / "profile.json" + card_path = root / "card.json" + profile_path.write_text(json.dumps(profile), encoding="utf-8") + card_path.write_text(json.dumps(card), encoding="utf-8") + errors = io.StringIO() + with ( + mock.patch.object( + cli, + "read_task_from_standard_input", + side_effect=AssertionError("task input was read"), + ), + contextlib.redirect_stderr(errors), + ): + exit_code = cli.main( + [ + "recommend", + "--preset", + "claude-cost-focused", + "--cost-profile", + str(profile_path), + "--qualification-card", + str(card_path), + ] + ) + + self.assertEqual(exit_code, 2) + self.assertEqual(errors.getvalue(), '{"error": "invalid_input"}\n') + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_release_workflow_structure.py b/tests/test_release_workflow_structure.py index b70e38d..7a66a41 100644 --- a/tests/test_release_workflow_structure.py +++ b/tests/test_release_workflow_structure.py @@ -29,6 +29,8 @@ def test_boundary_validators_run_goldens_from_exact_installed_wheel(self) -> Non self.assertIn('python -m venv "$installed_venv"', block) self.assertIn('"$installed_venv/bin/wclass" --version', block) self.assertIn('"$installed_venv/bin/wclass" classify', block) + self.assertIn('"$installed_venv/bin/wclass" review-cost-profile --help', block) + self.assertIn('"$installed_venv/bin/wclass" recommend --help', block) for policy_name in ( "agy-cost-focused", "claude-cost-focused", From 885baf7e78fd224d9d3019dbbce2ab030efdf727 Mon Sep 17 00:00:00 2001 From: Coden Date: Thu, 13 Aug 2026 20:33:07 +0900 Subject: [PATCH 2/2] fix: bind cost receipts to exact routes Reject non-integer schema versions and verify that recommendation receipt commands match their reviewed route fingerprints. --- HANDOFF.md | 11 +++-- src/weightclass/cli.py | 3 ++ src/weightclass/cost_recommendation.py | 19 ++++++-- tests/test_cost_recommendation.py | 65 ++++++++++++++++++++++++++ 4 files changed, 89 insertions(+), 9 deletions(-) diff --git a/HANDOFF.md b/HANDOFF.md index 9cb524c..98cddab 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -27,10 +27,10 @@ _Last updated: 2026-08-13 19:35 KST by Codex_ security/performance and release hardening described below. Custom selections remain explicitly unqualified; built-in routes and provider boundaries are unchanged. -- The current `main` worktree has an unreleased cost-recommendation feature: +- The current unreleased code adds the cost-recommendation feature: task-free `review-cost-profile` plus non-executing `recommend`. It is not - committed, tagged, published, or installed through Homebrew yet. Built-in - routes and every execution path remain unchanged. + tagged, published, or installed through Homebrew yet. Built-in routes and + every execution path remain unchanged. - The source-of-truth Homebrew formula was updated on `main` by `e18fd3988a6fa6ad642f44a084b436c6507a2f6b`. The matching tap commit is [`1244ed6`](https://github.com/ictechgy/homebrew-tap/commit/1244ed6f0baf6cf2c557bec356230e3ea199961d). @@ -415,8 +415,11 @@ _Last updated: 2026-08-13 19:35 KST by Codex_ abstention, configuration-before-task ordering, full qualification-card fingerprint binding, stale route binding, every fixed gate, all four provider capability/task-delivery receipts, and task-free profile review. + Review follow-up additionally rejects non-integer schema versions and a + route object whose displayed command is not bound by its supplied route + fingerprint. - Python 3.10.20 and Python 3.14.6 full `unittest` suites with - `ResourceWarning` as an error: 808 tests passed on each interpreter. + `ResourceWarning` as an error: 810 tests passed on each interpreter. - Ruff 0.16.2 check/format over 125 files, strict mypy over 105 source files, compileall, and `git diff --check` passed. - Offline wheel/sdist build and strict Twine checks passed. The artifacts diff --git a/src/weightclass/cli.py b/src/weightclass/cli.py index 7ee21b0..6728362 100644 --- a/src/weightclass/cli.py +++ b/src/weightclass/cli.py @@ -1323,8 +1323,11 @@ def recommend_from_standard_input( card, baseline_route=baseline_route, baseline_route_fingerprint=baseline_fingerprint, + baseline_allow_mixed_vendors=False, candidate_route=candidate_route, candidate_route_fingerprint=candidate_fingerprint, + candidate_allow_mixed_vendors=candidate_policy.allow_mixed_vendors, + candidate_posture=candidate_policy.posture, routing_reason_code=routing_reason_code, candidate_configuration_status=_packaged_configuration_status(preset, model, overrides), ) diff --git a/src/weightclass/cost_recommendation.py b/src/weightclass/cost_recommendation.py index 1fe73b0..04cc0e6 100644 --- a/src/weightclass/cost_recommendation.py +++ b/src/weightclass/cost_recommendation.py @@ -12,7 +12,7 @@ from .classification import Tier from .json_input import JsonInputError, load_json_object -from .router import Route, uses_argv_task_delivery +from .router import Posture, Route, native_route_fingerprint, uses_argv_task_delivery MAX_COST_INPUT_BYTES: Final = 262_144 MAX_COST_ROUTES: Final = 128 @@ -201,8 +201,7 @@ def parse_cost_profile(value: object) -> CostProfile: } ), ) - if profile["schema_version"] != 1: - raise CostRecommendationError() + _integer(profile["schema_version"], minimum=1, maximum=1) if not _boolean(profile["identifiers_not_task_derived"]): raise CostRecommendationError() if _boolean(profile["pricing_inferred"]) or _boolean(profile["actual_billing_claimed"]): @@ -266,8 +265,7 @@ def parse_qualification_card(value: object) -> QualificationCard: } ), ) - if card["schema_version"] != 1: - raise CostRecommendationError() + _integer(card["schema_version"], minimum=1, maximum=1) if not _boolean(card["identifiers_not_task_derived"]): raise CostRecommendationError() tier = _reviewable_label(card["tier"]) @@ -475,8 +473,11 @@ def build_recommendation_receipt( *, baseline_route: Route, baseline_route_fingerprint: str, + baseline_allow_mixed_vendors: bool, candidate_route: Route, candidate_route_fingerprint: str, + candidate_allow_mixed_vendors: bool, + candidate_posture: Posture | None, routing_reason_code: str, candidate_configuration_status: str, today: date | None = None, @@ -486,6 +487,14 @@ def build_recommendation_receipt( raise CostRecommendationError() if baseline_route.tier != candidate_route.tier: raise CostRecommendationError() + if baseline_route_fingerprint != native_route_fingerprint( + baseline_route, baseline_allow_mixed_vendors + ) or candidate_route_fingerprint != native_route_fingerprint( + candidate_route, + candidate_allow_mixed_vendors, + candidate_posture, + ): + raise CostRecommendationError() tier = baseline_route.tier vendor = candidate_route.vendor baseline_cost = profile.cost_for(baseline_route_fingerprint) diff --git a/tests/test_cost_recommendation.py b/tests/test_cost_recommendation.py index d9052c4..d96d8a4 100644 --- a/tests/test_cost_recommendation.py +++ b/tests/test_cost_recommendation.py @@ -12,6 +12,13 @@ from unittest import mock from weightclass import cli +from weightclass.cost_recommendation import ( + CostRecommendationError, + build_recommendation_receipt, + parse_cost_profile, + parse_qualification_card, +) +from weightclass.router import Route def _weightclass(*arguments: str, task: str) -> subprocess.CompletedProcess[str]: @@ -139,6 +146,64 @@ def _recommend( class CostRecommendationCliTests(unittest.TestCase): + def test_rejects_non_integer_schema_versions(self) -> None: + """Breaks if JSON booleans or floats can impersonate schema version 1.""" + baseline = {"route_fingerprint": "sha256:" + "1" * 64} + candidate = { + "route_fingerprint": "sha256:" + "2" * 64, + "vendor": "claude", + } + profile = _cost_profile(baseline, candidate) + card = _qualification_card(profile, baseline, candidate) + + for parser, document in ( + (parse_cost_profile, profile), + (parse_qualification_card, card), + ): + for invalid_version in (True, 1.0): + with self.subTest(parser=parser.__name__, value=invalid_version): + invalid_document = dict(document) + invalid_document["schema_version"] = invalid_version + + with self.assertRaises(CostRecommendationError): + parser(invalid_document) + + def test_rejects_route_objects_that_do_not_match_their_fingerprints(self) -> None: + """Breaks if a recommendation can display a command not bound by its fingerprint.""" + baseline = _route("--source-vendor", "claude") + candidate = _route("--preset", "claude-cost-focused") + profile_document = _cost_profile(baseline, candidate) + card_document = _qualification_card(profile_document, baseline, candidate) + forged_candidate = Route( + route_id="forged-candidate", + vendor="claude", + workflow="", + command=("claude", "--forged-candidate"), + tier="low", + ) + baseline_route = Route( + route_id=baseline["route"], + vendor=baseline["vendor"], + workflow="", + command=tuple(baseline["command"]), + tier=baseline["tier"], + ) + + with self.assertRaises(CostRecommendationError): + build_recommendation_receipt( + parse_cost_profile(profile_document), + parse_qualification_card(card_document), + baseline_route=baseline_route, + baseline_route_fingerprint=baseline["route_fingerprint"], + baseline_allow_mixed_vendors=False, + candidate_route=forged_candidate, + candidate_route_fingerprint=candidate["route_fingerprint"], + candidate_allow_mixed_vendors=False, + candidate_posture="balanced", + routing_reason_code="explicit.requested_tier", + candidate_configuration_status="measured_low_route_only", + ) + def test_reviews_a_cost_profile_without_reading_a_task(self) -> None: """Breaks if evidence binding requires an undocumented external hash tool.""" baseline = {"route_fingerprint": "sha256:" + "1" * 64}