diff --git a/docs/bundles/BUNDLE_70_FRESH_PERSONAL_HOLDOUT_RUNNER_R1.md b/docs/bundles/BUNDLE_70_FRESH_PERSONAL_HOLDOUT_RUNNER_R1.md new file mode 100644 index 00000000..ae269165 --- /dev/null +++ b/docs/bundles/BUNDLE_70_FRESH_PERSONAL_HOLDOUT_RUNNER_R1.md @@ -0,0 +1,90 @@ +# Bundle 70 — Fresh Personal Holdout Runner R1 + +## Goal + +Provide the local execution layer that turns the canonical Human Review Protocol R2 / Curation Calibration R3 contracts into a reproducible fresh personal holdout run. + +The runner must stop before human labels are collected unless all pre-label evidence is frozen and auditable. + +## Required execution order + +1. Verify the local checkout is on canonical branch `feature/bundle-0-bootstrap`, its HEAD exactly matches the supplied canonical SHA, and the working tree is clean. +2. Validate private local-library snapshot R1. +3. Deterministically generate a bounded candidate case pool without reading human labels or challenger scores. +4. Materialize real-library optimizer evidence and blind A/B assignments locally, isolating per-case technical failures so one invalid candidate cannot abort the full pool. +5. Build engineering-only `HoldoutCandidate` rows. +6. Freeze `HoldoutCaseSamplingPolicy` and select at least 24 personal holdout cases with four cases per set role plus a frozen fallback reservoir. +7. Freeze replacement policy and effective cohort. +8. Compute Bundle 67 competitive-curation shadow comparisons for the selected cases and frozen fallback reservoir before reviewer workspace publication. +9. Persist a private pre-registration manifest containing selection, assignments, challenger evidence, fingerprints, and authority=false. +10. Require zero replacement events in the R1 effective cohort before reviewer publication. +11. Before finalization, verify byte identity of the generated private manifest, reviewer packet, and CSV against their pre-finalization SHA-256 values. +12. Require reviewer case order/identity to match the frozen effective cohort exactly, with assignment and set-role metadata matching private frozen evidence. +13. Apply dual prior-exposure exclusion: reviewer-visible sequence matching plus stable track-ID matching from prior private manifests. +14. Finalize a reviewer-safe workspace bound to the exact preregistration/selection/cohort and exposure-registry fingerprints. +15. Publish a reviewer-safe packet containing only anonymous Plan A / Plan B track sequences and the four R2 curation dimensions. +16. Create the review CSV with system binding metadata but leave all human judgments and clean-attestation assertions empty; no ratings, preferences, confidence, timestamps, or exposure claims may be fabricated. + +## Critical isolation + +`HOLDOUT_SELECTION_INPUTS` may include only frozen policy plus engineering/technical candidate metadata. + +`HOLDOUT_SELECTION_INPUTS` must not include: +- human preference; +- human ratings; +- reviewer notes; +- competitive challenger scores; +- competitive challenger preference. + +The challenger comparison is computed only after holdout selection is frozen, but before reviewer workspace publication. Challenger evidence is also frozen for the fallback reservoir so a later protocol version can support bounded technical replacement without post-label challenger computation. + +For **Fresh Personal Holdout Run R1**, the effective cohort is immutable once reviewer workspace publication begins. A technical invalidity discovered after publication aborts/restarts the run; it does not authorize an in-review fallback substitution. + +## Freshness / prior-exposure rule + +A new `case_id` or changed display metadata is not evidence that a case is fresh. + +The formal run requires two independent historical exposure sources: + +1. prior blinded reviewer packet(s), used to reject exact reviewer-visible Plan A / Plan B sequence reuse; +2. prior private review manifest(s), used to reject exact stable track-ID sequence reuse even if display metadata changed. + +Before Case 1 may be opened, the workspace finalizer must fail closed if an effective holdout case reproduces: +- an exact previously exposed individual plan sequence; or +- an exact previously exposed A/B sequence pair. + +After successful finalization the new private manifest receives an opaque stable-exposure registry so subsequent fresh runs can exclude this holdout without exposing track identities to the reviewer. + +## Reviewer-safe dimensions + +- energy_flow +- dramaturgical_fit +- set_coherence +- alternative_usefulness + +Allowed preference values: +- plan_a +- plan_b +- tie +- abstain + +No transition execution is requested in this runner. + +The review CSV contains explicit R2 attestation fields, but the human-controlled fields remain empty at workspace freeze and must be completed only from the actual review session. + +## Privacy + +- local audio paths stay in private evidence only; +- no audio upload; +- no cloud MIR execution; +- reviewer packet must not expose absolute paths, optimizer strategy identity, shadow scores, challenger preference, or stable track IDs; +- prior-exposure sources are hashed/bound in private evidence and are not published through the reviewer packet. + +## Authority + +- optimizer ranking activation: NO +- PDM training: NO +- release: NO +- deploy: NO +- production activation: NO +- merge: NO without separate `MERGE GO` diff --git a/docs/runs/FRESH_PERSONAL_HOLDOUT_RUN_R1.md b/docs/runs/FRESH_PERSONAL_HOLDOUT_RUN_R1.md new file mode 100644 index 00000000..05c61244 --- /dev/null +++ b/docs/runs/FRESH_PERSONAL_HOLDOUT_RUN_R1.md @@ -0,0 +1,104 @@ +# Fresh Personal Holdout Run R1 + +## Purpose + +Execute a fresh personal blind curation holdout only after Bundle 70 is merged. + +## Preconditions + +- local APPLAYLIST checkout on canonical branch `feature/bundle-0-bootstrap`; +- local HEAD exactly matches the canonical SHA supplied to the runner; +- local working tree is clean; +- private `applaylist-local-library-snapshot-r1` JSON; +- local audio files remain readable; +- one or more prior blinded reviewer packets covering sequences already exposed to the reviewer, including the historical 12-case packet; +- one or more prior private review manifests carrying stable track-identity sequence evidence, including the historical real-library private manifest; +- no reviewer labels have been collected for the new run; +- sampling and blinding seeds are chosen before review. + +## Local command + +```bash +python scripts/applaylist_fresh_personal_holdout.py \ + --snapshot "$SNAPSHOT" \ + --output "$OUTPUT" \ + --database "$OUTPUT/APPLAYLIST_FRESH_PERSONAL_HOLDOUT_R1.sqlite" \ + --canonical-sha "$CANONICAL_SHA" \ + --generated-at "$GENERATED_AT" \ + --sampling-seed "$SAMPLING_SEED" \ + --blinding-seed "$BLINDING_SEED" \ + --exclude-reviewer-packet "$PRIOR_REVIEW_PACKET" \ + --exclude-private-manifest "$PRIOR_PRIVATE_MANIFEST" +``` + +Both exclusion arguments are repeatable when more than one earlier exposure source must be covered. + +The runner must fail closed before evidence generation if the checkout is not the exact canonical commit or the working tree is dirty. + +## Expected private outputs + +- `APPLAYLIST_FRESH_PERSONAL_HOLDOUT_R1.private.json` +- local SQLite database + +The private manifest includes frozen selection, replacement/effective-cohort provenance, blind assignments, challenger evidence for the selected + fallback reservoir, and after workspace finalization an opaque stable-exposure registry for future holdout runs. Challenger evidence must be frozen before the reviewer workspace is published. + +These outputs must not be published to a public repository because the private manifest is bound to local evidence and may contain private provenance. + +## Expected reviewer-safe outputs + +- `APPLAYLIST_FRESH_PERSONAL_HOLDOUT_REVIEWER_R1.json` +- `APPLAYLIST_FRESH_PERSONAL_HOLDOUT_REVIEW_R1.csv` + +The reviewer packet contains anonymous Plan A / Plan B sequences and only the R2 curation dimensions: + +- `energy_flow` +- `dramaturgical_fit` +- `set_coherence` +- `alternative_usefulness` + +The CSV contains system binding fields and explicit R2 clean-attestation columns, but all human-controlled fields must be empty when the workspace is frozen. + +## Prior-exposure exclusion + +A case is not fresh merely because it has a new case identifier or because display metadata changed. + +Before the reviewer workspace is finalized, the runner performs two independent exclusion checks: + +1. reviewer-visible sequence matching against prior blinded reviewer packets; +2. stable track-ID sequence matching against prior private manifests. + +It must fail closed if an effective holdout case reproduces: + +- any exact previously exposed individual plan sequence; or +- any exact previously exposed A/B sequence pair. + +The pre-finalization SHA-256 values of the generated private manifest, reviewer packet, and CSV are also verified before the finalizer is allowed to bind them. The reviewer case order/identity must exactly equal the frozen effective cohort, and reviewer assignment/set-role metadata must match the private frozen evidence. + +Do not open Case 1 if either exposure exclusion layer or any binding check failed. + +## Cohort immutability during review + +Fresh Personal Holdout Run R1 requires zero replacement events before reviewer publication. Once the reviewer workspace is finalized, the 24-case effective cohort is immutable for that run. + +If any selected case later becomes technically invalid, do **not** substitute a fallback during human review. Abort the run and create a new preregistered fresh holdout instead. The frozen fallback reservoir exists as pre-label provenance/future protocol support, not as authority for an in-review swap in R1. + +## Stop gate before Case 1 + +Before opening the reviewer packet, verify: + +- exact canonical SHA matches the run preregistration and local HEAD; +- canonical branch is `feature/bundle-0-bootstrap`; +- working tree was clean at run start; +- selected holdout has 24 effective cases; +- all six set roles are represented with four cases each; +- effective cohort has zero replacement events before publication; +- reviewer case order/identity exactly matches the frozen effective cohort; +- replacement policy and effective cohort fingerprints are frozen; +- challenger comparisons for selected + fallback cases are present in private evidence and absent from reviewer-safe outputs; +- reviewer-visible and stable track-ID prior-exposure registries were both applied successfully; +- reviewer packet is bound to the frozen preregistration/selection/effective-cohort fingerprints; +- all human review and clean-attestation fields are empty at freeze; +- algorithm identity is hidden; +- transition execution is not requested. + +If any check fails, do not review Case 1. diff --git a/scripts/applaylist_fresh_personal_holdout.py b/scripts/applaylist_fresh_personal_holdout.py new file mode 100644 index 00000000..73da71d1 --- /dev/null +++ b/scripts/applaylist_fresh_personal_holdout.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import argparse +import json +import subprocess +from pathlib import Path +from typing import Mapping + +from services.intelligence.fresh_holdout_reviewer_workspace import ( + finalize_fresh_holdout_reviewer_workspace, +) +from services.intelligence.fresh_personal_holdout_runner import ( + FreshPersonalHoldoutRunnerError, + materialize_fresh_personal_holdout_r1, +) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Materialize a fresh blinded personal curation holdout locally." + ) + parser.add_argument("--snapshot", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--database", required=True) + parser.add_argument("--canonical-sha", required=True) + parser.add_argument("--canonical-branch", default="feature/bundle-0-bootstrap") + parser.add_argument("--generated-at", required=True) + parser.add_argument("--sampling-seed", required=True) + parser.add_argument("--blinding-seed", required=True) + parser.add_argument( + "--exclude-reviewer-packet", + action="append", + required=True, + help="Prior blinded reviewer packet to exclude from visible-sequence exposure; repeatable.", + ) + parser.add_argument( + "--exclude-private-manifest", + action="append", + required=True, + help=( + "Prior private review manifest carrying stable track-identity exposure evidence; " + "repeatable." + ), + ) + parser.add_argument("--cases-per-role", type=int, default=8) + parser.add_argument("--candidate-scope-size", type=int, default=16) + parser.add_argument("--fallback-count", type=int, default=12) + return parser + + +def _git(*args: str) -> str: + completed = subprocess.run( + ("git", *args), + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + return completed.stdout.strip() + + +def verify_canonical_checkout(*, canonical_sha: str, canonical_branch: str) -> None: + """Fail closed unless the run is launched from the exact clean canonical checkout.""" + try: + head = _git("rev-parse", "HEAD") + branch = _git("rev-parse", "--abbrev-ref", "HEAD") + dirty = _git("status", "--porcelain") + except (subprocess.CalledProcessError, FileNotFoundError) as exc: + raise FreshPersonalHoldoutRunnerError( + "fresh holdout run requires a readable local Git checkout" + ) from exc + + expected_sha = str(canonical_sha).strip() + expected_branch = str(canonical_branch).strip() + if not expected_sha or not expected_branch: + raise FreshPersonalHoldoutRunnerError("canonical SHA/branch must not be empty") + if head != expected_sha: + raise FreshPersonalHoldoutRunnerError( + f"local HEAD {head} does not match declared canonical SHA {expected_sha}" + ) + if branch != expected_branch: + raise FreshPersonalHoldoutRunnerError( + f"local branch {branch} does not match canonical branch {expected_branch}" + ) + if dirty: + raise FreshPersonalHoldoutRunnerError( + "fresh holdout run requires a clean working tree before evidence generation" + ) + + +def verify_r1_fixed_effective_cohort(result: Mapping[str, str]) -> None: + """R1 publishes only a cohort with no replacement events already applied. + + Once the reviewer workspace is finalized, later technical invalidity must abort and + restart the run rather than silently swap in a fallback during human review. + """ + private_path = Path(str(result.get("private_manifest", "")).strip()) + if not private_path.is_file(): + raise FreshPersonalHoldoutRunnerError("fresh holdout private manifest is missing") + raw = json.loads(private_path.read_text(encoding="utf-8")) + cohort = raw.get("effective_cohort") if isinstance(raw, dict) else None + if not isinstance(cohort, dict): + raise FreshPersonalHoldoutRunnerError("fresh holdout effective cohort is missing") + events = cohort.get("replacement_events") + if events not in ([], ()): + raise FreshPersonalHoldoutRunnerError( + "Fresh Personal Holdout R1 requires zero replacement events before review publication" + ) + + +def main() -> int: + args = _parser().parse_args() + verify_canonical_checkout( + canonical_sha=args.canonical_sha, + canonical_branch=args.canonical_branch, + ) + snapshot_path = Path(args.snapshot) + result = materialize_fresh_personal_holdout_r1( + snapshot_path=snapshot_path, + output_dir=Path(args.output), + database_path=Path(args.database), + canonical_sha=args.canonical_sha, + generated_at=args.generated_at, + sampling_seed=args.sampling_seed, + blinding_seed=args.blinding_seed, + cases_per_role=args.cases_per_role, + candidate_scope_size=args.candidate_scope_size, + fallback_count=args.fallback_count, + ) + verify_r1_fixed_effective_cohort(result) + result = finalize_fresh_holdout_reviewer_workspace( + result, + snapshot_path=snapshot_path, + prior_reviewer_packet_paths=tuple(args.exclude_reviewer_packet), + prior_private_manifest_paths=tuple(args.exclude_private_manifest), + ) + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/intelligence/fresh_holdout_reviewer_workspace.py b/services/intelligence/fresh_holdout_reviewer_workspace.py new file mode 100644 index 00000000..448f501b --- /dev/null +++ b/services/intelligence/fresh_holdout_reviewer_workspace.py @@ -0,0 +1,624 @@ +from __future__ import annotations + +import csv +import hashlib +import json +from pathlib import Path +from typing import Any, Mapping, Sequence + +from services.intelligence.fresh_personal_holdout_runner import ( + FRESH_PERSONAL_HOLDOUT_PRIVATE_SCHEMA, + FRESH_PERSONAL_HOLDOUT_REVIEWER_SCHEMA, + FreshPersonalHoldoutRunnerError, +) + +REVIEWER_WORKSPACE_VERSION = "fresh-personal-holdout-reviewer-workspace-r1" + +_HUMAN_FIELDS = ( + "reviewer_ref", + "preference", + "energy_flow_plan_a", + "energy_flow_plan_b", + "dramaturgical_fit_plan_a", + "dramaturgical_fit_plan_b", + "set_coherence_plan_a", + "set_coherence_plan_b", + "alternative_usefulness_plan_a", + "alternative_usefulness_plan_b", + "confidence", + "prior_case_exposure", + "judgment_mode", + "transition_execution_used", + "transition_preview_heard", + "algorithm_identity_was_hidden", + "reason_codes", + "notes", + "observed_at", +) + + +def _load(path: Path) -> dict[str, Any]: + raw = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise FreshPersonalHoldoutRunnerError(f"workspace JSON must be an object: {path}") + return raw + + +def _sha256_json(value: object) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def _token(value: object, field: str) -> str: + text = str(value).strip() + if not text: + raise FreshPersonalHoldoutRunnerError(f"{field} must not be empty") + return text + + +def _plan_tuple(value: object, field: str) -> tuple[str, ...]: + if not isinstance(value, list) or not value: + raise FreshPersonalHoldoutRunnerError(f"{field} must be a non-empty plan array") + return tuple(_token(item, field) for item in value) + + +def _plan_fingerprint(plan: tuple[str, ...], *, namespace: str = "plan") -> str: + return f"{namespace}:" + _sha256_json(plan) + + +def _case_exposure_fingerprint(case: Mapping[str, Any]) -> str: + role = _token(case.get("set_role"), "set_role") + plan_a = _plan_tuple(case.get("plan_a"), "plan_a") + plan_b = _plan_tuple(case.get("plan_b"), "plan_b") + pair = tuple(sorted((_plan_fingerprint(plan_a), _plan_fingerprint(plan_b)))) + return "case-exposure:" + _sha256_json((role, pair)) + + +def _verify_prefinalization_hash( + *, + result: Mapping[str, str], + path: Path, + result_key: str, +) -> None: + expected = _token(result.get(result_key), result_key) + actual = hashlib.sha256(path.read_bytes()).hexdigest() + if actual != expected: + raise FreshPersonalHoldoutRunnerError( + f"pre-finalization file hash mismatch for {path.name}" + ) + + +def _prior_exposure_registry(paths: Sequence[str | Path]) -> dict[str, Any]: + if not paths: + raise FreshPersonalHoldoutRunnerError( + "fresh formal holdout requires at least one prior reviewer packet exclusion source" + ) + case_fingerprints: set[str] = set() + plan_fingerprints: set[str] = set() + sources: list[dict[str, str]] = [] + for raw_path in paths: + path = Path(raw_path).expanduser().resolve() + packet = _load(path) + cases = packet.get("cases") + if not isinstance(cases, list) or not cases: + raise FreshPersonalHoldoutRunnerError( + f"prior reviewer packet contains no cases: {path}" + ) + for case in cases: + if not isinstance(case, Mapping): + raise FreshPersonalHoldoutRunnerError("prior reviewer case must be an object") + case_fingerprints.add(_case_exposure_fingerprint(case)) + plan_fingerprints.add(_plan_fingerprint(_plan_tuple(case.get("plan_a"), "plan_a"))) + plan_fingerprints.add(_plan_fingerprint(_plan_tuple(case.get("plan_b"), "plan_b"))) + sources.append( + { + "path_sha256": hashlib.sha256(str(path).encode("utf-8")).hexdigest(), + "content_sha256": hashlib.sha256(path.read_bytes()).hexdigest(), + } + ) + registry_payload = { + "case_fingerprints": sorted(case_fingerprints), + "plan_fingerprints": sorted(plan_fingerprints), + "source_content_sha256": sorted(item["content_sha256"] for item in sources), + } + return { + "case_fingerprints": case_fingerprints, + "plan_fingerprints": plan_fingerprints, + "sources": sources, + "registry_fingerprint": "prior-exposure-registry:" + _sha256_json(registry_payload), + } + + +def _historical_private_sequences(raw: Mapping[str, Any]) -> tuple[tuple[str, tuple[str, ...], tuple[str, ...]], ...]: + cases = raw.get("cases") + if not isinstance(cases, list) or not cases: + raise FreshPersonalHoldoutRunnerError( + "historical private manifest contains no stable case sequence evidence" + ) + rows: list[tuple[str, tuple[str, ...], tuple[str, ...]]] = [] + for item in cases: + if not isinstance(item, Mapping): + raise FreshPersonalHoldoutRunnerError("historical private case must be an object") + case = item.get("case") + assignment = item.get("assignment") + if not isinstance(case, Mapping) or not isinstance(assignment, Mapping): + raise FreshPersonalHoldoutRunnerError( + "historical private case lacks case/assignment binding" + ) + greedy = case.get("greedy_plan") + beam = case.get("beam_plan") + if not isinstance(greedy, Mapping) or not isinstance(beam, Mapping): + raise FreshPersonalHoldoutRunnerError("historical private case lacks plan evidence") + plan_by_id = { + _token(greedy.get("plan_id"), "greedy plan_id"): _plan_tuple( + greedy.get("ordered_track_ids"), "greedy ordered_track_ids" + ), + _token(beam.get("plan_id"), "beam plan_id"): _plan_tuple( + beam.get("ordered_track_ids"), "beam ordered_track_ids" + ), + } + slot_a_id = _token(assignment.get("slot_a_plan_id"), "slot_a_plan_id") + slot_b_id = _token(assignment.get("slot_b_plan_id"), "slot_b_plan_id") + if slot_a_id not in plan_by_id or slot_b_id not in plan_by_id: + raise FreshPersonalHoldoutRunnerError( + "historical private assignment references unknown plans" + ) + rows.append( + ( + _token(case.get("set_role"), "set_role"), + plan_by_id[slot_a_id], + plan_by_id[slot_b_id], + ) + ) + return tuple(rows) + + +def _fresh_private_stable_registry(raw: Mapping[str, Any]) -> tuple[set[str], set[str]]: + stable = raw.get("stable_exposure_registry") + if not isinstance(stable, Mapping): + raise FreshPersonalHoldoutRunnerError( + "prior fresh private manifest lacks stable exposure registry" + ) + plans = stable.get("stable_plan_fingerprints") + cases = stable.get("stable_case_fingerprints") + if not isinstance(plans, list) or not isinstance(cases, list): + raise FreshPersonalHoldoutRunnerError( + "prior fresh stable exposure registry is malformed" + ) + return ( + {_token(item, "stable plan fingerprint") for item in plans}, + {_token(item, "stable case fingerprint") for item in cases}, + ) + + +def _prior_private_exposure_registry(paths: Sequence[str | Path]) -> dict[str, Any]: + if not paths: + raise FreshPersonalHoldoutRunnerError( + "fresh formal holdout requires at least one prior private stable-identity source" + ) + stable_plans: set[str] = set() + stable_cases: set[str] = set() + sources: list[dict[str, str]] = [] + for raw_path in paths: + path = Path(raw_path).expanduser().resolve() + raw = _load(path) + schema = _token(raw.get("schema"), "prior private schema") + if schema == "applaylist-private-runtime-music-evidence-r1": + for role, plan_a, plan_b in _historical_private_sequences(raw): + a_fp = _plan_fingerprint(plan_a, namespace="stable-plan") + b_fp = _plan_fingerprint(plan_b, namespace="stable-plan") + stable_plans.update((a_fp, b_fp)) + stable_cases.add( + "stable-case:" + _sha256_json((role, tuple(sorted((a_fp, b_fp))))) + ) + elif schema == FRESH_PERSONAL_HOLDOUT_PRIVATE_SCHEMA: + plans, cases = _fresh_private_stable_registry(raw) + stable_plans.update(plans) + stable_cases.update(cases) + else: + raise FreshPersonalHoldoutRunnerError( + f"unsupported prior private exposure schema: {schema}" + ) + sources.append( + { + "path_sha256": hashlib.sha256(str(path).encode("utf-8")).hexdigest(), + "content_sha256": hashlib.sha256(path.read_bytes()).hexdigest(), + } + ) + payload = { + "stable_plan_fingerprints": sorted(stable_plans), + "stable_case_fingerprints": sorted(stable_cases), + "source_content_sha256": sorted(item["content_sha256"] for item in sources), + } + return { + "stable_plan_fingerprints": stable_plans, + "stable_case_fingerprints": stable_cases, + "sources": sources, + "registry_fingerprint": "prior-private-exposure-registry:" + _sha256_json(payload), + } + + +def _snapshot_display_to_track_id(snapshot_path: str | Path) -> dict[str, str]: + path = Path(snapshot_path).expanduser().resolve() + raw = _load(path) + if raw.get("schema") != "applaylist-local-library-snapshot-r1": + raise FreshPersonalHoldoutRunnerError("unexpected snapshot schema for stable exposure") + tracks = raw.get("tracks") + if not isinstance(tracks, list) or not tracks: + raise FreshPersonalHoldoutRunnerError("snapshot contains no tracks") + mapping: dict[str, str] = {} + for item in tracks: + if not isinstance(item, Mapping): + raise FreshPersonalHoldoutRunnerError("snapshot track must be an object") + name = _token(item.get("display_name"), "display_name") + track_id = _token(item.get("track_id"), "track_id") + if name in mapping and mapping[name] != track_id: + raise FreshPersonalHoldoutRunnerError( + "snapshot display names are not unique enough for stable exposure binding" + ) + mapping[name] = track_id + return mapping + + +def _current_stable_exposure( + cases: Sequence[Mapping[str, Any]], + *, + snapshot_path: str | Path, +) -> dict[str, Any]: + names = _snapshot_display_to_track_id(snapshot_path) + stable_plans: set[str] = set() + stable_cases: set[str] = set() + case_rows: list[dict[str, Any]] = [] + for case in cases: + role = _token(case.get("set_role"), "set_role") + resolved: dict[str, tuple[str, ...]] = {} + for field in ("plan_a", "plan_b"): + display_plan = _plan_tuple(case.get(field), field) + try: + track_plan = tuple(names[name] for name in display_plan) + except KeyError as exc: + raise FreshPersonalHoldoutRunnerError( + "reviewer plan contains a display name absent from frozen snapshot" + ) from exc + resolved[field] = track_plan + a_fp = _plan_fingerprint(resolved["plan_a"], namespace="stable-plan") + b_fp = _plan_fingerprint(resolved["plan_b"], namespace="stable-plan") + case_fp = "stable-case:" + _sha256_json((role, tuple(sorted((a_fp, b_fp))))) + stable_plans.update((a_fp, b_fp)) + stable_cases.add(case_fp) + case_rows.append( + { + "case_id": _token(case.get("case_id"), "case_id"), + "stable_plan_a_fingerprint": a_fp, + "stable_plan_b_fingerprint": b_fp, + "stable_case_fingerprint": case_fp, + } + ) + return { + "stable_plan_fingerprints": stable_plans, + "stable_case_fingerprints": stable_cases, + "case_rows": case_rows, + } + + +def _assert_fresh_cases( + cases: list[dict[str, Any]], + reviewer_registry: Mapping[str, Any], + private_registry: Mapping[str, Any], + *, + snapshot_path: str | Path, +) -> dict[str, Any]: + prior_cases = set(reviewer_registry["case_fingerprints"]) + prior_plans = set(reviewer_registry["plan_fingerprints"]) + for case in cases: + if not isinstance(case, Mapping): + raise FreshPersonalHoldoutRunnerError("reviewer case must be an object") + if _case_exposure_fingerprint(case) in prior_cases: + raise FreshPersonalHoldoutRunnerError( + "fresh holdout selected an A/B case pair already exposed in prior review" + ) + for field in ("plan_a", "plan_b"): + plan = _plan_tuple(case.get(field), field) + if _plan_fingerprint(plan) in prior_plans: + raise FreshPersonalHoldoutRunnerError( + "fresh holdout selected a plan sequence already exposed in prior review" + ) + + stable = _current_stable_exposure(cases, snapshot_path=snapshot_path) + if stable["stable_plan_fingerprints"] & set(private_registry["stable_plan_fingerprints"]): + raise FreshPersonalHoldoutRunnerError( + "fresh holdout selected a stable track-id plan already exposed in prior review" + ) + if stable["stable_case_fingerprints"] & set(private_registry["stable_case_fingerprints"]): + raise FreshPersonalHoldoutRunnerError( + "fresh holdout selected a stable track-id A/B pair already exposed in prior review" + ) + return stable + + +def _workspace_session_id( + private: Mapping[str, Any], + reviewer: Mapping[str, Any], + *, + reviewer_exposure_registry_fingerprint: str, + private_exposure_registry_fingerprint: str, +) -> str: + material = { + "version": REVIEWER_WORKSPACE_VERSION, + "canonical_sha": private.get("canonical_sha"), + "preregistration_fingerprint": private.get("preregistration_fingerprint"), + "selection_manifest_fingerprint": (private.get("selection") or {}).get( + "manifest_fingerprint" + ), + "effective_cohort_id": (private.get("effective_cohort") or {}).get("cohort_id"), + "reviewer_packet_prebind_fingerprint": reviewer.get("packet_fingerprint"), + "reviewer_exposure_registry_fingerprint": reviewer_exposure_registry_fingerprint, + "private_exposure_registry_fingerprint": private_exposure_registry_fingerprint, + } + return "curation-session:" + _sha256_json(material)[:32] + + +def _write_review_csv( + *, + path: Path, + cases: list[dict[str, Any]], + session_id: str, + packet_fingerprint: str, +) -> None: + columns = [ + "case_index", + "case_id", + "set_role", + "assignment_id", + "dataset_role", + "curation_session_id", + "reviewer_packet_fingerprint", + "reviewer_ref", + "preference", + "energy_flow_plan_a", + "energy_flow_plan_b", + "dramaturgical_fit_plan_a", + "dramaturgical_fit_plan_b", + "set_coherence_plan_a", + "set_coherence_plan_b", + "alternative_usefulness_plan_a", + "alternative_usefulness_plan_b", + "confidence", + "prior_case_exposure", + "judgment_mode", + "transition_execution_used", + "transition_preview_heard", + "algorithm_identity_was_hidden", + "reason_codes", + "notes", + "observed_at", + ] + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=columns) + writer.writeheader() + for index, case in enumerate(cases, start=1): + row = { + "case_index": index, + "case_id": _token(case.get("case_id"), "case_id"), + "set_role": _token(case.get("set_role"), "set_role"), + "assignment_id": _token(case.get("assignment_id"), "assignment_id"), + "dataset_role": "personal_holdout", + "curation_session_id": session_id, + "reviewer_packet_fingerprint": packet_fingerprint, + } + for field in _HUMAN_FIELDS: + row[field] = "" + writer.writerow(row) + + +def finalize_fresh_holdout_reviewer_workspace( + result: Mapping[str, str], + *, + snapshot_path: str | Path, + prior_reviewer_packet_paths: Sequence[str | Path], + prior_private_manifest_paths: Sequence[str | Path], +) -> dict[str, str]: + """Bind reviewer-safe files to frozen preregistration and reject prior exposure.""" + private_path = Path(_token(result.get("private_manifest"), "private_manifest")) + reviewer_path = Path(_token(result.get("reviewer_packet"), "reviewer_packet")) + csv_path = Path(_token(result.get("review_csv"), "review_csv")) + + _verify_prefinalization_hash( + result=result, + path=private_path, + result_key="private_manifest_sha256", + ) + _verify_prefinalization_hash( + result=result, + path=reviewer_path, + result_key="reviewer_packet_sha256", + ) + _verify_prefinalization_hash( + result=result, + path=csv_path, + result_key="review_csv_sha256", + ) + + private = _load(private_path) + reviewer = _load(reviewer_path) + + if private.get("schema") != FRESH_PERSONAL_HOLDOUT_PRIVATE_SCHEMA: + raise FreshPersonalHoldoutRunnerError("unexpected private fresh-holdout schema") + if reviewer.get("schema") != FRESH_PERSONAL_HOLDOUT_REVIEWER_SCHEMA: + raise FreshPersonalHoldoutRunnerError("unexpected reviewer fresh-holdout schema") + if not private.get("challenger_frozen_before_reviewer_publication"): + raise FreshPersonalHoldoutRunnerError( + "challenger evidence is not frozen before reviewer publication" + ) + + selection = private.get("selection") or {} + policy = private.get("sampling_policy") or {} + cohort = private.get("effective_cohort") or {} + selected = selection.get("selected_case_ids") or [] + fallback = selection.get("fallback_case_ids") or [] + effective = cohort.get("effective_case_ids") or [] + expected_fallback = int(policy.get("fallback_count", -1)) + if len(selected) != 24: + raise FreshPersonalHoldoutRunnerError("frozen holdout selection must contain 24 cases") + if expected_fallback < 0 or len(fallback) != expected_fallback: + raise FreshPersonalHoldoutRunnerError( + "frozen fallback reservoir does not satisfy preregistered fallback_count" + ) + if len(effective) != 24: + raise FreshPersonalHoldoutRunnerError("effective cohort must contain exactly 24 cases") + + cases = reviewer.get("cases") + if not isinstance(cases, list) or len(cases) != 24: + raise FreshPersonalHoldoutRunnerError("reviewer workspace requires exactly 24 cases") + reviewer_case_ids = tuple(_token(case.get("case_id"), "case_id") for case in cases) + if reviewer_case_ids != tuple(_token(item, "effective_case_id") for item in effective): + raise FreshPersonalHoldoutRunnerError( + "reviewer case order/identity does not match frozen effective cohort" + ) + + assignments = private.get("assignments") + if not isinstance(assignments, list): + raise FreshPersonalHoldoutRunnerError("private manifest lacks assignment bindings") + assignment_by_case = { + _token(item.get("case_id"), "assignment case_id"): _token( + item.get("assignment_id"), "assignment_id" + ) + for item in assignments + if isinstance(item, Mapping) + } + specs = ((private.get("candidate_case_specs") or {}).get("case_specs") or []) + role_by_case = { + _token(item.get("case_spec_id"), "case_spec_id"): _token(item.get("set_role"), "set_role") + for item in specs + if isinstance(item, Mapping) + } + for case in cases: + case_id = _token(case.get("case_id"), "case_id") + if assignment_by_case.get(case_id) != _token(case.get("assignment_id"), "assignment_id"): + raise FreshPersonalHoldoutRunnerError( + "reviewer assignment does not match frozen private assignment" + ) + if role_by_case.get(case_id) != _token(case.get("set_role"), "set_role"): + raise FreshPersonalHoldoutRunnerError( + "reviewer set role does not match frozen candidate spec" + ) + + reviewer_registry = _prior_exposure_registry(prior_reviewer_packet_paths) + private_registry = _prior_private_exposure_registry(prior_private_manifest_paths) + stable_current = _assert_fresh_cases( + cases, + reviewer_registry, + private_registry, + snapshot_path=snapshot_path, + ) + + prereg = _token(private.get("preregistration_fingerprint"), "preregistration_fingerprint") + selection_fp = _token(selection.get("manifest_fingerprint"), "selection manifest fingerprint") + cohort_id = _token(cohort.get("cohort_id"), "effective cohort id") + canonical_sha = _token(private.get("canonical_sha"), "canonical_sha") + session_id = _workspace_session_id( + private, + reviewer, + reviewer_exposure_registry_fingerprint=reviewer_registry["registry_fingerprint"], + private_exposure_registry_fingerprint=private_registry["registry_fingerprint"], + ) + + reviewer.pop("packet_fingerprint", None) + reviewer.update( + { + "workspace_version": REVIEWER_WORKSPACE_VERSION, + "canonical_sha": canonical_sha, + "preregistration_fingerprint": prereg, + "selection_manifest_fingerprint": selection_fp, + "effective_cohort_id": cohort_id, + "curation_session_id": session_id, + "dataset_role": "personal_holdout", + "prior_exposure_registry_fingerprint": reviewer_registry["registry_fingerprint"], + "prior_private_exposure_registry_fingerprint": private_registry["registry_fingerprint"], + "explicit_human_fields_required": list(_HUMAN_FIELDS), + } + ) + packet_fingerprint = _sha256_json(reviewer) + reviewer["packet_fingerprint"] = packet_fingerprint + + reviewer_text = json.dumps(reviewer, indent=2, ensure_ascii=False, default=str) + "\n" + forbidden = ( + "challenger_evidence", + "left_assessment", + "right_assessment", + "left_score", + "right_score", + "shadow_right_path_preferred", + "shadow_left_path_preferred", + "greedy_recommend_next", + "bounded_beam", + "absolute_path", + ) + if any(token in reviewer_text for token in forbidden): + raise FreshPersonalHoldoutRunnerError( + "reviewer workspace contains private/model leakage" + ) + + reviewer_path.write_text(reviewer_text, encoding="utf-8") + _write_review_csv( + path=csv_path, + cases=cases, + session_id=session_id, + packet_fingerprint=packet_fingerprint, + ) + + stable_registry_payload = { + "registry_version": "stable-exposure-registry-r1", + "snapshot_ref": private.get("snapshot_ref"), + "stable_plan_fingerprints": sorted(stable_current["stable_plan_fingerprints"]), + "stable_case_fingerprints": sorted(stable_current["stable_case_fingerprints"]), + "case_rows": stable_current["case_rows"], + } + stable_registry_payload["registry_fingerprint"] = ( + "stable-exposure-registry:" + _sha256_json(stable_registry_payload) + ) + private["stable_exposure_registry"] = stable_registry_payload + private["reviewer_workspace_binding"] = { + "workspace_version": REVIEWER_WORKSPACE_VERSION, + "curation_session_id": session_id, + "reviewer_packet_fingerprint": packet_fingerprint, + "prior_exposure_registry_fingerprint": reviewer_registry["registry_fingerprint"], + "prior_exposure_sources": reviewer_registry["sources"], + "prior_private_exposure_registry_fingerprint": private_registry["registry_fingerprint"], + "prior_private_exposure_sources": private_registry["sources"], + "human_labels_present_at_freeze": False, + } + private_path.write_text( + json.dumps(private, indent=2, ensure_ascii=False, default=str) + "\n", + encoding="utf-8", + ) + + csv_rows = list(csv.DictReader(csv_path.open("r", encoding="utf-8", newline=""))) + if len(csv_rows) != 24: + raise FreshPersonalHoldoutRunnerError("review CSV requires exactly 24 rows") + for row in csv_rows: + if any(str(row.get(field, "")).strip() for field in _HUMAN_FIELDS): + raise FreshPersonalHoldoutRunnerError( + "review CSV fabricated a human evidence field" + ) + + finalized = dict(result) + finalized.update( + { + "curation_session_id": session_id, + "reviewer_packet_fingerprint": packet_fingerprint, + "prior_exposure_registry_fingerprint": reviewer_registry["registry_fingerprint"], + "prior_private_exposure_registry_fingerprint": private_registry["registry_fingerprint"], + "stable_exposure_registry_fingerprint": stable_registry_payload["registry_fingerprint"], + "private_manifest_sha256": hashlib.sha256(private_path.read_bytes()).hexdigest(), + "reviewer_packet_sha256": hashlib.sha256(reviewer_path.read_bytes()).hexdigest(), + "review_csv_sha256": hashlib.sha256(csv_path.read_bytes()).hexdigest(), + } + ) + return finalized + + +__all__ = [ + "REVIEWER_WORKSPACE_VERSION", + "finalize_fresh_holdout_reviewer_workspace", +] diff --git a/services/intelligence/fresh_personal_holdout_runner.py b/services/intelligence/fresh_personal_holdout_runner.py new file mode 100644 index 00000000..425db5ce --- /dev/null +++ b/services/intelligence/fresh_personal_holdout_runner.py @@ -0,0 +1,630 @@ +from __future__ import annotations + +import csv +import hashlib +import json +from dataclasses import asdict +from pathlib import Path +from typing import Any, Mapping, Sequence + +from core.intelligence.competitive_curation_contract import CompetitiveCurationPolicy +from core.intelligence.curated_real_library_review_contract import CuratedSetRole +from core.intelligence.human_review_preregistration_r2_contract import HoldoutReplacementPolicyR2 +from core.intelligence.human_review_protocol_r2_contract import ( + HoldoutCandidate, + HoldoutCaseSamplingPolicy, + ReviewDatasetRole, +) +from services.intelligence.competitive_curation import ( + compare_competitive_curation_paths, + track_curation_evidence_from_music_dna, +) +from services.intelligence.human_review_protocol_r2 import ( + build_effective_holdout_cohort_r2, + holdout_replacement_policy_fingerprint, + select_holdout_cases_r2, +) +from services.intelligence.real_library_pilot import ( + MaterializedCase, + RealLibraryPilotError, + _intent_for_case, + _load_json, + _sha256_json, + _track_inputs, + _validate_snapshot, + analyze_real_tracks, + materialize_cases, +) + +FRESH_PERSONAL_HOLDOUT_RUNNER_VERSION = "fresh-personal-holdout-runner-r1" +FRESH_PERSONAL_HOLDOUT_PRIVATE_SCHEMA = "applaylist-fresh-personal-holdout-private-r1" +FRESH_PERSONAL_HOLDOUT_REVIEWER_SCHEMA = "applaylist-fresh-personal-holdout-reviewer-r1" + + +class FreshPersonalHoldoutRunnerError(RuntimeError): + """Fail-closed error for fresh personal holdout execution.""" + + +def _token(value: object, field: str) -> str: + text = str(value).strip() + if not text: + raise FreshPersonalHoldoutRunnerError(f"{field} must not be empty") + return text + + +def _fingerprint(label: str, payload: object) -> str: + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return f"{label}:" + hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def _case_spec_pool( + snapshot_raw: Mapping[str, Any], + *, + sampling_seed: str, + cases_per_role: int = 8, + candidate_scope_size: int = 16, +) -> dict[str, Any]: + """Build deterministic case specs from snapshot identity only. + + This stage intentionally has no access to optimizer outputs, challenger scores, + or human evidence. Track ordering is hash-based from frozen snapshot identity. + """ + snapshot = _validate_snapshot(snapshot_raw) + seed = _token(sampling_seed, "sampling_seed") + tracks = _track_inputs(snapshot_raw) + track_ids = tuple(sorted(tracks)) + minimum_tracks = candidate_scope_size + 1 + if len(track_ids) < minimum_tracks: + raise FreshPersonalHoldoutRunnerError( + f"fresh holdout requires at least {minimum_tracks} tracks in the snapshot" + ) + if cases_per_role < 4: + raise FreshPersonalHoldoutRunnerError("cases_per_role must be at least 4") + + specs: list[dict[str, Any]] = [] + for role in CuratedSetRole: + ranked = sorted( + track_ids, + key=lambda track_id: hashlib.sha256( + f"{FRESH_PERSONAL_HOLDOUT_RUNNER_VERSION}|{seed}|{role.value}|seed|{track_id}".encode( + "utf-8" + ) + ).hexdigest(), + ) + for ordinal, seed_track_id in enumerate(ranked[:cases_per_role]): + remaining = [track_id for track_id in track_ids if track_id != seed_track_id] + candidate_scope = sorted( + remaining, + key=lambda track_id: hashlib.sha256( + ( + f"{FRESH_PERSONAL_HOLDOUT_RUNNER_VERSION}|{seed}|{role.value}|" + f"{ordinal}|scope|{track_id}" + ).encode("utf-8") + ).hexdigest(), + )[:candidate_scope_size] + case_spec_id = "fps_" + hashlib.sha256( + f"{snapshot.library_fingerprint}|{seed}|{role.value}|{ordinal}|{seed_track_id}".encode( + "utf-8" + ) + ).hexdigest()[:24] + specs.append( + { + "case_spec_id": case_spec_id, + "set_role": role.value, + "seed_track_id": seed_track_id, + "candidate_scope_track_ids": candidate_scope, + } + ) + return { + "schema": "applaylist-curated-case-selection-r1", + "snapshot_ref": [snapshot.snapshot_id, snapshot.snapshot_version], + "generator_version": FRESH_PERSONAL_HOLDOUT_RUNNER_VERSION, + "sampling_seed": seed, + "case_specs": specs, + } + + +def _snapshot_scope_fingerprint(snapshot_raw: Mapping[str, Any]) -> str: + tracks = _track_inputs(snapshot_raw) + payload = [ + {"track_id": item.track_id, "file_signature": item.file_signature} + for item in sorted(tracks.values(), key=lambda item: item.track_id) + ] + return _fingerprint("eligible-scope-r1", payload) + + +def _single_case_selection(selection_raw: Mapping[str, Any], spec: Mapping[str, Any]) -> dict[str, Any]: + return { + "schema": selection_raw["schema"], + "snapshot_ref": list(selection_raw["snapshot_ref"]), + "generator_version": selection_raw.get("generator_version"), + "sampling_seed": selection_raw.get("sampling_seed"), + "case_specs": [dict(spec)], + } + + +def _analyze_candidate_pool( + *, + snapshot_raw: Mapping[str, Any], + selection_raw: Mapping[str, Any], +) -> tuple[dict[str, Mapping[str, Any]], dict[str, Any], tuple[dict[str, str], ...]]: + """Analyze each candidate independently so one MIR failure cannot abort the pool.""" + evidence_by_case: dict[str, Mapping[str, Any]] = {} + merged_evidence: dict[str, Any] = {} + failures: list[dict[str, str]] = [] + for spec in selection_raw["case_specs"]: + case_id = _token(spec["case_spec_id"], "case_spec_id") + try: + case_evidence = analyze_real_tracks( + snapshot_raw=snapshot_raw, + selection_raw=_single_case_selection(selection_raw, spec), + ) + except RealLibraryPilotError as exc: + failures.append( + { + "case_id": case_id, + "set_role": str(spec["set_role"]), + "technical_invalidity_reason": "analysis_failed", + "detail": str(exc), + } + ) + continue + evidence_by_case[case_id] = case_evidence + merged_evidence.update(case_evidence) + return evidence_by_case, merged_evidence, tuple(failures) + + +def _materialize_candidate_pool( + *, + snapshot_raw: Mapping[str, Any], + selection_raw: Mapping[str, Any], + evidence_by_case: Mapping[str, Mapping[str, Any]], + initial_failures: Sequence[Mapping[str, str]], + database_path: str | Path, + generated_at: str, + blinding_seed: str, +) -> tuple[tuple[MaterializedCase, ...], tuple[dict[str, str], ...]]: + """Materialize each candidate independently so one invalid case cannot abort the pool.""" + materialized: list[MaterializedCase] = [] + failures: list[dict[str, str]] = [dict(item) for item in initial_failures] + failed_case_ids = {item["case_id"] for item in failures} + for spec in selection_raw["case_specs"]: + case_id = _token(spec["case_spec_id"], "case_spec_id") + case_evidence = evidence_by_case.get(case_id) + if case_evidence is None: + if case_id not in failed_case_ids: + failures.append( + { + "case_id": case_id, + "set_role": str(spec["set_role"]), + "technical_invalidity_reason": "analysis_failed", + "detail": "per-case MIR evidence missing after analysis boundary", + } + ) + failed_case_ids.add(case_id) + continue + try: + result = materialize_cases( + snapshot_raw=snapshot_raw, + selection_raw=_single_case_selection(selection_raw, spec), + evidence=case_evidence, + database_path=database_path, + generated_at=generated_at, + blinding_seed=blinding_seed, + ) + except RealLibraryPilotError as exc: + failures.append( + { + "case_id": case_id, + "set_role": str(spec["set_role"]), + "technical_invalidity_reason": "engineering_materialization_failed", + "detail": str(exc), + } + ) + continue + if len(result) != 1 or result[0].case.case_id != case_id: + raise FreshPersonalHoldoutRunnerError("single-case materialization identity mismatch") + materialized.append(result[0]) + return tuple(materialized), tuple(failures) + + +def _materialized_by_case(cases: Sequence[MaterializedCase]) -> dict[str, MaterializedCase]: + result: dict[str, MaterializedCase] = {} + for item in cases: + if item.case.case_id in result: + raise FreshPersonalHoldoutRunnerError("duplicate materialized case id") + result[item.case.case_id] = item + return result + + +def _holdout_candidates( + selection_raw: Mapping[str, Any], + cases: Sequence[MaterializedCase], + failures: Sequence[Mapping[str, str]], +) -> tuple[HoldoutCandidate, ...]: + valid = _materialized_by_case(cases) + failed = {item["case_id"]: item for item in failures} + rows: list[HoldoutCandidate] = [] + for spec in selection_raw["case_specs"]: + case_id = _token(spec["case_spec_id"], "case_spec_id") + role = CuratedSetRole(str(spec["set_role"]).strip().lower()) + if case_id in valid: + rows.append( + HoldoutCandidate( + candidate_id=f"holdout:{case_id}", + case_id=case_id, + set_role=role, + engineering_acceptance_passed=True, + ) + ) + else: + failure = failed.get(case_id) + rows.append( + HoldoutCandidate( + candidate_id=f"holdout:{case_id}", + case_id=case_id, + set_role=role, + engineering_acceptance_passed=False, + technical_invalidity_reason=( + failure["technical_invalidity_reason"] + if failure is not None + else "engineering_materialization_missing" + ), + ) + ) + return tuple(rows) + + +def _private_track_evidence(evidence: Mapping[str, Any]) -> tuple[Any, ...]: + rows = [] + for _, item in sorted(evidence.items()): + style_tags = (item.source.genre,) if item.source.genre else None + rows.append( + track_curation_evidence_from_music_dna( + music_dna=item.music_dna, + style_tags=style_tags, + vocal_presence=None, + ) + ) + return tuple(rows) + + +def _optimizer_alternative(item: MaterializedCase, *, strategy: str): + result = item.greedy_result if strategy == "greedy" else item.beam_result + path_id = item.case.greedy_plan.path_id if strategy == "greedy" else item.case.beam_plan.path_id + for alternative in result.alternatives: + if alternative.path_id == path_id: + return alternative + raise FreshPersonalHoldoutRunnerError( + f"materialized {strategy} path missing for {item.case.case_id}" + ) + + +def _challenger_record( + item: MaterializedCase, + *, + case_spec: Mapping[str, Any], + track_evidence: tuple[Any, ...], +) -> dict[str, Any]: + role = CuratedSetRole(str(case_spec["set_role"]).strip().lower()) + seed_track_id = _token(case_spec["seed_track_id"], "seed_track_id") + candidate_ids = tuple( + _token(value, "candidate_scope_track_id") + for value in case_spec["candidate_scope_track_ids"] + ) + intent = _intent_for_case( + case_id=item.case.case_id, + role=role, + seed_track_id=seed_track_id, + candidate_track_ids=candidate_ids, + ) + greedy = _optimizer_alternative(item, strategy="greedy") + beam = _optimizer_alternative(item, strategy="beam") + left, right, comparison = compare_competitive_curation_paths( + left=greedy, + right=beam, + intent=intent, + track_evidence=track_evidence, + policy=CompetitiveCurationPolicy(), + ) + return { + "case_id": item.case.case_id, + "left_assessment": asdict(left), + "right_assessment": asdict(right), + "comparison": asdict(comparison), + } + + +def _reviewer_case(item: MaterializedCase, *, names: Mapping[str, str]) -> dict[str, Any]: + assignment = item.assignment + case = item.case + plan_by_id = { + case.greedy_plan.plan_id: case.greedy_plan, + case.beam_plan.plan_id: case.beam_plan, + } + plan_a = plan_by_id[assignment.slot_a_plan_id] + plan_b = plan_by_id[assignment.slot_b_plan_id] + return { + "case_id": case.case_id, + "set_role": case.set_role.value, + "assignment_id": assignment.assignment_id, + "plan_a": [names[track_id] for track_id in plan_a.ordered_track_ids], + "plan_b": [names[track_id] for track_id in plan_b.ordered_track_ids], + "required_review_dimensions": [ + "energy_flow", + "dramaturgical_fit", + "set_coherence", + "alternative_usefulness", + ], + "allowed_preference": ["plan_a", "plan_b", "tie", "abstain"], + "transition_execution_required": False, + } + + +def _write_review_csv(path: Path, rows: Sequence[dict[str, Any]]) -> None: + columns = [ + "case_index", + "case_id", + "set_role", + "assignment_id", + "preference", + "energy_flow_plan_a", + "energy_flow_plan_b", + "dramaturgical_fit_plan_a", + "dramaturgical_fit_plan_b", + "set_coherence_plan_a", + "set_coherence_plan_b", + "alternative_usefulness_plan_a", + "alternative_usefulness_plan_b", + "confidence", + "reason_codes", + "notes", + "observed_at", + ] + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=columns) + writer.writeheader() + for index, row in enumerate(rows, start=1): + writer.writerow( + { + "case_index": index, + "case_id": row["case_id"], + "set_role": row["set_role"], + "assignment_id": row["assignment_id"], + } + ) + + +def materialize_fresh_personal_holdout_r1( + *, + snapshot_path: str | Path, + output_dir: str | Path, + database_path: str | Path, + canonical_sha: str, + generated_at: str, + sampling_seed: str, + blinding_seed: str, + cases_per_role: int = 8, + candidate_scope_size: int = 16, + fallback_count: int = 12, +) -> dict[str, str]: + """Materialize a fresh, frozen, reviewer-safe personal curation holdout.""" + snapshot_raw = _load_json(snapshot_path) + snapshot = _validate_snapshot(snapshot_raw) + canonical = _token(canonical_sha, "canonical_sha") + generated = _token(generated_at, "generated_at") + + selection_raw = _case_spec_pool( + snapshot_raw, + sampling_seed=sampling_seed, + cases_per_role=cases_per_role, + candidate_scope_size=candidate_scope_size, + ) + evidence_by_case, evidence, analysis_failures = _analyze_candidate_pool( + snapshot_raw=snapshot_raw, + selection_raw=selection_raw, + ) + cases, candidate_failures = _materialize_candidate_pool( + snapshot_raw=snapshot_raw, + selection_raw=selection_raw, + evidence_by_case=evidence_by_case, + initial_failures=analysis_failures, + database_path=database_path, + generated_at=generated, + blinding_seed=blinding_seed, + ) + if not cases: + raise FreshPersonalHoldoutRunnerError("candidate materialization produced no valid cases") + + policy = HoldoutCaseSamplingPolicy( + policy_id="fresh-personal-holdout-r1", + dataset_role=ReviewDatasetRole.PERSONAL_HOLDOUT, + canonical_sha=canonical, + snapshot_fingerprint=snapshot.library_fingerprint, + eligible_scope_fingerprint=_snapshot_scope_fingerprint(snapshot_raw), + source_case_generator_version=FRESH_PERSONAL_HOLDOUT_RUNNER_VERSION, + sampling_seed=_token(sampling_seed, "sampling_seed"), + role_quotas=tuple((role, 4) for role in CuratedSetRole), + fallback_count=fallback_count, + activation_authorized=False, + ) + selection = select_holdout_cases_r2( + policy=policy, + candidates=_holdout_candidates(selection_raw, cases, candidate_failures), + ) + prereg_payload = { + "runner_version": FRESH_PERSONAL_HOLDOUT_RUNNER_VERSION, + "canonical_sha": canonical, + "snapshot_ref": [snapshot.snapshot_id, snapshot.snapshot_version], + "snapshot_fingerprint": snapshot.library_fingerprint, + "sampling_policy": asdict(policy), + "selection": asdict(selection), + "candidate_failures": candidate_failures, + "generated_at": generated, + } + prereg_fingerprint = _fingerprint("fresh-personal-holdout-prereg-r1", prereg_payload) + replacement_policy = HoldoutReplacementPolicyR2( + policy_id="fresh-personal-holdout-replacement-r1", + selection_manifest_fingerprint=selection.manifest_fingerprint, + preregistration_manifest_fingerprint=prereg_fingerprint, + frozen_at=generated, + allowed_technical_invalidity_reasons=( + "audio_path_unreadable", + "analysis_failed", + "engineering_acceptance_failed", + "review_packet_materialization_failed", + ), + activation_authorized=False, + ) + cohort = build_effective_holdout_cohort_r2( + selection=selection, + replacement_policy=replacement_policy, + preregistration_manifest_fingerprint=prereg_fingerprint, + technical_invalidities=(), + ) + + by_case = _materialized_by_case(cases) + spec_by_case = { + _token(spec["case_spec_id"], "case_spec_id"): spec + for spec in selection_raw["case_specs"] + } + frozen_case_ids = tuple(selection.selected_case_ids) + tuple(selection.fallback_case_ids) + if any(case_id not in by_case for case_id in frozen_case_ids): + raise FreshPersonalHoldoutRunnerError( + "selected/fallback holdout references a non-materialized candidate" + ) + + track_evidence = _private_track_evidence(evidence) + challenger_rows = [ + _challenger_record( + by_case[case_id], + case_spec=spec_by_case[case_id], + track_evidence=track_evidence, + ) + for case_id in frozen_case_ids + ] + + names = {track_id: item.source.display_name for track_id, item in evidence.items()} + reviewer_rows = [ + _reviewer_case(by_case[case_id], names=names) + for case_id in cohort.effective_case_ids + ] + role_counts = { + role.value: sum(row["set_role"] == role.value for row in reviewer_rows) + for role in CuratedSetRole + } + if len(reviewer_rows) != 24 or any(count != 4 for count in role_counts.values()): + raise FreshPersonalHoldoutRunnerError( + "effective reviewer cohort must contain exactly 24 cases, four per set role" + ) + + track_provenance = [ + { + "track_id": track_id, + "absolute_path": item.source.absolute_path, + "inventory_file_signature": item.source.file_signature, + "content_sha256": item.content_sha256, + "analysis_revision": item.music_dna.identity.analysis_revision, + "genre": item.source.genre, + } + for track_id, item in sorted(evidence.items()) + ] + private_payload = { + "schema": FRESH_PERSONAL_HOLDOUT_PRIVATE_SCHEMA, + "runner_version": FRESH_PERSONAL_HOLDOUT_RUNNER_VERSION, + "generated_at": generated, + "canonical_sha": canonical, + "snapshot_ref": [snapshot.snapshot_id, snapshot.snapshot_version], + "snapshot_fingerprint": snapshot.library_fingerprint, + "privacy": { + "contains_local_absolute_paths": True, + "publishable_to_public_repo": False, + "storage_class": "CASER_PRIVATE_EVIDENCE", + }, + "track_provenance": track_provenance, + "candidate_case_specs": selection_raw, + "candidate_failures": list(candidate_failures), + "sampling_policy": asdict(policy), + "selection": asdict(selection), + "preregistration_fingerprint": prereg_fingerprint, + "replacement_policy": asdict(replacement_policy), + "replacement_policy_fingerprint": holdout_replacement_policy_fingerprint( + replacement_policy + ), + "effective_cohort": asdict(cohort), + "source_review_cases": [ + asdict(by_case[case_id].case) for case_id in frozen_case_ids + ], + "source_review_cases_frozen_before_reviewer_publication": True, + "assignments": [ + asdict(by_case[case_id].assignment) for case_id in frozen_case_ids + ], + "challenger_evidence": challenger_rows, + "challenger_frozen_before_reviewer_publication": True, + "activation_authorized": False, + "personal_dj_model_training_authorized": False, + } + reviewer_payload = { + "schema": FRESH_PERSONAL_HOLDOUT_REVIEWER_SCHEMA, + "protocol_version": "human-dj-review-r2", + "generated_at": generated, + "snapshot_ref": [snapshot.snapshot_id, snapshot.snapshot_version], + "algorithm_identity_hidden": True, + "sequence_only": True, + "transition_execution_required": False, + "required_review_dimensions": [ + "energy_flow", + "dramaturgical_fit", + "set_coherence", + "alternative_usefulness", + ], + "cases": reviewer_rows, + "activation_authorized": False, + "personal_dj_model_training_authorized": False, + } + reviewer_payload["packet_fingerprint"] = _sha256_json(reviewer_payload) + + output = Path(output_dir).expanduser().resolve() + output.mkdir(parents=True, exist_ok=True) + private_path = output / "APPLAYLIST_FRESH_PERSONAL_HOLDOUT_R1.private.json" + reviewer_path = output / "APPLAYLIST_FRESH_PERSONAL_HOLDOUT_REVIEWER_R1.json" + csv_path = output / "APPLAYLIST_FRESH_PERSONAL_HOLDOUT_REVIEW_R1.csv" + private_path.write_text( + json.dumps(private_payload, indent=2, ensure_ascii=False, default=str) + "\n", + encoding="utf-8", + ) + reviewer_path.write_text( + json.dumps(reviewer_payload, indent=2, ensure_ascii=False, default=str) + "\n", + encoding="utf-8", + ) + _write_review_csv(csv_path, reviewer_rows) + + private_text = private_path.read_text(encoding="utf-8") + reviewer_text = reviewer_path.read_text(encoding="utf-8") + if any(token in reviewer_text for token in ("absolute_path", "left_score", "right_score")): + raise FreshPersonalHoldoutRunnerError("reviewer packet leaked private/challenger evidence") + if any(path in reviewer_text for path in (item.source.absolute_path for item in evidence.values())): + raise FreshPersonalHoldoutRunnerError("reviewer packet leaked an absolute audio path") + if "challenger_evidence" not in private_text: + raise FreshPersonalHoldoutRunnerError("private preregistration is missing challenger evidence") + if "source_review_cases" not in private_text: + raise FreshPersonalHoldoutRunnerError("private preregistration is missing source review cases") + + return { + "private_manifest": str(private_path), + "reviewer_packet": str(reviewer_path), + "review_csv": str(csv_path), + "private_manifest_sha256": hashlib.sha256(private_path.read_bytes()).hexdigest(), + "reviewer_packet_sha256": hashlib.sha256(reviewer_path.read_bytes()).hexdigest(), + "review_csv_sha256": hashlib.sha256(csv_path.read_bytes()).hexdigest(), + "effective_case_count": str(len(cohort.effective_case_ids)), + "fallback_case_count": str(len(selection.fallback_case_ids)), + "preregistration_fingerprint": prereg_fingerprint, + } + + +__all__ = [ + "FRESH_PERSONAL_HOLDOUT_RUNNER_VERSION", + "FreshPersonalHoldoutRunnerError", + "materialize_fresh_personal_holdout_r1", +] diff --git a/tests/test_fresh_holdout_reviewer_workspace_r1.py b/tests/test_fresh_holdout_reviewer_workspace_r1.py new file mode 100644 index 00000000..67b7f8dc --- /dev/null +++ b/tests/test_fresh_holdout_reviewer_workspace_r1.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +import csv +import hashlib +import json +from pathlib import Path + +import pytest + +from services.intelligence.fresh_holdout_reviewer_workspace import ( + finalize_fresh_holdout_reviewer_workspace, +) +from services.intelligence.fresh_personal_holdout_runner import FreshPersonalHoldoutRunnerError + + +def _cases() -> list[dict]: + return [ + { + "case_id": f"case-{index}", + "set_role": ("opening", "build", "mid_set", "peak", "reset", "closing")[ + (index - 1) // 4 + ], + "assignment_id": f"assignment-{index}", + "plan_a": [f"A{index}-1", f"A{index}-2"], + "plan_b": [f"B{index}-1", f"B{index}-2"], + } + for index in range(1, 25) + ] + + +def _snapshot(tmp_path: Path) -> Path: + path = tmp_path / "snapshot.json" + names = sorted( + { + name + for case in _cases() + for field in ("plan_a", "plan_b") + for name in case[field] + } + ) + path.write_text( + json.dumps( + { + "schema": "applaylist-local-library-snapshot-r1", + "tracks": [ + {"track_id": f"trk::{name}", "display_name": name} + for name in names + ], + } + ), + encoding="utf-8", + ) + return path + + +def _prior_packet(tmp_path: Path, *, duplicate_current_plan: bool = False) -> Path: + path = tmp_path / "prior-reviewer.json" + plan_a = ["historical-a1", "historical-a2"] + plan_b = ["historical-b1", "historical-b2"] + if duplicate_current_plan: + plan_a = ["A1-1", "A1-2"] + path.write_text( + json.dumps( + { + "schema": "applaylist-blind-human-dj-review-packet-r1", + "cases": [ + { + "case_id": "historical-case", + "set_role": "opening", + "assignment_id": "historical-assignment", + "plan_a": plan_a, + "plan_b": plan_b, + } + ], + } + ), + encoding="utf-8", + ) + return path + + +def _prior_private(tmp_path: Path, *, duplicate_stable_plan: bool = False) -> Path: + path = tmp_path / "prior-private.json" + greedy_ids = ["old-track-a1", "old-track-a2"] + if duplicate_stable_plan: + greedy_ids = ["trk::A1-1", "trk::A1-2"] + path.write_text( + json.dumps( + { + "schema": "applaylist-private-runtime-music-evidence-r1", + "cases": [ + { + "case": { + "case_id": "historical-case", + "set_role": "opening", + "greedy_plan": { + "plan_id": "historical-greedy", + "ordered_track_ids": greedy_ids, + }, + "beam_plan": { + "plan_id": "historical-beam", + "ordered_track_ids": ["old-track-b1", "old-track-b2"], + }, + }, + "assignment": { + "case_id": "historical-case", + "assignment_id": "historical-assignment", + "slot_a_plan_id": "historical-greedy", + "slot_b_plan_id": "historical-beam", + }, + } + ], + } + ), + encoding="utf-8", + ) + return path + + +def _files(tmp_path: Path) -> dict[str, str]: + private_path = tmp_path / "private.json" + reviewer_path = tmp_path / "reviewer.json" + csv_path = tmp_path / "review.csv" + cases = _cases() + private = { + "schema": "applaylist-fresh-personal-holdout-private-r1", + "canonical_sha": "canonical-sha", + "snapshot_ref": ["snapshot", "r1"], + "preregistration_fingerprint": "prereg:fingerprint", + "sampling_policy": {"fallback_count": 12}, + "selection": { + "manifest_fingerprint": "selection:fingerprint", + "selected_case_ids": [case["case_id"] for case in cases], + "fallback_case_ids": [f"fallback-{index}" for index in range(12)], + }, + "effective_cohort": { + "cohort_id": "cohort:id", + "effective_case_ids": [case["case_id"] for case in cases], + }, + "candidate_case_specs": { + "case_specs": [ + {"case_spec_id": case["case_id"], "set_role": case["set_role"]} + for case in cases + ] + }, + "assignments": [ + {"case_id": case["case_id"], "assignment_id": case["assignment_id"]} + for case in cases + ], + "challenger_frozen_before_reviewer_publication": True, + "challenger_evidence": [{"case_id": "private-only"}], + } + reviewer = { + "schema": "applaylist-fresh-personal-holdout-reviewer-r1", + "packet_fingerprint": "prebind", + "cases": cases, + } + private_path.write_text(json.dumps(private), encoding="utf-8") + reviewer_path.write_text(json.dumps(reviewer), encoding="utf-8") + csv_path.write_text("old\n", encoding="utf-8") + return { + "private_manifest": str(private_path), + "reviewer_packet": str(reviewer_path), + "review_csv": str(csv_path), + "private_manifest_sha256": hashlib.sha256(private_path.read_bytes()).hexdigest(), + "reviewer_packet_sha256": hashlib.sha256(reviewer_path.read_bytes()).hexdigest(), + "review_csv_sha256": hashlib.sha256(csv_path.read_bytes()).hexdigest(), + } + + +def _finalize(tmp_path: Path) -> dict[str, str]: + return finalize_fresh_holdout_reviewer_workspace( + _files(tmp_path), + snapshot_path=_snapshot(tmp_path), + prior_reviewer_packet_paths=(_prior_packet(tmp_path),), + prior_private_manifest_paths=(_prior_private(tmp_path),), + ) + + +def test_finalize_binds_workspace_without_exposing_challenger(tmp_path: Path) -> None: + result = _finalize(tmp_path) + reviewer = Path(result["reviewer_packet"]).read_text(encoding="utf-8") + assert "prereg:fingerprint" in reviewer + assert "selection:fingerprint" in reviewer + assert "cohort:id" in reviewer + assert "prior-exposure-registry:" in reviewer + assert "prior-private-exposure-registry:" in reviewer + assert "challenger_evidence" not in reviewer + assert "left_score" not in reviewer + assert result["curation_session_id"].startswith("curation-session:") + + +def test_finalize_leaves_all_human_and_attestation_fields_empty(tmp_path: Path) -> None: + result = _finalize(tmp_path) + rows = list(csv.DictReader(Path(result["review_csv"]).open("r", encoding="utf-8"))) + assert len(rows) == 24 + human_fields = ( + "reviewer_ref", + "preference", + "energy_flow_plan_a", + "energy_flow_plan_b", + "dramaturgical_fit_plan_a", + "dramaturgical_fit_plan_b", + "set_coherence_plan_a", + "set_coherence_plan_b", + "alternative_usefulness_plan_a", + "alternative_usefulness_plan_b", + "confidence", + "prior_case_exposure", + "judgment_mode", + "transition_execution_used", + "transition_preview_heard", + "algorithm_identity_was_hidden", + "reason_codes", + "notes", + "observed_at", + ) + for row in rows: + assert row["dataset_role"] == "personal_holdout" + assert row["curation_session_id"].startswith("curation-session:") + assert all(row[field] == "" for field in human_fields) + + +def test_private_manifest_binds_final_reviewer_packet_and_stable_registry(tmp_path: Path) -> None: + result = _finalize(tmp_path) + private = json.loads(Path(result["private_manifest"]).read_text(encoding="utf-8")) + binding = private["reviewer_workspace_binding"] + assert binding["reviewer_packet_fingerprint"] == result["reviewer_packet_fingerprint"] + assert binding["prior_exposure_registry_fingerprint"] == result[ + "prior_exposure_registry_fingerprint" + ] + assert binding["prior_private_exposure_registry_fingerprint"] == result[ + "prior_private_exposure_registry_fingerprint" + ] + assert private["stable_exposure_registry"]["registry_fingerprint"] == result[ + "stable_exposure_registry_fingerprint" + ] + assert binding["human_labels_present_at_freeze"] is False + + +def test_finalize_requires_prior_visible_exposure_source(tmp_path: Path) -> None: + with pytest.raises(FreshPersonalHoldoutRunnerError, match="prior reviewer packet"): + finalize_fresh_holdout_reviewer_workspace( + _files(tmp_path), + snapshot_path=_snapshot(tmp_path), + prior_reviewer_packet_paths=(), + prior_private_manifest_paths=(_prior_private(tmp_path),), + ) + + +def test_finalize_requires_prior_private_stable_identity_source(tmp_path: Path) -> None: + with pytest.raises(FreshPersonalHoldoutRunnerError, match="prior private"): + finalize_fresh_holdout_reviewer_workspace( + _files(tmp_path), + snapshot_path=_snapshot(tmp_path), + prior_reviewer_packet_paths=(_prior_packet(tmp_path),), + prior_private_manifest_paths=(), + ) + + +def test_finalize_rejects_any_previously_exposed_visible_plan(tmp_path: Path) -> None: + with pytest.raises(FreshPersonalHoldoutRunnerError, match="plan sequence already exposed"): + finalize_fresh_holdout_reviewer_workspace( + _files(tmp_path), + snapshot_path=_snapshot(tmp_path), + prior_reviewer_packet_paths=( + _prior_packet(tmp_path, duplicate_current_plan=True), + ), + prior_private_manifest_paths=(_prior_private(tmp_path),), + ) + + +def test_finalize_rejects_stable_track_identity_even_if_display_names_differ(tmp_path: Path) -> None: + with pytest.raises(FreshPersonalHoldoutRunnerError, match="stable track-id plan"): + finalize_fresh_holdout_reviewer_workspace( + _files(tmp_path), + snapshot_path=_snapshot(tmp_path), + prior_reviewer_packet_paths=(_prior_packet(tmp_path),), + prior_private_manifest_paths=( + _prior_private(tmp_path, duplicate_stable_plan=True), + ), + ) + + +def test_finalize_rejects_prefinalization_reviewer_tamper(tmp_path: Path) -> None: + result = _files(tmp_path) + reviewer_path = Path(result["reviewer_packet"]) + reviewer_path.write_text(reviewer_path.read_text(encoding="utf-8") + " ", encoding="utf-8") + with pytest.raises(FreshPersonalHoldoutRunnerError, match="pre-finalization file hash mismatch"): + finalize_fresh_holdout_reviewer_workspace( + result, + snapshot_path=_snapshot(tmp_path), + prior_reviewer_packet_paths=(_prior_packet(tmp_path),), + prior_private_manifest_paths=(_prior_private(tmp_path),), + ) + + +def test_finalize_rejects_reviewer_cohort_not_equal_to_frozen_effective_cohort(tmp_path: Path) -> None: + result = _files(tmp_path) + reviewer_path = Path(result["reviewer_packet"]) + reviewer = json.loads(reviewer_path.read_text(encoding="utf-8")) + reviewer["cases"][0]["case_id"] = "tampered-case" + reviewer_path.write_text(json.dumps(reviewer), encoding="utf-8") + result["reviewer_packet_sha256"] = hashlib.sha256(reviewer_path.read_bytes()).hexdigest() + with pytest.raises(FreshPersonalHoldoutRunnerError, match="does not match frozen effective cohort"): + finalize_fresh_holdout_reviewer_workspace( + result, + snapshot_path=_snapshot(tmp_path), + prior_reviewer_packet_paths=(_prior_packet(tmp_path),), + prior_private_manifest_paths=(_prior_private(tmp_path),), + ) diff --git a/tests/test_fresh_personal_holdout_cli_r1.py b/tests/test_fresh_personal_holdout_cli_r1.py new file mode 100644 index 00000000..eac8cadb --- /dev/null +++ b/tests/test_fresh_personal_holdout_cli_r1.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import json + +import pytest + +from scripts import applaylist_fresh_personal_holdout as cli +from services.intelligence.fresh_personal_holdout_runner import FreshPersonalHoldoutRunnerError + + +def test_verify_canonical_checkout_accepts_exact_clean_checkout(monkeypatch) -> None: + values = { + ("rev-parse", "HEAD"): "canonical-sha", + ("rev-parse", "--abbrev-ref", "HEAD"): "feature/bundle-0-bootstrap", + ("status", "--porcelain"): "", + } + monkeypatch.setattr(cli, "_git", lambda *args: values[args]) + cli.verify_canonical_checkout( + canonical_sha="canonical-sha", + canonical_branch="feature/bundle-0-bootstrap", + ) + + +def test_verify_canonical_checkout_rejects_wrong_head(monkeypatch) -> None: + values = { + ("rev-parse", "HEAD"): "wrong-sha", + ("rev-parse", "--abbrev-ref", "HEAD"): "feature/bundle-0-bootstrap", + ("status", "--porcelain"): "", + } + monkeypatch.setattr(cli, "_git", lambda *args: values[args]) + with pytest.raises(FreshPersonalHoldoutRunnerError, match="does not match declared canonical SHA"): + cli.verify_canonical_checkout( + canonical_sha="canonical-sha", + canonical_branch="feature/bundle-0-bootstrap", + ) + + +def test_verify_canonical_checkout_rejects_dirty_tree(monkeypatch) -> None: + values = { + ("rev-parse", "HEAD"): "canonical-sha", + ("rev-parse", "--abbrev-ref", "HEAD"): "feature/bundle-0-bootstrap", + ("status", "--porcelain"): " M services/example.py", + } + monkeypatch.setattr(cli, "_git", lambda *args: values[args]) + with pytest.raises(FreshPersonalHoldoutRunnerError, match="clean working tree"): + cli.verify_canonical_checkout( + canonical_sha="canonical-sha", + canonical_branch="feature/bundle-0-bootstrap", + ) + + +def test_verify_r1_fixed_effective_cohort_accepts_no_replacements(tmp_path) -> None: + private = tmp_path / "private.json" + private.write_text( + json.dumps({"effective_cohort": {"replacement_events": []}}), + encoding="utf-8", + ) + cli.verify_r1_fixed_effective_cohort({"private_manifest": str(private)}) + + +def test_verify_r1_fixed_effective_cohort_rejects_replacement_event(tmp_path) -> None: + private = tmp_path / "private.json" + private.write_text( + json.dumps( + { + "effective_cohort": { + "replacement_events": [ + { + "invalid_case_id": "case-a", + "replacement_case_id": "case-b", + } + ] + } + } + ), + encoding="utf-8", + ) + with pytest.raises(FreshPersonalHoldoutRunnerError, match="zero replacement events"): + cli.verify_r1_fixed_effective_cohort({"private_manifest": str(private)}) diff --git a/tests/test_fresh_personal_holdout_runner_r1.py b/tests/test_fresh_personal_holdout_runner_r1.py new file mode 100644 index 00000000..59aaa121 --- /dev/null +++ b/tests/test_fresh_personal_holdout_runner_r1.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import hashlib + +import pytest + +from core.intelligence.curated_real_library_review_contract import CuratedSetRole +from services.intelligence import fresh_personal_holdout_runner as runner +from services.intelligence.fresh_personal_holdout_runner import ( + FreshPersonalHoldoutRunnerError, + _case_spec_pool, + _write_review_csv, +) + + +def _snapshot(track_count: int = 40) -> dict: + tracks = [] + for index in range(track_count): + signature = hashlib.sha256(f"track-{index}".encode()).hexdigest() + tracks.append( + { + "track_id": f"trk_{signature[:24]}", + "absolute_path": f"/private/library/{index}.mp3", + "file_signature": signature, + "display_name": f"Artist {index} - Track {index}", + "artist": f"Artist {index}", + "genre": "Techno", + "energy": 7.0, + } + ) + return { + "schema": "applaylist-local-library-snapshot-r1", + "snapshot_id": "snapshot-fresh", + "snapshot_version": "local-library-subset-r1", + "library_fingerprint": "snapshot:fresh:fingerprint", + "created_date": "2026-08-23", + "scope": {"kind": "REAL_INVENTORY_BACKED_SUBSET"}, + "privacy": {"publishable_to_public_repo": False}, + "tracks": tracks, + } + + +def test_case_pool_is_deterministic_balanced_and_has_no_model_inputs() -> None: + first = _case_spec_pool(_snapshot(), sampling_seed="seed-1", cases_per_role=8, candidate_scope_size=16) + replay = _case_spec_pool(_snapshot(), sampling_seed="seed-1", cases_per_role=8, candidate_scope_size=16) + assert first == replay + assert len(first["case_specs"]) == 8 * len(tuple(CuratedSetRole)) + for role in CuratedSetRole: + assert sum(row["set_role"] == role.value for row in first["case_specs"]) == 8 + text = str(first).lower() + assert "challenger" not in text + assert "preference" not in text + assert "rating" not in text + + +def test_case_pool_changes_with_seed() -> None: + left = _case_spec_pool(_snapshot(), sampling_seed="seed-a") + right = _case_spec_pool(_snapshot(), sampling_seed="seed-b") + assert left != right + + +def test_case_pool_requires_bounded_minimum_library() -> None: + with pytest.raises(FreshPersonalHoldoutRunnerError, match="at least 17 tracks"): + _case_spec_pool(_snapshot(16), sampling_seed="seed") + + +def test_candidate_analysis_isolates_one_case_mir_failure(monkeypatch) -> None: + selection = { + "schema": "applaylist-curated-case-selection-r1", + "snapshot_ref": ["snapshot-fresh", "local-library-subset-r1"], + "generator_version": "fresh-personal-holdout-runner-r1", + "sampling_seed": "seed", + "case_specs": [ + { + "case_spec_id": "case-bad", + "set_role": "opening", + "seed_track_id": "trk-bad", + "candidate_scope_track_ids": ["trk-x"], + }, + { + "case_spec_id": "case-good", + "set_role": "opening", + "seed_track_id": "trk-good", + "candidate_scope_track_ids": ["trk-y"], + }, + ], + } + + def fake_analyze_real_tracks(*, snapshot_raw, selection_raw): + del snapshot_raw + case_id = selection_raw["case_specs"][0]["case_spec_id"] + if case_id == "case-bad": + raise runner.RealLibraryPilotError("unreadable candidate audio") + return {"trk-good": object()} + + monkeypatch.setattr(runner, "analyze_real_tracks", fake_analyze_real_tracks) + evidence_by_case, merged, failures = runner._analyze_candidate_pool( + snapshot_raw={}, + selection_raw=selection, + ) + + assert "case-bad" not in evidence_by_case + assert evidence_by_case["case-good"] == {"trk-good": merged["trk-good"]} + assert len(failures) == 1 + assert failures[0]["case_id"] == "case-bad" + assert failures[0]["technical_invalidity_reason"] == "analysis_failed" + + +def test_review_csv_contains_only_curation_fields_and_empty_human_labels(tmp_path) -> None: + path = tmp_path / "review.csv" + rows = [ + { + "case_id": "case-1", + "set_role": "opening", + "assignment_id": "assignment-1", + } + ] + _write_review_csv(path, rows) + text = path.read_text(encoding="utf-8") + assert "transition_smoothness" not in text + assert "phrase_alignment" not in text + assert "energy_flow_plan_a" in text + assert "dramaturgical_fit_plan_b" in text + assert "set_coherence_plan_a" in text + assert "alternative_usefulness_plan_b" in text + assert "plan_a,plan_b" not in text + assert "case-1,opening,assignment-1" in text + + +def test_snapshot_case_pool_never_exposes_absolute_paths() -> None: + pool = _case_spec_pool(_snapshot(), sampling_seed="seed") + assert "/private/library/" not in str(pool) diff --git a/tests/test_fresh_personal_holdout_runner_security_r1.py b/tests/test_fresh_personal_holdout_runner_security_r1.py new file mode 100644 index 00000000..36503439 --- /dev/null +++ b/tests/test_fresh_personal_holdout_runner_security_r1.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import inspect + +from services.intelligence import fresh_personal_holdout_runner as runner + + +def test_holdout_selection_call_has_no_human_or_challenger_inputs() -> None: + source = inspect.getsource(runner.materialize_fresh_personal_holdout_r1) + selection_call = source.split("select_holdout_cases_r2(", 1)[1].split(")", 1)[0].lower() + for forbidden in ("preference", "rating", "review", "challenger", "shadow"): + assert forbidden not in selection_call + + +def test_reviewer_packet_builder_has_no_strategy_or_challenger_parameters() -> None: + parameters = set(inspect.signature(runner._reviewer_case).parameters) + assert parameters == {"item", "names"} + + +def test_private_manifest_freezes_source_review_cases_for_unblinding() -> None: + source = inspect.getsource(runner.materialize_fresh_personal_holdout_r1) + assert '"source_review_cases"' in source + assert "asdict(by_case[case_id].case)" in source + assert '"source_review_cases_frozen_before_reviewer_publication": True' in source + + +def test_runner_has_no_activation_path() -> None: + source = inspect.getsource(runner).lower() + assert "activation_authorized\": true" not in source + assert "personal_dj_model_training_authorized\": true" not in source