diff --git a/.github/workflows/music-plan-contract.yml b/.github/workflows/music-plan-contract.yml new file mode 100644 index 00000000..cf3ecd98 --- /dev/null +++ b/.github/workflows/music-plan-contract.yml @@ -0,0 +1,72 @@ +name: MusicPlan Contract + +on: + pull_request: + branches: [main] + paths: + - "schemas/music-plan-v1.schema.json" + - "src/music_plan.py" + - "src/music_provider.py" + - "scripts/music_plan_mutation_probe.py" + - "scripts/music_plan_holdout_evaluator.py" + - "tests/test_music_plan.py" + - "tests/fixtures/music_plan/**" + - "docs/music-planning-layer-v1.md" + - ".github/workflows/music-plan-contract.yml" + +permissions: + contents: read + +jobs: + validate: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install focused validation dependencies + run: python -m pip install 'pytest>=8,<10' 'jsonschema>=4.23,<5' + - name: Compile model-free contracts + run: python -m py_compile src/music_plan.py src/music_provider.py scripts/music_plan_mutation_probe.py scripts/music_plan_holdout_evaluator.py + - name: Run MusicPlan contract tests + run: python -m pytest -q tests/test_music_plan.py + - name: Run controlled negative mutation probe + run: python -m scripts.music_plan_mutation_probe + - name: Verify private holdout evaluator protocol + run: | + python - <<'PY' + import json + import subprocess + import sys + from pathlib import Path + + plan = json.loads(Path('tests/fixtures/music_plan/full.json').read_text(encoding='utf-8')) + request = {"id": "public-protocol-smoke", "input": {"plan": plan, "expected_valid": True}} + r = subprocess.run( + [sys.executable, '-m', 'scripts.music_plan_holdout_evaluator'], + input=json.dumps(request), text=True, capture_output=True, check=True, + ) + response = json.loads(r.stdout) + assert response == {"passed": True}, response + print('holdout evaluator protocol OK') + PY + - name: Verify Draft 2020-12 schema and reference fixtures + run: | + python - <<'PY' + import json + from pathlib import Path + from jsonschema import Draft202012Validator + + schema_path = Path('schemas/music-plan-v1.schema.json') + schema = json.loads(schema_path.read_text(encoding='utf-8')) + Draft202012Validator.check_schema(schema) + validator = Draft202012Validator(schema) + for path in sorted(Path('tests/fixtures/music_plan').glob('*.json')): + data = json.loads(path.read_text(encoding='utf-8')) + validator.validate(data) + print(f'OK {path}') + PY diff --git a/docs/music-planning-layer-v1.md b/docs/music-planning-layer-v1.md new file mode 100644 index 00000000..ba26a143 --- /dev/null +++ b/docs/music-planning-layer-v1.md @@ -0,0 +1,150 @@ +# StoryCore Music Planning Layer v1 + +Status: **experimental contract; provider-neutral; no provider activated**. + +## Goal + +Insert an editable symbolic planning layer between narrative intent and any music +generator. StoryCore should be able to inspect, compare, revise and version a +score plan without coupling the project to one model or one licence regime. + +```text +story / scene / shot intent + | + v +MusicPlan v1 + | + v +Draft 2020-12 schema + deterministic invariants + | + v +provider capability / licence check + | + v +requested vs executed delta + | + v +candidate artifact + portable handoff + | + v +independent validation later + | + v +last-known-good promotion outside provider adapter +``` + +The design is inspired by open-source workflows that expose an editable +intermediate representation, but this contract is StoryCore-owned and does not +require or redistribute third-party model weights. + +## Three execution modes + +- `full`: strongest symbolic plan; use where continuity, motifs, timing or + reproducibility matter. +- `guided`: StoryCore fixes high-level structure and synchronization while the + provider may elaborate details. +- `free`: narrative/time brief remains binding, but detailed symbolic planning + is optional. It must not be represented as equivalent to a `full` render. + +## Validation before inference + +The model-free gate composes the repository's existing `jsonschema` dependency +with StoryCore-specific checks. It: + +- validates the Draft 2020-12 schema itself; +- validates required fields and bounded values in a requested plan; +- rejects invalid section timing and overlaps; +- checks unique IDs and motif references; +- checks cue bounds; +- rejects declared non-commercial model weights on commercial targets; +- treats known unknown/unqualified model-weight licence markers as ineligible + for commercial use. + +The reference fixtures cover `full`, `guided`, and `free`. No LLM, music model, +network request or weight download is needed for this gate. + +## Explicit degradation + +`execution_delta(requested, executed)` is recursive. It records mode changes, +removed or added fields, changed scalar values, list-length changes and nested +changes. Examples include: + +```text +mode:full->guided +dropped:motifs +changed:duration_seconds +changed:sync_cues.length +changed:sync_cues[0].time_seconds +``` + +A provider must never silently change requested state. + +`MockMusicProvider` is deterministic and performs no inference. A fully capable +mock keeps the requested state unchanged. A limited mock may use an explicit +fallback and list unsupported fields, but any material delta requires a +non-empty degradation reason or the preparation fails closed. + +## Portable provider handoff + +`build_provider_handoff()` creates a data-only interchange envelope containing: + +- plan ID; +- provider/version/model; +- adapter/harness and hardware identity; +- requested/executed modes; +- unsupported fields, deltas, reason and acknowledgement; +- candidate artifact identity and optional digests; +- verification state and evidence references. + +It does not import Botte Secrète, execute a provider or write memory. Provider +results remain candidates. The envelope always carries +`activation_allowed=false`, `promoted=false` and +`memory_write_performed=false`. A `verified` handoff requires evidence. + +## Commercial boundary + +Code, model weights, datasets and assets retain separate provenance/licences. +Permissive adapter code cannot make restricted model weights commercially +usable. Non-commercial or unknown/unqualified weights remain outside the +StoryCore/Obolune commercial path until independently qualified. + +## Benchmark identity + +Future measurements must retain the complete path: + +`story fixture x model/provider x adapter/harness x hardware x parameters` + +A model-only score is insufficient because the harness can materially change +plan preservation, failures, quality, latency and resource use. + +## Current proof boundary + +The isolated `MusicPlan Contract` workflow runs on Python 3.10 and 3.12. It +installs only focused test/schema dependencies, compiles the MusicPlan/provider +contracts, executes the focused regression suite, checks the Draft 2020-12 +schema, and validates all three reference fixtures against it. + +Exact-head CI remains the authority for PR claims. This contract does **not** +prove real audio quality, a production provider, output rights beyond the +explicit licence policy, hardware performance, end-to-end StoryCore integration, +or last-known-good audio recovery. + +## Relationship to Botte Secrète + +StoryCore remains standalone. A future Botte integration should consume the +portable handoff as data rather than importing Botte as a hard runtime +dependency. Conceptually: + +- MusicPlan + narrative refs -> context snapshot; +- provider capabilities/licence/hardware -> constraints; +- provider handoff -> execution delta + candidate artifact; +- independent validators -> evidence; +- accepted soundtrack -> recovery point; +- measured provider runs -> Capability Atlas observations. + +## Next bounded slice + +Before any real provider is connected, add deterministic artifact-digest and +independent-verification fixtures around the portable handoff. Real local or +external music inference waits for explicit licence/hardware qualification and +must remain non-promoting by default. diff --git a/schemas/music-plan-v1.schema.json b/schemas/music-plan-v1.schema.json new file mode 100644 index 00000000..e73d000d --- /dev/null +++ b/schemas/music-plan-v1.schema.json @@ -0,0 +1,121 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://storycore.local/schemas/music-plan-v1.schema.json", + "title": "StoryCore MusicPlan v1", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "plan_id", + "mode", + "duration_seconds", + "sections", + "sync_cues", + "provenance" + ], + "properties": { + "schema_version": {"const": 1}, + "plan_id": {"type": "string", "minLength": 1, "maxLength": 128}, + "mode": {"enum": ["full", "guided", "free"]}, + "title": {"type": "string", "maxLength": 256}, + "narrative_intent": {"type": "string", "maxLength": 4000}, + "duration_seconds": {"type": "number", "exclusiveMinimum": 0, "maximum": 7200}, + "tempo_bpm": {"type": ["number", "null"], "minimum": 20, "maximum": 300}, + "meter": {"type": ["string", "null"], "maxLength": 32}, + "tonal_center": {"type": ["string", "null"], "maxLength": 64}, + "sections": { + "type": "array", + "minItems": 1, + "maxItems": 128, + "items": { + "type": "object", + "additionalProperties": false, + "required": ["id", "start_seconds", "end_seconds", "purpose"], + "properties": { + "id": {"type": "string", "minLength": 1, "maxLength": 128}, + "start_seconds": {"type": "number", "minimum": 0}, + "end_seconds": {"type": "number", "exclusiveMinimum": 0}, + "purpose": {"type": "string", "minLength": 1, "maxLength": 1000}, + "energy": {"type": ["number", "null"], "minimum": 0, "maximum": 1}, + "tension": {"type": ["number", "null"], "minimum": 0, "maximum": 1}, + "dialogue_safe": {"type": "boolean", "default": false}, + "motif_refs": { + "type": "array", + "items": {"type": "string", "minLength": 1, "maxLength": 128}, + "uniqueItems": true, + "default": [] + }, + "instrument_roles": { + "type": "array", + "items": {"type": "string", "minLength": 1, "maxLength": 128}, + "uniqueItems": true, + "default": [] + } + } + } + }, + "motifs": { + "type": "array", + "maxItems": 64, + "items": { + "type": "object", + "additionalProperties": false, + "required": ["id", "intent"], + "properties": { + "id": {"type": "string", "minLength": 1, "maxLength": 128}, + "intent": {"type": "string", "minLength": 1, "maxLength": 1000}, + "symbolic_hint": {"type": ["string", "null"], "maxLength": 2000} + } + }, + "default": [] + }, + "sync_cues": { + "type": "array", + "maxItems": 256, + "items": { + "type": "object", + "additionalProperties": false, + "required": ["id", "time_seconds", "intent"], + "properties": { + "id": {"type": "string", "minLength": 1, "maxLength": 128}, + "time_seconds": {"type": "number", "minimum": 0}, + "intent": {"type": "string", "minLength": 1, "maxLength": 1000}, + "story_ref": {"type": ["string", "null"], "maxLength": 512} + } + } + }, + "lyrics": { + "type": ["object", "null"], + "additionalProperties": false, + "required": ["language", "text"], + "properties": { + "language": {"type": "string", "minLength": 2, "maxLength": 35}, + "text": {"type": "string", "maxLength": 20000} + } + }, + "provenance": { + "type": "object", + "additionalProperties": false, + "required": ["commercial_target", "dependencies"], + "properties": { + "commercial_target": {"type": "boolean"}, + "dependencies": { + "type": "array", + "maxItems": 128, + "items": { + "type": "object", + "additionalProperties": false, + "required": ["name", "kind", "license"], + "properties": { + "name": {"type": "string", "minLength": 1, "maxLength": 256}, + "kind": {"enum": ["code", "model-weights", "dataset", "asset", "other"]}, + "license": {"type": "string", "minLength": 1, "maxLength": 256}, + "version": {"type": ["string", "null"], "maxLength": 128}, + "source": {"type": ["string", "null"], "maxLength": 1000} + } + } + } + } + } + } +} diff --git a/scripts/music_plan_holdout_evaluator.py b/scripts/music_plan_holdout_evaluator.py new file mode 100644 index 00000000..3526cc6b --- /dev/null +++ b/scripts/music_plan_holdout_evaluator.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""JSON-stdin evaluator for private MusicPlan holdouts. + +This file contains no holdout cases. It is designed to be invoked by Botte +Secrete's private holdout runner. The private payload supplies a MusicPlan and a +hidden expected-validity bit; stdout returns only {"passed": bool}. +""" + +from __future__ import annotations + +import json +import sys +from typing import Any + +from src.music_plan import validate_music_plan +from src.music_provider import MockMusicProvider, MusicProviderContractError + + +def evaluate(payload: dict[str, Any]) -> bool: + wrapper = payload.get("input") + if not isinstance(wrapper, dict): + return False + plan = wrapper.get("plan") + expected_valid = wrapper.get("expected_valid") + if not isinstance(plan, dict) or not isinstance(expected_valid, bool): + return False + + validation = validate_music_plan(plan) + validator_valid = validation.valid + + provider_valid = True + try: + MockMusicProvider().prepare(plan) + except MusicProviderContractError: + provider_valid = False + except Exception: + return False + + observed_valid = validator_valid and provider_valid + return observed_valid is expected_valid + + +def main() -> int: + try: + payload = json.loads(sys.stdin.read()) + except json.JSONDecodeError: + print(json.dumps({"passed": False})) + return 0 + if not isinstance(payload, dict): + print(json.dumps({"passed": False})) + return 0 + print(json.dumps({"passed": evaluate(payload)})) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/music_plan_mutation_probe.py b/scripts/music_plan_mutation_probe.py new file mode 100644 index 00000000..69dc253d --- /dev/null +++ b/scripts/music_plan_mutation_probe.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Negative-control probe for the MusicPlan contract. + +This is deliberately separate from the ordinary unit suite. It starts from the +known-good full fixture, injects bounded invalid mutations, and succeeds only if +the public validation/provider boundary rejects every mutant. +""" + +from __future__ import annotations + +import copy +import json +from pathlib import Path + +from src.music_plan import validate_music_plan +from src.music_provider import MockMusicProvider, MusicProviderContractError + +FIXTURE = Path(__file__).resolve().parents[1] / "tests" / "fixtures" / "music_plan" / "full.json" + + +def _base() -> dict: + return json.loads(FIXTURE.read_text(encoding="utf-8")) + + +def _mutants() -> list[tuple[str, dict]]: + cases: list[tuple[str, dict]] = [] + + missing_id = copy.deepcopy(_base()) + missing_id.pop("plan_id", None) + cases.append(("missing-plan-id", missing_id)) + + invalid_energy = copy.deepcopy(_base()) + invalid_energy["sections"][0]["energy"] = 2 + cases.append(("energy-out-of-range", invalid_energy)) + + unknown_license = copy.deepcopy(_base()) + unknown_license["provenance"]["dependencies"].append({ + "name": "mutation-unknown-weights", + "kind": "model-weights", + "license": "unknown", + }) + cases.append(("unknown-commercial-model-license", unknown_license)) + + unresolved_motif = copy.deepcopy(_base()) + unresolved_motif["sections"][0]["motif_refs"] = ["mutation-missing-motif"] + cases.append(("unresolved-motif", unresolved_motif)) + + overlapping = copy.deepcopy(_base()) + overlapping["sections"][1]["start_seconds"] = overlapping["sections"][0]["start_seconds"] + cases.append(("overlapping-sections", overlapping)) + + return cases + + +def main() -> int: + failures: list[str] = [] + provider = MockMusicProvider() + + base_validation = validate_music_plan(_base()) + if not base_validation.valid: + print("FAIL positive control: reference fixture no longer validates") + return 1 + print("PASS positive control: reference fixture validates") + + for name, mutant in _mutants(): + validation = validate_music_plan(mutant) + validator_rejected = not validation.valid + provider_rejected = False + try: + provider.prepare(mutant) + except MusicProviderContractError: + provider_rejected = True + except Exception as exc: # fail closed, but distinguish contract leakage + failures.append(f"{name}: leaked {type(exc).__name__} instead of MusicProviderContractError") + print(f"FAIL {name}: unexpected exception {type(exc).__name__}") + continue + + if validator_rejected and provider_rejected: + print(f"PASS mutation rejected: {name}") + else: + failures.append( + f"{name}: validator_rejected={validator_rejected}, provider_rejected={provider_rejected}" + ) + print(f"FAIL mutation survived: {name}") + + if failures: + print("\nNEGATIVE CONTROL FAILED") + for failure in failures: + print(f"- {failure}") + return 1 + + print("\nNEGATIVE CONTROL PASSED: every controlled mutant was rejected") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/music_plan.py b/src/music_plan.py new file mode 100644 index 00000000..33e2c6bf --- /dev/null +++ b/src/music_plan.py @@ -0,0 +1,216 @@ +"""Provider-neutral MusicPlan v1 validation helpers. + +Validation is model-free but uses the repository's existing ``jsonschema`` +dependency for Draft 2020-12 structural checks, then applies StoryCore-specific +cross-field and commercial-use invariants. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from jsonschema import Draft202012Validator + +NON_COMMERCIAL_MARKERS = ("CC BY-NC", "BY-NC", "NON-COMMERCIAL", "NONCOMMERCIAL") +UNKNOWN_LICENSE_MARKERS = frozenset({"unknown", "unspecified", "unqualified", "tbd", "n/a", "none"}) +DEFAULT_SCHEMA_PATH = Path(__file__).resolve().parents[1] / "schemas" / "music-plan-v1.schema.json" + + +@dataclass(frozen=True) +class MusicPlanValidation: + valid: bool + errors: tuple[str, ...] + warnings: tuple[str, ...] = () + + +def _is_non_commercial(license_name: str) -> bool: + upper = license_name.upper().replace("_", "-") + return any(marker in upper for marker in NON_COMMERCIAL_MARKERS) + + +def _is_unknown_license(license_name: str) -> bool: + return license_name.strip().lower() in UNKNOWN_LICENSE_MARKERS + + +def _schema_errors(plan: dict[str, Any], schema_path: Path = DEFAULT_SCHEMA_PATH) -> list[str]: + schema = json.loads(schema_path.read_text(encoding="utf-8")) + Draft202012Validator.check_schema(schema) + validator = Draft202012Validator(schema) + errors: list[str] = [] + for issue in sorted(validator.iter_errors(plan), key=lambda item: list(item.absolute_path)): + path = ".".join(str(part) for part in issue.absolute_path) or "$" + errors.append(f"schema:{path}: {issue.message}") + return errors + + +def validate_music_plan(plan: dict[str, Any]) -> MusicPlanValidation: + """Validate MusicPlan structure and deterministic invariants without a model.""" + errors: list[str] = _schema_errors(plan) + warnings: list[str] = [] + + if plan.get("schema_version") != 1: + errors.append("schema_version must be 1") + if plan.get("mode") not in {"full", "guided", "free"}: + errors.append("mode must be full, guided, or free") + + duration = plan.get("duration_seconds") + if not isinstance(duration, (int, float)) or isinstance(duration, bool) or duration <= 0: + errors.append("duration_seconds must be a positive number") + duration = None + + sections = plan.get("sections") + if not isinstance(sections, list) or not sections: + errors.append("sections must be a non-empty list") + sections = [] + + section_ids: set[str] = set() + motif_refs: set[str] = set() + previous_end = 0.0 + for index, section in enumerate(sections): + if not isinstance(section, dict): + errors.append(f"sections[{index}] must be an object") + continue + sid = section.get("id") + if not isinstance(sid, str) or not sid: + errors.append(f"sections[{index}].id must be non-empty") + elif sid in section_ids: + errors.append(f"duplicate section id: {sid}") + else: + section_ids.add(sid) + start = section.get("start_seconds") + end = section.get("end_seconds") + if not isinstance(start, (int, float)) or isinstance(start, bool): + errors.append(f"sections[{index}].start_seconds must be numeric") + continue + if not isinstance(end, (int, float)) or isinstance(end, bool): + errors.append(f"sections[{index}].end_seconds must be numeric") + continue + if start < 0 or end <= start: + errors.append(f"sections[{index}] has invalid time bounds") + if start < previous_end: + errors.append(f"sections[{index}] overlaps the preceding section") + previous_end = max(previous_end, float(end)) + if duration is not None and end > duration: + errors.append(f"sections[{index}] exceeds duration_seconds") + refs = section.get("motif_refs", []) + if isinstance(refs, list): + motif_refs.update(ref for ref in refs if isinstance(ref, str)) + + motifs = plan.get("motifs", []) + motif_ids: set[str] = set() + if isinstance(motifs, list): + for index, motif in enumerate(motifs): + if not isinstance(motif, dict): + errors.append(f"motifs[{index}] must be an object") + continue + mid = motif.get("id") + if not isinstance(mid, str) or not mid: + errors.append(f"motifs[{index}].id must be non-empty") + elif mid in motif_ids: + errors.append(f"duplicate motif id: {mid}") + else: + motif_ids.add(mid) + unresolved = sorted(motif_refs - motif_ids) + if unresolved: + errors.append("unresolved motif_refs: " + ", ".join(unresolved)) + + cues = plan.get("sync_cues") + if not isinstance(cues, list): + errors.append("sync_cues must be a list") + cues = [] + cue_ids: set[str] = set() + for index, cue in enumerate(cues): + if not isinstance(cue, dict): + errors.append(f"sync_cues[{index}] must be an object") + continue + cid = cue.get("id") + if not isinstance(cid, str) or not cid: + errors.append(f"sync_cues[{index}].id must be non-empty") + elif cid in cue_ids: + errors.append(f"duplicate sync cue id: {cid}") + else: + cue_ids.add(cid) + time_seconds = cue.get("time_seconds") + if not isinstance(time_seconds, (int, float)) or isinstance(time_seconds, bool) or time_seconds < 0: + errors.append(f"sync_cues[{index}].time_seconds must be non-negative") + elif duration is not None and time_seconds > duration: + errors.append(f"sync_cues[{index}] exceeds duration_seconds") + + provenance = plan.get("provenance") + if not isinstance(provenance, dict): + errors.append("provenance must be an object") + else: + commercial = provenance.get("commercial_target") + if not isinstance(commercial, bool): + errors.append("provenance.commercial_target must be boolean") + dependencies = provenance.get("dependencies") + if not isinstance(dependencies, list): + errors.append("provenance.dependencies must be a list") + else: + for index, dep in enumerate(dependencies): + if not isinstance(dep, dict): + errors.append(f"provenance.dependencies[{index}] must be an object") + continue + license_name = dep.get("license") + if not isinstance(license_name, str) or not license_name.strip(): + errors.append(f"provenance.dependencies[{index}].license is required") + continue + if commercial and dep.get("kind") == "model-weights": + name = dep.get("name", index) + if _is_unknown_license(license_name): + errors.append( + f"commercial target cannot use model weights with unknown licence: {name}" + ) + elif _is_non_commercial(license_name): + errors.append( + f"commercial target cannot use non-commercial model weights: {name}" + ) + + mode = plan.get("mode") + if mode == "full" and not motifs: + warnings.append("full mode has no symbolic motifs") + if mode == "free" and motifs: + warnings.append("free mode carries motifs that a provider may intentionally ignore") + + return MusicPlanValidation(not errors, tuple(errors), tuple(warnings)) + + +def _diff_values(requested: Any, executed: Any, path: str, deltas: list[str]) -> None: + if isinstance(requested, dict) and isinstance(executed, dict): + requested_keys = set(requested) + executed_keys = set(executed) + for key in sorted(requested_keys - executed_keys): + child = f"{path}.{key}" if path else key + if child not in {"schema_version", "plan_id", "provenance"}: + deltas.append(f"dropped:{child}") + for key in sorted(executed_keys - requested_keys): + child = f"{path}.{key}" if path else key + deltas.append(f"added:{child}") + for key in sorted(requested_keys & executed_keys): + child = f"{path}.{key}" if path else key + if child == "mode": + continue + _diff_values(requested[key], executed[key], child, deltas) + return + + if isinstance(requested, list) and isinstance(executed, list): + if len(requested) != len(executed): + deltas.append(f"changed:{path}.length") + for index, (left, right) in enumerate(zip(requested, executed)): + _diff_values(left, right, f"{path}[{index}]", deltas) + return + + if requested != executed: + deltas.append(f"changed:{path}") + + +def execution_delta(requested: dict[str, Any], executed: dict[str, Any]) -> tuple[str, ...]: + """Return material provider changes, including nested and value-level deltas.""" + deltas: list[str] = [] + if requested.get("mode") != executed.get("mode"): + deltas.append(f"mode:{requested.get('mode')}->{executed.get('mode')}") + _diff_values(requested, executed, "", deltas) + return tuple(dict.fromkeys(deltas)) diff --git a/src/music_provider.py b/src/music_provider.py new file mode 100644 index 00000000..cf623780 --- /dev/null +++ b/src/music_provider.py @@ -0,0 +1,183 @@ +"""Deterministic provider contract for MusicPlan v1. + +The provider adapter is deliberately model-free: it proves how a future music +backend must report requested versus executed state without silently degrading a +plan. It never promotes artifacts, downloads weights, or performs network I/O. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Iterable + +from src.music_plan import execution_delta, validate_music_plan + + +class MusicProviderContractError(ValueError): + """Raised when a provider response violates the explicit-degradation contract.""" + + +@dataclass(frozen=True) +class ProviderResult: + provider_id: str + requested_mode: str + executed_mode: str + requested_plan_id: str + unsupported_fields: tuple[str, ...] + deltas: tuple[str, ...] + reason: str + acknowledged: bool + artifact_id: str + artifact_status: str = "candidate" + activation_allowed: bool = False + promoted: bool = False + executed_external_model: bool = False + + def as_dict(self) -> dict[str, Any]: + return { + "provider_id": self.provider_id, + "requested_mode": self.requested_mode, + "executed_mode": self.executed_mode, + "requested_plan_id": self.requested_plan_id, + "unsupported_fields": list(self.unsupported_fields), + "deltas": list(self.deltas), + "reason": self.reason, + "acknowledged": self.acknowledged, + "artifact_id": self.artifact_id, + "artifact_status": self.artifact_status, + "activation_allowed": self.activation_allowed, + "promoted": self.promoted, + "executed_external_model": self.executed_external_model, + } + + +def build_provider_handoff( + result: ProviderResult, + *, + provider_version: str = "", + model: str = "", + harness: str = "music-plan-v1", + hardware: str = "", + input_digest: str = "", + candidate_digest: str = "", + evidence_refs: Iterable[str] = (), + verification_state: str = "unverified", +) -> dict[str, Any]: + """Build a data-only interchange envelope for a later validation harness. + + This function does not import Botte Secrète, execute a provider, promote an + artifact, or write memory. It merely carries the bounded provider outcome + and provenance dimensions that another harness may verify independently. + """ + allowed_states = {"unverified", "partially_verified", "verified", "failed"} + if verification_state not in allowed_states: + raise MusicProviderContractError("unsupported verification_state") + refs = tuple(str(ref) for ref in evidence_refs if str(ref).strip()) + if verification_state == "verified" and not refs: + raise MusicProviderContractError("verified handoff requires evidence references") + if result.artifact_status != "candidate": + raise MusicProviderContractError("provider handoff accepts candidate artifacts only") + if result.activation_allowed or result.promoted: + raise MusicProviderContractError("provider result cannot carry activation or promotion authority") + + return { + "schema_version": "storycore.music-provider-handoff/v1", + "plan_id": result.requested_plan_id, + "provider": { + "id": result.provider_id, + "version": provider_version, + "model": model, + "harness": harness, + "hardware": hardware, + }, + "execution": { + "requested_mode": result.requested_mode, + "executed_mode": result.executed_mode, + "unsupported_fields": list(result.unsupported_fields), + "deltas": list(result.deltas), + "reason": result.reason, + "acknowledged": result.acknowledged, + "executed_external_model": result.executed_external_model, + }, + "artifact": { + "id": result.artifact_id, + "status": result.artifact_status, + "input_digest": input_digest, + "candidate_digest": candidate_digest, + }, + "verification": { + "state": verification_state, + "evidence_refs": list(refs), + }, + "activation_allowed": False, + "promoted": False, + "memory_write_performed": False, + } + + +class MockMusicProvider: + """Small deterministic provider used to prove the adapter contract.""" + + def __init__( + self, + provider_id: str = "mock-music-provider", + *, + supported_modes: tuple[str, ...] = ("full", "guided", "free"), + unsupported_fields: tuple[str, ...] = (), + fallback_mode: str | None = None, + degradation_reason: str = "", + ) -> None: + self.provider_id = provider_id + self.supported_modes = supported_modes + self.unsupported_fields = unsupported_fields + self.fallback_mode = fallback_mode + self.degradation_reason = degradation_reason + + def prepare(self, plan: dict[str, Any]) -> ProviderResult: + validation = validate_music_plan(plan) + if not validation.valid: + raise MusicProviderContractError( + "invalid MusicPlan: " + "; ".join(validation.errors) + ) + + requested_mode = str(plan["mode"]) + executed = dict(plan) + if requested_mode not in self.supported_modes: + if not self.fallback_mode or self.fallback_mode not in self.supported_modes: + raise MusicProviderContractError( + f"provider does not support requested mode {requested_mode!r} and has no valid fallback" + ) + executed["mode"] = self.fallback_mode + + for field in self.unsupported_fields: + executed.pop(field, None) + + deltas = execution_delta(plan, executed) + acknowledged = bool(deltas) + reason = self.degradation_reason.strip() if deltas else "" + if deltas and not reason: + raise MusicProviderContractError( + "provider changed requested state without an explicit degradation reason" + ) + + return ProviderResult( + provider_id=self.provider_id, + requested_mode=requested_mode, + executed_mode=str(executed["mode"]), + requested_plan_id=str(plan["plan_id"]), + unsupported_fields=tuple( + field for field in self.unsupported_fields if field in plan + ), + deltas=deltas, + reason=reason, + acknowledged=acknowledged, + artifact_id=f"candidate:{self.provider_id}:{plan['plan_id']}", + ) + + +__all__ = [ + "MockMusicProvider", + "MusicProviderContractError", + "ProviderResult", + "build_provider_handoff", +] diff --git a/tests/fixtures/music_plan/free.json b/tests/fixtures/music_plan/free.json new file mode 100644 index 00000000..c299b0e1 --- /dev/null +++ b/tests/fixtures/music_plan/free.json @@ -0,0 +1,14 @@ +{ + "schema_version": 1, + "plan_id": "fixture-free-v1", + "mode": "free", + "title": "Ambient interlude", + "duration_seconds": 20, + "sections": [ + {"id": "interlude", "start_seconds": 0, "end_seconds": 20, "purpose": "open ambience", "energy": 0.25, "tension": 0.2, "dialogue_safe": true, "motif_refs": [], "instrument_roles": []} + ], + "motifs": [], + "sync_cues": [], + "lyrics": null, + "provenance": {"commercial_target": false, "dependencies": []} +} diff --git a/tests/fixtures/music_plan/full.json b/tests/fixtures/music_plan/full.json new file mode 100644 index 00000000..497e939d --- /dev/null +++ b/tests/fixtures/music_plan/full.json @@ -0,0 +1,23 @@ +{ + "schema_version": 1, + "plan_id": "fixture-full-v1", + "mode": "full", + "title": "Arrival and reveal", + "narrative_intent": "Build restrained anticipation, then reveal the destination without masking dialogue.", + "duration_seconds": 30, + "tempo_bpm": 92, + "meter": "4/4", + "tonal_center": "D minor", + "sections": [ + {"id": "arrival", "start_seconds": 0, "end_seconds": 18, "purpose": "quiet approach", "energy": 0.35, "tension": 0.55, "dialogue_safe": true, "motif_refs": ["arrival-motif"], "instrument_roles": ["soft strings", "pulse"]}, + {"id": "reveal", "start_seconds": 18, "end_seconds": 30, "purpose": "visual reveal", "energy": 0.72, "tension": 0.3, "dialogue_safe": false, "motif_refs": ["arrival-motif"], "instrument_roles": ["strings", "low brass"]} + ], + "motifs": [ + {"id": "arrival-motif", "intent": "recognizable four-note identity", "symbolic_hint": "short rising phrase, sparse rhythm"} + ], + "sync_cues": [ + {"id": "door-open", "time_seconds": 18, "intent": "accent the reveal", "story_ref": "scene-01/shot-04"} + ], + "lyrics": null, + "provenance": {"commercial_target": true, "dependencies": []} +} diff --git a/tests/fixtures/music_plan/guided.json b/tests/fixtures/music_plan/guided.json new file mode 100644 index 00000000..aa6f9ba3 --- /dev/null +++ b/tests/fixtures/music_plan/guided.json @@ -0,0 +1,17 @@ +{ + "schema_version": 1, + "plan_id": "fixture-guided-v1", + "mode": "guided", + "title": "Chase transition", + "duration_seconds": 24, + "sections": [ + {"id": "setup", "start_seconds": 0, "end_seconds": 8, "purpose": "establish momentum", "energy": 0.45, "tension": 0.5, "dialogue_safe": true, "motif_refs": [], "instrument_roles": ["percussion"]}, + {"id": "chase", "start_seconds": 8, "end_seconds": 24, "purpose": "accelerate chase", "energy": 0.9, "tension": 0.85, "dialogue_safe": false, "motif_refs": [], "instrument_roles": ["percussion", "bass"]} + ], + "motifs": [], + "sync_cues": [ + {"id": "cut-chase", "time_seconds": 8, "intent": "mark transition to chase", "story_ref": "scene-02/shot-01"} + ], + "lyrics": null, + "provenance": {"commercial_target": true, "dependencies": []} +} diff --git a/tests/test_music_plan.py b/tests/test_music_plan.py new file mode 100644 index 00000000..7c6ffe03 --- /dev/null +++ b/tests/test_music_plan.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +import copy +import json +from pathlib import Path + +import pytest + +from src.music_plan import execution_delta, validate_music_plan +from src.music_provider import ( + MockMusicProvider, + MusicProviderContractError, + build_provider_handoff, +) + +FIXTURES = Path(__file__).parent / "fixtures" / "music_plan" + + +def _load(name: str) -> dict: + return json.loads((FIXTURES / name).read_text(encoding="utf-8")) + + +def test_three_reference_modes_validate_without_model() -> None: + for name in ("full.json", "guided.json", "free.json"): + result = validate_music_plan(_load(name)) + assert result.valid, (name, result.errors) + + +def test_required_plan_id_is_enforced_by_schema() -> None: + plan = _load("full.json") + plan.pop("plan_id") + result = validate_music_plan(plan) + assert not result.valid + assert any("schema:$" in error and "plan_id" in error for error in result.errors) + + +def test_energy_range_is_enforced_by_schema() -> None: + plan = _load("full.json") + plan["sections"][0]["energy"] = 2 + result = validate_music_plan(plan) + assert not result.valid + assert any("energy" in error and "greater than the maximum" in error for error in result.errors) + + +def test_overlapping_sections_fail_closed() -> None: + plan = _load("guided.json") + plan["sections"][1]["start_seconds"] = 7 + result = validate_music_plan(plan) + assert not result.valid + assert any("overlaps" in error for error in result.errors) + + +def test_unresolved_motif_reference_fails_closed() -> None: + plan = _load("full.json") + plan["sections"][0]["motif_refs"] = ["missing-motif"] + result = validate_music_plan(plan) + assert not result.valid + assert any("unresolved motif_refs" in error for error in result.errors) + + +def test_commercial_target_blocks_noncommercial_model_weights() -> None: + plan = _load("full.json") + plan["provenance"]["dependencies"].append({ + "name": "example-nc-weights", + "kind": "model-weights", + "license": "CC BY-NC 4.0", + }) + result = validate_music_plan(plan) + assert not result.valid + assert any("non-commercial model weights" in error for error in result.errors) + + +def test_commercial_target_blocks_unknown_model_weight_licence() -> None: + plan = _load("full.json") + plan["provenance"]["dependencies"].append({ + "name": "unqualified-weights", + "kind": "model-weights", + "license": "unknown", + }) + result = validate_music_plan(plan) + assert not result.valid + assert any("unknown licence" in error for error in result.errors) + + +def test_noncommercial_research_fixture_can_remain_noncommercial() -> None: + plan = _load("free.json") + plan["provenance"]["dependencies"].append({ + "name": "example-nc-weights", + "kind": "model-weights", + "license": "CC BY-NC 4.0", + }) + result = validate_music_plan(plan) + assert result.valid + + +def test_provider_degradation_is_explicit() -> None: + requested = _load("full.json") + executed = copy.deepcopy(requested) + executed["mode"] = "free" + executed.pop("motifs") + delta = execution_delta(requested, executed) + assert "mode:full->free" in delta + assert "dropped:motifs" in delta + + +def test_duration_shortening_is_reported() -> None: + requested = _load("full.json") + executed = copy.deepcopy(requested) + executed["duration_seconds"] = requested["duration_seconds"] / 2 + for section in executed["sections"]: + section["start_seconds"] /= 2 + section["end_seconds"] /= 2 + for cue in executed["sync_cues"]: + cue["time_seconds"] /= 2 + assert validate_music_plan(executed).valid + delta = execution_delta(requested, executed) + assert "changed:duration_seconds" in delta + assert "changed:sections[0].end_seconds" in delta + assert any(item.startswith("changed:sync_cues[0].time_seconds") for item in delta) + + +def test_emptied_sync_cues_are_reported() -> None: + requested = _load("full.json") + executed = copy.deepcopy(requested) + executed["sync_cues"] = [] + assert validate_music_plan(executed).valid + delta = execution_delta(requested, executed) + assert "changed:sync_cues.length" in delta + + +def test_nested_cue_change_is_reported() -> None: + requested = _load("full.json") + executed = copy.deepcopy(requested) + executed["sync_cues"][0]["time_seconds"] += 1 + assert validate_music_plan(executed).valid + delta = execution_delta(requested, executed) + assert "changed:sync_cues[0].time_seconds" in delta + + +def test_full_capability_mock_keeps_requested_state() -> None: + result = MockMusicProvider().prepare(_load("full.json")) + assert result.requested_mode == "full" + assert result.executed_mode == "full" + assert result.deltas == () + assert result.unsupported_fields == () + assert result.acknowledged is False + assert result.reason == "" + assert result.artifact_status == "candidate" + assert result.activation_allowed is False + assert result.promoted is False + assert result.executed_external_model is False + + +def test_limited_mock_reports_mode_and_field_degradation() -> None: + provider = MockMusicProvider( + provider_id="mock-limited", + supported_modes=("guided", "free"), + fallback_mode="guided", + unsupported_fields=("motifs",), + degradation_reason="mock provider lacks full symbolic motif control", + ) + result = provider.prepare(_load("full.json")) + assert result.executed_mode == "guided" + assert "mode:full->guided" in result.deltas + assert "dropped:motifs" in result.deltas + assert result.unsupported_fields == ("motifs",) + assert result.acknowledged is True + assert result.reason + assert result.artifact_status == "candidate" + assert result.activation_allowed is False + assert result.promoted is False + assert result.executed_external_model is False + + +def test_provider_handoff_is_data_only_and_non_activating() -> None: + provider = MockMusicProvider( + provider_id="mock-limited", + supported_modes=("guided",), + fallback_mode="guided", + unsupported_fields=("motifs",), + degradation_reason="fixture capability boundary", + ) + result = provider.prepare(_load("full.json")) + handoff = build_provider_handoff( + result, + provider_version="test-v1", + model="none", + harness="music-plan-contract-tests", + hardware="github-hosted-cpu", + input_digest="sha256:input", + candidate_digest="sha256:candidate", + evidence_refs=("ci:fixture",), + verification_state="partially_verified", + ) + assert handoff["provider"]["id"] == "mock-limited" + assert handoff["provider"]["harness"] == "music-plan-contract-tests" + assert handoff["execution"]["deltas"] == list(result.deltas) + assert handoff["artifact"]["status"] == "candidate" + assert handoff["verification"]["evidence_refs"] == ["ci:fixture"] + assert handoff["activation_allowed"] is False + assert handoff["promoted"] is False + assert handoff["memory_write_performed"] is False + + +def test_verified_handoff_requires_evidence() -> None: + result = MockMusicProvider().prepare(_load("full.json")) + with pytest.raises(MusicProviderContractError, match="requires evidence"): + build_provider_handoff(result, verification_state="verified") + + +def test_silent_provider_degradation_is_rejected() -> None: + provider = MockMusicProvider( + provider_id="mock-bad", + supported_modes=("guided",), + fallback_mode="guided", + ) + with pytest.raises(MusicProviderContractError, match="explicit degradation reason"): + provider.prepare(_load("full.json")) + + +def test_provider_without_valid_fallback_fails_closed() -> None: + provider = MockMusicProvider( + provider_id="mock-no-fallback", + supported_modes=("guided",), + ) + with pytest.raises(MusicProviderContractError, match="no valid fallback"): + provider.prepare(_load("full.json"))