From c316283c8fc612990f74246958edaab9aaa4ec08 Mon Sep 17 00:00:00 2001 From: Timothy Wayne Gregg Date: Sat, 3 Oct 2026 08:50:44 -0400 Subject: [PATCH] Fit optional native Score cuts and Choice weights on calibration --- README.md | 8 +- docs/ARCHITECTURE.md | 30 +++ docs/ROADMAP.md | 8 +- schemas/hierarchy-artifact.schema.json | 45 ++++ schemas/hierarchy-source.schema.json | 45 ++++ schemas/program.schema.json | 45 ++++ src/typewright/cli.py | 8 +- src/typewright/compiler.py | 6 +- src/typewright/models.py | 39 +++- src/typewright/policy.py | 121 +++++++++- src/typewright/runtime.py | 17 ++ tests/test_policy_fitting.py | 298 +++++++++++++++++++++++++ 12 files changed, 653 insertions(+), 17 deletions(-) create mode 100644 tests/test_policy_fitting.py diff --git a/README.md b/README.md index 0ae0f76..7b403b8 100644 --- a/README.md +++ b/README.md @@ -59,7 +59,7 @@ budget-bounded, frozen artifact** that an application can run without DSPy. | Concern | DSPy 3.4 (experimental, per its docs and PRs) | Typewright | | --- | --- | --- | | Use Jev in a Python program | Yes: native client, async, and generative LMs mixed in one program. | Not its job. Jev only (mock and TypeSafe backends), no async. | -| Tune decision knobs | `ReAnchor` fits Noul thresholds, Score cuts, and Choice weights. | Fits Noul thresholds and review gates. Score cuts and Choice weights are planned ([#92](https://github.com/Jev-Engineering/TypeWright/issues/92)). | +| Tune decision knobs | `ReAnchor` fits Noul thresholds, Score cuts, and Choice weights. | Fits Noul thresholds and review gates; flat `compile --fit-decision-knobs` opts in to native Score cuts and Choice selection weights from cached calibration predictions. | | Search wording and structure | GEPA over instructions. Program decomposition (via generated code) is in an open upstream PR. | GEPA plus bounded structural proposals as validated JSON. No generated code. | | Held-out discipline | You supply the data; `ReAnchor` uses folds. | Four disjoint splits with leakage checks; test only after freeze. | | Review and abstention | None documented. | Fitted review gates with a minimum accepted count and an error constraint; `review_required` on every output. | @@ -365,7 +365,9 @@ comparisons, paired cluster inference, and replayable evidence. **Implemented:** typed declarations and JSON artifacts; a native TypeSafe SDK adapter; DSPy synthesis and mutation; a standalone GEPA adapter; bounded add/drop/revise and numeric-decomposition proposals; validation-based selection; -separate empirical policy fitting; projected inputs; request budgets; an optional +separate empirical policy fitting, including optional native Score cuts and Choice +selection weights from cached calibration predictions with flat +`compile --fit-decision-knobs` (off by default); projected inputs; request budgets; an optional sensitive-output cache; reports; schema export; hardening proposals; a no-key fixture; the hierarchy v1 source and frozen-graph format with strict evidence replay; and a preregistered study runner. @@ -373,7 +375,7 @@ replay; and a preregistered study runner. **Not claimed:** measured hierarchy accuracy gains; a completed independent live study; production certification; general multi-stage agent orchestration; arbitrary code generation; MIPRO integration; learned posterior calibration; a -supervised weight fitter; automatic release approval; a TypeScript runtime; +supervised composition-weight fitter; automatic release approval; a TypeScript runtime; pretrained weights; or a hosted service. The artifact stays JSON all the way down: no `eval`, generated Python, dynamic diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 4118702..a1ecc85 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -179,6 +179,36 @@ accepted count and an empirical error constraint. If no gate qualifies, it sets `force_review=true`. A separate probability recalibrator such as isotonic regression or temperature scaling is **not implemented**. +Flat `compile --fit-decision-knobs` (Python `CompileOptions(fit_decision_knobs=True)`) +opts in to optional `native-policy/v1` extensions that fit strictly ordered `score_cuts` (K-1 +cuts in 0..K-1) or positive `choice_weights` for exactly the fixed labels. Absent +extensions serialize exactly as older policies, preserving existing artifact +checksums and behavior. Score cuts map the continuous native expectation (or +rescaled composite value) to an ordinal level; Choice multipliers change only +selection, with native ties preserved. Raw distributions, P(true), and vendor +confidence remain separate and unchanged. Direct fitted gates use the native +probability of the selected level or weighted-selected label, and name that +selection provenance. Composite Score retains its explicitly heuristic minimum +component gate; no joint probability is created. + +Without that opt-in, decision knobs remain unchanged and report `disabled`; +existing Noul thresholds and review-gate fitting retain their behavior. Hierarchy +compilation keeps its existing routing and final review gates, and rejects this +flat-only flag. + +The native fitter makes no model calls and imports no DSPy. Only cached calibration +answers and labels enter bounded two-pass coordinate search. Five grouped folds +are constructed deterministically from IDs/groups, independent of outcomes. Each +held-out fold needs at least `max(2, min_samples)` observations. The objective is +Choice misclassification or normalized Score MAE. No held-out fold may worsen and +the aggregate loss must strictly improve; ties, sparse folds, or unsuccessful +full-calibration refits report `decision_fit.status="unchanged"` and preserve the +original selection knobs. Review-gate fitting remains a separate operational +step and may still set `force_review`. Reports include fold losses, train/held-out +counts, chosen policies and synthetic provenance. These checks are conservative +heuristics, not an independent evaluation or a claim of gains. Validation selects +prompts; it and test data never enter fitting. Freeze still precedes test. + The reported Wilson interval is descriptive; it is not a corrected guarantee under adaptive threshold selection or distribution shift. Per-output review flags are not a joint guarantee for the whole decision vector. Production needs its own diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 3064042..44fd721 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -20,9 +20,11 @@ rather than capabilities claimed by the current release. 3. **Richer policy and calibration.** Add task-level utility matrices, statistically justified selective-risk control, optional probability recalibration, and supervised composition-weight fitting with carefully isolated data. Fitting - Score cuts and Choice weights on the calibration split, where today only Noul - thresholds are fitted, is tracked in - [#92](https://github.com/Jev-Engineering/TypeWright/issues/92). + Score cuts and Choice selection weights is now implemented natively from cached + calibration predictions in flat programs, explicitly enabled with + `compile --fit-decision-knobs`, with conservative grouped five-fold checks and an + explicit unchanged result (#92; owner approved native fitting for #95). + This is not posterior calibration or a measured gain. 4. **Scale and operations.** Async batched evaluation, explicitly authorized retry accounting, drift monitoring, signed artifact attestations, and a separately reviewed TypeScript runtime. Resumable non-pickle records already exist for diff --git a/schemas/hierarchy-artifact.schema.json b/schemas/hierarchy-artifact.schema.json index 2a10d01..73a19f6 100644 --- a/schemas/hierarchy-artifact.schema.json +++ b/schemas/hierarchy-artifact.schema.json @@ -466,6 +466,51 @@ "default": false, "title": "Force Review", "type": "boolean" + }, + "fitting_version": { + "anyOf": [ + { + "const": "native-policy/v1", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Fitting Version" + }, + "score_cuts": { + "anyOf": [ + { + "items": { + "minimum": 0, + "type": "number" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Score Cuts" + }, + "choice_weights": { + "anyOf": [ + { + "additionalProperties": { + "exclusiveMinimum": 0, + "type": "number" + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Choice Weights" } }, "title": "Policy", diff --git a/schemas/hierarchy-source.schema.json b/schemas/hierarchy-source.schema.json index 2db9410..7d704bb 100644 --- a/schemas/hierarchy-source.schema.json +++ b/schemas/hierarchy-source.schema.json @@ -390,6 +390,51 @@ "default": false, "title": "Force Review", "type": "boolean" + }, + "fitting_version": { + "anyOf": [ + { + "const": "native-policy/v1", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Fitting Version" + }, + "score_cuts": { + "anyOf": [ + { + "items": { + "minimum": 0, + "type": "number" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Score Cuts" + }, + "choice_weights": { + "anyOf": [ + { + "additionalProperties": { + "exclusiveMinimum": 0, + "type": "number" + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Choice Weights" } }, "title": "Policy", diff --git a/schemas/program.schema.json b/schemas/program.schema.json index dc9e96d..3c3e832 100644 --- a/schemas/program.schema.json +++ b/schemas/program.schema.json @@ -154,6 +154,51 @@ "default": false, "title": "Force Review", "type": "boolean" + }, + "fitting_version": { + "anyOf": [ + { + "const": "native-policy/v1", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Fitting Version" + }, + "score_cuts": { + "anyOf": [ + { + "items": { + "minimum": 0, + "type": "number" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Score Cuts" + }, + "choice_weights": { + "anyOf": [ + { + "additionalProperties": { + "exclusiveMinimum": 0, + "type": "number" + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Choice Weights" } }, "title": "Policy", diff --git a/src/typewright/cli.py b/src/typewright/cli.py index 86dcaa1..2867b25 100644 --- a/src/typewright/cli.py +++ b/src/typewright/cli.py @@ -109,6 +109,8 @@ def build_parser(): compile_p.add_argument("--seed", type=int, default=7) compile_p.add_argument("--max-calibration-error", type=float, default=0.05) compile_p.add_argument("--min-calibration-samples", type=int, default=10) + compile_p.add_argument("--fit-decision-knobs", action="store_true", + help="Opt in to offline flat Score-cut and Choice-weight fitting.") backend_args(compile_p) teacher_args(compile_p) run = sub.add_parser("run", help="Run a compiled JSON artifact. Does not execute application actions.") @@ -130,7 +132,7 @@ def build_parser(): demo = sub.add_parser("demo", help="No-key, no-network end-to-end synthetic smoke test.") demo.add_argument("--out", type=Path, default=Path("runs/demo")) demo.add_argument("--starter", choices=["flat", "hierarchy"], default="flat") - inspect_p = sub.add_parser("inspect", help="Validate artifact checksum and print its typed plan.") + inspect_p = sub.add_parser("inspect", help="Validate artifact checksum and print its typed plan, including fitted policy knobs.") inspect_p.add_argument("artifact", type=Path) harden = sub.add_parser("harden", help="Create robustness proposals that require human label review.") harden.add_argument("spec", type=Path) @@ -209,6 +211,8 @@ def dispatch(args): splits = {name: read_hierarchy_jsonl(getattr(args, name), draft) for name in ("train", "validation", "calibration", "test")} validate_hierarchy_compile_inputs(source, splits) + if args.fit_decision_knobs: + raise ConfigurationError("Decision-knob fitting currently supports flat programs only.") options = HierarchyCompileOptions(architect=args.architect, optimizer=args.optimizer, structural_rounds=args.structural_rounds, max_metric_calls=args.max_metric_calls, seed=args.seed, max_calibration_error=args.max_calibration_error, @@ -238,7 +242,7 @@ def dispatch(args): options = CompileOptions(architect=args.architect, optimizer=args.optimizer, structural_rounds=args.structural_rounds, max_metric_calls=args.max_metric_calls, seed=args.seed, max_calibration_error=args.max_calibration_error, - min_calibration_samples=args.min_calibration_samples) + min_calibration_samples=args.min_calibration_samples, fit_decision_knobs=args.fit_decision_knobs) teacher = make_teacher(args) if (args.architect == "dspy" or args.optimizer == "gepa" or args.structural_rounds) else None backend = make_backend(args) try: diff --git a/src/typewright/compiler.py b/src/typewright/compiler.py index 9e6fee5..da53fcb 100644 --- a/src/typewright/compiler.py +++ b/src/typewright/compiler.py @@ -23,8 +23,11 @@ class CompileOptions: seed: int = 7 max_calibration_error: float = 0.05 min_calibration_samples: int = 10 + fit_decision_knobs: bool = False def __post_init__(self): + if type(self.fit_decision_knobs) is not bool: + raise ConfigurationError("fit_decision_knobs must be a boolean.") if self.architect not in {"template", "dspy"} or self.optimizer not in {"none", "gepa"}: raise ConfigurationError("Unknown architect or optimizer.") if not 0 <= self.structural_rounds <= 10: @@ -106,7 +109,8 @@ def compile(self, source: UseCase, *, train: list[Example], validation: list[Exa _, calibration_predictions = evaluate(best, calibration, self.backend) best, calibration_fit = fit_policies( best, calibration, calibration_predictions, - max_error=options.max_calibration_error, min_samples=options.min_calibration_samples) + max_error=options.max_calibration_error, min_samples=options.min_calibration_samples, + fit_decision_knobs=options.fit_decision_knobs) fitted_predictions = [dict(r, decisions=apply_policy(best, r["answers"])) for r in calibration_predictions] calibration_report = report_from_results(best, calibration, fitted_predictions) best.provenance = { diff --git a/src/typewright/models.py b/src/typewright/models.py index f14d7b0..c043d8d 100644 --- a/src/typewright/models.py +++ b/src/typewright/models.py @@ -2,9 +2,9 @@ import math import re from pathlib import Path -from typing import Any, Literal +from typing import Annotated, Any, Literal -from pydantic import BaseModel, ConfigDict, Field, JsonValue, model_validator +from pydantic import BaseModel, ConfigDict, Field, JsonValue, model_serializer, model_validator from .errors import CandidateError, DataError from .io import atomic_json, fingerprint, load_document @@ -107,6 +107,32 @@ class Policy(StrictModel): noul_threshold: float = Field(default=0.5, gt=0, lt=1) min_gate: float = Field(default=0.8, ge=0, le=1) force_review: bool = False + fitting_version: Literal["native-policy/v1"] | None = None + score_cuts: list[Annotated[float, Field(strict=True, ge=0)]] | None = None + choice_weights: dict[str, Annotated[float, Field(strict=True, gt=0)]] | None = None + + @model_validator(mode="after") + def check_fitted_policy(self): + knobs = self.score_cuts is not None or self.choice_weights is not None + if knobs != (self.fitting_version is not None): + raise ValueError("Fitted knobs require an explicit native-policy/v1 version.") + if self.score_cuts is not None: + if not self.score_cuts or any(a >= b for a, b in zip(self.score_cuts, self.score_cuts[1:])): + raise ValueError("Score cuts must be nonempty and strictly ordered.") + if self.choice_weights is not None: + if not self.choice_weights or any(v <= 0 for v in self.choice_weights.values()): + raise ValueError("Choice weights must be positive.") + return self + + @model_serializer(mode="wrap") + def serialize_compatible(self, handler): + # Omit absent extensions even in nested dumps: existing artifact bytes and + # content hashes retain their original meaning. + data = handler(self) + for key in ("fitting_version", "score_cuts", "choice_weights"): + if data.get(key) is None: + data.pop(key, None) + return data class Program(StrictModel): @@ -132,6 +158,15 @@ def check_program(self): used = set() for name, binding in self.bindings.items(): decision = self.decisions[name] + policy = self.policies[name] + if policy.score_cuts is not None: + if decision.type != "score" or len(policy.score_cuts) != len(decision.criteria) - 1: + raise ValueError("Score cuts must match the declared Score scale.") + if any(not 0 <= cut <= len(decision.criteria) - 1 for cut in policy.score_cuts): + raise ValueError("Score cuts must lie within the declared scale.") + if policy.choice_weights is not None: + if decision.type != "choice" or set(policy.choice_weights) != set(decision.criteria): + raise ValueError("Choice weights must match the fixed labels.") if binding.kind == "question": if binding.question not in self.questions: raise ValueError(f"Unknown question in binding {name}.") diff --git a/src/typewright/policy.py b/src/typewright/policy.py index 405aef3..2e06276 100644 --- a/src/typewright/policy.py +++ b/src/typewright/policy.py @@ -3,23 +3,126 @@ from .data import Example from .metrics import is_correct, wilson_upper +from .io import fingerprint from .models import Policy, Program from .runtime import apply_policy +def _loss(program, name, rows, results): + declaration = program.decisions[name] + values = [apply_policy(program, r["answers"])[name]["value"] for r in results] + if declaration.type == "choice": + return sum(value != row.expected[name] for value, row in zip(values, rows)) / len(rows) + return sum(abs(value - row.expected[name]) for value, row in zip(values, rows)) / ( + len(rows) * (len(declaration.criteria) - 1)) + + +def _search(program, name, rows, results): + """Bounded coordinate search; only strictly better candidates replace baseline.""" + declaration = program.decisions[name] + current = program.model_copy(deep=True) + best_loss = _loss(current, name, rows, results) + base = current.policies[name].model_dump() + if declaration.type == "choice": + knobs = dict(base.get("choice_weights", dict.fromkeys(declaration.criteria, 1.0))) + coordinates = list(declaration.criteria) + grid = [.25, .5, 1.0, 2.0, 4.0] + field = "choice_weights" + else: + knobs = list(base.get("score_cuts", [i + .5 for i in range(len(declaration.criteria) - 1)])) + coordinates = list(range(len(knobs))) + raw_program = program.model_copy(deep=True) + raw_policy = base.copy() + raw_policy.pop("score_cuts", None) + raw_policy.pop("fitting_version", None) + raw_program.policies[name] = Policy(**raw_policy) + values = sorted({apply_policy(raw_program, r["answers"])[name]["value"] for r in results}) + midpoints = [(a + b) / 2 for a, b in zip(values, values[1:])] + # At most 33 observed boundaries plus endpoints and conventional cuts. + grid = sorted({0.0, float(len(declaration.criteria) - 1), *knobs, + *[midpoints[i] for i in range(0, len(midpoints), max(1, len(midpoints)//32 + 1))]}) + field = "score_cuts" + for _ in range(2): + for coordinate in coordinates: + for value in grid: + proposed = knobs.copy() + proposed[coordinate] = value + if field == "score_cuts" and any(a >= b for a, b in zip(proposed, proposed[1:])): + continue + candidate = current.model_copy(deep=True) + candidate.policies[name] = Policy(**{**base, "fitting_version": "native-policy/v1", field: proposed}) + loss = _loss(candidate, name, rows, results) + if loss < best_loss - 1e-12: + current, knobs, best_loss = candidate, proposed, loss + return current.policies[name] + + +def _fit_knobs(program, name, rows, results, min_samples): + detail = {"status": "unchanged", "calibration_n": len(rows), "folds": [], + "synthetic": any(r.get("synthetic", False) for r in results), + "objective": "misclassification_rate" if program.decisions[name].type == "choice" + else "normalized_mae", + "method": "native-policy/v1; bounded coordinate search; five grouped held-out folds", + "min_fold_samples": max(2, min_samples)} + # Group boundaries are retained inside calibration, too. IDs determine stable + # order without using outcomes or input text to construct the folds. + units = {} + for i, row in enumerate(rows): + key = ("group", row.group) if row.group else ("id", row.id) + units.setdefault(key, []).append(i) + folds = [[] for _ in range(5)] + for j, key in enumerate(sorted(units, key=fingerprint)): + folds[j % 5].extend(units[key]) + if any(len(fold) < max(2, min_samples) for fold in folds): + detail["reason"] = "Insufficient independent held-out calibration samples." + return program.policies[name], detail + for fold in folds: + held_out = set(fold) + train = [i for i in range(len(rows)) if i not in held_out] + candidate = program.model_copy(deep=True) + candidate.policies[name] = _search(program, name, [rows[i] for i in train], [results[i] for i in train]) + gold = [rows[i] for i in fold] + cached = [results[i] for i in fold] + detail["folds"].append({"train_n": len(train), "held_out_n": len(fold), + "default_loss": _loss(program, name, gold, cached), + "candidate_loss": _loss(candidate, name, gold, cached), + "policy": candidate.policies[name].model_dump()}) + checks = detail["folds"] + if (any(f["candidate_loss"] > f["default_loss"] + 1e-12 for f in checks) + or sum(f["held_out_n"] * (f["default_loss"] - f["candidate_loss"]) for f in checks) <= 1e-12): + detail["reason"] = "No conservative held-out calibration improvement; retain default." + return program.policies[name], detail + fitted = _search(program, name, rows, results) + if fitted == program.policies[name]: + detail["reason"] = "Full calibration refit did not beat default." + return fitted, detail + detail["status"] = "fitted" + detail["policy"] = fitted.model_dump() + return fitted, detail + + def fit_policies(program: Program, rows: list[Example], results: list[dict[str, Any]], *, - max_error: float = 0.05, min_samples: int = 10) -> tuple[Program, dict[str, Any]]: - """Fit thresholds on the calibration split only. No new model calls. + max_error: float = 0.05, min_samples: int = 10, + fit_decision_knobs: bool = False) -> tuple[Program, dict[str, Any]]: + """Fit decision knobs and gates on cached calibration predictions only. Select the largest observed accepted set within an empirical error target. This is an operational heuristic; it is NOT a distribution-free guarantee. """ if not 0 <= max_error <= 1 or min_samples < 1 or not rows or len(rows) != len(results): raise ValueError("Invalid calibration data or policy configuration.") + if type(fit_decision_knobs) is not bool: + raise ValueError("fit_decision_knobs must be a boolean.") fitted = program.model_copy(deep=True) report = {} for name, decision in fitted.decisions.items(): initial_policy = fitted.policies[name] + knob_report = None + if fit_decision_knobs and decision.type in {"choice", "score"}: + initial_policy, knob_report = _fit_knobs(program, name, rows, results, min_samples) + elif decision.type in {"choice", "score"}: + knob_report = {"status": "disabled", "reason": "Decision fitting requires explicit opt-in.", + "calibration_n": len(rows), "folds": []} threshold = initial_policy.noul_threshold if decision.type == "noul": p_values = [r["decisions"][name]["p_true"] for r in results] @@ -29,7 +132,8 @@ def fit_policies(program: Program, rows: list[Example], results: list[dict[str, grid.update(min(.999999, p + 1e-9) for p in p_values if p < 1) threshold = min(grid, key=lambda t: ( sum((p >= t) != row.expected[name] for p, row in zip(p_values, rows)), abs(t-.5))) - fitted.policies[name] = Policy(noul_threshold=threshold, min_gate=0, force_review=False) + fitted.policies[name] = Policy(**{**initial_policy.model_dump(), "noul_threshold": threshold, + "min_gate": 0, "force_review": False}) predictions = [apply_policy(fitted, r["answers"])[name] for r in results] correct = [is_correct(fitted, name, pred["value"], row.expected[name]) for pred, row in zip(predictions, rows)] @@ -42,12 +146,17 @@ def fit_policies(program: Program, rows: list[Example], results: list[dict[str, options.append((len(accepted), -errors, gate, errors)) if options: count, _, gate, errors = max(options) - fitted.policies[name] = Policy(noul_threshold=threshold, min_gate=gate) + fitted.policies[name] = Policy(**{**initial_policy.model_dump(), "noul_threshold": threshold, + "min_gate": gate, "force_review": False}) report[name] = {"status": "fitted", "accepted_n": count, "empirical_error": errors/count, "wilson_upper_95_descriptive": wilson_upper(errors, count)} else: - fitted.policies[name] = Policy(noul_threshold=threshold, min_gate=1.0, force_review=True) + fitted.policies[name] = Policy(**{**initial_policy.model_dump(), "noul_threshold": threshold, + "min_gate": 1.0, "force_review": True}) report[name] = {"status": "review_only", "reason": "No threshold met sample/error requirements."} - report[name].update(policy=fitted.policies[name].model_dump(), min_samples=min_samples, + if knob_report is not None: + report[name]["decision_fit"] = knob_report + report[name].update(calibration_n=len(rows), synthetic=any(r.get("synthetic", False) for r in results), + policy=fitted.policies[name].model_dump(), min_samples=min_samples, empirical_error_target=max_error) return fitted, report diff --git a/src/typewright/runtime.py b/src/typewright/runtime.py index 1e96a10..80cbecb 100644 --- a/src/typewright/runtime.py +++ b/src/typewright/runtime.py @@ -112,6 +112,16 @@ def apply_policy(program: Program, answers: dict[str, dict[str, Any]]) -> dict[s vendor_confidence = answer.get("confidence") if declaration.type == "choice": value = answer["choice"] + if policy.choice_weights is not None: + # Multipliers affect selection only. Native probabilities remain + # untouched; ties preserve the native selection when possible. + weighted = {label: p * policy.choice_weights[label] + for label, p in distributions.items()} + best = max(weighted.values()) + if weighted[value] < best: + value = next(label for label in declaration.criteria if weighted[label] == best) + gate = distributions[value] + basis = "native_probability_of_weighted_selected_option" elif declaration.type == "noul": p_true = answer["noul"] value = p_true >= policy.noul_threshold @@ -133,6 +143,13 @@ def apply_policy(program: Program, answers: dict[str, dict[str, Any]]) -> dict[s gate = min(gates) basis = "minimum_component_gate_heuristic" # No invented distribution or composite posterior confidence. + if policy.score_cuts is not None: + value = sum(value >= cut for cut in policy.score_cuts) + if distributions is not None: + gate = distributions[str(value)] + basis = "native_probability_of_cut_selected_level" + else: + basis = "minimum_component_gate_heuristic_after_score_cuts" decisions[name] = { "type": declaration.type, "value": value, "probabilities": distributions, "p_true": p_true, "vendor_confidence": vendor_confidence, diff --git a/tests/test_policy_fitting.py b/tests/test_policy_fitting.py new file mode 100644 index 0000000..228662e --- /dev/null +++ b/tests/test_policy_fitting.py @@ -0,0 +1,298 @@ +"""Synthetic policy mechanics; these are not measured Jev gains.""" +import json +import subprocess +import sys + +import pytest +from pydantic import ValidationError + +from typewright.architect import template_program +from typewright.data import Example +from typewright.io import fingerprint +from typewright.models import Decision, Policy, Program, StateField, UseCase +from typewright.policy import fit_policies +from typewright.runtime import apply_policy + + +def fixture(kind, n=100): + criteria = {"a": "A", "b": "B"} if kind == "choice" else ["low", "high"] + source = UseCase(name="synthetic", state={"text": StateField()}, + decisions={"output": Decision(type=kind, goal="Synthetic test", criteria=criteria)}) + program = template_program(source) + qid = program.bindings["output"].question + rows, results = [], [] + for i in range(n): + if kind == "choice": + # A shifted decision boundary: native argmax is A, policy should pick B. + probabilities = {"a": .55, "b": .45} if i < n * .8 else {"a": .9, "b": .1} + answer = {"type": kind, "choice": "a", "probabilities": probabilities, "confidence": .7} + gold = "b" if i < n * .8 else "a" + else: + # A high ordinal level despite a continuous native expectation of .3. + answer = {"type": kind, "score": .3, "probabilities": {"0": .7, "1": .3}, "confidence": .7} + gold = 1 + rows.append(Example(id=f"cal-{i}", state={"text": str(i)}, expected={"output": gold})) + answers = {qid: answer} + results.append({"answers": answers, "decisions": apply_policy(program, answers), "synthetic": True}) + return program, rows, results + + +@pytest.mark.parametrize("kind", ["choice", "score"]) +def test_synthetic_shift_fits_with_held_out_improvement(kind): + program, rows, results = fixture(kind) + fitted, report = fit_policies(program, rows, results, fit_decision_knobs=True, max_error=1) + detail = report["output"]["decision_fit"] + assert detail["status"] == "fitted" + assert detail["synthetic"] is True + assert detail["calibration_n"] == 100 + assert len(detail["folds"]) == 5 + assert all(f["candidate_loss"] < f["default_loss"] for f in detail["folds"]) + prediction = apply_policy(fitted, results[0]["answers"])["output"] + assert prediction["value"] == rows[0].expected["output"] + assert prediction["probabilities"] == results[0]["decisions"]["output"]["probabilities"] + assert prediction["gate_score"] == (.45 if kind == "choice" else .3) + assert "native_probability" in prediction["gate_basis"] + assert program.policies["output"].fitting_version is None + + +@pytest.mark.parametrize("kind", ["choice", "score"]) +def test_ties_and_small_samples_keep_defaults(kind): + program, rows, results = fixture(kind) + for row in rows: + row.expected["output"] = "a" if kind == "choice" else .3 + fitted, report = fit_policies(program, rows, results, fit_decision_knobs=True) + assert report["output"]["decision_fit"]["status"] == "unchanged" + assert fitted.policies["output"].fitting_version is None + fitted, report = fit_policies(program, rows[:4], results[:4], fit_decision_knobs=True) + assert report["output"]["decision_fit"]["status"] == "unchanged" + assert fitted.policies["output"].force_review + + +def test_grouped_fold_overfit_declined(): + program, rows, results = fixture("choice") + # Four folds favor B; one independent group favors native A. Each fold has + # enough support, but the held-out harmed fold rejects the fitted policy. + for i, row in enumerate(rows): + row.group = f"group-{i // 20}" + if i < 20: + row.expected["output"] = "a" + fitted, report = fit_policies(program, rows, results, fit_decision_knobs=True) + detail = report["output"]["decision_fit"] + assert detail["status"] == "unchanged" + assert len(detail["folds"]) == 5 + assert any(f["candidate_loss"] > f["default_loss"] for f in detail["folds"]) + assert fitted.policies["output"].choice_weights is None + + +@pytest.mark.parametrize("kwargs", [ + {"choice_weights": {"a": 0}}, {"score_cuts": [1, .5]}, + {"score_cuts": [float("nan")]}, {"choice_weights": {"a": float("inf")}}, +]) +def test_strict_knobs(kwargs): + with pytest.raises(ValidationError): + Policy(fitting_version="native-policy/v1", **kwargs) + + +def test_contract_bound_knobs(): + program, _, _ = fixture("score") + data = program.model_dump() + for cuts in ([.2, .8], [-.1], [1.1]): + data["policies"]["output"] = {"fitting_version": "native-policy/v1", "score_cuts": cuts} + with pytest.raises(ValidationError): + Program.model_validate(data) + program, _, _ = fixture("choice") + data = program.model_dump() + data["policies"]["output"] = {"fitting_version": "native-policy/v1", "choice_weights": {"a": 1}} + with pytest.raises(ValidationError): + Program.model_validate(data) + + +def test_legacy_checksum_serialization_and_new_roundtrip(tmp_path): + program, rows, results = fixture("choice") + legacy = program.model_dump(mode="json") + assert set(legacy["policies"]["output"]) == {"noul_threshold", "min_gate", "force_review"} + path = tmp_path / "legacy.json" + path.write_text(json.dumps({"program": legacy, "sha256": fingerprint(legacy)})) + loaded = Program.load(path) + assert loaded.content_hash == program.content_hash + assert loaded.model_dump(mode="json") == legacy + assert apply_policy(loaded, results[0]["answers"]) == results[0]["decisions"] + fitted, _ = fit_policies(program, rows, results, fit_decision_knobs=True, max_error=1) + fitted.save(path) + assert Program.load(path) == fitted + + +def test_runtime_import_boundary(): + subprocess.run([sys.executable, "-c", "import typewright.runtime; import sys; " + "assert 'dspy' not in sys.modules; assert 'gepa' not in sys.modules"], check=True) + + +def test_compile_validation_and_test_cannot_influence_fit(source, splits, backend, monkeypatch): + import typewright.compiler as module + from typewright.compiler import CompileOptions, Compiler + seen = [] + original = module.fit_policies + + def spy(program, rows, results, **kwargs): + assert [r.id for r in rows] == [r.id for r in splits["calibration"]] + seen.append(original(program, rows, results, **kwargs)[1]) + return original(program, rows, results, **kwargs) + + monkeypatch.setattr(module, "fit_policies", spy) + options = CompileOptions(min_calibration_samples=3) + Compiler(backend, options=options).compile(source, **splits) + # With template/no optimizer, neither validation nor test labels select any + # prompt. Deliberately corrupt those labels to prove they cannot fit knobs. + for split in ("validation", "test"): + for row in splits[split]: + row.expected["urgent"] = not row.expected["urgent"] + row.expected["department"] = next(iter(source.decisions["department"].criteria)) + row.expected["frustration"] = 0 + Compiler(backend, options=options).compile(source, **splits) + assert seen[0] == seen[1] + + +def test_fitted_inspect_and_schema(tmp_path, capsys): + from typewright.cli import main + program, rows, results = fixture("score") + fitted, _ = fit_policies(program, rows, results, fit_decision_knobs=True, max_error=1) + path = tmp_path / "fitted.json" + fitted.save(path) + assert main(["inspect", str(path)]) == 0 + policy = json.loads(capsys.readouterr().out)["policies"]["output"] + assert policy["fitting_version"] == "native-policy/v1" + assert policy["score_cuts"] == fitted.policies["output"].score_cuts + assert main(["schema", "--out", str(tmp_path / "schemas")]) == 0 + schema = json.loads((tmp_path / "schemas/program.schema.json").read_text()) + assert {"score_cuts", "choice_weights", "fitting_version"} <= set(schema["$defs"]["Policy"]["properties"]) + + +def test_choice_weight_tie_preserves_native_label(): + program, _, results = fixture("choice") + program.policies["output"] = Policy(fitting_version="native-policy/v1", choice_weights={"a": 1, "b": 1}) + answers = results[0]["answers"] + answer = next(iter(answers.values())) + answer.update(choice="b", probabilities={"a": .5, "b": .5}) + assert apply_policy(program, answers)["output"]["value"] == "b" + + +def test_composite_score_cuts_keep_heuristic_gate(): + from typewright.models import Binding, Question + program, _, results = fixture("score") + qid = program.bindings["output"].question + program.questions["extra"] = Question(type="noul", instructions="Synthetic component") + program.bindings["output"] = Binding(kind="weighted_mean", weights={qid: 1, "extra": 1}) + program.policies["output"] = Policy(fitting_version="native-policy/v1", score_cuts=[.2]) + answers = {**results[0]["answers"], "extra": {"type": "noul", "noul": .3}} + prediction = apply_policy(program, answers)["output"] + assert prediction["value"] == 1 + assert prediction["probabilities"] is None + assert prediction["vendor_confidence"] is None + assert prediction["gate_score"] == .7 + assert prediction["gate_basis"] == "minimum_component_gate_heuristic_after_score_cuts" + + +def test_checked_in_pre_extension_artifact_keeps_hash_and_payload(tmp_path): + from pathlib import Path + # This committed artifact predates the extension; its stored checksum and + # payload are independent of the new Policy serializer. + path = Path(__file__).resolve().parents[1] / "examples/compiled_demo/program.s1.json" + original = json.loads(path.read_text()) + assert all(set(policy) == {"noul_threshold", "min_gate", "force_review"} + for policy in original["program"]["policies"].values()) + loaded = Program.load(path) + assert fingerprint(loaded.model_dump(mode="json")) == original["sha256"] + output = tmp_path / "saved.json" + loaded.save(output) + assert json.loads(output.read_text()) == original + + +@pytest.mark.parametrize("kind", ["choice", "score"]) +def test_successful_compile_fit_isolated_from_validation_and_test(kind, monkeypatch): + import typewright.compiler as module + from typewright.backends import ManagedBackend, MockBackend + from typewright.compiler import CompileOptions, Compiler + from typewright.metrics import report_from_results + + program, calibration, cached = fixture(kind) + source = UseCase(name=program.name, model=program.model, state=program.state, decisions=program.decisions) + splits = {"calibration": calibration} + for split in ("train", "validation", "test"): + splits[split] = [Example(id=f"{split}-{i}", state={"text": f"{split}-{i}"}, + expected={"output": "a" if kind == "choice" else 0}) for i in range(10)] + seen = [] + original_fit = module.fit_policies + + def cached_evaluation(selected, rows, backend): + answers = cached[0]["answers"] + results = [{"answers": answers, "decisions": apply_policy(selected, answers), + "synthetic": True, "latency_ms": 0, "cache_hit": True} for row in rows] + return report_from_results(selected, rows, results), results + + def capture(selected, rows, results, **kwargs): + assert [row.id for row in rows] == [row.id for row in calibration] + fitted, report = original_fit(selected, rows, results, **kwargs) + assert report["output"]["decision_fit"]["status"] == "fitted" + seen.append((fitted.policies["output"].model_dump(), report)) + return fitted, report + + monkeypatch.setattr(module, "evaluate", cached_evaluation) + monkeypatch.setattr(module, "fit_policies", capture) + backend = ManagedBackend(MockBackend()) + try: + compiler = Compiler(backend, options=CompileOptions(min_calibration_samples=10, fit_decision_knobs=True)) + compiler.compile(source, **splits) + for split in ("validation", "test"): + for row in splits[split]: + row.state["text"] += " VALIDATION_TEST_SENTINEL" + row.expected["output"] = "b" if kind == "choice" else 1 + compiler.compile(source, **splits) + finally: + backend.close() + assert seen[0] == seen[1] + + +def test_decision_fitting_opt_in_default_preserves_selection(): + program, rows, results = fixture("choice") + fitted, report = fit_policies(program, rows, results, max_error=1) + assert report["output"]["decision_fit"]["status"] == "disabled" + assert fitted.policies["output"].choice_weights is None + assert apply_policy(fitted, results[0]["answers"])["output"]["value"] == "a" + + +def test_cli_opt_in_reaches_flat_compiler(tmp_path, monkeypatch): + import typewright.cli as cli + original = cli.Compiler + captured = [] + + class Capture(original): + def compile(self, source, **splits): + captured.append(self.options.fit_decision_knobs) + return super().compile(source, **splits) + + monkeypatch.setattr(cli, "Compiler", Capture) + from pathlib import Path + root = Path(__file__).resolve().parents[1] / "examples/support_triage" + args = ["compile", str(root / "usecase.yaml"), "--out", str(tmp_path / "compile"), + "--fit-decision-knobs"] + for split in ("train", "validation", "calibration", "test"): + args += [f"--{split}", str(root / f"{split}.jsonl")] + assert cli.main(args) == 0 + assert captured == [True] + + +def test_cli_rejects_hierarchy_fitting_before_calls(tmp_path, monkeypatch, capsys): + import typewright.cli as cli + from pathlib import Path + root = Path(__file__).resolve().parents[1] / "examples/hierarchy/support" + + def forbidden_backend(args): + raise AssertionError("Unsupported fitting must fail before any backend is made") + + monkeypatch.setattr(cli, "make_backend", forbidden_backend) + args = ["compile", str(root / "source.json"), "--out", str(tmp_path / "compile"), + "--fit-decision-knobs"] + for split in ("train", "validation", "calibration", "test"): + args += [f"--{split}", str(root / f"{split}.jsonl")] + assert cli.main(args) == 2 + assert "flat programs only" in capsys.readouterr().err