taste-application — Skillopedia

Taste Application The second half of the pipeline. taste-distillation measures references into a style pack; this generates against that pack and cuts the result. Execution and Delivery Contract The original implementation ships here in ; no separate checkout is required. Install for local processing. Install only for provider execution. Live uploads, submissions and downloads require explicit in addition to credentials; set it only for the user's authorized run. remains credential-free and produces labelled placeholders. An ambiguous provider timeout is not retried as a new paid job. Inspect…

):\n yield value[1:].split('.')\n elif isinstance(value, dict):\n for child in value.values():\n yield from references_in(child)\n elif isinstance(value, list):\n for child in value:\n yield from references_in(child)\n\n\ndef validate_graph(graph):\n \"\"\"Verify dependencies and every declared input; endpoint schemas need live QA.\"\"\"\n contents = graph['contents']\n nodes, inputs = contents['nodes'], contents['schema']['input']\n used = set()\n for name, node in nodes.items():\n if node['id'] != name:\n raise ValueError(f'Node id mismatch: {name}')\n for dependency in node.get('depends', []):\n if dependency != 'input' and dependency not in nodes:\n raise ValueError(f'Unknown dependency: {dependency}')\n ancestors = {}\n\n def visit(name, stack):\n if name == 'input':\n return set()\n if name in stack:\n raise ValueError('Dependency cycle')\n if name not in ancestors:\n deps = nodes[name].get('depends', [])\n ancestors[name] = set(deps).union(*(visit(dep, stack | {name}) for dep in deps))\n return ancestors[name]\n\n for name, node in nodes.items():\n reachable = visit(name, set())\n for ref in references_in({'input': node.get('input'), 'fields': node.get('fields')}):\n if ref[0] not in reachable:\n raise ValueError(f'{name} references undeclared dependency: {ref[0]}')\n if ref[0] == 'input':\n if len(ref) \u003c 2 or ref[1] not in inputs:\n raise ValueError('Unknown workflow input')\n used.add(ref[1])\n for ref in references_in(contents.get('output', {})):\n if ref[0] not in nodes:\n raise ValueError('Unknown output node')\n if used != set(inputs):\n raise ValueError(f'Unwired workflow inputs: {sorted(set(inputs) - used)}')\n\n\ndef load_graph(kind):\n if kind not in ('apply', 'apply-motion', 'distill', 'prop3d'):\n raise ValueError('Unknown workflow kind')\n graph = json.loads((WORKFLOWS / f'taste-{kind}.json').read_text())\n validate_graph(graph)\n return graph\n\n\ndef main():\n parser = argparse.ArgumentParser(description=__doc__)\n parser.add_argument('--kind', choices=('apply', 'apply-bundle', 'distill'), required=True)\n parser.add_argument('--config', type=Path, required=True)\n parser.add_argument('--out', type=Path, required=True)\n args = parser.parse_args()\n try:\n if args.kind == 'apply-bundle':\n from tasteforge.integration import build_application_bundle, load_application_request\n config = load_application_request(args.config)\n else:\n config = json.loads(args.config.read_text())\n if not isinstance(config, dict):\n raise ValueError('config must be a JSON object')\n local_only = False\n if args.kind == 'apply-bundle':\n local_only = config.get('local_only', False)\n if type(local_only) is not bool:\n raise ValueError('local_only must be an exact boolean')\n if local_only:\n if set(config) != {'local_only', 'integration'}:\n raise ValueError('local-only request permits only local_only and integration fields')\n payload = build_application_bundle(config['integration'], None, local_only=True)\n else:\n load_graph('apply' if args.kind == 'apply-bundle' else args.kind)\n compile_input = compile_application_input if args.kind != 'distill' else prepare_distillation_input\n payload = compile_input(config)\n if args.kind == 'apply-bundle':\n payload = build_application_bundle(config.get('integration'), payload)\n with args.out.open('x') as output:\n output.write(json.dumps(payload, indent=2) + '\\n')\n except (OSError, ValueError, KeyError) as error:\n parser.exit(2, f'Offline compilation failed: {error}\\n')\n\n\nif __name__ == '__main__':\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":6735,"content_sha256":"dbf2c85e7360388a8bd038f43f47162879c8a5cf259751fcfc8af8affecbe563"},{"filename":"SOURCE.md","content":"# Source and verification\n\nRecovered from the user's latest `ecc-taste-skills_1.zip` attachment to the\nClaude conversation **Video workflow architecture**. Archive SHA-256:\n`5e0dc440df4dcf6b2082a7dd59e1d6e9cc11d10166d4e1a19dc6c96478f4d2c8`.\n\nThe archive's `README-MERGE.md` identifies the two standalone skill script\ndirectories as the implementation. This import preserves the measured grade,\nreference cadence, median-edge UI crop, scale-to-cover normalization,\nalpha-bounded overlay plates and seeded placement logic from that source.\nRaw media, signed provider responses and project files are not bundled.\n\nFocused continuation fixes address observed execution failures: script paths\noutside the source directory, retained editable shot media, explicit output\nFPS, provider tier forwarding, existing-take passthrough, measured zero\nbackground targets, packed PBR textures, Blender slotted actions and exact\nResolve overlay readback. Original and retopologized meshes are retained as\nseparate assets. Provider calls require explicit live opt-in and ambiguous\nsubmissions are not automatically repeated.\n\n`taste-distillation` retains its own `taste/` helpers so that skill can be\ninstalled independently, as the original bundle intended. The transport copies\nare checked for equality by regression tests. ECC owns the reusable `tasteforge` engine, including interview/schema\ncontracts, workflow planning, asset receipts and the Resolve adapter. The\nlegacy `taste.resolve` import delegates to that same adapter. `ito-video`\nconsumes the packaged ECC engine as an example project.\n\nVerification uses `tests/test_taste_*.py` and the dedicated taste workflow CI.\nActual application checks additionally exercised a full textured GLB in\nBlender 5.1 and overlay placement in Resolve Studio 21. These are distinct\nfrom the offline test suite and from artistic approval of a finished video.\n\nThe metadata-only `tasteforge/fixtures/flashethereal` fixture comes from the earlier `tasteforge (4).zip` archive, SHA-256 `ef06a606d3b528fbd939b05fadc25bf6674073a1e05a01e3aa6b9c9416fd6284`. It includes no source media or `look.cube`; original source media stays outside the package.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":2169,"content_sha256":"9fff1b35e4e191973af3b606929e2e10cd2b21d88ffd433b88d6d52726217608"},{"filename":"tests/test_apply.py","content":"\"\"\"Failing-first tests for applying a pack to local media (deterministic only).\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nimport tempfile\nimport shutil\nimport unittest\nfrom pathlib import Path\nfrom unittest.mock import Mock\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nfrom tasteforge import apply as apply_mod # noqa: E402\nfrom tasteforge import pack as pack_mod # noqa: E402\nfrom tasteforge import schema, timeline # noqa: E402\n\nFIXTURE = Path(__import__(\"tasteforge\").__file__).resolve().parent / \"fixtures\" / \"flashethereal\"\n\nMEDIA = {\n \"clips\": [\n {\"path\": \"/tmp/media/shot_a.mov\", \"duration\": 6.2, \"name\": \"shot_a\"},\n {\"path\": \"/tmp/media/shot_b.mov\", \"duration\": 4.8, \"name\": \"shot_b\"},\n {\"path\": \"/tmp/media/shot_c.mov\", \"duration\": 8.1, \"name\": \"shot_c\"},\n ]\n}\n\n\nclass ApplyLocalTests(unittest.TestCase):\n def test_apply_local_report_is_schema_valid(self):\n sp = pack_mod.load(FIXTURE)\n report = apply_mod.apply_local(sp, MEDIA[\"clips\"])\n problems = schema.validate(report, schema.APPLICATION_REPORT_SCHEMA)\n self.assertEqual(problems, [])\n\n def test_report_claims_no_provider(self):\n report = apply_mod.apply_local(pack_mod.load(FIXTURE), MEDIA[\"clips\"])\n self.assertEqual(report[\"provider\"], \"none\")\n self.assertTrue(report[\"dry_run\"])\n self.assertEqual(report[\"mode\"], \"local-deterministic\")\n\n def test_planned_shots_follow_cadence_and_fill_duration(self):\n sp = pack_mod.load(FIXTURE)\n report = apply_mod.apply_local(sp, MEDIA[\"clips\"], duration=20.0)\n durations = [s[\"duration\"] for s in report[\"planned_shots\"]]\n self.assertGreater(len(durations), 3, \"77-cut cadence must not plan 3 shots\")\n self.assertLessEqual(sum(durations), 20.0 + max(durations))\n self.assertEqual(len(report[\"timeline_events\"]), len(durations))\n\n def test_apply_local_deterministic(self):\n sp = pack_mod.load(FIXTURE)\n a = apply_mod.apply_local(sp, MEDIA[\"clips\"], duration=12.0)\n b = apply_mod.apply_local(sp, MEDIA[\"clips\"], duration=12.0)\n a.pop(\"generated\"), b.pop(\"generated\")\n self.assertEqual(a, b)\n\n def test_events_reference_local_paths(self):\n report = apply_mod.apply_local(pack_mod.load(FIXTURE), MEDIA[\"clips\"], duration=8.0)\n for event in report[\"timeline_events\"]:\n self.assertTrue(event[\"path\"].startswith(\"/tmp/media/\"))\n self.assertGreater(event[\"frames\"], 0)\n\n def test_provider_generation_fails_closed(self):\n with self.assertRaises(apply_mod.ProviderDisabledError):\n apply_mod.apply_generate(pack_mod.load(FIXTURE), brief=\"x\")\n\n\nclass StrictApplyTests(unittest.TestCase):\n def setUp(self):\n self.pack = Mock(name=\"pack\")\n self.pack.name = \"test-pack\"\n self.pack.read_json.return_value = {\"fps\": 24, \"shots\": [{\"duration\": 1.0}]}\n\n def clips(self, *durations):\n return [{\"path\": f\"/tmp/strict-{i}.mov\", \"duration\": d}\n for i, d in enumerate(durations)]\n\n def test_default_still_repeats(self):\n report = apply_mod.apply_local(self.pack, self.clips(1), duration=3)\n self.assertEqual(len(report[\"timeline_events\"]), 3)\n\n def test_strict_rejects_insufficient_unique_clips(self):\n with self.assertRaisesRegex(ValueError, \"unique\"):\n apply_mod.apply_local(self.pack, self.clips(8), duration=3, no_repeat=True)\n\n def test_strict_normalizes_relative_absolute_and_symlink_aliases(self):\n with tempfile.TemporaryDirectory() as td:\n source = Path(td) / \"source.mov\"\n source.touch()\n alias = Path(td) / \"alias.mov\"\n alias.symlink_to(source)\n media = [{\"path\": str(p), \"duration\": 5}\n for p in (source, source.parent / \"..\" / source.parent.name / source.name, alias)]\n with self.assertRaisesRegex(ValueError, \"unique\"):\n apply_mod.apply_local(self.pack, media, duration=2, no_repeat=True)\n\n def test_strict_rejects_short_sources(self):\n with self.assertRaisesRegex(ValueError, \"source|short\"):\n apply_mod.apply_local(self.pack, self.clips(.5, .5), duration=2, no_repeat=True)\n\n def test_strict_quantization_never_rounds_source_capacity_up(self):\n self.pack.read_json.return_value = {\"fps\": 10, \"shots\": [{\"duration\": .16}]}\n with self.assertRaisesRegex(ValueError, \"source|short\"):\n apply_mod.apply_local(self.pack, self.clips(.16), duration=.16, no_repeat=True)\n\n def test_successful_strict_plan_fills_tail_and_uses_each_source_once(self):\n report = apply_mod.apply_local(self.pack, self.clips(2, 2, 2),\n duration=2.25, fps=20, no_repeat=True)\n events = report[\"timeline_events\"]\n self.assertEqual(sum(e[\"frames\"] for e in events), 45)\n self.assertEqual(len({e[\"path\"] for e in events}), len(events))\n self.assertEqual([e[\"offset_frames\"] for e in events], [0, 20, 40])\n self.assertTrue(all(e[\"fps\"] == 20 for e in events))\n self.assertEqual(report[\"planned_shots\"][-1][\"end\"], 2.25)\n self.assertEqual(schema.validate(report, schema.APPLICATION_REPORT_SCHEMA), [])\n\n def test_strict_stops_when_rounded_cadence_has_filled_target(self):\n self.pack.read_json.return_value = {\"fps\": 30, \"shots\": [{\"duration\": .1006}]}\n report = apply_mod.apply_local(self.pack, self.clips(*([1] * 1100)),\n duration=100, no_repeat=True)\n events = report[\"timeline_events\"]\n self.assertEqual(sum(e[\"frames\"] for e in events), 3000)\n self.assertTrue(all(e[\"frames\"] > 0 for e in events))\n self.assertEqual(report[\"planned_shots\"][-1][\"end\"], 100)\n\n def test_strict_accepts_exact_frame_source_duration_float_boundaries(self):\n for fps, frames in ((24, 4), (23.976, 4), (29.97, 5), (59.94, 10)):\n duration = float(frames / timeline.fps_fraction(fps))\n with self.subTest(fps=fps):\n self.pack.read_json.return_value = {\"fps\": fps, \"shots\": [{\"duration\": duration}]}\n report = apply_mod.apply_local(self.pack, self.clips(duration),\n duration=duration, no_repeat=True)\n event = report[\"timeline_events\"][0]\n self.assertEqual(event[\"frames\"], frames)\n self.assertLessEqual(event[\"duration\"], duration)\n\n def test_strict_frame_duration_metadata_does_not_round_past_source(self):\n report = apply_mod.apply_local(self.pack, self.clips(.016668),\n duration=1 / 60, fps=60, no_repeat=True)\n event = report[\"timeline_events\"][0]\n self.assertEqual(event[\"frames\"], 1)\n self.assertLessEqual(event[\"duration\"], .016668)\n self.assertEqual(event[\"duration\"], 1 / 60)\n\n def test_strict_preserves_manifest_order_instead_of_reassigning_short_source(self):\n self.pack.read_json.return_value = {\"fps\": 24, \"shots\": [{\"duration\": 2}]}\n with self.assertRaisesRegex(ValueError, \"source.*short\"):\n apply_mod.apply_local(self.pack, self.clips(1, 2), duration=3, no_repeat=True)\n report = apply_mod.apply_local(self.pack, self.clips(2, 1), duration=3, no_repeat=True)\n self.assertEqual([e[\"path\"] for e in report[\"timeline_events\"]],\n [str(Path(c[\"path\"]).resolve()) for c in self.clips(2, 1)])\n self.assertEqual(sum(e[\"frames\"] for e in report[\"timeline_events\"]), 72)\n\n def test_strict_is_deterministic_and_does_not_mutate_inputs(self):\n media = self.clips(2, 2, 2)\n before = [dict(clip) for clip in media]\n first = apply_mod.apply_local(self.pack, media, duration=2.25, no_repeat=True)\n second = apply_mod.apply_local(self.pack, media, duration=2.25, no_repeat=True)\n self.assertEqual(first[\"timeline_events\"], second[\"timeline_events\"])\n self.assertEqual(media, before)\n\n def test_plan_shots_rejects_invalid_target_and_cadence(self):\n for invalid in (True, 0, -1, float(\"nan\"), float(\"inf\"), None):\n with self.subTest(value=invalid):\n with self.assertRaises(ValueError):\n apply_mod.plan_shots({}, invalid)\n with self.assertRaises(ValueError):\n apply_mod.plan_shots({\"shots\": [{\"duration\": invalid}]}, 2)\n\n def test_missing_or_empty_cadence_is_not_defaulted(self):\n with tempfile.TemporaryDirectory() as td:\n pack = pack_mod.load(shutil.copytree(FIXTURE, Path(td) / \"pack\"))\n pack.cadence_path.unlink()\n with self.assertRaisesRegex(ValueError, \"cadence\"):\n apply_mod.apply_local(pack, self.clips(2), duration=2)\n pack.cadence_path.write_text(\"{}\")\n with self.assertRaisesRegex(ValueError, \"cadence\"):\n apply_mod.apply_local(pack, self.clips(2), duration=2)\n with self.assertRaisesRegex(ValueError, \"cadence\"):\n apply_mod.plan_shots({\"shots\": []}, 2)\n self.assertEqual(apply_mod.plan_shots({\"shots\": [], \"mean_shot\": 2.0}, 2), [2.0])\n\n def test_explicit_fps_overrides_pack_in_default_mode(self):\n report = apply_mod.apply_local(self.pack, self.clips(2), duration=1, fps=30)\n self.assertEqual(report[\"timeline_events\"][0][\"frames\"], 30)\n\n def test_invalid_target_fps_and_media_duration_are_rejected(self):\n for invalid in (0, -1, float(\"nan\"), float(\"inf\"), True, False, \"invalid\"):\n for field in (\"duration\", \"fps\", \"media\"):\n with self.subTest(field=field, value=invalid), self.assertRaises(ValueError):\n kwargs = {field: invalid} if field != \"media\" else {}\n media = self.clips(invalid if field == \"media\" else 2)\n apply_mod.apply_local(self.pack, media, **kwargs)\n\n def test_invalid_cadence_fps_is_not_masked_by_default(self):\n for invalid in (0, float(\"nan\"), float(\"inf\"), True):\n with self.subTest(value=invalid), self.assertRaises(ValueError):\n self.pack.read_json.return_value = {\"fps\": invalid}\n apply_mod.apply_local(self.pack, self.clips(2))\n\n def test_strict_rejects_subframe_target(self):\n with self.assertRaisesRegex(ValueError, \"frame\"):\n apply_mod.apply_local(self.pack, self.clips(2), duration=.001, no_repeat=True)\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":10584,"content_sha256":"7107f3f02f9e94e513d9bbfeca6b71b2e4110da7539401e1dec9f9cfcc1267dc"},{"filename":"tests/test_assets.py","content":"\"\"\"Local asset handoff rejects unsupported provenance and changed files.\"\"\"\nimport json\nimport struct\nimport tempfile\nimport unittest\nfrom pathlib import Path\n\nfrom tasteforge.assets import ingest_assets, validate_assets\n\n\nclass AssetTests(unittest.TestCase):\n def setUp(self):\n self.tmp = tempfile.TemporaryDirectory()\n self.addCleanup(self.tmp.cleanup)\n self.root = Path(self.tmp.name).resolve()\n (self.root / 'clip.mp4').write_bytes(b'existing media')\n self.config = self.root / 'input.json'\n self.receipt = self.root / 'receipt.json'\n self.asset = dict(id='clip', modality='video', path='clip.mp4',\n origin='local_passthrough')\n\n def ingest(self, assets=None, **extra):\n self.config.write_text(json.dumps(dict(assets=assets or [self.asset], **extra)))\n return ingest_assets(self.config, self.receipt)\n\n def test_roundtrip_and_lineage(self):\n (self.root / 'genre.json').write_text('{}')\n result = self.ingest(input_artifacts=['genre.json'], genre_spec='genre.json')\n self.assertEqual(validate_assets(self.receipt), result)\n self.assertEqual(result['provider_calls'], 0)\n self.assertIs(result['provider_execution'], False)\n self.assertEqual(result['assets'][0]['bytes'], 14)\n self.assertEqual(result['genre_spec']['sha256'], result['input_artifacts'][0]['sha256'])\n\n def test_external_result_requires_evidence_and_identifier(self):\n self.asset['origin'] = 'external_result'\n with self.assertRaises(ValueError):\n self.ingest()\n (self.root / 'provider.json').write_text('{\"status\":\"completed\"}')\n self.asset['provider_provenance'] = dict(provider='fal', request_id='abc',\n evidence_path='provider.json')\n result = self.ingest()\n self.assertEqual(result['assets'][0]['provider_provenance']['request_id'], 'abc')\n self.assertFalse(result['provider_execution'])\n (self.root / 'provider.json').write_text('{}')\n with self.assertRaises(ValueError):\n validate_assets(self.receipt)\n\n def test_recovered_does_not_infer_provider(self):\n self.asset['origin'] = 'recovered_unverified'\n result = self.ingest()\n self.assertNotIn('provider_provenance', result['assets'][0])\n\n def test_changed_media_and_lineage_rejected(self):\n self.ingest()\n (self.root / 'clip.mp4').write_bytes(b'changed')\n with self.assertRaises(ValueError):\n validate_assets(self.receipt)\n\n def test_glb_header(self):\n path = self.root / 'model.glb'\n path.write_bytes(struct.pack('\u003c4sII', b'glTF', 2, 12))\n self.asset.update(modality='3d_asset', path='model.glb')\n self.ingest()\n self.receipt.unlink()\n for header in [(b'xxxx', 2, 12), (b'glTF', 1, 12), (b'glTF', 2, 99)]:\n path.write_bytes(struct.pack('\u003c4sII', *header))\n with self.assertRaises(ValueError):\n self.ingest()\n\n def test_duplicate_invalid_remote_empty_and_special(self):\n with self.assertRaises(ValueError):\n self.ingest([self.asset, self.asset])\n for update in [dict(modality='audio'), dict(path='https://example.org/a.mp4'),\n dict(path='.'), dict(id=''), dict(origin='generated')]:\n with self.assertRaises(ValueError):\n self.ingest([{**self.asset, **update}])\n\n def test_symlink_and_symlink_parent(self):\n (self.root / 'link.mp4').symlink_to(self.root / 'clip.mp4')\n with self.assertRaises(ValueError):\n self.ingest([{**self.asset, 'path': 'link.mp4'}])\n (self.root / 'alias').symlink_to(self.root, target_is_directory=True)\n with self.assertRaises(ValueError):\n self.ingest([{**self.asset, 'path': 'alias/clip.mp4'}])\n\n def test_no_overwrites_or_collisions(self):\n self.ingest()\n original = self.receipt.read_bytes()\n with self.assertRaises(ValueError):\n self.ingest()\n self.assertEqual(original, self.receipt.read_bytes())\n with self.assertRaises(ValueError):\n ingest_assets(self.config, self.config)\n\n def test_tampered_receipt_operation_and_duplicates(self):\n result = self.ingest()\n for patch in [dict(provider_calls=1), dict(provider_calls=False),\n dict(provider_execution=True), dict(assets=result['assets'] * 2)]:\n self.receipt.write_text(json.dumps({**result, **patch}))\n with self.assertRaises(ValueError):\n validate_assets(self.receipt)\n\n def test_bundle_binds_exact_request_and_revalidates(self):\n from tasteforge.workflow import run_workflow\n workflow = self.root / 'workflow.json'\n workflow.write_text(json.dumps(dict(\n schema_version=1, run_id='assets-test', seed=42,\n genres=[dict(number=1, slug='flash', label='Flash', references=['clip.mp4'],\n signature=dict(materials=['chrome'], motion=['orbit'],\n composition=['center'], avoid=['mud']))])))\n bundle = self.root / 'bundle'\n run_workflow(workflow, bundle, probe=lambda path: dict(\n duration=6.0, width=1920, height=1080, fps=24.0, codec='fixture',\n sample_times=[0.75, 2.25, 3.75], scene_changes=[0.75, 2.25]))\n self.asset['request_id'] = 'assets-test-1-video'\n result = self.ingest(bundle_dir='bundle')\n self.assertEqual(result['assets'][0]['genre_slug'], 'flash')\n self.assertEqual(validate_assets(self.receipt), result)\n self.receipt.unlink()\n self.asset['request_id'] = 'assets-test-1-image'\n with self.assertRaises(ValueError):\n self.ingest(bundle_dir='bundle')\n self.asset['request_id'] = 'assets-test-1-video'\n self.asset['genre_slug'] = 'invented'\n with self.assertRaises(ValueError):\n self.ingest(bundle_dir='bundle')\n result['assets'][0]['style_fingerprint'] = 'fabricated'\n self.receipt.write_text(json.dumps(result))\n with self.assertRaises(ValueError):\n validate_assets(self.receipt)\n\n def test_invalid_provenance_and_lineage_shapes(self):\n for provenance in [None, {}, dict(provider='fal', evidence_path='clip.mp4'),\n dict(provider='fal', request_id='', evidence_path='clip.mp4'),\n dict(provider='fal', request_id='id', evidence_path='missing.json')]:\n with self.assertRaises(ValueError):\n self.ingest([{**self.asset, 'origin': 'external_result',\n 'provider_provenance': provenance}])\n with self.assertRaises(ValueError):\n self.ingest(input_artifacts='clip.mp4')\n with self.assertRaises(ValueError):\n self.ingest([{**self.asset, 'provider_provenance': {'provider': 'fal'}}])\n with self.assertRaises(ValueError):\n self.ingest([{**self.asset, 'genre_slug': 'invented'}])\n\n def test_special_file_and_missing_output_directory(self):\n import os\n os.mkfifo(self.root / 'pipe')\n with self.assertRaises(ValueError):\n self.ingest([{**self.asset, 'path': 'pipe'}])\n self.receipt = self.root / 'missing' / 'receipt.json'\n with self.assertRaises(ValueError):\n self.ingest()\n\n def test_mutated_lineage_and_invalid_receipt_bindings(self):\n (self.root / 'genre.json').write_text('{}')\n result = self.ingest(genre_spec='genre.json')\n (self.root / 'genre.json').write_text('{\"changed\":true}')\n with self.assertRaises(ValueError):\n validate_assets(self.receipt)\n for patch in [dict(schema='wrong'), dict(input_artifacts=None),\n dict(genre_spec=None)]:\n self.receipt.write_text(json.dumps({**result, **patch}))\n with self.assertRaises(ValueError):\n validate_assets(self.receipt)\n\n def test_invalid_json_shapes(self):\n for value in [[], {}, {'assets': []}, {'assets': [None]}]:\n self.config.write_text(json.dumps(value))\n with self.assertRaises(ValueError):\n ingest_assets(self.config, self.receipt)\n\n\nif __name__ == '__main__':\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":8375,"content_sha256":"41e58206c2197b10e33ddee8dd578f7a36f176322f79ac3edcfa9a27f230a3e3"},{"filename":"tests/test_cli.py","content":"\"\"\"Failing-first tests for the tasteforge CLI (python3 -m tasteforge).\"\"\"\n\nfrom __future__ import annotations\n\nimport io\nimport json\nimport shutil\nimport subprocess\nimport sys\nimport tempfile\nimport unittest\nfrom contextlib import redirect_stderr\nfrom pathlib import Path\nfrom unittest import mock\n\nfrom tasteforge import cli\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nFIXTURE = Path(__import__(\"tasteforge\").__file__).resolve().parent / \"fixtures\" / \"flashethereal\"\n\nANSWERS = {\n \"palette\": \"near-black void, bone white, violet bloom\",\n \"grain\": \"fine 35mm grain\",\n \"lighting\": \"single hard key\",\n \"focal_length\": \"35mm\",\n \"camera_motion\": \"locked off\",\n \"subject_framing\": \"centered, headroom\",\n \"grade_description\": \"crushed blacks\",\n \"mood_adjectives\": \"holy, crystalline\",\n \"avoid\": \"plastic highlights\",\n \"brief\": \"courier in night traffic\",\n}\n\nMEDIA = {\n \"clips\": [\n {\"path\": \"/tmp/media/a.mov\", \"duration\": 5.0, \"name\": \"a\"},\n {\"path\": \"/tmp/media/b.mov\", \"duration\": 4.0, \"name\": \"b\"},\n ]\n}\n\n\ndef run_cli(*args, expect=0):\n proc = subprocess.run(\n [sys.executable, \"-m\", \"tasteforge\", *args],\n capture_output=True,\n text=True,\n cwd=REPO_ROOT,\n check=False,\n )\n return proc\n\n\nclass CliTests(unittest.TestCase):\n def test_missing_ffmpeg_or_ffprobe_is_bounded_without_traceback(self):\n with tempfile.TemporaryDirectory() as td:\n root = Path(td)\n reference = root / \"reference.mov\"\n reference.write_bytes(b\"local-reference\")\n config = root / \"workflow.json\"\n config.write_text(json.dumps({\n \"schema_version\": 1,\n \"run_id\": \"missing-tools\",\n \"seed\": 15,\n \"dry_run\": True,\n \"resolve_duration\": 6.0,\n \"genres\": [{\n \"number\": 1,\n \"slug\": \"flash-ethereal\",\n \"label\": \"Flash Ethereal\",\n \"references\": [str(reference)],\n \"signature\": {\n \"materials\": [\"glass\"],\n \"motion\": [\"flash\"],\n \"composition\": [\"center\"],\n \"avoid\": [\"mud\"],\n },\n }],\n }), encoding=\"utf-8\")\n fake_bin = root / \"bin\"\n fake_bin.mkdir()\n ffprobe = fake_bin / \"ffprobe\"\n ffprobe.write_text(\n \"#!/bin/sh\\nprintf '%s\\\\n' \"\n \"'{\\\"streams\\\":[{\\\"codec_type\\\":\\\"video\\\",\\\"duration\\\":\\\"1\\\",\"\n \"\\\"avg_frame_rate\\\":\\\"24/1\\\"}],\\\"format\\\":{\\\"duration\\\":\\\"1\\\"}}'\\n\",\n encoding=\"utf-8\",\n )\n ffprobe.chmod(0o700)\n\n for label, path_value in ((\"ffprobe\", \"\"), (\"ffmpeg\", str(fake_bin))):\n with self.subTest(tool=label):\n proc = subprocess.run(\n [sys.executable, \"-m\", \"tasteforge\", \"multimodal\",\n \"--config\", str(config), \"--out-dir\", str(root / f\"out-{label}\")],\n capture_output=True,\n text=True,\n cwd=REPO_ROOT,\n env={\"PATH\": path_value},\n check=False,\n )\n self.assertEqual(proc.returncode, cli.EXIT_INVALID)\n self.assertEqual(proc.stderr, \"ERROR local media processing unavailable\\n\")\n self.assertNotIn(\"Traceback\", proc.stderr)\n\n def test_corrupt_media_process_failure_is_bounded_and_redacted(self):\n failure = subprocess.CalledProcessError(\n 1,\n [\"ffprobe\", \"https://provider.invalid/?token=secret-value\"],\n stderr=\"provider response secret-value\",\n )\n stderr = io.StringIO()\n with mock.patch(\n \"tasteforge.cli.workflow_mod.run_workflow\", side_effect=failure\n ), redirect_stderr(stderr):\n status = cli.main([\n \"multimodal\", \"--config\", \"corrupt.json\", \"--out-dir\", \"out\"\n ])\n message = stderr.getvalue()\n self.assertEqual(status, cli.EXIT_INVALID)\n self.assertEqual(message, \"ERROR local media processing failed\\n\")\n self.assertNotIn(\"Traceback\", message)\n self.assertNotIn(\"secret-value\", message)\n self.assertNotIn(\"provider.invalid\", message)\n\n def test_multimodal_command_routes_file_contract_and_validates_bundle(self):\n with tempfile.TemporaryDirectory() as td:\n config = Path(td) / \"workflow.json\"\n config.write_text(\"{}\", encoding=\"utf-8\")\n out = Path(td) / \"out\"\n expected = {\"provider_calls\": 0, \"provider_execution\": False}\n with mock.patch(\n \"tasteforge.cli.workflow_mod.run_workflow\", return_value=expected\n ) as run, mock.patch(\"tasteforge.cli.contract_mod.validate_bundle\") as validate:\n status = cli.main([\n \"multimodal\", \"--config\", str(config), \"--out-dir\", str(out)\n ])\n self.assertEqual(status, 0)\n run.assert_called_once_with(config, out)\n validate.assert_called_once_with(out)\n def test_provenance_subcommand(self):\n proc = run_cli(\"provenance\", \"--json\")\n self.assertEqual(proc.returncode, 0, proc.stderr)\n data = json.loads(proc.stdout)\n self.assertIn(\"generations\", data)\n\n def test_inspect_subcommand(self):\n proc = run_cli(\"inspect\", str(FIXTURE), \"--json\")\n self.assertEqual(proc.returncode, 0, proc.stderr)\n data = json.loads(proc.stdout)\n self.assertEqual(data[\"name\"], \"flashethereal\")\n\n def test_validate_subcommand_ok_and_fail(self):\n proc = run_cli(\"validate\", str(FIXTURE))\n self.assertEqual(proc.returncode, 0, proc.stderr)\n with tempfile.TemporaryDirectory() as td:\n bad = Path(td) / \"badpack\"\n bad.mkdir()\n (bad / \"pack.json\").write_text(\"{}\")\n proc = run_cli(\"validate\", str(bad))\n self.assertNotEqual(proc.returncode, 0)\n\n def test_interview_distill_apply_export_roundtrip(self):\n with tempfile.TemporaryDirectory() as td:\n answers_p = Path(td) / \"answers.json\"\n profile_p = Path(td) / \"profile.json\"\n spec_p = Path(td) / \"spec.json\"\n report_p = Path(td) / \"report.json\"\n media_p = Path(td) / \"media.json\"\n events_p = Path(td) / \"events.json\"\n answers_p.write_text(json.dumps(ANSWERS))\n media_p.write_text(json.dumps(MEDIA))\n\n proc = run_cli(\"interview\", \"--answers\", str(answers_p),\n \"--genre\", \"flashethereal\", \"--out\", str(profile_p))\n self.assertEqual(proc.returncode, 0, proc.stderr)\n self.assertTrue(profile_p.exists())\n\n proc = run_cli(\"distill\", \"--profile\", str(profile_p),\n \"--pack\", str(FIXTURE), \"--out\", str(spec_p))\n self.assertEqual(proc.returncode, 0, proc.stderr)\n spec = json.loads(spec_p.read_text())\n self.assertTrue(spec[\"source\"][\"dry_run\"])\n\n proc = run_cli(\"apply\", \"--pack\", str(FIXTURE), \"--media\", str(media_p),\n \"--duration\", \"10\", \"--out\", str(report_p))\n self.assertEqual(proc.returncode, 0, proc.stderr)\n report = json.loads(report_p.read_text())\n self.assertEqual(report[\"provider\"], \"none\")\n events_p.write_text(json.dumps({\"clips\": report[\"timeline_events\"]}))\n\n proc = run_cli(\"export\", \"--events\", str(events_p),\n \"--out-dir\", td, \"--title\", \"cli-test\")\n self.assertEqual(proc.returncode, 0, proc.stderr)\n self.assertTrue((Path(td) / \"cli-test.edl\").exists())\n self.assertTrue((Path(td) / \"cli-test.fcpxml\").exists())\n\n def test_apply_strict_cli_emits_exact_frame_report(self):\n with tempfile.TemporaryDirectory() as td:\n media = {\"clips\": [{\"path\": f\"/tmp/clip-{i}.mov\", \"duration\": 5}\n for i in range(20)]}\n media_p = Path(td) / \"media.json\"\n media_p.write_text(json.dumps(media))\n out = Path(td) / \"report.json\"\n proc = run_cli(\"apply\", \"--pack\", str(FIXTURE), \"--media\", str(media_p),\n \"--duration\", \"2.25\", \"--fps\", \"20\", \"--no-repeat\", \"--out\", str(out))\n self.assertEqual(proc.returncode, 0, proc.stderr)\n report = json.loads(out.read_text())\n events = report[\"timeline_events\"]\n self.assertEqual(sum(e[\"frames\"] for e in events), 45)\n self.assertTrue(all(e[\"fps\"] == 20 for e in events))\n self.assertEqual(len(events), len({e[\"path\"] for e in events}))\n\n def test_apply_invalid_or_insufficient_strict_input_creates_no_report(self):\n with tempfile.TemporaryDirectory() as td:\n media_p = Path(td) / \"media.json\"\n media_p.write_text(json.dumps(MEDIA))\n out = Path(td) / \"report.json\"\n for options in ((\"--duration\", \"20\", \"--no-repeat\"),\n (\"--duration\", \"0\"), (\"--fps\", \"nan\")):\n with self.subTest(options=options):\n proc = run_cli(\"apply\", \"--pack\", str(FIXTURE), \"--media\", str(media_p),\n \"--out\", str(out), *options)\n self.assertEqual(proc.returncode, 1, proc.stderr)\n self.assertNotIn(\"Traceback\", proc.stderr)\n self.assertFalse(out.exists())\n\n def test_default_output_names_reject_path_traversal(self):\n with tempfile.TemporaryDirectory() as td:\n # pack.json name with traversal: default report path must not be derived from it\n pack = Path(td) / \"pack\"\n shutil.copytree(FIXTURE, pack)\n manifest = json.loads((pack / \"pack.json\").read_text())\n manifest[\"name\"] = \"../../escaped\"\n (pack / \"pack.json\").write_text(json.dumps(manifest))\n media_p = Path(td) / \"media.json\"\n media_p.write_text(json.dumps(MEDIA))\n proc = run_cli(\"apply\", \"--pack\", str(pack), \"--media\", str(media_p), \"--duration\", \"4\")\n self.assertEqual(proc.returncode, 1, proc.stderr)\n self.assertNotIn(\"Traceback\", proc.stderr)\n self.assertIn(\"pack name\", proc.stderr)\n self.assertFalse((REPO_ROOT.parent / \"escaped_apply_report.json\").exists())\n self.assertFalse((REPO_ROOT / \"out\").exists() and any(REPO_ROOT.glob(\"out/*escaped*\")))\n # profile genre with traversal: default spec path must not be derived from it\n answers_p = Path(td) / \"answers.json\"\n answers_p.write_text(json.dumps(ANSWERS))\n profile_p = Path(td) / \"profile.json\"\n proc = run_cli(\"interview\", \"--answers\", str(answers_p), \"--genre\", \"flashethereal\", \"--out\", str(profile_p))\n self.assertEqual(proc.returncode, 0, proc.stderr)\n profile = json.loads(profile_p.read_text())\n profile[\"genre\"] = \"../escaped\"\n profile_p.write_text(json.dumps(profile))\n proc = run_cli(\"distill\", \"--profile\", str(profile_p), \"--pack\", str(FIXTURE))\n self.assertEqual(proc.returncode, 1, proc.stderr)\n self.assertNotIn(\"Traceback\", proc.stderr)\n self.assertIn(\"profile genre\", proc.stderr)\n self.assertFalse((REPO_ROOT.parent / \"escaped-spec.json\").exists())\n # an explicit --out still works with an odd genre\n spec_p = Path(td) / \"spec.json\"\n proc = run_cli(\"distill\", \"--profile\", str(profile_p), \"--pack\", str(FIXTURE), \"--out\", str(spec_p))\n self.assertEqual(proc.returncode, 0, proc.stderr)\n\n def test_live_provider_flags_fail_closed(self):\n with tempfile.TemporaryDirectory() as td:\n profile_p = Path(td) / \"profile.json\"\n run_cli(\"interview\", \"--answers\", self._write(td, ANSWERS),\n \"--genre\", \"g\", \"--out\", str(profile_p))\n proc = run_cli(\"distill\", \"--profile\", str(profile_p), \"--live\")\n self.assertNotEqual(proc.returncode, 0)\n self.assertIn(\"separately authorized\", proc.stderr + proc.stdout)\n media_p = Path(td) / \"media.json\"\n media_p.write_text(json.dumps(MEDIA))\n proc = run_cli(\"apply\", \"--pack\", str(FIXTURE),\n \"--media\", str(media_p), \"--live\")\n self.assertNotEqual(proc.returncode, 0)\n self.assertIn(\"separately authorized\", proc.stderr + proc.stdout)\n\n @staticmethod\n def _write(td, obj):\n p = Path(td) / \"answers.json\"\n p.write_text(json.dumps(obj))\n return str(p)\n\n\[email protected](shutil.which(\"ffmpeg\") and shutil.which(\"ffprobe\"), \"ffmpeg tools unavailable\")\nclass RealMediaCliIntegrationTests(unittest.TestCase):\n def setUp(self):\n self.tmp = tempfile.TemporaryDirectory()\n self.root = Path(self.tmp.name)\n self.media = self.root / \"reference.mp4\"\n ffmpeg = shutil.which(\"ffmpeg\")\n assert ffmpeg is not None\n generated = subprocess.run(\n [\n ffmpeg, \"-v\", \"error\", \"-f\", \"lavfi\", \"-i\",\n \"color=c=blue:s=64x64:r=12:d=1\", \"-c:v\", \"mpeg4\", \"-y\", str(self.media),\n ],\n capture_output=True,\n text=True,\n check=False,\n )\n if generated.returncode != 0:\n self.skipTest(\"local ffmpeg cannot generate the integration fixture\")\n\n def tearDown(self):\n self.tmp.cleanup()\n\n def _config(self, reference: Path) -> Path:\n genres = []\n values = [\n (1, \"flash-ethereal\", \"Flash Ethereal\", \"glass\", \"flash\", \"center\", \"mud\"),\n (2, \"3d-cyber-glitch\", \"3D Cyber Glitch\", \"chrome\", \"orbit\", \"full\", \"corner\"),\n (3, \"fluid-sketch\", \"Fluid Sketch\", \"ink\", \"bleed\", \"space\", \"grid\"),\n ]\n for number, slug, label, material, motion, composition, avoid in values:\n genres.append({\n \"number\": number,\n \"slug\": slug,\n \"label\": label,\n \"references\": [str(reference)],\n \"signature\": {\n \"materials\": [material],\n \"motion\": [motion],\n \"composition\": [composition],\n \"avoid\": [avoid],\n },\n })\n config = self.root / \"workflow.json\"\n config.write_text(json.dumps({\n \"schema_version\": 1,\n \"run_id\": \"real-tools\",\n \"seed\": 15,\n \"dry_run\": True,\n \"resolve_duration\": 6.0,\n \"genres\": genres,\n }), encoding=\"utf-8\")\n return config\n\n def test_real_ffmpeg_ffprobe_cli_emits_and_validates_bundle(self):\n out = self.root / \"out\"\n proc = run_cli(\n \"multimodal\", \"--config\", str(self._config(self.media)), \"--out-dir\", str(out)\n )\n self.assertEqual(proc.returncode, 0, proc.stderr)\n receipt = json.loads(proc.stdout)\n self.assertEqual(receipt[\"provider_calls\"], 0)\n self.assertFalse(receipt[\"provider_execution\"])\n self.assertTrue((out / \"receipt.json\").is_file())\n\n def test_real_corrupt_media_cli_failure_is_bounded_and_redacted(self):\n corrupt = self.root / \"corrupt.mov\"\n corrupt.write_bytes(b\"not-media-secret-marker\")\n proc = run_cli(\n \"multimodal\", \"--config\", str(self._config(corrupt)),\n \"--out-dir\", str(self.root / \"corrupt-out\"),\n )\n self.assertEqual(proc.returncode, cli.EXIT_INVALID)\n self.assertEqual(proc.stderr, \"ERROR local media processing failed\\n\")\n self.assertNotIn(\"Traceback\", proc.stderr)\n self.assertNotIn(\"not-media-secret-marker\", proc.stderr)\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":16127,"content_sha256":"9325927a0232c11988c785a866df60bec40bd1e8ff69edc8f564e2521baccbf6"},{"filename":"tests/test_distill.py","content":"\"\"\"Failing-first tests for offline distillation and the fail-closed live path.\"\"\"\n\nfrom __future__ import annotations\n\nimport json\nimport sys\nimport unittest\nfrom pathlib import Path\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nfrom tasteforge import distill, interview, schema # noqa: E402\n\nFIXTURE = Path(__import__(\"tasteforge\").__file__).resolve().parent / \"fixtures\" / \"flashethereal\"\n\n\ndef _profile():\n return interview.conduct(\n {\n \"palette\": \"near-black void, bone white, violet bloom\",\n \"grain\": \"fine 35mm grain\",\n \"lighting\": \"single hard key\",\n \"focal_length\": \"35mm\",\n \"camera_motion\": \"locked off\",\n \"subject_framing\": \"centered, headroom\",\n \"grade_description\": \"crushed blacks\",\n \"mood_adjectives\": \"holy, crystalline\",\n \"avoid\": \"plastic highlights\",\n \"brief\": \"courier in night traffic\",\n },\n genre=\"flashethereal\",\n )\n\n\nclass LocalDistillTests(unittest.TestCase):\n def test_distill_local_produces_valid_spec(self):\n spec = distill.distill_local(_profile())\n problems = schema.validate(spec, schema.SPEC_SCHEMA)\n self.assertEqual(problems, [])\n\n def test_distill_local_is_deterministic(self):\n a = distill.distill_local(_profile())\n b = distill.distill_local(_profile())\n a[\"source\"].pop(\"generated\"), b[\"source\"].pop(\"generated\")\n self.assertEqual(a, b)\n\n def test_distill_local_labels_dry_run_and_carries_answers(self):\n spec = distill.distill_local(_profile())\n self.assertTrue(spec[\"source\"][\"dry_run\"])\n self.assertEqual(spec[\"source\"][\"provider\"], \"none\")\n self.assertEqual(spec[\"lighting\"], \"single hard key\")\n self.assertEqual(spec[\"mood_adjectives\"], [\"holy\", \"crystalline\"])\n\n def test_grounding_from_grade_states_measurements(self):\n grade = json.loads((FIXTURE / \"grade.json\").read_text())\n cadence = json.loads((FIXTURE / \"cadence.json\").read_text())\n text = distill.grounding_from_grade(grade, cadence)\n self.assertIn(\"MEASURED GROUND TRUTH\", text)\n self.assertIn(\"black point\", text)\n self.assertIn(\"#131215\", text) # dominant palette hex survives\n self.assertIn(\"cuts/min\", text)\n # The banned-words contract from the recovered grounding prompt.\n self.assertIn(\"Do not contradict\", text)\n\n def test_grounding_from_empty_inputs_is_empty(self):\n self.assertEqual(distill.grounding_from_grade({}, {}), \"\")\n\n\nclass FailClosedLiveTests(unittest.TestCase):\n def test_distill_live_raises_provider_disabled(self):\n with self.assertRaises(distill.ProviderDisabledError) as ctx:\n distill.distill_live(_profile())\n self.assertIn(\"separately authorized\", str(ctx.exception))\n\n def test_no_network_module_imported(self):\n import sys as _sys\n\n _sys.modules.pop(\"fal_client\", None)\n try:\n distill.distill_live(_profile())\n except distill.ProviderDisabledError:\n pass\n self.assertNotIn(\"fal_client\", _sys.modules)\n self.assertNotIn(\"urllib.request\", _sys.modules)\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":3286,"content_sha256":"70a51b84d0a629f7c7a10dca4a42691619d68de68d0dbabb0de263bf4869c8e1"},{"filename":"tests/test_integration.py","content":"\"\"\"Synthetic, local-only acceptance tests for preserving an existing edit.\"\"\"\n\nimport copy\nimport hashlib\nimport json\nimport subprocess\nimport sys\nimport tempfile\nimport unittest\nfrom pathlib import Path\nfrom unittest import mock\n\nfrom tasteforge.integration import build_application_bundle, validate_application_bundle\n\n\nSCRIPT = Path(__file__).resolve().parents[1] / \"scripts\" / \"workflow_graphs.py\"\nPAYLOAD = {\"source_video\": \"https://media.example/source.mov\", \"compiled_prompt\": \"Test motion\"}\n\n\nclass ApplicationBundleTests(unittest.TestCase):\n def setUp(self):\n self.temp = tempfile.TemporaryDirectory()\n self.addCleanup(self.temp.cleanup)\n self.root = Path(self.temp.name).resolve()\n self.rate = {\"numerator\": 30, \"denominator\": 1}\n self.source = self.artifact(\"source.mov\", b\"synthetic original video\")\n self.audio = self.artifact(\"music.wav\", b\"synthetic music\")\n self.snapshot = {\n \"project\": \"Synthetic project\", \"timeline\": \"Original timeline\",\n \"settings\": {\"timelineFrameRate\": 30.0},\n \"timeline_readback\": {\n \"video1\": [self.clip(self.source[\"path\"], 0, 120, 3, 2)],\n \"video2\": [self.clip(\"/synthetic/contour.mov\", 25, 38, 0, 0)],\n \"video3\": [self.clip(\"/synthetic/disabled-bloom.mov\", 20, 45, 0, 0, False)],\n \"audio1\": [self.clip(self.audio[\"path\"], 0, 117, 0, 1)],\n },\n }\n self.config = {\n \"baseline\": {\n \"project_file\": self.artifact(\"project.drp\", b\"synthetic native project\"),\n \"snapshot_file\": self.artifact(\"snapshot.json\", self.snapshot),\n \"project_name\": \"Synthetic project\", \"timeline_name\": \"Original timeline\",\n \"fps\": self.rate, \"timeline_range\": [0, 120],\n },\n \"source\": self.binding(self.source, \"video1\", 125, [3, 123], [0, 120]),\n \"audio\": [self.binding(self.audio, \"audio1\", 118, [0, 117], [0, 117])],\n \"protected_intervals\": [{\"range\": [24, 40], \"reason\": \"Original hand treatment\"}],\n }\n\n def artifact(self, name, content):\n data = json.dumps(content).encode() if isinstance(content, dict) else content\n path = self.root / name\n path.write_bytes(data)\n return {\"path\": str(path), \"bytes\": len(data), \"sha256\": hashlib.sha256(data).hexdigest()}\n\n @staticmethod\n def clip(path, start, end, left, right, enabled=True):\n return {\"path\": path, \"name\": Path(path).name, \"start\": start, \"end\": end,\n \"left_offset\": left, \"right_offset\": right, \"enabled\": enabled,\n \"properties\": {\"Opacity\": 88.0, \"CompositeMode\": 0}}\n\n def binding(self, media, track, frames, source_range, timeline_range):\n return {\"media\": media, \"track\": track, \"clip_index\": 0, \"media_frames\": frames,\n \"fps\": self.rate, \"source_range\": source_range, \"timeline_range\": timeline_range}\n\n def approved_insert(self):\n media = self.artifact(\"candidate.mov\", b\"synthetic generated variation\")\n candidate = {\n \"id\": \"take-1\", \"media\": media, \"media_frames\": 30, \"fps\": self.rate,\n \"origin\": \"provider_generated\", \"relationship\": \"generated_variation\",\n \"source_sha256\": self.source[\"sha256\"], \"review_status\": \"approved\",\n \"compiled_input_sha256\": hashlib.sha256(\n json.dumps(PAYLOAD, sort_keys=True, separators=(\",\", \":\"), allow_nan=False).encode()\n ).hexdigest(),\n }\n candidate[\"generation_receipt\"] = self.artifact(\"generation.json\", {\n \"request_id\": \"synthetic-request\", \"source_url\": PAYLOAD[\"source_video\"],\n \"source_sha256\": self.source[\"sha256\"], \"candidate_sha256\": media[\"sha256\"],\n \"compiled_input_sha256\": candidate[\"compiled_input_sha256\"],\n })\n insert = {\"candidate_id\": \"take-1\", \"candidate_range\": [2, 14],\n \"timeline_range\": [60, 72], \"retime\": \"none\"}\n approval = {\"status\": \"approved\", \"candidate_sha256\": media[\"sha256\"],\n \"source_sha256\": self.source[\"sha256\"],\n \"compiled_input_sha256\": candidate[\"compiled_input_sha256\"],\n \"candidate_range\": [2, 14], \"timeline_range\": [60, 72]}\n approval[\"edit_context_sha256\"] = hashlib.sha256(json.dumps(\n {key: self.config[key] for key in (\"baseline\", \"source\", \"audio\", \"protected_intervals\")},\n sort_keys=True, separators=(\",\", \":\"), allow_nan=False).encode()).hexdigest()\n insert[\"approval_file\"] = self.artifact(\"approval.json\", approval)\n self.config[\"candidates\"] = [candidate]\n self.config[\"inserts\"] = [insert]\n return candidate, insert, approval\n\n def test_default_bundle_preserves_baseline_audio_and_entire_protected_stack(self):\n before = copy.deepcopy(self.config)\n bundle = build_application_bundle(self.config, PAYLOAD)\n self.assertEqual(self.config, before)\n self.assertEqual(bundle[\"mode\"], \"preserve_native_timeline\")\n self.assertEqual(bundle[\"baseline\"], before[\"baseline\"])\n self.assertEqual(bundle[\"audio\"], before[\"audio\"])\n self.assertEqual(bundle[\"inserts\"], [])\n self.assertEqual(bundle[\"provider_calls\"], 0)\n self.assertIs(type(bundle[\"provider_calls\"]), int)\n self.assertIs(bundle[\"provider_execution\"], False)\n self.assertIs(bundle[\"submit\"], False)\n self.assertEqual(bundle[\"provider_input\"], PAYLOAD)\n self.assertEqual({c[\"track\"] for c in bundle[\"protected_stack\"]},\n {\"video1\", \"video2\", \"video3\", \"audio1\"})\n disabled = next(c for c in bundle[\"protected_stack\"] if c[\"track\"] == \"video3\")\n self.assertEqual(disabled[\"clip\"], self.snapshot[\"timeline_readback\"][\"video3\"][0])\n self.assertEqual(validate_application_bundle(bundle), None)\n bundle[\"audio\"][0][\"timeline_range\"][1] = 116\n self.assertEqual(self.config, before)\n\n def test_deterministic_bundle_and_no_provider_calls(self):\n with mock.patch(\"socket.socket\", side_effect=AssertionError(\"network forbidden\")), \\\n mock.patch(\"subprocess.run\", side_effect=AssertionError(\"process forbidden\")):\n self.assertEqual(build_application_bundle(self.config, PAYLOAD),\n build_application_bundle(self.config, PAYLOAD))\n\n def test_approved_insert_is_an_additive_video_only_proposal(self):\n self.approved_insert()\n bundle = build_application_bundle(self.config, PAYLOAD)\n self.assertEqual(len(bundle[\"inserts\"]), 1)\n self.assertEqual(bundle[\"insert_policy\"], \"new_video_track_preserve_baseline_audio\")\n self.assertEqual(bundle[\"audio\"], self.config[\"audio\"])\n validate_application_bundle(bundle)\n\n def test_pending_rejected_and_historical_urls_never_become_inserts(self):\n candidate, _, _ = self.approved_insert()\n for state in [\"pending\", \"rejected\", \"unknown\"]:\n candidate[\"review_status\"] = state\n with self.subTest(state=state), self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n candidate[\"review_status\"] = \"pending\"\n self.config[\"inserts\"] = []\n self.assertEqual(build_application_bundle(self.config, PAYLOAD)[\"inserts\"], [])\n candidate[\"media\"] = {\"url\": \"https://media.example/historical.mov\"}\n with self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_protected_overlap_rejected_including_one_frame(self):\n _, insert, approval = self.approved_insert()\n for span in [[12, 25], [39, 51], [24, 40]]:\n insert[\"timeline_range\"] = span\n insert[\"candidate_range\"] = [0, span[1] - span[0]]\n insert[\"approval_file\"] = self.artifact(\"approval.json\", {\n **approval, \"timeline_range\": span, \"candidate_range\": insert[\"candidate_range\"]})\n with self.subTest(span=span), self.assertRaisesRegex(ValueError, \"protected\"):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_approval_binds_both_source_and_exact_placement(self):\n _, insert, approval = self.approved_insert()\n for field, value in [(\"status\", \"pending\"), (\"source_sha256\", \"0\" * 64),\n (\"candidate_sha256\", \"1\" * 64),\n (\"compiled_input_sha256\", \"2\" * 64),\n (\"timeline_range\", [72, 84]), (\"candidate_range\", [3, 15])]:\n altered = {**approval, field: value}\n insert[\"approval_file\"] = self.artifact(\"approval.json\", altered)\n with self.subTest(field=field), self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_approval_cannot_transfer_to_another_edit_or_timebase(self):\n self.approved_insert()\n for field in [\"baseline\", \"fps\", \"protected\"]:\n cfg = copy.deepcopy(self.config)\n if field == \"baseline\":\n cfg[\"baseline\"][\"project_file\"] = self.artifact(\"other.drp\", b\"another edit\")\n elif field == \"fps\":\n cfg[\"baseline\"][\"fps\"][\"numerator\"] = 60\n changed_snapshot = {**self.snapshot, \"settings\": {\"timelineFrameRate\": 60.0}}\n cfg[\"baseline\"][\"snapshot_file\"] = self.artifact(\"other-snapshot.json\", changed_snapshot)\n else:\n cfg[\"protected_intervals\"][0][\"range\"] = [24, 41]\n with self.subTest(field=field), self.assertRaisesRegex(ValueError, \"approval\"):\n build_application_bundle(cfg, PAYLOAD)\n\n def test_original_claim_wrong_source_or_wrong_input_is_rejected(self):\n candidate, _, _ = self.approved_insert()\n for field, value in [(\"origin\", \"original\"), (\"relationship\", \"original_hgx\"),\n (\"source_sha256\", \"0\" * 64), (\"compiled_input_sha256\", \"1\" * 64)]:\n prior = candidate[field]\n candidate[field] = value\n with self.subTest(field=field), self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n candidate[field] = prior\n\n def test_exact_integer_frames_and_rational_fps(self):\n for bad in [True, False, 1.0, float(\"nan\"), float(\"inf\"), -1, \"30\"]:\n for key in [\"numerator\", \"denominator\"]:\n cfg = copy.deepcopy(self.config)\n cfg[\"baseline\"][\"fps\"][key] = bad\n with self.subTest(key=key, bad=bad), self.assertRaises(ValueError):\n build_application_bundle(cfg, PAYLOAD)\n cfg = copy.deepcopy(self.config)\n cfg[\"protected_intervals\"][0][\"range\"][0] = bad\n with self.subTest(frame=bad), self.assertRaises(ValueError):\n build_application_bundle(cfg, PAYLOAD)\n\n def test_source_binding_must_match_native_clip_and_capacity(self):\n for field, value in [(\"source_range\", [0, 120]), (\"timeline_range\", [1, 121]),\n (\"media_frames\", 124), (\"clip_index\", True),\n (\"track\", \"video2\"), (\"fps\", {\"numerator\": 24, \"denominator\": 1})]:\n cfg = copy.deepcopy(self.config)\n cfg[\"source\"][field] = value\n with self.subTest(field=field), self.assertRaises(ValueError):\n build_application_bundle(cfg, PAYLOAD)\n\n def test_audio_cannot_be_dropped_retimed_or_extended(self):\n for audio in [[], self.config[\"audio\"] * 2,\n [{**self.config[\"audio\"][0], \"timeline_range\": [0, 120]}]]:\n with self.subTest(audio=audio), self.assertRaises(ValueError):\n build_application_bundle({**self.config, \"audio\": audio}, PAYLOAD)\n\n def test_mismatched_fps_duration_retime_and_overlapping_inserts(self):\n candidate, insert, approval = self.approved_insert()\n candidate[\"fps\"] = {\"numerator\": 30000, \"denominator\": 1001}\n with self.assertRaisesRegex(ValueError, \"retime ambiguity\"):\n build_application_bundle(self.config, PAYLOAD)\n candidate[\"fps\"] = self.rate\n for field, value in [(\"retime\", \"fit\"), (\"candidate_range\", [2, 15]),\n (\"candidate_range\", [20, 32]), (\"timeline_range\", [115, 127])]:\n original = insert[field]\n insert[field] = value\n insert[\"approval_file\"] = self.artifact(\"approval.json\", {\n **approval, \"timeline_range\": insert[\"timeline_range\"],\n \"candidate_range\": insert[\"candidate_range\"]})\n reason = \"retime ambiguity\" if field == \"retime\" or value == [2, 15] else \"bounds\"\n with self.subTest(field=field), self.assertRaisesRegex(ValueError, reason):\n build_application_bundle(self.config, PAYLOAD)\n insert[field] = original\n insert[\"approval_file\"] = self.artifact(\"approval.json\", approval)\n self.config[\"inserts\"].append(copy.deepcopy(insert))\n with self.assertRaisesRegex(ValueError, \"proposals overlap\"):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_missing_mutated_or_symlink_artifacts_are_rejected(self):\n path = Path(self.source[\"path\"])\n original = path.read_bytes()\n path.write_bytes(b\"changed original\")\n with self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n path.unlink()\n with self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n target = self.root / \"target.mov\"\n target.write_bytes(original)\n path.symlink_to(target)\n with self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_bundle_tampering_flags_omissions_and_extra_fields_fail(self):\n bundle = build_application_bundle(self.config, PAYLOAD)\n for field, value in [(\"provider_calls\", False), (\"provider_execution\", True),\n (\"submit\", True), (\"dry_run\", False), (\"protected_stack\", []),\n (\"mode\", \"replace_timeline\"), (\"extra\", \"unbound\")]:\n with self.subTest(field=field), self.assertRaises(ValueError):\n validate_application_bundle({**bundle, field: value})\n del bundle[\"audio\"]\n with self.assertRaises(ValueError):\n validate_application_bundle(bundle)\n\n def test_unresolved_hash_numeric_size_and_duplicate_json_fields_are_rejected(self):\n for field, value in [(\"sha256\", None), (\"sha256\", \"https://media.example/a.mov\"),\n (\"bytes\", True), (\"bytes\", 1.5)]:\n cfg = copy.deepcopy(self.config)\n cfg[\"source\"][\"media\"][field] = value\n with self.subTest(field=field), self.assertRaises(ValueError):\n build_application_bundle(cfg, PAYLOAD)\n self.config[\"baseline\"][\"snapshot_file\"] = self.artifact(\n \"snapshot.json\", b'{\"project\":\"first\",\"project\":\"second\"}')\n with self.assertRaisesRegex(ValueError, \"duplicate\"):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_parent_symlink_special_file_and_cloud_placeholder_are_not_read(self):\n path = Path(self.source[\"path\"])\n alias = self.root / \"alias\"\n alias.symlink_to(self.root, target_is_directory=True)\n cfg = copy.deepcopy(self.config)\n cfg[\"source\"][\"media\"][\"path\"] = str(alias / path.name)\n with self.assertRaises(ValueError):\n build_application_bundle(cfg, PAYLOAD)\n import os\n path.unlink()\n os.mkfifo(path)\n with self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n path.unlink()\n path.write_bytes(b\"synthetic original video\")\n actual_stat = os.stat\n\n def cloud_stat(target, *args, **kwargs):\n info = actual_stat(target, *args, **kwargs)\n if target == path.name:\n return mock.Mock(st_mode=info.st_mode, st_flags=0x40000000)\n return info\n\n with mock.patch(\"tasteforge.integration.os.stat\", side_effect=cloud_stat), \\\n mock.patch(\"tasteforge.integration.os.read\", wraps=os.read) as read:\n with self.assertRaisesRegex(ValueError, \"resident\"):\n build_application_bundle(self.config, PAYLOAD)\n # Only baseline project/snapshot were read; the placeholder never opened.\n self.assertTrue(read.called)\n\n def test_generation_receipt_drift_and_unselected_history_are_distinct(self):\n candidate, _, _ = self.approved_insert()\n evidence = json.loads(Path(candidate[\"generation_receipt\"][\"path\"]).read_text())\n for field in [\"request_id\", \"source_url\", \"source_sha256\", \"candidate_sha256\",\n \"compiled_input_sha256\"]:\n bad = {**evidence, field: \"\"}\n candidate[\"generation_receipt\"] = self.artifact(\"generation.json\", bad)\n with self.subTest(field=field), self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n self.config[\"candidates\"] = []\n self.config[\"inserts\"] = []\n self.config[\"historical_receipts\"] = [self.artifact(\"history.json\", {\n \"request_id\": \"old\", \"output_url\": \"https://media.example/unknown.mov\"})]\n bundle = build_application_bundle(self.config, PAYLOAD)\n self.assertEqual(bundle[\"inserts\"], [])\n self.assertEqual(bundle[\"candidates\"], [])\n\n def test_source_subset_and_ntsc_rate_preserve_exact_frame_mapping(self):\n self.config[\"source\"][\"source_range\"] = [13, 33]\n self.config[\"source\"][\"timeline_range\"] = [10, 30]\n self.rate.update(numerator=30000, denominator=1001)\n self.snapshot[\"settings\"][\"timelineFrameRate\"] = \"29.97\"\n self.config[\"baseline\"][\"snapshot_file\"] = self.artifact(\"snapshot.json\", self.snapshot)\n bundle = build_application_bundle(self.config, PAYLOAD)\n self.assertEqual(bundle[\"source\"][\"timeline_range\"], [10, 30])\n self.assertEqual(bundle[\"baseline\"][\"fps\"], self.rate)\n\n def test_missing_or_conflicting_native_fps_is_rejected(self):\n for settings in [{}, {\"timelineFrameRate\": \"24\"}, {\"timelineFrameRate\": True}]:\n self.snapshot[\"settings\"] = settings\n self.config[\"baseline\"][\"snapshot_file\"] = self.artifact(\"snapshot.json\", self.snapshot)\n with self.subTest(settings=settings), self.assertRaisesRegex(ValueError, \"fps\"):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_exponent_fps_is_rejected_before_fraction_allocation(self):\n self.snapshot[\"settings\"][\"timelineFrameRate\"] = \"1e1000000000\"\n self.config[\"baseline\"][\"snapshot_file\"] = self.artifact(\"snapshot.json\", self.snapshot)\n with mock.patch(\"tasteforge.integration.Fraction\", side_effect=AssertionError(\"unsafe allocation\")):\n with self.assertRaisesRegex(ValueError, \"fps\"):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_fileless_native_generator_is_preserved_without_becoming_a_source(self):\n self.snapshot[\"timeline_readback\"][\"video4\"] = [{\n **self.clip(\"unused\", 24, 40, 0, 0), \"path\": None, \"name\": \"Native title\"}]\n self.config[\"baseline\"][\"snapshot_file\"] = self.artifact(\"snapshot.json\", self.snapshot)\n bundle = build_application_bundle(self.config, PAYLOAD)\n generator = next(row for row in bundle[\"protected_stack\"] if row[\"track\"] == \"video4\")\n self.assertIsNone(generator[\"clip\"][\"path\"])\n self.config[\"source\"][\"track\"] = \"video4\"\n with self.assertRaises(ValueError):\n build_application_bundle(self.config, PAYLOAD)\n\n def test_parent_directory_substitution_during_read_is_rejected(self):\n from tasteforge import integration\n import os\n parent = self.root / \"reference\"\n parent.mkdir()\n record = self.artifact(\"reference/source.mov\", b\"original bytes\")\n actual_read = os.read\n replaced = False\n\n def replace_parent(descriptor, length):\n nonlocal replaced\n data = actual_read(descriptor, length)\n if not replaced:\n replaced = True\n parent.rename(self.root / \"moved-reference\")\n parent.mkdir()\n (parent / \"source.mov\").write_bytes(b\"different bytes\")\n return data\n\n with mock.patch(\"tasteforge.integration.os.read\", side_effect=replace_parent):\n with self.assertRaisesRegex(ValueError, \"changed\"):\n integration._artifact(record)\n\n def test_invalid_cli_bundle_creates_no_output_and_does_not_overwrite(self):\n config = {\"source_video\": PAYLOAD[\"source_video\"], \"brief\": \"Synthetic test\",\n \"style_steer\": \"Original motion\", \"integration\": self.config}\n cfg_file = self.root / \"request.json\"\n cfg_file.write_text(json.dumps(config))\n output = self.root / \"bundle.json\"\n output.write_text(\"original file\")\n command = [sys.executable, str(SCRIPT), \"--kind\", \"apply-bundle\", \"--config\",\n str(cfg_file), \"--out\", str(output)]\n proc = subprocess.run(command, capture_output=True, text=True)\n self.assertEqual(proc.returncode, 2)\n self.assertEqual(output.read_text(), \"original file\")\n output.unlink()\n config[\"integration\"][\"source\"][\"media\"][\"sha256\"] = \"0\" * 64\n cfg_file.write_text(json.dumps(config))\n proc = subprocess.run(command, capture_output=True, text=True)\n self.assertEqual(proc.returncode, 2)\n self.assertFalse(output.exists())\n self.assertNotIn(\"Traceback\", proc.stderr)\n\n def test_bundle_cli_refuses_symlink_request_without_output(self):\n cfg = {\"source_video\": PAYLOAD[\"source_video\"], \"brief\": \"Synthetic\",\n \"style_steer\": \"Synthetic\", \"integration\": self.config}\n real = self.root / \"request.json\"\n real.write_text(json.dumps(cfg))\n alias = self.root / \"alias.json\"\n alias.symlink_to(real)\n out = self.root / \"bundle.json\"\n proc = subprocess.run([sys.executable, str(SCRIPT), \"--kind\", \"apply-bundle\",\n \"--config\", str(alias), \"--out\", str(out)],\n capture_output=True, text=True)\n self.assertEqual(proc.returncode, 2)\n self.assertFalse(out.exists())\n\n def test_growing_artifact_is_rejected_before_accumulating_unbounded_data(self):\n from tasteforge import integration\n record = self.config[\"baseline\"][\"project_file\"]\n with mock.patch(\"tasteforge.integration.os.read\", side_effect=[b\"x\" * (record[\"bytes\"] + 1), b\"\"]):\n with self.assertRaisesRegex(ValueError, \"byte count exceeded\"):\n integration._artifact(record)\n\n def test_cli_bundle_compilation_and_legacy_payload_are_separate(self):\n config = {\"source_video\": PAYLOAD[\"source_video\"], \"brief\": \"Synthetic test\",\n \"style_steer\": \"Original motion\", \"integration\": self.config}\n config_file = self.root / \"request.json\"\n config_file.write_text(json.dumps(config))\n for kind in [\"apply\", \"apply-bundle\"]:\n proc = subprocess.run([sys.executable, str(SCRIPT), \"--kind\", kind,\n \"--config\", str(config_file), \"--out\", str(self.root / kind)],\n capture_output=True, text=True)\n self.assertEqual(proc.returncode, 0, proc.stderr)\n plain = json.loads((self.root / \"apply\").read_text())\n bundle = json.loads((self.root / \"apply-bundle\").read_text())\n self.assertEqual(set(plain), {\"source_video\", \"compiled_prompt\"})\n self.assertEqual(bundle[\"provider_input\"], plain)\n validate_application_bundle(bundle)\n\n def run_local_cli(self, request, *, suffix=\"local\"):\n config = self.root / (suffix + \"-request.json\")\n output = self.root / (suffix + \"-bundle.json\")\n config.write_text(json.dumps(request))\n proc = subprocess.run([sys.executable, str(SCRIPT), \"--kind\", \"apply-bundle\",\n \"--config\", str(config), \"--out\", str(output)],\n capture_output=True, text=True)\n return proc, output\n\n def test_local_only_cli_compiles_preservation_without_hosted_source(self):\n proc, output = self.run_local_cli({\"local_only\": True, \"integration\": self.config})\n self.assertEqual(proc.returncode, 0, proc.stderr)\n bundle = json.loads(output.read_text())\n self.assertIs(bundle[\"local_only\"], True)\n self.assertIsNone(bundle[\"provider_input\"])\n self.assertIsNone(bundle[\"compiled_input_sha256\"])\n self.assertEqual(bundle[\"provider_input_status\"], \"not_prepared_local_only\")\n self.assertEqual(bundle[\"insert_policy\"], \"none_preserve_baseline\")\n self.assertEqual(bundle[\"inserts\"], [])\n self.assertEqual(bundle[\"candidates\"], [])\n self.assertEqual(bundle[\"baseline\"], self.config[\"baseline\"])\n self.assertEqual(bundle[\"source\"], self.config[\"source\"])\n self.assertEqual(bundle[\"audio\"], self.config[\"audio\"])\n self.assertEqual(len(bundle[\"protected_stack\"]), 4)\n self.assertIs(bundle[\"submit\"], False)\n self.assertIs(bundle[\"provider_execution\"], False)\n validate_application_bundle(bundle)\n\n def test_local_only_flag_is_exact_boolean_in_cli_and_api(self):\n for i, bad in enumerate([None, 0, 1, \"true\", \"false\", [], {}]):\n with self.subTest(flag=bad), self.assertRaisesRegex(ValueError, \"local_only\"):\n build_application_bundle(self.config, None, local_only=bad)\n request = {\"local_only\": bad, \"integration\": self.config,\n \"source_video\": PAYLOAD[\"source_video\"], \"brief\": \"Synthetic\",\n \"style_steer\": \"Synthetic\"}\n proc, output = self.run_local_cli(request, suffix=f\"flag-{i}\")\n self.assertEqual(proc.returncode, 2, proc.stderr)\n self.assertIn(\"local_only\", proc.stderr)\n self.assertFalse(output.exists())\n\n def test_local_only_rejects_mixed_provider_request_fields(self):\n for i, extra in enumerate([\n {\"source_video\": PAYLOAD[\"source_video\"], \"brief\": \"Synthetic\", \"style_steer\": \"Synthetic\"},\n {\"source_video\": None}, {\"provider_input\": PAYLOAD}, {\"provider_input\": None},\n {\"compiled_prompt\": \"Synthetic\"}, {\"brief\": \"Uncompiled provider brief\"},\n ]):\n proc, output = self.run_local_cli(\n {\"local_only\": True, \"integration\": self.config, **extra}, suffix=f\"mixed-{i}\")\n self.assertEqual(proc.returncode, 2, proc.stderr)\n self.assertIn(\"local-only request\", proc.stderr)\n self.assertFalse(output.exists())\n for supplied in [PAYLOAD, {}, \"https://media.example/source.mov\"]:\n with self.subTest(supplied=supplied), self.assertRaisesRegex(ValueError, \"provider input\"):\n build_application_bundle(self.config, supplied, local_only=True)\n\n def test_local_only_rejects_candidates_and_inserts_before_reading_them(self):\n for field in [\"candidates\", \"inserts\"]:\n cfg = {**self.config, field: [{\"unresolved\": \"https://media.example/old.mov\"}]}\n with self.subTest(field=field), self.assertRaisesRegex(ValueError, \"local-only.*candidates|local-only.*inserts\"):\n build_application_bundle(cfg, None, local_only=True)\n\n def test_local_only_bundle_cannot_switch_modes_or_gain_provider_fields(self):\n bundle = build_application_bundle(self.config, None, local_only=True)\n for field, value in [(\"local_only\", False), (\"local_only\", 1),\n (\"provider_input\", PAYLOAD), (\"provider_input\", {}),\n (\"compiled_input_sha256\", \"0\" * 64),\n (\"provider_input_status\", \"prepared\"),\n (\"insert_policy\", \"new_video_track_preserve_baseline_audio\")]:\n with self.subTest(field=field), self.assertRaises(ValueError):\n validate_application_bundle({**bundle, field: value})\n without_flag = {k: v for k, v in bundle.items() if k != \"local_only\"}\n with self.assertRaises(ValueError):\n validate_application_bundle(without_flag)\n normal = build_application_bundle(self.config, PAYLOAD)\n with self.assertRaises(ValueError):\n validate_application_bundle({**normal, \"local_only\": True})\n\n def test_local_only_still_revalidates_native_evidence(self):\n bundle = build_application_bundle(self.config, None, local_only=True)\n before = copy.deepcopy(self.config)\n self.assertEqual(build_application_bundle(self.config, None, local_only=True), bundle)\n self.assertEqual(self.config, before)\n Path(self.source[\"path\"]).write_bytes(b\"changed media\")\n with self.assertRaises(ValueError):\n validate_application_bundle(bundle)\n\n def test_normal_bundle_default_false_retains_legacy_behavior(self):\n normal = build_application_bundle(self.config, PAYLOAD)\n explicit = build_application_bundle(self.config, PAYLOAD, local_only=False)\n self.assertEqual(normal, explicit)\n self.assertNotIn(\"local_only\", normal)\n self.assertNotIn(\"provider_input_status\", normal)\n with self.assertRaises(ValueError):\n build_application_bundle(self.config, None, local_only=False)\n for local_only in [False, \"omitted\"]:\n request = {\"integration\": self.config, \"brief\": \"Synthetic\", \"style_steer\": \"Synthetic\"}\n if local_only is False:\n request[\"local_only\"] = False\n proc, output = self.run_local_cli(request, suffix=f\"normal-{local_only}\")\n self.assertEqual(proc.returncode, 2)\n self.assertFalse(output.exists())\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":30206,"content_sha256":"3f3a54302c7470efe84b74b7404352e6771ca1d2c996eba91b03c177914733f9"},{"filename":"tests/test_interview.py","content":"\"\"\"Failing-first tests for the deterministic taste interview/profile.\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nimport unittest\nfrom pathlib import Path\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nfrom tasteforge import interview, schema # noqa: E402\n\n\nclass QuestionSetTests(unittest.TestCase):\n def test_questions_cover_required_axes(self):\n ids = {q.id for q in interview.QUESTIONS}\n for required in (\n \"palette\", \"grain\", \"lighting\", \"focal_length\", \"camera_motion\",\n \"subject_framing\", \"grade_description\", \"mood_adjectives\", \"avoid\",\n \"brief\",\n ):\n self.assertIn(required, ids)\n\n def test_every_question_has_prompt_and_id(self):\n for q in interview.QUESTIONS:\n self.assertTrue(q.id)\n self.assertTrue(q.prompt)\n\n\nclass ConductTests(unittest.TestCase):\n def _answers(self):\n return {\n \"palette\": \"near-black void with bone-white highlights and one violet bloom\",\n \"grain\": \"fine 35mm grain\",\n \"lighting\": \"single hard key, backgrounds unlit\",\n \"focal_length\": \"35mm, mild compression\",\n \"camera_motion\": \"locked off with slow push-ins\",\n \"subject_framing\": \"centered subjects, generous headroom\",\n \"grade_description\": \"crushed blacks, blown highlights, cool mids\",\n \"mood_adjectives\": \"holy, crystalline, distant\",\n \"avoid\": \"over-saturated skin, plastic highlights, drifting camera\",\n \"brief\": \"a courier weaves through night traffic\",\n }\n\n def test_conduct_produces_valid_profile(self):\n profile = interview.conduct(self._answers(), genre=\"flashethereal\")\n problems = schema.validate(profile, schema.TASTE_PROFILE_SCHEMA)\n self.assertEqual(problems, [])\n self.assertEqual(profile[\"genre\"], \"flashethereal\")\n\n def test_missing_answers_are_flagged_not_invented(self):\n answers = self._answers()\n del answers[\"lighting\"]\n profile = interview.conduct(answers, genre=\"flashethereal\")\n self.assertIn(\"lighting\", profile[\"unanswered\"])\n self.assertNotIn(\"lighting\", profile[\"constraints\"][\"look\"])\n # but the profile is still schema-valid\n self.assertEqual(schema.validate(profile, schema.TASTE_PROFILE_SCHEMA), [])\n\n def test_profile_deterministic(self):\n a = interview.conduct(self._answers(), genre=\"g\")\n b = interview.conduct(self._answers(), genre=\"g\")\n a.pop(\"created\"), b.pop(\"created\")\n self.assertEqual(a, b)\n\n def test_constraints_split_look_and_content(self):\n profile = interview.conduct(self._answers(), genre=\"flashethereal\")\n look = profile[\"constraints\"][\"look\"]\n self.assertIn(\"lighting\", look)\n self.assertIn(\"mood_adjectives\", look)\n self.assertIn(\"avoid\", look)\n self.assertEqual(\n profile[\"constraints\"][\"content\"][\"brief\"],\n \"a courier weaves through night traffic\",\n )\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":3119,"content_sha256":"ea9be4c613366a2d3d34f4be3d260eddb6dd931437eaabac2f4fada379647342"},{"filename":"tests/test_media_utilities.py","content":"import io\nimport json\nimport pathlib\nimport shutil\nimport subprocess\nimport sys\nimport tempfile\nimport unittest\nfrom unittest.mock import Mock, call, patch\n\nsys.path.insert(0, str(pathlib.Path(__file__).parents[1] / \"scripts\"))\nfrom tasteforge.media import capcut, stills\n\n\nclass MediaUtilitiesTests(unittest.TestCase):\n def test_reject_bad_geometry_before_execution(self):\n runner = Mock()\n for kwargs in ({\"width\": 0}, {\"fps\": float(\"nan\")}, {\"duration\": -1}):\n with self.assertRaises(ValueError):\n stills.make_clip(\"missing.png\", \"out.mp4\", runner=runner, **kwargs)\n runner.assert_not_called()\n\n def test_still_failure_propagates_and_preserves_destination(self):\n with tempfile.TemporaryDirectory() as directory:\n source = pathlib.Path(directory) / \"in.png\"\n source.touch()\n target = pathlib.Path(directory) / \"out.mp4\"\n target.write_bytes(b\"old\")\n runner = Mock(side_effect=subprocess.CalledProcessError(1, \"ffmpeg\"))\n with self.assertRaises(FileExistsError):\n stills.make_clip(source, target, runner=runner)\n runner.assert_not_called()\n with self.assertRaises(subprocess.CalledProcessError):\n stills.make_clip(source, target, overwrite=True, runner=runner)\n self.assertEqual(target.read_bytes(), b\"old\")\n\n def test_crop_changes_filter(self):\n first = stills.filter_graph(\n (0, 0, 1, 1), (0.1, 0.1, 0.8, 0.8), 60, 320, 180, 30\n )\n second = stills.filter_graph(\n (0.2, 0.1, 0.5, 0.6), (0.1, 0.1, 0.8, 0.8), 60, 320, 180, 30\n )\n self.assertNotEqual(first, second)\n\n def test_capcut_preflight_before_draft_creation(self):\n cc = Mock()\n with self.assertRaises(ValueError):\n capcut.export_draft([], \"/tmp/drafts\", \"../escape\", cc=cc)\n cc.DraftFolder.assert_not_called()\n\n def test_capcut_overwrite_refused_before_app_mutation(self):\n cc = Mock()\n with self.assertRaises(ValueError):\n capcut.export_draft(\n [\"clip.mp4\"], \"/tmp/drafts\", \"draft\", overwrite=True, cc=cc\n )\n cc.DraftFolder.assert_not_called()\n\n def test_crop_height_changes_aspect_fit(self):\n first = stills.filter_graph(\n (0.25, 0.25, 0.5, 0.5), (0, 0, 1, 1), 60, 320, 180, 30\n )\n second = stills.filter_graph(\n (0.25, 0.125, 0.5, 0.75), (0, 0, 1, 1), 60, 320, 180, 30\n )\n self.assertNotEqual(first, second)\n\n def test_capcut_duration_failure_prevents_creation(self):\n cc = Mock()\n with tempfile.TemporaryDirectory() as directory:\n source = pathlib.Path(directory) / \"clip.mp4\"\n source.touch()\n with self.assertRaises(ValueError):\n capcut.export_draft(\n [source], directory, \"draft\", cc=cc, probe=lambda _: float(\"nan\")\n )\n cc.DraftFolder.assert_not_called()\n\n @unittest.skipUnless(\n shutil.which(\"ffmpeg\") and shutil.which(\"ffprobe\"), \"FFmpeg required\"\n )\n def test_real_ffmpeg_still_and_feedback(self):\n try:\n import PIL # noqa: F401\n from tasteforge.media.glitch import render\n except ImportError:\n self.skipTest(\"numpy and Pillow required for effect smoke\")\n with tempfile.TemporaryDirectory() as directory:\n root = pathlib.Path(directory)\n image = root / \"source.ppm\"\n image.write_bytes(b\"P6\\n32 18\\n255\\n\" + bytes([100, 20, 200]) * 32 * 18)\n clip = root / \"still.mp4\"\n stills.make_clip(image, clip, duration=0.5, width=32, height=18, fps=10)\n final = root / \"feedback.mp4\"\n self.assertEqual(render(clip, 0, 0.5, final, 42, \"feedback\", 32, 18, 10), 5)\n result = subprocess.run(\n [\n \"ffprobe\",\n \"-v\",\n \"error\",\n \"-count_frames\",\n \"-show_entries\",\n \"stream=width,height,nb_read_frames,r_frame_rate\",\n \"-of\",\n \"json\",\n str(final),\n ],\n check=True,\n capture_output=True,\n text=True,\n )\n stream = json.loads(result.stdout)[\"streams\"][0]\n self.assertEqual(\n (stream[\"width\"], stream[\"height\"], stream[\"nb_read_frames\"]),\n (32, 18, \"5\"),\n )\n self.assertEqual(stream[\"r_frame_rate\"], \"10/1\")\n\n def test_empty_decoder_fails(self):\n try:\n from tasteforge.media.glitch import transform_stream\n except ImportError:\n self.skipTest(\"numpy required\")\n with self.assertRaises(RuntimeError):\n transform_stream(\n Mock(stdout=io.BytesIO()),\n Mock(stdin=io.BytesIO()),\n 1,\n 32,\n 18,\n 30,\n 42,\n \"drift\",\n )\n\n def test_seeded_effect_is_repeatable(self):\n try:\n from tasteforge.media.glitch import transform_stream\n except ImportError:\n self.skipTest(\"numpy required\")\n frame = bytes(range(256)) * (32 * 18 * 3 // 256) + bytes(range(192))\n results = []\n for _ in range(2):\n destination = io.BytesIO()\n transform_stream(\n Mock(stdout=io.BytesIO(frame)),\n Mock(stdin=destination),\n 1,\n 32,\n 18,\n 30,\n 42,\n \"drift\",\n )\n results.append(destination.getvalue())\n self.assertEqual(results[0], results[1])\n\n def test_capcut_success_preserves_order_and_frame_rate(self):\n cc = Mock()\n with tempfile.TemporaryDirectory() as directory:\n sources = [\n pathlib.Path(directory) / name for name in (\"first.mp4\", \"second.mp4\")\n ]\n for source in sources:\n source.touch()\n receipt = capcut.export_draft(\n sources,\n directory,\n \"Review 2\",\n width=320,\n height=180,\n fps=24,\n cc=cc,\n probe=Mock(side_effect=[1.25, 2.5]),\n )\n cc.DraftFolder.return_value.create_draft.assert_called_once_with(\n \"Review 2\", 320, 180, fps=24, allow_replace=False\n )\n self.assertEqual(\n cc.trange.call_args_list,\n [call(\"0.000000s\", \"1.250000s\"), call(\"1.250000s\", \"2.500000s\")],\n )\n self.assertEqual(\n [entry.args[0] for entry in cc.VideoSegment.call_args_list],\n [str(source.resolve()) for source in sources],\n )\n self.assertEqual(receipt[\"duration\"], 3.75)\n self.assertEqual(receipt[\"segments\"], 2)\n cc.DraftFolder.return_value.create_draft.return_value.save.assert_called_once()\n\n def test_capcut_save_failure_is_not_reported_as_success(self):\n cc = Mock()\n cc.DraftFolder.return_value.create_draft.return_value.save.side_effect = (\n OSError(\"disk full\")\n )\n with tempfile.TemporaryDirectory() as directory:\n source = pathlib.Path(directory) / \"clip.mp4\"\n source.touch()\n with self.assertRaisesRegex(OSError, \"disk full\"):\n capcut.export_draft(\n [source], directory, \"new-draft\", cc=cc, probe=lambda _: 1\n )\n\n def test_capcut_empty_list_rejected(self):\n with self.assertRaises(ValueError):\n capcut.export_draft([], \"/tmp/drafts\", \"valid-name\", cc=Mock())\n\n def test_concat_paths_resolve_against_list_not_current_directory(self):\n with tempfile.TemporaryDirectory() as directory:\n concat = pathlib.Path(directory) / \"concat.txt\"\n concat.write_text(\"# heading\\nfile 'with space.mp4'\\n\\nfile second.mp4\\n\")\n self.assertEqual(\n capcut.read_concat(concat),\n [\n (pathlib.Path(directory) / \"with space.mp4\").resolve(),\n (pathlib.Path(directory) / \"second.mp4\").resolve(),\n ],\n )\n concat.write_text(\"file first.mp4 unexpected\\n\")\n with self.assertRaises(ValueError):\n capcut.read_concat(concat)\n\n def test_probe_failure_and_valid_duration(self):\n with patch.object(\n capcut.subprocess, \"run\", return_value=Mock(stdout=\"1.125\\n\")\n ) as runner:\n self.assertEqual(capcut.duration_of(\"source.mp4\"), 1.125)\n self.assertTrue(runner.call_args.kwargs[\"check\"])\n with patch.object(\n capcut.subprocess,\n \"run\",\n side_effect=subprocess.CalledProcessError(1, \"ffprobe\"),\n ):\n with self.assertRaises(subprocess.CalledProcessError):\n capcut.duration_of(\"source.mp4\")\n\n def test_media_cli_parameter_forwarding(self):\n with patch.object(stills, \"make_clip\") as renderer:\n stills.main(\n [\n \"source.png\",\n \"final.mp4\",\n \"--duration\",\n \".5\",\n \"--width\",\n \"320\",\n \"--height\",\n \"180\",\n \"--fps\",\n \"24\",\n \"--overwrite\",\n ]\n )\n self.assertEqual(renderer.call_args.kwargs[\"fps\"], 24)\n self.assertTrue(renderer.call_args.kwargs[\"overwrite\"])\n with (\n patch.object(capcut, \"read_concat\", return_value=[\"source.mp4\"]),\n patch.object(capcut, \"export_draft\") as exporter,\n ):\n capcut.main(\n [\n \"list.txt\",\n \"--drafts\",\n \"/tmp/drafts\",\n \"--name\",\n \"review\",\n \"--fps\",\n \"24\",\n ]\n )\n self.assertEqual(exporter.call_args.kwargs[\"files\"], [\"source.mp4\"])\n self.assertEqual(exporter.call_args.kwargs[\"name\"], \"review\")\n self.assertEqual(exporter.call_args.kwargs[\"fps\"], 24)\n\n def test_crop_bounds_missing_source_and_odd_geometry(self):\n for crop in ((0, 0, 1), (0, 0, float(\"nan\"), 1), (0.5, 0, 1, 1)):\n with self.assertRaises(ValueError):\n stills.filter_graph(crop, (0, 0, 1, 1), 10, 320, 180, 30)\n with self.assertRaises(ValueError):\n stills.make_clip(\"missing.png\", \"out.mp4\", width=319)\n with self.assertRaises(FileNotFoundError):\n stills.make_clip(\"missing.png\", \"out.mp4\")\n\n def test_output_empty_rejected_and_successful_replace_atomic(self):\n from tasteforge.media.common import output_file\n\n with tempfile.TemporaryDirectory() as directory:\n target = pathlib.Path(directory) / \"out.mp4\"\n target.write_bytes(b\"previous\")\n with self.assertRaises(RuntimeError):\n with output_file(target, overwrite=True):\n pass\n self.assertEqual(target.read_bytes(), b\"previous\")\n with output_file(target, overwrite=True) as temporary:\n temporary.write_bytes(b\"complete\")\n self.assertEqual(target.read_bytes(), b\"previous\")\n self.assertEqual(target.read_bytes(), b\"complete\")\n self.assertEqual(list(pathlib.Path(directory).glob(\".media-*\")), [])\n\n def test_glitch_validation_and_cli(self):\n try:\n from tasteforge.media import glitch\n except ImportError:\n self.skipTest(\"numpy required\")\n with tempfile.TemporaryDirectory() as directory:\n source = pathlib.Path(directory) / \"source.mp4\"\n source.touch()\n for options in ({\"start\": -1}, {\"mode\": \"unknown\"}, {\"seed\": -1}):\n params = dict(\n source=source, start=0, duration=1, output=\"out.mp4\", seed=42\n )\n params.update(options)\n with self.assertRaises(ValueError):\n glitch.render(**params)\n with self.assertRaises(FileNotFoundError):\n glitch.render(source / \"missing\", 0, 1, \"out.mp4\", 42)\n with (\n patch.object(glitch, \"render\", return_value=12) as renderer,\n patch(\"builtins.print\"),\n ):\n glitch.main(\n [\"source.mp4\", \"1\", \".5\", \"out.mp4\", \"42\", \"mosh\", \"--fps\", \"24\"]\n )\n self.assertEqual(renderer.call_args.kwargs[\"mode\"], \"mosh\")\n self.assertEqual(renderer.call_args.kwargs[\"fps\"], 24)\n\n def test_glitch_truncated_frame_and_mosh_are_explicit(self):\n try:\n from tasteforge.media.glitch import transform_stream\n except ImportError:\n self.skipTest(\"numpy required\")\n with self.assertRaisesRegex(RuntimeError, \"truncated\"):\n transform_stream(\n Mock(stdout=io.BytesIO(b\"truncated\")),\n Mock(stdin=io.BytesIO()),\n 1,\n 32,\n 18,\n 30,\n 42,\n \"drift\",\n )\n frame = bytes(range(256)) * 6 + bytes(range(192))\n destination = io.BytesIO()\n self.assertEqual(\n transform_stream(\n Mock(stdout=io.BytesIO(frame * 3)),\n Mock(stdin=destination),\n 0.1,\n 32,\n 18,\n 30,\n 42,\n \"mosh\",\n ),\n 3,\n )\n self.assertEqual(len(destination.getvalue()), len(frame) * 3)\n self.assertNotEqual(destination.getvalue(), frame * 3)\n\n def test_encoder_launch_failure_reaps_decoder(self):\n try:\n from tasteforge.media import glitch\n except ImportError:\n self.skipTest(\"numpy required\")\n decoder = Mock(stdin=None, stdout=io.BytesIO(), poll=Mock(return_value=None))\n with tempfile.TemporaryDirectory() as directory:\n source = pathlib.Path(directory) / \"source.mp4\"\n source.touch()\n target = pathlib.Path(directory) / \"out.mp4\"\n with patch.object(\n glitch.subprocess,\n \"Popen\",\n side_effect=[decoder, OSError(\"encoder unavailable\")],\n ):\n with self.assertRaisesRegex(OSError, \"encoder unavailable\"):\n glitch.render(source, 0, 1, target, 42)\n decoder.kill.assert_called_once()\n decoder.wait.assert_called_once()\n self.assertTrue(decoder.stdout.closed)\n self.assertFalse(target.exists())\n\n def test_still_cannot_replace_source_through_same_path_or_symlink(self):\n with tempfile.TemporaryDirectory() as directory:\n source = pathlib.Path(directory) / \"source.png\"\n source.write_bytes(b\"original\")\n alias = pathlib.Path(directory) / \"alias.png\"\n alias.symlink_to(source)\n renderer = Mock()\n for output in (source, alias):\n with self.assertRaises(ValueError):\n stills.make_clip(source, output, overwrite=True, runner=renderer)\n renderer.assert_not_called()\n self.assertEqual(source.read_bytes(), b\"original\")\n self.assertTrue(alias.is_symlink())\n\n def test_glitch_cannot_replace_source_through_same_path_or_symlink(self):\n try:\n from tasteforge.media import glitch\n except ImportError:\n self.skipTest(\"numpy required\")\n with tempfile.TemporaryDirectory() as directory:\n source = pathlib.Path(directory) / \"source.mp4\"\n source.write_bytes(b\"original\")\n alias = pathlib.Path(directory) / \"alias.mp4\"\n alias.symlink_to(source)\n with patch.object(glitch.subprocess, \"Popen\") as process:\n for output in (source, alias):\n with self.assertRaises(ValueError):\n glitch.render(source, 0, 1, output, 42, overwrite=True)\n process.assert_not_called()\n self.assertEqual(source.read_bytes(), b\"original\")\n self.assertTrue(alias.is_symlink())\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":16706,"content_sha256":"87a7de6d38b1ab2e455ecc5517fa8e4bff5c2d0c0cf71687b0274d0f3f169308"},{"filename":"tests/test_multimodal_contract.py","content":"import hashlib\nimport json\nimport tempfile\nimport unittest\nfrom pathlib import Path\n\nfrom tasteforge.contract import (\n ContractError,\n validate_artifact_receipt,\n validate_effect_recipe,\n validate_genre_specs,\n validate_manifests,\n validate_provenance,\n)\n\n\nclass GenreContractTests(unittest.TestCase):\n def test_genre_spec_requires_explicit_dry_run_true(self):\n spec = {\n \"number\": 1,\n \"slug\": \"flash-ethereal\",\n \"style_fingerprint\": \"a\" * 64,\n \"signature\": {\n \"materials\": [\"glass bloom\"],\n \"motion\": [\"hard-cut flash\"],\n \"composition\": [\"centered subject\"],\n \"avoid\": [\"muddy shadows\"],\n },\n \"dry_run\": False,\n }\n with self.assertRaisesRegex(ContractError, \"dry-run|dry_run\"):\n validate_genre_specs([spec])\n\n def test_empty_avoid_signature_is_rejected(self):\n spec = {\n \"number\": 1,\n \"slug\": \"flash-ethereal\",\n \"style_fingerprint\": \"a\" * 64,\n \"signature\": {\n \"materials\": [\"glass bloom\"],\n \"motion\": [\"hard-cut flash\"],\n \"composition\": [\"centered subject\"],\n \"avoid\": [],\n },\n \"dry_run\": True,\n }\n with self.assertRaisesRegex(ContractError, \"avoid|empty\"):\n validate_genre_specs([spec])\n\n def test_collapsing_references_into_one_generic_style_is_rejected(self):\n generic = {\n \"signature\": {\n \"materials\": [\"cinematic\"],\n \"motion\": [\"dynamic\"],\n \"composition\": [\"beautiful\"],\n \"avoid\": [],\n },\n \"style_fingerprint\": \"same\",\n }\n specs = [\n {**generic, \"number\": 1, \"slug\": \"flash-ethereal\"},\n {**generic, \"number\": 2, \"slug\": \"3d-cyber-glitch\"},\n {**generic, \"number\": 3, \"slug\": \"fluid-sketch\"},\n ]\n with self.assertRaisesRegex(ContractError, \"collapsed|distinct\"):\n validate_genre_specs(specs)\n\n\nclass ResolveRecipeContractTests(unittest.TestCase):\n def _valid_recipe(self):\n recipe = {\n \"dry_run\": True,\n \"provider_calls\": 0,\n \"provider_execution\": False,\n \"seed\": 41,\n \"rng_algorithm\": \"python.random.Random/v1\",\n \"periodic\": False,\n \"timeline_duration\": 6.0,\n \"events\": [\n {\"time\": 0.2, \"duration\": 0.2, \"effect\": \"bloom\", \"placement\": self._placement()},\n {\"time\": 1.1, \"duration\": 0.2, \"effect\": \"bloom\", \"placement\": self._placement()},\n {\"time\": 2.7, \"duration\": 0.2, \"effect\": \"bloom\", \"placement\": self._placement()},\n {\"time\": 5.5, \"duration\": 0.2, \"effect\": \"bloom\", \"placement\": self._placement()},\n ],\n }\n for event in recipe[\"events\"]:\n event[\"evidence\"] = {\n \"reference_sha256\": \"a\" * 64,\n \"time\": 0.5,\n \"source_duration\": 6.0,\n }\n return recipe\n\n def test_numeric_timeline_and_evidence_values_must_be_finite_reals(self):\n cases = (\n (\"timeline_duration\", None, float(\"nan\")),\n (\"timeline_duration\", None, float(\"inf\")),\n (\"timeline_duration\", None, True),\n (\"time\", 0, float(\"nan\")),\n (\"time\", 0, float(\"inf\")),\n (\"time\", 0, True),\n (\"duration\", 0, float(\"nan\")),\n (\"duration\", 0, float(\"inf\")),\n (\"duration\", 0, True),\n )\n for field, event_index, unsafe in cases:\n with self.subTest(field=field, unsafe=unsafe):\n recipe = self._valid_recipe()\n target = recipe if event_index is None else recipe[\"events\"][event_index]\n target[field] = unsafe\n with self.assertRaisesRegex(ContractError, \"finite|timeline|duration|start\"):\n validate_effect_recipe(recipe)\n\n def test_effect_evidence_time_must_be_within_finite_source_duration(self):\n for field, unsafe in (\n (\"time\", float(\"nan\")),\n (\"time\", float(\"inf\")),\n (\"time\", True),\n (\"time\", 6.1),\n (\"source_duration\", float(\"nan\")),\n (\"source_duration\", float(\"inf\")),\n (\"source_duration\", True),\n ):\n with self.subTest(field=field, unsafe=unsafe):\n recipe = self._valid_recipe()\n recipe[\"events\"][0][\"evidence\"][field] = unsafe\n with self.assertRaisesRegex(ContractError, \"evidence|source duration\"):\n validate_effect_recipe(recipe)\n\n def test_effect_recipe_requires_exact_disabled_provider_state(self):\n for field, unsafe in (\n (\"dry_run\", False),\n (\"provider_calls\", 1),\n (\"provider_calls\", False),\n (\"provider_execution\", True),\n ):\n with self.subTest(field=field):\n recipe = self._valid_recipe()\n recipe[field] = unsafe\n with self.assertRaisesRegex(ContractError, \"dry-run|provider\"):\n validate_effect_recipe(recipe)\n\n def test_anchor_evidence_time_must_be_within_finite_source_duration(self):\n for field, unsafe in (\n (\"evidence_time\", float(\"nan\")),\n (\"evidence_time\", float(\"inf\")),\n (\"evidence_time\", True),\n (\"evidence_time\", 6.1),\n (\"source_duration\", float(\"nan\")),\n (\"source_duration\", float(\"inf\")),\n (\"source_duration\", True),\n ):\n with self.subTest(field=field, unsafe=unsafe):\n recipe = self._valid_recipe()\n event = recipe[\"events\"][1]\n event.update({\n \"effect\": \"cv_wireframe_lock\",\n \"requires_subject_anchor\": True,\n \"subject_anchor\": {\n \"mode\": \"segmentation_track\",\n \"target\": \"primary_subject\",\n \"source_ref_sha256\": \"a\" * 64,\n \"evidence_time\": 0.5,\n \"source_duration\": 6.0,\n \"lost_policy\": \"disable_effect_until_track_recovers\",\n },\n })\n event[\"subject_anchor\"][field] = unsafe\n with self.assertRaisesRegex(ContractError, \"anchor evidence|source duration\"):\n validate_effect_recipe(recipe)\n\n def test_event_start_before_zero_is_rejected(self):\n recipe = self._valid_recipe()\n recipe[\"events\"][0][\"time\"] = -0.01\n with self.assertRaisesRegex(ContractError, \"timeline|start\"):\n validate_effect_recipe(recipe)\n\n def test_event_end_after_timeline_is_rejected(self):\n recipe = self._valid_recipe()\n recipe[\"events\"][-1].update({\"time\": 5.9, \"duration\": 0.2})\n with self.assertRaisesRegex(ContractError, \"timeline|end\"):\n validate_effect_recipe(recipe)\n\n def test_cv_anchor_continue_without_anchor_policy_is_rejected(self):\n recipe = self._valid_recipe()\n recipe[\"events\"][1].update({\n \"effect\": \"cv_wireframe_lock\",\n \"requires_subject_anchor\": True,\n \"subject_anchor\": {\n \"mode\": \"segmentation_track\",\n \"target\": \"primary_subject\",\n \"source_ref_sha256\": \"a\" * 64,\n \"evidence_time\": 0.0,\n \"lost_policy\": \"continue_without_anchor\",\n },\n })\n with self.assertRaisesRegex(ContractError, \"lost|anchor|fail\"):\n validate_effect_recipe(recipe)\n\n def test_repeating_interval_cycle_is_rejected_as_periodic(self):\n recipe = {\n \"seed\": 41,\n \"rng_algorithm\": \"python.random.Random/v1\",\n \"periodic\": False,\n \"timeline_duration\": 8.0,\n \"events\": [\n {\"time\": 1.0, \"effect\": \"bloom\"},\n {\"time\": 2.0, \"effect\": \"bloom\"},\n {\"time\": 4.0, \"effect\": \"bloom\"},\n {\"time\": 5.0, \"effect\": \"bloom\"},\n {\"time\": 7.0, \"effect\": \"bloom\"},\n ],\n }\n recipe = self._complete_recipe(recipe)\n with self.assertRaisesRegex(ContractError, \"periodic\"):\n validate_effect_recipe(recipe)\n\n def test_seed_without_declared_rng_algorithm_is_rejected(self):\n recipe = {\n \"seed\": 41,\n \"periodic\": False,\n \"events\": [\n {\"time\": 1.0, \"effect\": \"bloom\"},\n {\"time\": 2.2, \"effect\": \"bloom\"},\n {\"time\": 4.9, \"effect\": \"bloom\"},\n {\"time\": 8.3, \"effect\": \"bloom\"},\n ],\n }\n recipe = self._complete_recipe(recipe)\n with self.assertRaisesRegex(ContractError, \"seed|algorithm\"):\n validate_effect_recipe(recipe)\n\n def test_unseeded_schedule_is_rejected(self):\n recipe = {\n \"periodic\": False,\n \"rng_algorithm\": \"python.random.Random/v1\",\n \"events\": [\n {\"time\": 1.0, \"effect\": \"bloom\"},\n {\"time\": 2.2, \"effect\": \"bloom\"},\n {\"time\": 4.9, \"effect\": \"bloom\"},\n ],\n }\n recipe = self._complete_recipe(recipe)\n with self.assertRaisesRegex(ContractError, \"seed\"):\n validate_effect_recipe(recipe)\n\n def test_cv_effect_with_placeholder_anchor_is_rejected(self):\n recipe = {\n \"seed\": 41,\n \"rng_algorithm\": \"python.random.Random/v1\",\n \"periodic\": False,\n \"timeline_duration\": 9.0,\n \"events\": [\n {\"time\": 1.0, \"duration\": 0.2, \"effect\": \"bloom\"},\n {\n \"time\": 2.2,\n \"duration\": 0.2,\n \"effect\": \"cv_boxes\",\n \"requires_subject_anchor\": True,\n \"subject_anchor\": {\"mode\": \"frame_center\"},\n },\n {\"time\": 4.9, \"duration\": 0.2, \"effect\": \"bloom\"},\n {\"time\": 8.3, \"duration\": 0.2, \"effect\": \"bloom\"},\n ],\n }\n recipe = self._complete_recipe(recipe)\n with self.assertRaisesRegex(ContractError, \"subject anchor\"):\n validate_effect_recipe(recipe)\n\n def test_cv_effect_cannot_bypass_anchor_by_clearing_requirement_flag(self):\n recipe = {\n \"seed\": 41,\n \"rng_algorithm\": \"python.random.Random/v1\",\n \"periodic\": False,\n \"timeline_duration\": 9.0,\n \"events\": [\n {\"time\": 1.0, \"duration\": 0.2, \"effect\": \"bloom\", \"placement\": self._placement()},\n {\"time\": 2.2, \"duration\": 0.2, \"effect\": \"cv_wireframe_lock\",\n \"requires_subject_anchor\": False, \"placement\": self._placement()},\n {\"time\": 4.9, \"duration\": 0.2, \"effect\": \"bloom\", \"placement\": self._placement()},\n {\"time\": 8.3, \"duration\": 0.2, \"effect\": \"bloom\", \"placement\": self._placement()},\n ],\n }\n recipe = self._complete_recipe(recipe)\n with self.assertRaisesRegex(ContractError, \"subject anchor\"):\n validate_effect_recipe(recipe)\n\n def _complete_recipe(self, recipe):\n recipe.update({\n \"dry_run\": True,\n \"provider_calls\": 0,\n \"provider_execution\": False,\n })\n recipe.setdefault(\"timeline_duration\", 10.0)\n for event in recipe[\"events\"]:\n event.setdefault(\"duration\", 0.2)\n event.setdefault(\"placement\", self._placement())\n event.setdefault(\"evidence\", {\n \"reference_sha256\": \"a\" * 64,\n \"time\": 0.5,\n \"source_duration\": 6.0,\n })\n return recipe\n\n @staticmethod\n def _placement():\n return {\n \"safe_area\": 0.08,\n \"max_coverage\": 0.35,\n \"occlusion_policy\": \"preserve_subject_face_and_readable_type\",\n }\n\n\nclass ProvenanceContractTests(unittest.TestCase):\n def test_reference_evidence_times_must_be_finite_and_within_source_duration(self):\n for field, unsafe in (\n (\"times\", [float(\"nan\")]),\n (\"times\", [float(\"inf\")]),\n (\"times\", [True]),\n (\"times\", [6.1]),\n (\"source_duration\", float(\"nan\")),\n (\"source_duration\", float(\"inf\")),\n (\"source_duration\", True),\n ):\n with self.subTest(field=field, unsafe=unsafe):\n evidence = {\n \"reference_sha256\": \"a\" * 64,\n \"times\": [0.5],\n \"source_duration\": 6.0,\n }\n evidence[field] = unsafe\n payload = {\"rules\": [{\n \"rule_id\": \"genre-1-materials\",\n \"rule\": [\"glass bloom\"],\n \"evidence\": [evidence],\n }]}\n with self.assertRaisesRegex(ContractError, \"time evidence|source duration\"):\n validate_provenance(payload)\n\n def test_rule_without_reference_time_evidence_is_rejected(self):\n payload = {\n \"rules\": [{\n \"rule_id\": \"genre-1-materials\",\n \"rule\": [\"glass bloom\"],\n \"evidence\": [{\"reference_sha256\": \"a\" * 64, \"times\": []}],\n }],\n }\n with self.assertRaisesRegex(ContractError, \"time evidence\"):\n validate_provenance(payload)\n\n\nclass ManifestContractTests(unittest.TestCase):\n def test_boolean_provider_calls_is_rejected_for_manifest_and_request(self):\n for unsafe_scope in (\"manifest\", \"request\"):\n with self.subTest(scope=unsafe_scope), tempfile.TemporaryDirectory() as tmp:\n root = Path(tmp)\n for modality in (\"image\", \"video\", \"3d_asset\"):\n request = {\n \"prompt\": modality,\n \"dry_run\": True,\n \"submit\": False,\n \"provider_calls\": False if unsafe_scope == \"request\" and modality == \"video\" else 0,\n \"provider_execution\": False,\n \"provider_call_mode\": \"disabled\",\n }\n payload = {\n \"modality\": modality,\n \"dry_run\": True,\n \"submit\": False,\n \"provider_calls\": False if unsafe_scope == \"manifest\" and modality == \"video\" else 0,\n \"provider_execution\": False,\n \"requests\": [request],\n }\n (root / f\"{modality}.json\").write_text(json.dumps(payload), encoding=\"utf-8\")\n with self.assertRaisesRegex(ContractError, \"dry-run boundary\"):\n validate_manifests(root)\n\n def test_request_with_dry_run_false_is_rejected(self):\n with tempfile.TemporaryDirectory() as tmp:\n root = Path(tmp)\n for modality in (\"image\", \"video\", \"3d_asset\"):\n (root / f\"{modality}.json\").write_text(json.dumps({\n \"modality\": modality,\n \"dry_run\": True,\n \"submit\": False,\n \"provider_calls\": 0,\n \"provider_execution\": False,\n \"requests\": [{\n \"prompt\": modality,\n \"dry_run\": modality != \"video\",\n \"submit\": False,\n \"provider_calls\": 0,\n \"provider_execution\": False,\n \"provider_call_mode\": \"disabled\",\n }],\n }), encoding=\"utf-8\")\n with self.assertRaisesRegex(ContractError, \"dry-run boundary\"):\n validate_manifests(root)\n\n def test_missing_3d_asset_manifest_is_rejected(self):\n with tempfile.TemporaryDirectory() as tmp:\n root = Path(tmp)\n for modality in (\"image\", \"video\"):\n (root / f\"{modality}.json\").write_text(json.dumps({\n \"modality\": modality,\n \"dry_run\": True,\n \"provider_calls\": 0,\n \"requests\": [{\"prompt\": modality}],\n }), encoding=\"utf-8\")\n with self.assertRaisesRegex(ContractError, \"3d_asset\"):\n validate_manifests(root)\n\n def test_manifest_with_provider_execution_enabled_is_rejected(self):\n with tempfile.TemporaryDirectory() as tmp:\n root = Path(tmp)\n for modality in (\"image\", \"video\", \"3d_asset\"):\n (root / f\"{modality}.json\").write_text(json.dumps({\n \"modality\": modality,\n \"dry_run\": True,\n \"provider_calls\": 0,\n \"provider_execution\": modality == \"video\",\n \"requests\": [{\n \"genre_number\": 1,\n \"style_fingerprint\": \"a\" * 64,\n \"prompt\": modality,\n \"submit\": False,\n \"provider_call_mode\": \"disabled\",\n \"provider_execution\": False,\n }],\n }), encoding=\"utf-8\")\n with self.assertRaisesRegex(ContractError, \"dry-run boundary\"):\n validate_manifests(root)\n\n\nclass ArtifactReceiptContractTests(unittest.TestCase):\n def test_unbound_emitted_artifact_is_rejected(self):\n with tempfile.TemporaryDirectory() as tmp:\n root = Path(tmp)\n (root / \"unbound.json\").write_text(\"{}\\n\", encoding=\"utf-8\")\n with self.assertRaisesRegex(ContractError, \"unbound emitted artifact\"):\n validate_artifact_receipt(root, {\"evidence_artifacts\": []})\n\n def test_provider_execution_or_missing_provenance_is_rejected(self):\n with tempfile.TemporaryDirectory() as tmp:\n root = Path(tmp)\n artifact = root / \"artifact.json\"\n artifact.write_text(\"{}\\n\", encoding=\"utf-8\")\n entry = {\n \"path\": \"artifact.json\",\n \"bytes\": artifact.stat().st_size,\n \"sha256\": \"a\" * 64,\n \"genre_numbers\": [],\n \"modalities\": [],\n \"provider_execution\": True,\n \"provenance\": [],\n }\n with self.assertRaisesRegex(ContractError, \"provider execution\"):\n validate_artifact_receipt(root, {\"evidence_artifacts\": [entry]})\n\n def test_artifact_byte_tampering_is_rejected(self):\n with tempfile.TemporaryDirectory() as tmp:\n root = Path(tmp)\n artifact = root / \"artifact.json\"\n artifact.write_text(\"{}\\n\", encoding=\"utf-8\")\n entry = {\n \"path\": \"artifact.json\",\n \"bytes\": artifact.stat().st_size,\n \"sha256\": \"0\" * 64,\n \"genre_numbers\": [1],\n \"modalities\": [\"image\"],\n \"provider_execution\": False,\n \"provenance\": [{\n \"reference_path\": \"/reference.mov\",\n \"reference_sha256\": \"b\" * 64,\n \"reference_times\": [0.5],\n \"time_basis\": \"media_seconds\",\n }],\n }\n with self.assertRaisesRegex(ContractError, \"SHA-256\"):\n validate_artifact_receipt(root, {\"evidence_artifacts\": [entry]})\n\n def test_unknown_provenance_source_is_rejected(self):\n with tempfile.TemporaryDirectory() as tmp:\n root = Path(tmp)\n artifact = root / \"artifact.json\"\n artifact.write_text(\"{}\\n\", encoding=\"utf-8\")\n digest = hashlib.sha256(artifact.read_bytes()).hexdigest()\n receipt = {\n \"references\": [],\n \"evidence_files\": [],\n \"evidence_artifacts\": [{\n \"path\": \"artifact.json\",\n \"bytes\": artifact.stat().st_size,\n \"sha256\": digest,\n \"genre_numbers\": [1],\n \"modalities\": [\"image\"],\n \"provider_execution\": False,\n \"provenance\": [{\n \"reference_path\": \"/unknown.mov\",\n \"reference_sha256\": \"b\" * 64,\n \"reference_times\": [0.5],\n \"time_basis\": \"media_seconds\",\n }],\n }],\n }\n with self.assertRaisesRegex(ContractError, \"unknown provenance source\"):\n validate_artifact_receipt(root, receipt)\n\n def test_receipt_digest_mismatch_is_rejected(self):\n with tempfile.TemporaryDirectory() as tmp:\n receipt = {\"evidence_artifacts\": [], \"receipt_sha256\": \"0\" * 64}\n with self.assertRaisesRegex(ContractError, \"receipt SHA-256\"):\n validate_artifact_receipt(tmp, receipt)\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":21158,"content_sha256":"581e73a1ad9be860cdfef62a055507fe17979f39a2a0af8e890c2a163fb1a43a"},{"filename":"tests/test_multimodal_workflow.py","content":"import hashlib\nimport json\nimport tempfile\nimport unittest\nfrom pathlib import Path\n\nfrom tasteforge.contract import ContractError, validate_bundle\nfrom tasteforge.workflow import parse_feature_output, run_workflow\n\n\nclass FeatureExtractionTests(unittest.TestCase):\n def test_ffmpeg_metadata_becomes_timestamped_style_and_scene_features(self):\n output = \"\"\"\nframe:0 pts:0 pts_time:0.75\nlavfi.signalstats.YAVG=51\nlavfi.signalstats.SATAVG=40\nlavfi.signalstats.HUEAVG=15\nlavfi.scene_score=0.42\nframe:1 pts:1 pts_time:2.25\nlavfi.signalstats.YAVG=204\nlavfi.signalstats.SATAVG=70\nlavfi.signalstats.HUEAVG=20\nlavfi.scene_score=0.08\n\"\"\"\n parsed = parse_feature_output(output)\n self.assertEqual(parsed[\"scene_changes\"], [0.75])\n self.assertEqual([sample[\"time\"] for sample in parsed[\"style_samples\"]], [0.75, 2.25])\n self.assertAlmostEqual(parsed[\"style_samples\"][0][\"luma\"], 0.2)\n self.assertAlmostEqual(parsed[\"style_samples\"][1][\"luma\"], 0.8)\n self.assertEqual(parsed[\"style_samples\"][0][\"hue\"], 15.0)\n\n\nclass MultimodalWorkflowTests(unittest.TestCase):\n def setUp(self):\n self.tmp = tempfile.TemporaryDirectory()\n self.root = Path(self.tmp.name)\n self.references = []\n for name, payload in (\n (\"flash.mov\", b\"flash-ethereal-reference\"),\n (\"cyber.mov\", b\"3d-cyber-glitch-reference\"),\n (\"fluid.mov\", b\"fluid-sketch-reference\"),\n ):\n path = self.root / name\n path.write_bytes(payload)\n self.references.append(path)\n self.editorial = self.root / \"FINAL_canvas.edl\"\n self.editorial.write_text(\"TITLE: FINAL_canvas\\nFCM: NON-DROP FRAME\\n\", encoding=\"utf-8\")\n\n self.config = {\n \"schema_version\": 1,\n \"run_id\": \"fixture-run\",\n \"seed\": 20260819,\n \"evidence_files\": [str(self.editorial)],\n \"genres\": [\n {\n \"number\": 1,\n \"slug\": \"flash-ethereal\",\n \"label\": \"Flash Ethereal\",\n \"references\": [str(self.references[0])],\n \"signature\": {\n \"materials\": [\"glass bloom\", \"white phosphor\"],\n \"motion\": [\"hard-cut flash\", \"slow orbital drift\"],\n \"composition\": [\"high-key centered subject\"],\n \"avoid\": [\"muddy shadows\"],\n },\n },\n {\n \"number\": 2,\n \"slug\": \"3d-cyber-glitch\",\n \"label\": \"3D Cyber Glitch\",\n \"references\": [str(self.references[1])],\n \"signature\": {\n \"materials\": [\"wireframe chrome\", \"scanline emissive\"],\n \"motion\": [\"depth orbit\", \"macroblock rupture\"],\n \"composition\": [\"full-frame 3D interstitial\"],\n \"avoid\": [\"decorative corner mesh\"],\n },\n },\n {\n \"number\": 3,\n \"slug\": \"fluid-sketch\",\n \"label\": \"Fluid Sketch\",\n \"references\": [str(self.references[2])],\n \"signature\": {\n \"materials\": [\"ink wash\", \"graphite edge\"],\n \"motion\": [\"nonlinear contour flow\", \"paper bleed\"],\n \"composition\": [\"negative-space drawing field\"],\n \"avoid\": [\"rigid neon grid\"],\n },\n },\n ],\n }\n self.config_path = self.root / \"workflow.json\"\n self.config_path.write_text(json.dumps(self.config), encoding=\"utf-8\")\n\n def tearDown(self):\n self.tmp.cleanup()\n\n @staticmethod\n def fake_probe(path: Path) -> dict:\n return {\n \"duration\": 6.0,\n \"width\": 1920,\n \"height\": 1080,\n \"fps\": 24.0,\n \"codec\": \"fixture\",\n \"sample_times\": [0.75, 2.25, 3.75, 5.25],\n \"style_samples\": [\n {\"time\": 0.75, \"luma\": 0.2, \"saturation\": 0.4},\n {\"time\": 2.25, \"luma\": 0.8, \"saturation\": 0.7},\n ],\n \"scene_changes\": [0.75, 2.25, 5.25],\n }\n\n def test_file_driven_run_emits_distinct_genres_and_all_modality_manifests(self):\n out = self.root / \"out\"\n receipt = run_workflow(self.config_path, out, probe=self.fake_probe)\n\n self.assertTrue(receipt[\"dry_run\"])\n self.assertEqual(receipt[\"provider_calls\"], 0)\n self.assertEqual(len(receipt[\"references\"]), 3)\n self.assertEqual(len(receipt[\"evidence_files\"]), 1)\n self.assertEqual(receipt[\"evidence_files\"][0][\"kind\"], \"editorial\")\n self.assertEqual(len(receipt[\"evidence_files\"][0][\"sha256\"]), 64)\n self.assertTrue(all(len(ref[\"sha256\"]) == 64 for ref in receipt[\"references\"]))\n emitted = {\n path.relative_to(out).as_posix()\n for path in out.rglob(\"*\")\n if path.is_file() and path.name != \"receipt.json\"\n }\n bound = {artifact[\"path\"] for artifact in receipt[\"evidence_artifacts\"]}\n self.assertEqual(bound, emitted)\n self.assertTrue({\n \"manifests/image.json\", \"manifests/video.json\", \"manifests/3d_asset.json\"\n }.issubset(bound))\n self.assertTrue(receipt[\"evidence_artifacts\"])\n for artifact in receipt[\"evidence_artifacts\"]:\n self.assertGreater(artifact[\"bytes\"], 0)\n self.assertEqual(len(artifact[\"sha256\"]), 64)\n self.assertIn(\"genre_numbers\", artifact)\n self.assertIn(\"modalities\", artifact)\n self.assertFalse(artifact[\"provider_execution\"])\n self.assertTrue(artifact[\"provenance\"])\n for source in artifact[\"provenance\"]:\n self.assertTrue(source[\"reference_path\"])\n self.assertEqual(len(source[\"reference_sha256\"]), 64)\n self.assertIn(\"reference_times\", source)\n self.assertIn(source[\"time_basis\"], {\"media_seconds\", \"whole_file\"})\n\n specs = [json.loads(path.read_text()) for path in sorted((out / \"genres\").glob(\"*.json\"))]\n self.assertEqual([spec[\"number\"] for spec in specs], [1, 2, 3])\n self.assertEqual(len({spec[\"style_fingerprint\"] for spec in specs}), 3)\n self.assertEqual({spec[\"label\"] for spec in specs}, {\n \"Flash Ethereal\", \"3D Cyber Glitch\", \"Fluid Sketch\"\n })\n for spec in specs:\n temporal = spec[\"measured_features\"][\"temporal\"]\n style = spec[\"measured_features\"][\"style\"]\n self.assertEqual(temporal[\"scene_change_count\"], 3)\n self.assertGreater(temporal[\"scene_interval_variance\"], 0)\n self.assertAlmostEqual(style[\"luma_mean\"], 0.5)\n self.assertAlmostEqual(style[\"saturation_mean\"], 0.55)\n\n for modality in (\"image\", \"video\", \"3d_asset\"):\n manifest = json.loads((out / \"manifests\" / f\"{modality}.json\").read_text())\n self.assertEqual(manifest[\"modality\"], modality)\n self.assertEqual(manifest.get(\"provider_calls\"), 0)\n self.assertIs(manifest.get(\"provider_execution\"), False)\n self.assertIs(manifest.get(\"dry_run\"), True)\n self.assertIs(manifest.get(\"submit\"), False)\n self.assertEqual(len(manifest[\"requests\"]), 3)\n self.assertTrue(all(request[\"prompt\"] for request in manifest[\"requests\"]))\n self.assertTrue(all(request[\"dry_run\"] for request in manifest[\"requests\"]))\n self.assertTrue(all(request[\"provider_call_mode\"] == \"disabled\" for request in manifest[\"requests\"]))\n self.assertTrue(all(request.get(\"provider_calls\") == 0 for request in manifest[\"requests\"]))\n self.assertTrue(all(request.get(\"provider_execution\") is False for request in manifest[\"requests\"]))\n self.assertTrue(all(request.get(\"dry_run\") is True for request in manifest[\"requests\"]))\n self.assertTrue(all(request.get(\"submit\") is False for request in manifest[\"requests\"]))\n self.assertTrue(all(request[\"endpoint_candidate\"] for request in manifest[\"requests\"]))\n self.assertTrue(all(request[\"request_body\"][\"prompt\"] == request[\"prompt\"]\n for request in manifest[\"requests\"]))\n\n validate_bundle(out)\n recipe = json.loads((out / \"resolve\" / \"effect_recipe.json\").read_text())\n self.assertEqual(recipe[\"seed\"], self.config[\"seed\"])\n self.assertFalse(recipe[\"periodic\"])\n cv_events = [event for event in recipe[\"events\"] if event[\"requires_subject_anchor\"]]\n self.assertTrue(cv_events)\n self.assertTrue(all(event[\"subject_anchor\"][\"source_ref_sha256\"] for event in cv_events))\n self.assertTrue(all(event[\"placement\"][\"max_coverage\"] \u003c= 0.35 for event in recipe[\"events\"]))\n\n def test_probe_and_hash_use_stable_bytes_and_fail_on_source_mutation(self):\n original = self.references[0].read_bytes()\n observed = []\n\n def mutating_probe(snapshot: Path) -> dict:\n observed.append(snapshot.read_bytes())\n self.references[0].write_bytes(b\"mutated-during-probe\")\n return self.fake_probe(snapshot)\n\n with self.assertRaisesRegex(ValueError, \"mutat|changed|stable\"):\n run_workflow(self.config_path, self.root / \"race-out\", probe=mutating_probe)\n\n self.assertEqual(observed, [original])\n\n def test_symlinked_output_root_is_rejected_before_writes(self):\n real_output = self.root / \"real-output\"\n real_output.mkdir()\n linked_output = self.root / \"linked-output\"\n linked_output.symlink_to(real_output, target_is_directory=True)\n\n with self.assertRaisesRegex(ValueError, \"symlink|output\"):\n run_workflow(self.config_path, linked_output, probe=self.fake_probe)\n\n self.assertEqual(list(real_output.iterdir()), [])\n\n def test_symlinked_output_intermediate_is_rejected_without_escape(self):\n out = self.root / \"out\"\n out.mkdir()\n victim = self.root / \"victim\"\n victim.mkdir()\n (out / \"manifests\").symlink_to(victim, target_is_directory=True)\n\n with self.assertRaisesRegex(ValueError, \"symlink|output\"):\n run_workflow(self.config_path, out, probe=self.fake_probe)\n\n self.assertEqual(list(victim.iterdir()), [])\n\n def test_non_directory_output_intermediate_is_rejected(self):\n out = self.root / \"out\"\n out.mkdir()\n (out / \"genres\").write_text(\"not a directory\", encoding=\"utf-8\")\n\n with self.assertRaisesRegex(ValueError, \"directory|output\"):\n run_workflow(self.config_path, out, probe=self.fake_probe)\n\n self.assertEqual((out / \"genres\").read_text(encoding=\"utf-8\"), \"not a directory\")\n\n def test_short_timeline_never_emits_out_of_bounds_forced_events(self):\n self.config[\"seed\"] = 15\n self.config[\"resolve_duration\"] = 6.0\n self.config_path.write_text(json.dumps(self.config), encoding=\"utf-8\")\n out = self.root / \"short-out\"\n\n run_workflow(self.config_path, out, probe=self.fake_probe)\n\n recipe = json.loads((out / \"resolve\" / \"effect_recipe.json\").read_text())\n timeline = recipe[\"timeline_duration\"]\n self.assertGreaterEqual(len(recipe[\"events\"]), 3)\n for event in recipe[\"events\"]:\n self.assertGreaterEqual(event[\"time\"], 0)\n self.assertLessEqual(event[\"time\"] + event[\"duration\"], timeline)\n\n def test_workflow_rejects_dry_run_false_before_output(self):\n self.config[\"dry_run\"] = False\n self.config_path.write_text(json.dumps(self.config), encoding=\"utf-8\")\n out = self.root / \"not-dry-run\"\n\n with self.assertRaisesRegex(ValueError, \"dry_run|dry-run\"):\n run_workflow(self.config_path, out, probe=self.fake_probe)\n\n self.assertFalse(out.exists())\n\n def test_bundle_receipt_requires_exact_disabled_provider_state(self):\n for field, unsafe in (\n (\"dry_run\", False),\n (\"provider_calls\", False),\n (\"provider_calls\", 1),\n (\"provider_execution\", True),\n ):\n with self.subTest(field=field, unsafe=unsafe):\n out = self.root / f\"receipt-provider-{field}-{unsafe!s}\"\n run_workflow(self.config_path, out, probe=self.fake_probe)\n receipt_path = out / \"receipt.json\"\n receipt = json.loads(receipt_path.read_text(encoding=\"utf-8\"))\n receipt[field] = unsafe\n digest_payload = dict(receipt)\n digest_payload.pop(\"receipt_sha256\")\n receipt[\"receipt_sha256\"] = hashlib.sha256(\n json.dumps(digest_payload, sort_keys=True, separators=(\",\", \":\")).encode(\"utf-8\")\n ).hexdigest()\n receipt_path.write_text(json.dumps(receipt), encoding=\"utf-8\")\n\n with self.assertRaisesRegex(ContractError, \"dry-run boundary\"):\n validate_bundle(out)\n\n def test_receipt_binds_source_duration_and_rejects_out_of_range_times(self):\n for label in (\"duration\", \"time\"):\n with self.subTest(field=label):\n out = self.root / f\"receipt-bound-{label}\"\n run_workflow(self.config_path, out, probe=self.fake_probe)\n receipt_path = out / \"receipt.json\"\n receipt = json.loads(receipt_path.read_text(encoding=\"utf-8\"))\n if label == \"duration\":\n receipt[\"references\"][0][\"source_duration\"] = 7.0\n else:\n artifact = next(\n item for item in receipt[\"evidence_artifacts\"]\n if item[\"provenance\"][0][\"time_basis\"] == \"media_seconds\"\n )\n artifact[\"provenance\"][0][\"reference_times\"] = [6.1]\n digest_payload = dict(receipt)\n digest_payload.pop(\"receipt_sha256\")\n receipt[\"receipt_sha256\"] = hashlib.sha256(\n json.dumps(digest_payload, sort_keys=True, separators=(\",\", \":\")).encode(\"utf-8\")\n ).hexdigest()\n receipt_path.write_text(json.dumps(receipt), encoding=\"utf-8\")\n\n with self.assertRaisesRegex(ContractError, \"duration|time|evidence\"):\n validate_bundle(out)\n\n def test_recipe_evidence_duration_must_match_cited_receipt_source(self):\n out = self.root / \"recipe-source-duration\"\n run_workflow(self.config_path, out, probe=self.fake_probe)\n recipe_path = out / \"resolve\" / \"effect_recipe.json\"\n recipe = json.loads(recipe_path.read_text(encoding=\"utf-8\"))\n event = next(item for item in recipe[\"events\"] if item.get(\"requires_subject_anchor\"))\n event[\"evidence\"].update({\"time\": 99.0, \"source_duration\": 100.0})\n event[\"subject_anchor\"].update({\"evidence_time\": 99.0, \"source_duration\": 100.0})\n recipe_path.write_text(json.dumps(recipe, indent=2, sort_keys=True) + \"\\n\", encoding=\"utf-8\")\n\n receipt_path = out / \"receipt.json\"\n receipt = json.loads(receipt_path.read_text(encoding=\"utf-8\"))\n artifact = next(\n item for item in receipt[\"evidence_artifacts\"]\n if item[\"path\"] == \"resolve/effect_recipe.json\"\n )\n artifact[\"bytes\"] = recipe_path.stat().st_size\n artifact[\"sha256\"] = hashlib.sha256(recipe_path.read_bytes()).hexdigest()\n digest_payload = dict(receipt)\n digest_payload.pop(\"receipt_sha256\")\n receipt[\"receipt_sha256\"] = hashlib.sha256(\n json.dumps(digest_payload, sort_keys=True, separators=(\",\", \":\")).encode(\"utf-8\")\n ).hexdigest()\n receipt_path.write_text(json.dumps(receipt), encoding=\"utf-8\")\n\n with self.assertRaisesRegex(ContractError, \"source duration|reference\"):\n validate_bundle(out)\n\n def test_receipt_rehash_rejects_mutated_available_source(self):\n out = self.root / \"mutation-out\"\n run_workflow(self.config_path, out, probe=self.fake_probe)\n\n self.references[0].write_bytes(b\"mutated-after-receipt\")\n\n with self.assertRaisesRegex(ContractError, \"source|reference|SHA-256|mutat\"):\n validate_bundle(out)\n\n def test_explicit_allow_unavailable_policy_supports_offline_validation(self):\n out = self.root / \"offline-out\"\n receipt = run_workflow(self.config_path, out, probe=self.fake_probe)\n self.assertEqual(receipt.get(\"source_availability_policy\"), \"allow_unavailable\")\n for source in [*self.references, self.editorial]:\n source.unlink()\n\n validate_bundle(out)\n\n def test_validation_rejects_symlinked_bundle_root(self):\n out = self.root / \"real-bundle\"\n run_workflow(self.config_path, out, probe=self.fake_probe)\n linked = self.root / \"linked-bundle\"\n linked.symlink_to(out, target_is_directory=True)\n\n with self.assertRaisesRegex(ContractError, \"symlink\"):\n validate_bundle(linked)\n\n def test_validation_rejects_symlinked_bundle_intermediate(self):\n out = self.root / \"bundle\"\n run_workflow(self.config_path, out, probe=self.fake_probe)\n external = self.root / \"external-manifests\"\n (out / \"manifests\").rename(external)\n (out / \"manifests\").symlink_to(external, target_is_directory=True)\n\n with self.assertRaisesRegex(ContractError, \"symlink\"):\n validate_bundle(out)\n\n def test_receipt_is_deterministic_across_output_directories(self):\n first = run_workflow(self.config_path, self.root / \"first\", probe=self.fake_probe)\n second = run_workflow(self.config_path, self.root / \"second\", probe=self.fake_probe)\n\n self.assertEqual(first, second)\n self.assertEqual(first[\"receipt_sha256\"], second[\"receipt_sha256\"])\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":18035,"content_sha256":"3a5d19007ef7dae077f87e71391db53d9a50643c8b9710972c7eaf6bbbb2048b"},{"filename":"tests/test_offline_fixture.py","content":"\"\"\"Failing-first tests: offline fixture path + documented provenance.\"\"\"\n\nfrom __future__ import annotations\n\nimport json\nimport sys\nimport unittest\nfrom pathlib import Path\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nFIXTURE = Path(__import__(\"tasteforge\").__file__).resolve().parent / \"fixtures\" / \"flashethereal\"\nPROVENANCE_MD = REPO_ROOT.parent / \"SOURCE.md\"\n\n\nclass OfflineFixtureTests(unittest.TestCase):\n def test_fixture_contains_recovered_metadata_only(self):\n names = {p.name for p in FIXTURE.iterdir()}\n self.assertIn(\"pack.json\", names)\n self.assertIn(\"grade.json\", names)\n self.assertIn(\"cadence.json\", names)\n self.assertIn(\"spec.json\", names)\n self.assertIn(\"grounding.txt\", names)\n # Deliberately excluded heavy/binary recovered artifacts.\n self.assertNotIn(\"look.cube\", names)\n self.assertFalse(any(n.endswith(\".glb\") for n in names))\n self.assertFalse(any(n.endswith(\".png\") for n in names))\n self.assertNotIn(\".DS_Store\", names)\n\n def test_cadence_statistics_are_self_consistent(self):\n cad = json.loads((FIXTURE / \"cadence.json\").read_text())\n durs = [s[\"duration\"] for s in cad[\"shots\"]]\n self.assertEqual(cad[\"n_shots\"], len(durs))\n self.assertAlmostEqual(cad[\"mean_shot\"], sum(durs) / len(durs), places=2)\n self.assertGreater(cad[\"cuts_per_min\"], 50)\n\n def test_grade_has_zone_structure(self):\n grade = json.loads((FIXTURE / \"grade.json\").read_text())\n self.assertEqual(len(grade[\"zones\"]), 5)\n self.assertEqual(len(grade[\"palette\"][0]), 2)\n self.assertEqual(len(grade[\"l_cdf\"]), 256)\n\n def test_pack_manifest_matches_recovered_values(self):\n manifest = json.loads((FIXTURE / \"pack.json\").read_text())\n self.assertEqual(manifest[\"name\"], \"flashethereal\")\n self.assertEqual(len(manifest[\"refs\"]), 3)\n self.assertEqual(manifest[\"mint\"][\"lut_size\"], 33)\n self.assertTrue(manifest[\"distill\"][\"dry_run\"])\n\n\nclass ProvenanceDocTests(unittest.TestCase):\n def test_provenance_md_documents_lineage_and_exclusions(self):\n text = PROVENANCE_MD.read_text()\n self.assertIn(\"5e0dc440df4dcf6b2082a7dd59e1d6e9cc11d10166d4e1a19dc6c96478f4d2c8\", text)\n self.assertIn(\"Raw media\", text)\n\n def test_readme_documents_operator_workflow(self):\n readme = (Path(__import__(\"tasteforge\").__file__).resolve().parent / \"README.md\").read_text()\n for cmd in (\"inspect\", \"validate\", \"interview\", \"distill\", \"apply\", \"export\"):\n self.assertIn(cmd, readme)\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":2688,"content_sha256":"760ca26ae1e08f7221cf8f3c6642e80b2bc6042f07577a72a1033ef409d01ef4"},{"filename":"tests/test_pack.py","content":"\"\"\"Failing-first tests for offline style-pack inspect/validate.\"\"\"\n\nfrom __future__ import annotations\n\nimport json\nimport sys\nimport tempfile\nimport unittest\nfrom pathlib import Path\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nfrom tasteforge import pack as pack_mod # noqa: E402\n\nFIXTURE = Path(__import__(\"tasteforge\").__file__).resolve().parent / \"fixtures\" / \"flashethereal\"\n\n\nclass FixturePackTests(unittest.TestCase):\n def test_fixture_pack_loads_and_inspecks(self):\n sp = pack_mod.load(FIXTURE)\n self.assertEqual(sp.name, \"flashethereal\")\n report = sp.inspect()\n self.assertEqual(report[\"name\"], \"flashethereal\")\n self.assertEqual(report[\"manifest_version\"], 1)\n self.assertEqual(len(report[\"refs\"]), 3)\n self.assertTrue(report[\"artifacts\"][\"grade\"])\n self.assertTrue(report[\"artifacts\"][\"cadence\"])\n self.assertTrue(report[\"artifacts\"][\"spec\"])\n\n def test_inspect_reports_validation_status(self):\n report = pack_mod.load(FIXTURE).inspect()\n self.assertEqual(report[\"validation\"][\"status\"], \"valid\")\n self.assertEqual(report[\"validation\"][\"errors\"], [])\n # The fixture deliberately ships metadata only; missing stills must be\n # a warning, never silently ignored.\n self.assertTrue(\n any(\"stills\" in w for w in report[\"validation\"][\"warnings\"])\n )\n\n def test_inspect_includes_cadence_and_grade_summary(self):\n report = pack_mod.load(FIXTURE).inspect()\n self.assertAlmostEqual(report[\"cadence\"][\"mean_shot\"], 0.78, places=1)\n self.assertGreater(report[\"cadence\"][\"n_shots\"], 50)\n self.assertIn(\"contrast\", report[\"grade\"])\n\n def test_fixture_spec_is_dry_run(self):\n sp = pack_mod.load(FIXTURE)\n spec = sp.read_json(sp.spec_path)\n self.assertTrue(spec[\"source\"][\"dry_run\"])\n\n\nclass BrokenPackTests(unittest.TestCase):\n def _write(self, tmp, manifest) -> Path:\n d = Path(tmp) / \"brokenpack\"\n d.mkdir(parents=True)\n (d / \"pack.json\").write_text(json.dumps(manifest))\n return d\n\n def test_missing_manifest_raises(self):\n with tempfile.TemporaryDirectory() as td:\n with self.assertRaises(FileNotFoundError):\n pack_mod.load(Path(td) / \"nowhere\")\n\n def test_invalid_manifest_reports_errors(self):\n with tempfile.TemporaryDirectory() as td:\n d = self._write(td, {\"name\": \"broken\", \"version\": 99})\n report = pack_mod.load(d).inspect()\n self.assertEqual(report[\"validation\"][\"status\"], \"invalid\")\n self.assertTrue(report[\"validation\"][\"errors\"])\n\n def test_corrupt_cadence_reported(self):\n with tempfile.TemporaryDirectory() as td:\n d = self._write(\n td,\n {\n \"name\": \"broken\",\n \"version\": 1,\n \"created\": \"2026-08-16T05:53:05Z\",\n \"updated\": \"2026-08-16T07:10:42Z\",\n \"refs\": [],\n \"artifacts\": {},\n },\n )\n (d / \"cadence.json\").write_text(json.dumps({\"shots\": \"nope\"}))\n report = pack_mod.load(d).inspect()\n self.assertEqual(report[\"validation\"][\"status\"], \"invalid\")\n self.assertTrue(any(\"cadence\" in e for e in report[\"validation\"][\"errors\"]))\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":3481,"content_sha256":"98779024825599ea4fea7776b5d4d269ce9a2023a158784377d3207398d32450"},{"filename":"tests/test_provenance.py","content":"\"\"\"Failing-first tests for tasteforge provenance and provider-reference policy.\n\nContract:\n- exact lineage of the recovered TasteForge sources is recorded as data;\n- a provider workflow may only ever be referenced, never claimed as saved;\n- the Claude cloud session is recorded honestly (selected, transcript absent).\n\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nimport unittest\nfrom pathlib import Path\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nfrom tasteforge import provenance # noqa: E402\n\nCANONICAL = \"recovered/tasteforge-flow-20260818\"\nLATEST_SHA = \"ef06a606d3b528fbd939b05fadc25bf6674073a1e05a01e3aa6b9c9416fd6284\"\nSESSION = \"redacted-local-session\"\n\n\nclass LineageTests(unittest.TestCase):\n def test_lineage_report_names_canonical_source(self):\n report = provenance.lineage_report()\n self.assertEqual(report[\"canonical_source\"][\"path\"], CANONICAL)\n self.assertTrue(report[\"canonical_source\"][\"read_only\"])\n\n def test_all_five_generations_recorded_with_digests(self):\n gens = provenance.lineage_report()[\"generations\"]\n self.assertEqual(len(gens), 5)\n latest = [g for g in gens if g[\"archive\"] == \"tasteforge (4).zip\"][0]\n self.assertEqual(latest[\"sha256\"], LATEST_SHA)\n self.assertEqual(latest[\"status\"], \"latest\")\n for g in gens[:-1]:\n self.assertEqual(g[\"status\"], \"prior\")\n\n def test_generation_deltas_explain_lineage(self):\n gens = provenance.lineage_report()[\"generations\"]\n self.assertTrue(all(g.get(\"delta\") for g in gens))\n latest = gens[-1]\n self.assertIn(\"grade\", latest[\"delta\"])\n\n def test_claude_session_recorded_honestly(self):\n sess = provenance.lineage_report()[\"claude_session\"]\n self.assertEqual(sess[\"id\"], SESSION)\n self.assertFalse(sess[\"transcript_available\"])\n self.assertTrue(sess[\"selection_evidence_local\"])\n\n def test_lineage_report_passes_schema(self):\n from tasteforge import schema\n\n problems = schema.validate(\n provenance.lineage_report(), schema.PROVENANCE_SCHEMA\n )\n self.assertEqual(problems, [])\n\n\nclass ProviderReferenceTests(unittest.TestCase):\n def test_provider_reference_is_pointer_only(self):\n ref = provenance.provider_reference(\"fal\")\n self.assertEqual(ref[\"kind\"], \"provider-workflow-reference\")\n self.assertEqual(ref[\"provider\"], \"fal\")\n self.assertTrue(ref[\"reference_only\"])\n self.assertFalse(ref[\"persisted_workflow_state\"])\n self.assertFalse(ref[\"authorizes_execution\"])\n\n def test_saved_workflow_state_is_rejected(self):\n record = {\"kind\": \"provider-workflow-reference\", \"provider\": \"fal\",\n \"reference_only\": True, \"persisted_workflow_state\": True,\n \"authorizes_execution\": False}\n with self.assertRaises(provenance.SavedWorkflowClaimError):\n provenance.assert_no_saved_provider_workflow([record])\n\n def test_clean_records_pass(self):\n provenance.assert_no_saved_provider_workflow([provenance.provider_reference(\"fal\")])\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":3194,"content_sha256":"c2e6105d44c8fe12d4ab9386714d2afbef99202f905a62c7956415e277da79a8"},{"filename":"tests/test_providers.py","content":"\"\"\"Failing-first tests: provider adapters must fail closed, always.\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nimport unittest\nfrom pathlib import Path\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nfrom tasteforge import providers # noqa: E402\n\n\nclass RegistryFailClosedTests(unittest.TestCase):\n def test_default_registry_has_no_providers(self):\n self.assertEqual(providers.list_providers(), [])\n\n def test_get_unknown_provider_raises_not_authorized(self):\n with self.assertRaises(providers.ProviderNotAuthorizedError) as ctx:\n providers.get(\"fal\")\n self.assertIn(\"separately authorized\", str(ctx.exception))\n\n def test_env_flag_alone_does_not_enable(self):\n import os\n\n old = os.environ.get(\"TASTEFORGE_ALLOW_PROVIDERS\")\n os.environ[\"TASTEFORGE_ALLOW_PROVIDERS\"] = \"1\"\n try:\n with self.assertRaises(providers.ProviderNotAuthorizedError):\n providers.get(\"fal\")\n finally:\n if old is None:\n del os.environ[\"TASTEFORGE_ALLOW_PROVIDERS\"]\n else:\n os.environ[\"TASTEFORGE_ALLOW_PROVIDERS\"] = old\n\n def test_explicit_registration_requires_authorization_flag(self):\n with self.assertRaises(providers.ProviderNotAuthorizedError):\n providers.register(\n \"fal\",\n callable_factory=lambda: (_ for _ in ()).throw(AssertionError(\"never\")),\n )\n\n def test_no_network_modules_imported(self):\n for mod in (\"fal_client\", \"requests\", \"http.client\", \"urllib.request\"):\n self.assertNotIn(mod, sys.modules, f\"{mod} must not be imported by tasteforge\")\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":1773,"content_sha256":"b2d1c6dc358b0aeef4d9e09c69f8560fc2084f922f71e95fe9d1883484576b94"},{"filename":"tests/test_resolve.py","content":"# ruff: noqa: N802 -- fake methods preserve Resolve API names\n\"\"\"Resolve adapter regression tests; no connection to Resolve is made.\"\"\"\n\nimport json\nimport tempfile\nimport unittest\nfrom pathlib import Path\nfrom types import SimpleNamespace\nfrom unittest.mock import patch\n\nfrom tasteforge.resolve import allocate_placements, apply_placements, probe_asset\n\n\nclass Item:\n def __init__(self, request):\n self.request = request\n self.start = request[\"recordFrame\"]\n self.frames = request[\"endFrame\"] + 1\n self.props = {\"Opacity\": 100, \"CompositeMode\": 0}\n\n def GetStart(self):\n return self.start\n\n def GetEnd(self):\n return None if self.start is None else self.start + self.frames\n\n def GetDuration(self):\n return self.frames\n\n def GetClipEnabled(self):\n return True\n\n def GetMediaPoolItem(self):\n return self.request[\"mediaPoolItem\"]\n\n def GetProperty(self, key=None):\n return self.props.copy() if key is None else self.props[key]\n\n def SetProperty(self, key, value):\n self.props[key] = value\n return True\n\n\nclass Media:\n def __init__(self, path):\n self.path = path\n\n def GetClipProperty(self, key):\n return self.path\n\n\nclass Timeline:\n def __init__(self):\n self.tracks = {1: []}\n\n def GetName(self):\n return \"target\"\n\n def GetSetting(self, key):\n return \"30\"\n\n def GetTrackCount(self, kind):\n return len(self.tracks) if kind == \"video\" else 0\n\n def GetItemListInTrack(self, kind, track):\n return self.tracks[track]\n\n def AddTrack(self, kind):\n self.tracks[len(self.tracks) + 1] = []\n return True\n\n\nclass Pool:\n def __init__(self, timeline, fault=None, host_mode=\"inclusive\"):\n self.timeline, self.fault, self.calls = timeline, fault, []\n self.host_mode = host_mode\n\n def ImportMedia(self, paths):\n return [Media(paths[0])]\n\n def AppendToTimeline(self, requests):\n request = requests[0]\n self.calls.append(request)\n item = Item(request)\n if self.host_mode == \"exclusive\":\n item.frames -= 1\n self.timeline.tracks[request[\"trackIndex\"]].append(item)\n if self.fault == \"null\":\n item.start = None\n if self.fault == \"shift\":\n item.start += 1\n if self.fault == \"trim\":\n item.frames -= 1\n if self.fault == \"later\" and len(self.calls) == 2:\n self.timeline.tracks[2][0].frames -= 1\n if self.fault == \"disabled\":\n item.GetClipEnabled = lambda: False\n if self.fault == \"property\":\n item.SetProperty = lambda key, value: True\n if self.fault == \"path\":\n item.request[\"mediaPoolItem\"].path = \"/wrong.mov\"\n if self.fault == \"track\":\n self.timeline.tracks[request[\"trackIndex\"]].remove(item)\n if self.fault == \"base\":\n self.timeline.tracks[1].append(Item(request))\n return [item]\n\n\nclass ResolveTests(unittest.TestCase):\n def setUp(self):\n self.tmp = tempfile.TemporaryDirectory()\n self.addCleanup(self.tmp.cleanup)\n self.path = Path(self.tmp.name) / \"asset.mov\"\n self.path.write_bytes(b\"fixture\")\n self.events = [\n dict(\n id=\"a\",\n asset=str(self.path),\n record_frame=0,\n frames=10,\n opacity=88,\n composite=22,\n ),\n dict(\n id=\"b\",\n asset=str(self.path),\n record_frame=5,\n frames=10,\n opacity=100,\n composite=0,\n ),\n dict(\n id=\"c\",\n asset=str(self.path),\n record_frame=10,\n frames=5,\n opacity=50,\n composite=22,\n ),\n ]\n self.probe = lambda path: dict(fps=30, frames=20, has_alpha=True)\n\n def plan(self, events=None, **kwargs):\n return allocate_placements(\n self.events if events is None else events,\n fps=30,\n base_track_count=1,\n probe=self.probe,\n **kwargs,\n )\n\n def apply(self, fault=None, source_end_mode=\"inclusive\", host_mode=\"inclusive\"):\n tl = Timeline()\n pool = Pool(tl, fault, host_mode)\n result = apply_placements(\n tl,\n pool,\n self.events,\n source_timeline=\"source\",\n source_end_mode=source_end_mode,\n fps=30,\n base_track_count=1,\n probe=self.probe,\n )\n return result, pool\n\n def test_overlap_coloring_and_inclusive_source_end(self):\n plan = self.plan()\n self.assertEqual([p[\"track\"] for p in plan], [2, 3, 2])\n receipt, pool = self.apply()\n self.assertEqual(pool.calls[0][\"endFrame\"], 9)\n self.assertEqual(receipt[\"placements\"][0][\"actual\"][\"end\"], 10)\n self.assertTrue(receipt[\"preservation\"][\"base_tracks_match\"])\n self.assertNotIn(\"track\", self.events[0])\n\n def test_readback_failure_never_returns_receipt(self):\n for fault in (\n \"null\",\n \"shift\",\n \"trim\",\n \"later\",\n \"base\",\n \"disabled\",\n \"property\",\n \"path\",\n \"track\",\n ):\n with self.subTest(fault=fault), self.assertRaises(RuntimeError):\n self.apply(fault)\n\n def test_occupied_overlay_tracks_rejected_before_append(self):\n tl = Timeline()\n tl.tracks[2] = [object()]\n pool = Pool(tl)\n with self.assertRaises(ValueError):\n apply_placements(\n tl,\n pool,\n self.events,\n source_timeline=\"source\",\n source_end_mode=\"inclusive\",\n fps=30,\n base_track_count=1,\n probe=self.probe,\n )\n self.assertEqual(pool.calls, [])\n\n def test_invalid_contract(self):\n for key, value in [\n (\"frames\", 1.5),\n (\"frames\", True),\n (\"frames\", 0),\n (\"record_frame\", -1),\n (\"opacity\", float(\"nan\")),\n (\"opacity\", 101),\n (\"composite\", None),\n (\"asset\", self.tmp.name),\n ]:\n with self.subTest(key=key, value=value), self.assertRaises(ValueError):\n self.plan([{**self.events[0], key: value}])\n with self.assertRaises(ValueError):\n self.plan([self.events[0], self.events[0]])\n\n def test_metadata_gates(self):\n for metadata in [\n dict(fps=24, frames=20, has_alpha=True),\n dict(fps=30, frames=2, has_alpha=True),\n dict(fps=30, frames=20, has_alpha=False),\n ]:\n with self.subTest(metadata=metadata), self.assertRaises(ValueError):\n allocate_placements(\n [{**self.events[0], \"requires_alpha\": True}],\n fps=30,\n base_track_count=1,\n probe=lambda p: metadata,\n )\n\n def test_timeline_fps_mismatch_before_mutation(self):\n tl = Timeline()\n tl.GetSetting = lambda key: \"24\"\n pool = Pool(tl)\n with self.assertRaises(ValueError):\n apply_placements(\n tl,\n pool,\n self.events,\n source_timeline=\"source\",\n source_end_mode=\"inclusive\",\n fps=30,\n base_track_count=1,\n probe=self.probe,\n )\n self.assertEqual(pool.calls, [])\n\n def test_exclusive_host_and_receipt(self):\n receipt, pool = self.apply(source_end_mode=\"exclusive\", host_mode=\"exclusive\")\n self.assertEqual(pool.calls[0][\"endFrame\"], 10)\n self.assertEqual(receipt[\"source_end_mode\"], \"exclusive\")\n self.assertEqual(receipt[\"placements\"][0][\"actual\"][\"duration\"], 10)\n\n def test_mode_mismatch_fails_without_retry(self):\n for mode, host in [(\"inclusive\", \"exclusive\"), (\"exclusive\", \"inclusive\")]:\n tl = Timeline()\n pool = Pool(tl, host_mode=host)\n with self.assertRaises(RuntimeError):\n apply_placements(\n tl,\n pool,\n self.events,\n source_timeline=\"source\",\n source_end_mode=mode,\n fps=30,\n base_track_count=1,\n probe=self.probe,\n )\n self.assertEqual(len(pool.calls), 1)\n\n def test_mode_must_be_explicit_and_valid(self):\n with self.assertRaises(ValueError):\n self.apply(source_end_mode=\"auto\")\n with self.assertRaises(TypeError):\n apply_placements(\n Timeline(),\n None,\n self.events,\n source_timeline=\"source\",\n fps=30,\n base_track_count=1,\n probe=self.probe,\n )\n\n\nclass ProbeTests(unittest.TestCase):\n def test_ffprobe_alpha_and_frame_count(self):\n stream = dict(\n avg_frame_rate=\"30000/1001\",\n r_frame_rate=\"30000/1001\",\n nb_read_frames=\"42\",\n pix_fmt=\"yuva444p10le\",\n )\n with patch(\n \"tasteforge.resolve.subprocess.run\",\n return_value=SimpleNamespace(stdout=json.dumps(dict(streams=[stream]))),\n ) as run:\n result = probe_asset(Path(\"/asset.mov\"))\n self.assertEqual(result, dict(fps=\"30000/1001\", frames=42, has_alpha=True))\n self.assertIn(\"-count_frames\", run.call_args.args[0])\n\n def test_probe_rejects_no_video_and_ambiguous_rate(self):\n for streams in [[], [dict(avg_frame_rate=\"24\", r_frame_rate=\"30\")]]:\n with (\n patch(\n \"tasteforge.resolve.subprocess.run\",\n return_value=SimpleNamespace(\n stdout=json.dumps(dict(streams=streams))\n ),\n ),\n self.assertRaises(ValueError),\n ):\n probe_asset(Path(\"/asset.mov\"))\n","content_type":"text/x-python; charset=utf-8","language":"python","size":10281,"content_sha256":"6ef7be2771028cd679630c1184e908bbb23ca5dae5522aa046e17bfd29c6d5e6"},{"filename":"tests/test_schema.py","content":"\"\"\"Failing-first tests for the tasteforge schema subset validator.\n\nContract (from the recovered TasteForge gen4 source, canonicalized):\n- hand-rolled JSON-Schema subset: type, required, properties, items, enum,\n minimum/minimum, minItems, pattern; no third-party dependency.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nimport unittest\nfrom pathlib import Path\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nfrom tasteforge import schema # noqa: E402\n\n\nclass ValidatorTests(unittest.TestCase):\n def setUp(self):\n self.simple = {\n \"type\": \"object\",\n \"required\": [\"name\", \"count\"],\n \"properties\": {\n \"name\": {\"type\": \"string\", \"pattern\": \"^[a-z][a-z0-9_-]*$\"},\n \"count\": {\"type\": \"integer\", \"minimum\": 0},\n \"tags\": {\"type\": \"array\", \"items\": {\"type\": \"string\"}, \"minItems\": 1},\n \"mode\": {\"type\": \"string\", \"enum\": [\"local\", \"dry-run\"]},\n },\n }\n\n def test_accepts_valid_instance(self):\n problems = schema.validate(\n {\"name\": \"flashethereal\", \"count\": 3, \"tags\": [\"a\"], \"mode\": \"local\"},\n self.simple,\n )\n self.assertEqual(problems, [])\n\n def test_rejects_missing_required(self):\n problems = schema.validate({\"count\": 3}, self.simple)\n self.assertTrue(any(\"required\" in p and \"name\" in p for p in problems))\n\n def test_rejects_wrong_type(self):\n problems = schema.validate({\"name\": \"x\", \"count\": \"three\"}, self.simple)\n self.assertTrue(any(\"count\" in p and \"type\" in p for p in problems))\n\n def test_rejects_bad_pattern(self):\n problems = schema.validate({\"name\": \"Bad Name!\", \"count\": 0}, self.simple)\n self.assertTrue(any(\"name\" in p and \"pattern\" in p for p in problems))\n\n def test_rejects_bad_enum(self):\n problems = schema.validate({\"name\": \"x\", \"count\": 0, \"mode\": \"live\"}, self.simple)\n self.assertTrue(any(\"mode\" in p and \"enum\" in p for p in problems))\n\n def test_rejects_below_minimum(self):\n problems = schema.validate({\"name\": \"x\", \"count\": -1}, self.simple)\n self.assertTrue(any(\"count\" in p and \"minimum\" in p for p in problems))\n\n def test_rejects_bad_items_and_min_items(self):\n problems = schema.validate({\"name\": \"x\", \"count\": 0, \"tags\": [1, 2]}, self.simple)\n self.assertTrue(any(\"tags[0]\" in p for p in problems))\n problems = schema.validate({\"name\": \"x\", \"count\": 0, \"tags\": []}, self.simple)\n self.assertTrue(any(\"tags\" in p and \"minItems\" in p for p in problems))\n\n def test_non_object_root_rejected(self):\n problems = schema.validate([\"not\", \"an\", \"object\"], self.simple)\n self.assertTrue(problems)\n\n\nclass ExportedSchemasTests(unittest.TestCase):\n EXPORTED = [\n \"TASTE_PROFILE_SCHEMA\",\n \"PACK_MANIFEST_SCHEMA\",\n \"GRADE_SCHEMA\",\n \"CADENCE_SCHEMA\",\n \"SPEC_SCHEMA\",\n \"TIMELINE_EVENT_SCHEMA\",\n \"APPLICATION_REPORT_SCHEMA\",\n \"PROVENANCE_SCHEMA\",\n ]\n\n def test_all_exported_schemas_exist_and_are_objects(self):\n for name in self.EXPORTED:\n with self.subTest(schema=name):\n s = getattr(schema, name)\n self.assertIsInstance(s, dict)\n self.assertEqual(s.get(\"type\"), \"object\")\n self.assertIn(\"required\", s)\n self.assertIn(\"properties\", s)\n\n def test_application_report_forbids_provider_generation(self):\n s = schema.APPLICATION_REPORT_SCHEMA\n self.assertEqual(\n s[\"properties\"][\"provider\"].get(\"enum\"), [\"none\"],\n \"application reports must only ever claim provider=none in this lane\",\n )\n self.assertEqual(\n s[\"properties\"][\"dry_run\"].get(\"enum\"), [True],\n \"application reports must never claim a live provider run\",\n )\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":3982,"content_sha256":"1675ac74acfa431cb9443ced4a501f3c6389d408f4d383259192e899d3958dcb"},{"filename":"tests/test_timeline_export.py","content":"\"\"\"Failing-first tests for the timeline timebase and EDL/FCPXML export.\n\nNumeric expectations are canonicalized from the recovered gen4\n``taste/timeline.py`` self-checks and the recovered ``flashethereal-cut.edl``.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nimport tempfile\nimport unittest\nimport xml.etree.ElementTree as ET\nfrom fractions import Fraction\nfrom pathlib import Path\n\nREPO_ROOT = Path(__file__).resolve().parents[1] / \"scripts\"\nsys.path.insert(0, str(REPO_ROOT))\n\nfrom tasteforge import export, timeline # noqa: E402\n\nCLIPS = [\n {\"path\": \"/tmp/media/a.mov\", \"duration\": 1.5, \"name\": \"alpha\"},\n {\"path\": \"/tmp/media/b.mov\", \"duration\": 2.25, \"name\": \"beta\"},\n {\"path\": \"/tmp/media/c.mov\", \"duration\": 0.75, \"name\": \"gamma\"},\n]\n\n\nclass TimebaseTests(unittest.TestCase):\n def test_ntsc_snap(self):\n self.assertEqual(timeline.fps_fraction(29.97), Fraction(30000, 1001))\n self.assertEqual(timeline.fps_fraction(23.976), Fraction(24000, 1001))\n self.assertEqual(timeline.fps_fraction(24), Fraction(24, 1))\n\n def test_negative_fps_rejected(self):\n with self.assertRaises(ValueError):\n timeline.fps_fraction(-3)\n\n def test_seconds_to_frames_rounds_half_away_from_zero(self):\n self.assertEqual(timeline.seconds_to_frames(0.5, 24), 12)\n self.assertEqual(timeline.seconds_to_frames(1.25, 24), 30)\n\n def test_rational_time_strings(self):\n self.assertEqual(timeline.frames_to_rational(1, 29.97), \"1001/30000s\")\n self.assertEqual(timeline.frames_to_rational(30, 29.97), \"1001/1000s\")\n self.assertEqual(timeline.frames_to_rational(120, 24), \"5s\")\n self.assertEqual(timeline.seconds_to_rational(2.5, 24), \"5/2s\")\n\n def test_timecode_drop_frame(self):\n self.assertEqual(timeline.frames_to_timecode(1800, 29.97), \"00:01:00:02\")\n self.assertEqual(timeline.frames_to_timecode(17982, 29.97), \"00:10:00:00\")\n self.assertEqual(timeline.frames_to_timecode(24, 24), \"00:00:01:00\")\n\n\nclass EDLTests(unittest.TestCase):\n def test_build_edl_header_and_events(self):\n text = export.build_edl(CLIPS, fps=24.0, title=\"unittest-cut\")\n lines = text.splitlines()\n self.assertEqual(lines[0], \"TITLE: UNITTEST-CUT\")\n self.assertIn(\"FCM: NON-DROP FRAME\", lines)\n event_lines = [ln for ln in lines if ln[:1].isdigit()]\n self.assertEqual(len(event_lines), 3)\n self.assertIn(\"* FROM CLIP NAME: a.mov\", text)\n\n def test_edl_roundtrip_parses(self):\n text = export.build_edl(CLIPS, fps=24.0, title=\"rt\")\n events = export.parse_edl(text)\n self.assertEqual(len(events), 3)\n self.assertEqual(events[0][\"name\"], \"a.mov\")\n self.assertEqual(events[-1][\"record_in\"], 90) # 1.5+2.25 s at 24fps\n self.assertEqual(events[-1][\"duration_frames\"], 18)\n\n def test_ntsc_edl_is_drop_frame(self):\n text = export.build_edl(CLIPS, fps=29.97, title=\"ntsc\")\n self.assertIn(\"FCM: DROP FRAME\", text)\n\n\nclass FCPXMLTests(unittest.TestCase):\n def test_build_fcpxml_structure(self):\n text = export.build_fcpxml(CLIPS, fps=24.0, title=\"tf-test\")\n root = ET.fromstring(text)\n self.assertEqual(root.tag, \"fcpxml\")\n self.assertIn(\"version\", root.attrib)\n assets = root.findall(\"./resources/asset\")\n self.assertEqual(len(assets), 3)\n clips = root.findall(\".//asset-clip\")\n self.assertEqual(len(clips), 3)\n\n def test_fcpxml_durations_are_rational_and_exact(self):\n text = export.build_fcpxml(CLIPS, fps=24.0, title=\"tf-test\")\n root = ET.fromstring(text)\n clips = root.findall(\".//asset-clip\")\n # 1.5s @24 -> \"7/2s\"? No: quantized frames=36 -> \"3/2s\"\n self.assertEqual(clips[0].get(\"duration\"), \"3/2s\")\n total = sum(Fraction(c.get(\"duration\").rstrip(\"s\")) for c in clips)\n self.assertEqual(total, Fraction(int(4.5 * 24), 24))\n\n def test_zero_or_negative_duration_rejected(self):\n with self.assertRaises(ValueError):\n export.build_fcpxml([{\"path\": \"x\", \"duration\": 0}], fps=24)\n with self.assertRaises(ValueError):\n export.build_edl([], fps=24)\n\n\nclass WriteTimelineTests(unittest.TestCase):\n def test_write_timeline_emits_both_formats(self):\n with tempfile.TemporaryDirectory() as td:\n edl, fcpxml = export.write_timeline(CLIPS, out_dir=td, title=\"wt\")\n self.assertTrue(Path(edl).exists())\n self.assertTrue(Path(fcpxml).exists())\n self.assertIn(\"TITLE: WT\", Path(edl).read_text())\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":4623,"content_sha256":"c38d73c9427612d183f3eb492c377b1795c0adbcfe90e883619aafd320ae4e90"},{"filename":"workflows/README.md","content":"# Offline Fal workflow clones\n\nThese importable templates were derived from the September 7, 2026 exports of the existing application, distillation and prop workflows. Account identifiers, sharing state, timestamps and example/default inputs are excluded. They do not change the live originals. Import them as new workflows, then verify the imported input schema and endpoint contracts before any authorized paid run.\n\n- `taste-apply.json` uses first/middle/last frames from **your own content** as generation references. Its required `compiled_prompt` binds the content brief and taste steering to every generator. Merge output is explicitly 30 fps. This generates a reinterpretation from still references; it does not preserve the original footage or its motion.\n- `taste-apply-motion.json` is an optional generated reinterpretation variant. Each generator also receives `video_urls: [\"$input.source_video\"]` as a motion reference, using the list field verified in the live Seedance UI/export. It keeps the same compiled input contract, 30 fps merge and disabled generated audio. Using a motion reference still generates new footage; it is not passthrough. Import it separately and confirm the run budget before submission.\n- `taste-distill.json` requires three reference URLs and supplied measured grounding for the actual reference set. It retains middle-frame extraction and vision analysis, removes inherited Flash Ethereal measurements and the raw collage-to-mesh branch. Three still frames cannot measure cadence or motion; those measurements must come from local frame/video analysis.\n- `taste-prop3d.json` retains the separate plate-to-PBR-mesh workflow. Supply a clean isolated prop description with plate rules. Preserve the full textured GLB when passing it to Blender; a reduced geometry proxy is not evidence that materials survived.\n\nFor actual existing-footage passthrough and editorial application, use the local pipeline `--takes` path and `forge.py`; these graph variants are generation paths.\n\nAll templates have blank required inputs. Endpoint/model IDs and output paths are preserved from the observed exports, including Seedance reference-to-video and Gemini 2.5 Flash. Their inclusion is **not** evidence of current availability or a successful provider run. Validate live schemas before submission. `generate_audio: false` explicitly disables generated audio on all three application generators. This field was verified in the September 7 live Seedance 2.5 UI export. `audio_urls: []` separately supplies no reference audio. Verify the resulting media streams during delivery checks.\n\n## Compile application inputs locally\n\nCreate a private configuration JSON outside the repository:\n\n```json\n{\n \"source_video\": \"https://example.org/your-own-content.mp4\",\n \"brief\": \"Describe the content and action\",\n \"style_steer\": \"Describe measured structure, lighting, framing and motion\"\n}\n```\n\n```sh\npython3 skills/taste-application/scripts/workflow_graphs.py \\\n --kind apply --config /absolute/private/apply-config.json \\\n --out /absolute/private/apply-input.json\n```\n\nThe output object contains `source_video` and `compiled_prompt` and works with either application template; the optional motion variant needs no additional compiler option. The compiler separates WHAT, HOW and mandatory GRADE sections. Rendering stays neutral so the measured grade can be applied once downstream. There are no unsupported graph string-concatenation expressions.\n\n## Compile distillation inputs locally\n\n```json\n{\n \"genre\": \"Your actual reference genre\",\n \"references\": [\n \"https://example.org/reference-a.mp4\",\n \"https://example.org/reference-b.mp4\",\n \"https://example.org/reference-c.mp4\"\n ],\n \"measured_grounding\": \"Supply results and provenance from actual local analysis. Do not copy another reference set's measurements.\"\n}\n```\n\nUse the same command with `--kind distill`. Its output object contains `reference_1`, `reference_2`, `reference_3`, and `measured_grounding` for the corresponding template. The compiler requires nonempty genre and grounding, adds no numerical measurements, performs no network calls and refuses to overwrite an output file. It cannot establish whether supplied measurements are truthful; retain the referenced analysis report for review.\n\n`validate_graph` checks input consumption, output node references, dependency wiring and cycles. It does not validate provider-specific model schemas or make a live submission. Local compiled inputs may contain private media URLs and should not be committed.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":4553,"content_sha256":"1c1c7d4fb67c224695ab6d4c88a4defaaf419cb96bee72e3a3b645a348189d8a"},{"filename":"workflows/taste-apply-motion.json","content":"{\n \"name\": \"taste-apply-motion-30fps\",\n \"title\": \"taste-apply-motion-30fps\",\n \"contents\": {\n \"name\": \"workflow\",\n \"nodes\": {\n \"node-xfirst\": {\n \"type\": \"run\",\n \"id\": \"node-xfirst\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 340,\n \"y\": -320\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.source_video\",\n \"frame_type\": \"first\"\n }\n },\n \"node-gen1\": {\n \"type\": \"run\",\n \"id\": \"node-gen1\",\n \"depends\": [\n \"input\",\n \"node-xfirst\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 760,\n \"y\": -320\n }\n },\n \"app\": \"bytedance/seedance-2.5/reference-to-video\",\n \"input\": {\n \"prompt\": \"$input.compiled_prompt\",\n \"image_urls\": [\n \"$node-xfirst.images.0.url\"\n ],\n \"duration\": \"5\",\n \"resolution\": \"720p\",\n \"audio_urls\": [],\n \"generate_audio\": false,\n \"video_urls\": [\n \"$input.source_video\"\n ]\n }\n },\n \"node-xmid\": {\n \"type\": \"run\",\n \"id\": \"node-xmid\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 340,\n \"y\": 0\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.source_video\",\n \"frame_type\": \"middle\"\n }\n },\n \"node-gen2\": {\n \"type\": \"run\",\n \"id\": \"node-gen2\",\n \"depends\": [\n \"input\",\n \"node-xmid\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 760,\n \"y\": 0\n }\n },\n \"app\": \"bytedance/seedance-2.5/reference-to-video\",\n \"input\": {\n \"prompt\": \"$input.compiled_prompt\",\n \"image_urls\": [\n \"$node-xmid.images.0.url\"\n ],\n \"duration\": \"5\",\n \"resolution\": \"720p\",\n \"audio_urls\": [],\n \"generate_audio\": false,\n \"video_urls\": [\n \"$input.source_video\"\n ]\n }\n },\n \"node-xlast\": {\n \"type\": \"run\",\n \"id\": \"node-xlast\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 340,\n \"y\": 320\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.source_video\",\n \"frame_type\": \"last\"\n }\n },\n \"node-gen3\": {\n \"type\": \"run\",\n \"id\": \"node-gen3\",\n \"depends\": [\n \"input\",\n \"node-xlast\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 760,\n \"y\": 320\n }\n },\n \"app\": \"bytedance/seedance-2.5/reference-to-video\",\n \"input\": {\n \"prompt\": \"$input.compiled_prompt\",\n \"image_urls\": [\n \"$node-xlast.images.0.url\"\n ],\n \"duration\": \"5\",\n \"resolution\": \"720p\",\n \"audio_urls\": [],\n \"generate_audio\": false,\n \"video_urls\": [\n \"$input.source_video\"\n ]\n }\n },\n \"node-merge\": {\n \"type\": \"run\",\n \"id\": \"node-merge\",\n \"depends\": [\n \"node-gen1\",\n \"node-gen2\",\n \"node-gen3\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 1180,\n \"y\": 0\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/merge-videos\",\n \"input\": {\n \"video_urls\": [\n \"$node-gen1.video.url\",\n \"$node-gen2.video.url\",\n \"$node-gen3.video.url\"\n ],\n \"target_fps\": 30\n }\n },\n \"output\": {\n \"type\": \"display\",\n \"id\": \"output\",\n \"depends\": [\n \"node-merge\"\n ],\n \"input\": {},\n \"metadata\": {\n \"position\": {\n \"x\": 1600,\n \"y\": 0\n }\n },\n \"fields\": {\n \"reel\": \"$node-merge.video\",\n \"take_1\": \"$node-gen1.video\",\n \"take_2\": \"$node-gen2.video\",\n \"take_3\": \"$node-gen3.video\"\n }\n }\n },\n \"output\": {\n \"reel\": \"$node-merge.video\",\n \"take_1\": \"$node-gen1.video\",\n \"take_2\": \"$node-gen2.video\",\n \"take_3\": \"$node-gen3.video\"\n },\n \"schema\": {\n \"input\": {\n \"source_video\": {\n \"name\": \"video_url\",\n \"label\": \"Own-content source video\",\n \"description\": \"Content reference; taste structure comes from compiled_prompt.\",\n \"required\": true,\n \"defaultValue\": \"\",\n \"examples\": [],\n \"ui\": {},\n \"type\": \"string\",\n \"modelId\": \"node-xfirst\"\n },\n \"compiled_prompt\": {\n \"name\": \"text\",\n \"label\": \"Compiled WHAT / HOW / GRADE prompt\",\n \"description\": \"Compile offline with workflow_graphs.py.\",\n \"required\": true,\n \"defaultValue\": \"\",\n \"examples\": [],\n \"ui\": {\n \"field\": \"textarea\"\n },\n \"type\": \"string\",\n \"modelId\": \"node-gen1\"\n }\n },\n \"output\": {}\n },\n \"version\": \"1\",\n \"metadata\": {\n \"input\": {\n \"position\": {\n \"x\": 0,\n \"y\": 0\n }\n }\n }\n }\n}\n","content_type":"application/json; charset=utf-8","language":"json","size":5504,"content_sha256":"577e2005e651ce02c8f3839605a8d9460241beff6bc7996e070e3e9d6f1c4a45"},{"filename":"workflows/taste-apply.json","content":"{\n \"name\": \"taste-apply-30fps\",\n \"title\": \"taste-apply-30fps\",\n \"contents\": {\n \"name\": \"workflow\",\n \"nodes\": {\n \"node-xfirst\": {\n \"type\": \"run\",\n \"id\": \"node-xfirst\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 340,\n \"y\": -320\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.source_video\",\n \"frame_type\": \"first\"\n }\n },\n \"node-gen1\": {\n \"type\": \"run\",\n \"id\": \"node-gen1\",\n \"depends\": [\n \"input\",\n \"node-xfirst\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 760,\n \"y\": -320\n }\n },\n \"app\": \"bytedance/seedance-2.5/reference-to-video\",\n \"input\": {\n \"prompt\": \"$input.compiled_prompt\",\n \"image_urls\": [\n \"$node-xfirst.images.0.url\"\n ],\n \"duration\": \"5\",\n \"resolution\": \"720p\",\n \"audio_urls\": [],\n \"generate_audio\": false\n }\n },\n \"node-xmid\": {\n \"type\": \"run\",\n \"id\": \"node-xmid\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 340,\n \"y\": 0\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.source_video\",\n \"frame_type\": \"middle\"\n }\n },\n \"node-gen2\": {\n \"type\": \"run\",\n \"id\": \"node-gen2\",\n \"depends\": [\n \"input\",\n \"node-xmid\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 760,\n \"y\": 0\n }\n },\n \"app\": \"bytedance/seedance-2.5/reference-to-video\",\n \"input\": {\n \"prompt\": \"$input.compiled_prompt\",\n \"image_urls\": [\n \"$node-xmid.images.0.url\"\n ],\n \"duration\": \"5\",\n \"resolution\": \"720p\",\n \"audio_urls\": [],\n \"generate_audio\": false\n }\n },\n \"node-xlast\": {\n \"type\": \"run\",\n \"id\": \"node-xlast\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 340,\n \"y\": 320\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.source_video\",\n \"frame_type\": \"last\"\n }\n },\n \"node-gen3\": {\n \"type\": \"run\",\n \"id\": \"node-gen3\",\n \"depends\": [\n \"input\",\n \"node-xlast\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 760,\n \"y\": 320\n }\n },\n \"app\": \"bytedance/seedance-2.5/reference-to-video\",\n \"input\": {\n \"prompt\": \"$input.compiled_prompt\",\n \"image_urls\": [\n \"$node-xlast.images.0.url\"\n ],\n \"duration\": \"5\",\n \"resolution\": \"720p\",\n \"audio_urls\": [],\n \"generate_audio\": false\n }\n },\n \"node-merge\": {\n \"type\": \"run\",\n \"id\": \"node-merge\",\n \"depends\": [\n \"node-gen1\",\n \"node-gen2\",\n \"node-gen3\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 1180,\n \"y\": 0\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/merge-videos\",\n \"input\": {\n \"video_urls\": [\n \"$node-gen1.video.url\",\n \"$node-gen2.video.url\",\n \"$node-gen3.video.url\"\n ],\n \"target_fps\": 30\n }\n },\n \"output\": {\n \"type\": \"display\",\n \"id\": \"output\",\n \"depends\": [\n \"node-merge\"\n ],\n \"input\": {},\n \"metadata\": {\n \"position\": {\n \"x\": 1600,\n \"y\": 0\n }\n },\n \"fields\": {\n \"reel\": \"$node-merge.video\",\n \"take_1\": \"$node-gen1.video\",\n \"take_2\": \"$node-gen2.video\",\n \"take_3\": \"$node-gen3.video\"\n }\n }\n },\n \"output\": {\n \"reel\": \"$node-merge.video\",\n \"take_1\": \"$node-gen1.video\",\n \"take_2\": \"$node-gen2.video\",\n \"take_3\": \"$node-gen3.video\"\n },\n \"schema\": {\n \"input\": {\n \"source_video\": {\n \"name\": \"video_url\",\n \"label\": \"Own-content source video\",\n \"description\": \"Content reference; taste structure comes from compiled_prompt.\",\n \"required\": true,\n \"defaultValue\": \"\",\n \"examples\": [],\n \"ui\": {},\n \"type\": \"string\",\n \"modelId\": \"node-xfirst\"\n },\n \"compiled_prompt\": {\n \"name\": \"text\",\n \"label\": \"Compiled WHAT / HOW / GRADE prompt\",\n \"description\": \"Compile offline with workflow_graphs.py.\",\n \"required\": true,\n \"defaultValue\": \"\",\n \"examples\": [],\n \"ui\": {\n \"field\": \"textarea\"\n },\n \"type\": \"string\",\n \"modelId\": \"node-gen1\"\n }\n },\n \"output\": {}\n },\n \"version\": \"1\",\n \"metadata\": {\n \"input\": {\n \"position\": {\n \"x\": 0,\n \"y\": 0\n }\n }\n }\n }\n}\n","content_type":"application/json; charset=utf-8","language":"json","size":5271,"content_sha256":"a471a621bc456effae8f31fd567720620280536fcf42dffe49e493a221177847"},{"filename":"workflows/taste-distill.json","content":"{\n \"name\": \"taste-distill\",\n \"title\": \"taste-distill\",\n \"contents\": {\n \"name\": \"workflow\",\n \"nodes\": {\n \"output\": {\n \"type\": \"display\",\n \"id\": \"output\",\n \"depends\": [\n \"node-vision\"\n ],\n \"input\": {},\n \"metadata\": {\n \"position\": {\n \"x\": 2800,\n \"y\": 0\n }\n },\n \"fields\": {\n \"output\": \"$node-vision.output\"\n }\n },\n \"node-reference1\": {\n \"type\": \"run\",\n \"id\": \"node-reference1\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 1000,\n \"y\": 1200\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.reference_1\",\n \"frame_type\": \"middle\"\n }\n },\n \"node-vision\": {\n \"type\": \"run\",\n \"id\": \"node-vision\",\n \"depends\": [\n \"node-reference1\",\n \"node-reference2\",\n \"node-reference3\",\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 1900,\n \"y\": 0\n }\n },\n \"app\": \"openrouter/router/vision\",\n \"input\": {\n \"image_urls\": [\n \"$node-reference1.images.0.url\",\n \"$node-reference2.images.0.url\",\n \"$node-reference3.images.0.url\"\n ],\n \"prompt\": \"These frames are sampled from separate reference videos that share one visual identity. Distill the identity COMMON to all of them, not the content of any single frame. Return ONLY a JSON object, no prose and no markdown fences, with exactly these keys: palette_description (string), grain (string), lighting (string), focal_length (string), camera_motion (string), subject_framing (string), grade_description (string), mood_adjectives (array of strings), avoid (array of strings). Describe only what is visually verifiable across the set. The 'avoid' list names things that would break this look if introduced.\",\n \"system_prompt\": \"$input.measured_grounding\",\n \"model\": \"google/gemini-2.5-flash\"\n }\n },\n \"node-reference3\": {\n \"type\": \"run\",\n \"id\": \"node-reference3\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 1000,\n \"y\": 600\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.reference_3\",\n \"frame_type\": \"middle\"\n }\n },\n \"node-reference2\": {\n \"type\": \"run\",\n \"id\": \"node-reference2\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 1000,\n \"y\": 0\n }\n },\n \"app\": \"fal-ai/ffmpeg-api/extract-frame\",\n \"input\": {\n \"video_url\": \"$input.reference_2\",\n \"frame_type\": \"middle\"\n }\n }\n },\n \"output\": {\n \"output\": \"$node-vision.output\"\n },\n \"schema\": {\n \"input\": {\n \"reference_1\": {\n \"name\": \"video_url\",\n \"label\": \"Video_url Field\",\n \"description\": \"A video_url field\",\n \"required\": true,\n \"defaultValue\": \"\",\n \"examples\": [],\n \"ui\": {},\n \"type\": \"string\",\n \"modelId\": \"node-reference1\"\n },\n \"reference_2\": {\n \"name\": \"video_url_1\",\n \"label\": \"Video_url_1 Field\",\n \"description\": \"A video_url field\",\n \"required\": true,\n \"defaultValue\": \"\",\n \"examples\": [],\n \"ui\": {},\n \"type\": \"string\",\n \"modelId\": \"node-reference2\"\n },\n \"reference_3\": {\n \"name\": \"video_url_1\",\n \"label\": \"Video_url_1 Field\",\n \"description\": \"A video_url field\",\n \"required\": true,\n \"defaultValue\": \"\",\n \"examples\": [],\n \"ui\": {},\n \"type\": \"string\",\n \"modelId\": \"node-reference3\"\n },\n \"measured_grounding\": {\n \"name\": \"measured_grounding\",\n \"label\": \"Supplied genre and measured grounding\",\n \"description\": \"Compile offline from measured evidence for these references.\",\n \"type\": \"string\",\n \"ui\": {\n \"field\": \"textarea\"\n },\n \"examples\": [],\n \"modelId\": \"node-vision\",\n \"defaultValue\": \"\",\n \"required\": true\n }\n },\n \"output\": {}\n },\n \"version\": \"1\",\n \"metadata\": {\n \"input\": {\n \"position\": {\n \"x\": 0,\n \"y\": 0\n }\n }\n }\n }\n}\n","content_type":"application/json; charset=utf-8","language":"json","size":4690,"content_sha256":"58c3abb1839211f70f454ad7b0df6100bfaf5208e21d850240c7156715562db9"},{"filename":"workflows/taste-prop3d.json","content":"{\n \"name\": \"taste-prop3d\",\n \"title\": \"taste-prop3d\",\n \"contents\": {\n \"name\": \"workflow\",\n \"nodes\": {\n \"node-plate\": {\n \"type\": \"run\",\n \"id\": \"node-plate\",\n \"depends\": [\n \"input\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 380,\n \"y\": 0\n }\n },\n \"app\": \"fal-ai/nano-banana-pro\",\n \"input\": {\n \"prompt\": \"$input.prop\",\n \"num_images\": 1,\n \"aspect_ratio\": \"1:1\",\n \"output_format\": \"png\"\n }\n },\n \"node-3d\": {\n \"type\": \"run\",\n \"id\": \"node-3d\",\n \"depends\": [\n \"node-plate\"\n ],\n \"metadata\": {\n \"position\": {\n \"x\": 800,\n \"y\": 0\n }\n },\n \"app\": \"fal-ai/hunyuan-3d/v3.1/pro/image-to-3d\",\n \"input\": {\n \"input_image_url\": \"$node-plate.images.0.url\",\n \"generate_type\": \"Normal\",\n \"enable_pbr\": true,\n \"face_count\": 300000\n }\n },\n \"output\": {\n \"type\": \"display\",\n \"id\": \"output\",\n \"depends\": [\n \"node-3d\"\n ],\n \"input\": {},\n \"metadata\": {\n \"position\": {\n \"x\": 1220,\n \"y\": 0\n }\n },\n \"fields\": {\n \"mesh_glb\": \"$node-3d.model_glb\",\n \"mesh_obj\": \"$node-3d.model_urls.obj\",\n \"preview\": \"$node-3d.thumbnail\",\n \"plate\": \"$node-plate.images.0\"\n }\n }\n },\n \"output\": {\n \"mesh_glb\": \"$node-3d.model_glb\",\n \"mesh_obj\": \"$node-3d.model_urls.obj\",\n \"preview\": \"$node-3d.thumbnail\",\n \"plate\": \"$node-plate.images.0\"\n },\n \"schema\": {\n \"input\": {\n \"prop\": {\n \"name\": \"prompt\",\n \"label\": \"Prop description (append the plate rules)\",\n \"description\": \"Prop description (append the plate rules)\",\n \"required\": true,\n \"defaultValue\": \"\",\n \"examples\": [],\n \"ui\": {\n \"field\": \"textarea\"\n },\n \"type\": \"string\",\n \"modelId\": \"node-plate\"\n }\n },\n \"output\": {}\n },\n \"version\": \"1\",\n \"metadata\": {\n \"input\": {\n \"position\": {\n \"x\": 0,\n \"y\": 0\n }\n }\n }\n }\n}\n","content_type":"application/json; charset=utf-8","language":"json","size":2302,"content_sha256":"ef16a86801696a9212ce2208d73d574712c3dda7ac64a574f56184f126930cd0"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Taste Application","type":"text"}]},{"type":"paragraph","content":[{"text":"The second half of the pipeline. ","type":"text"},{"text":"taste-distillation","type":"text","marks":[{"type":"strong"}]},{"text":" measures references into a style pack; this generates against that pack and cuts the result.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Execution and Delivery Contract","type":"text"}]},{"type":"paragraph","content":[{"text":"The original implementation ships here in ","type":"text"},{"text":"scripts/","type":"text","marks":[{"type":"code_inline"}]},{"text":"; no separate ","type":"text"},{"text":"ito-video","type":"text","marks":[{"type":"code_inline"}]},{"text":" checkout is required. Install ","type":"text"},{"text":"scripts/requirements.txt","type":"text","marks":[{"type":"code_inline"}]},{"text":" for local processing. Install ","type":"text"},{"text":"scripts/requirements-live.txt","type":"text","marks":[{"type":"code_inline"}]},{"text":" only for provider execution. Live uploads, submissions and downloads require explicit ","type":"text"},{"text":"TASTE_FORGE_ALLOW_LIVE=1","type":"text","marks":[{"type":"code_inline"}]},{"text":" in addition to credentials; set it only for the user's authorized run. ","type":"text"},{"text":"--dry-run","type":"text","marks":[{"type":"code_inline"}]},{"text":" remains credential-free and produces labelled placeholders. An ambiguous provider timeout is not retried as a new paid job. Inspect the provider request before deciding whether another submission is warranted.","type":"text"}]},{"type":"paragraph","content":[{"text":"Use existing completed takes without any provider calls:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"python scripts/pipeline.py --genre example --root stylepacks \\\n --takes media/take-a.mp4 media/take-b.mp4 --duration 12 --fps 30 \\\n --out out/review-v1.mp4","type":"text"}]},{"type":"paragraph","content":[{"text":"--duration","type":"text","marks":[{"type":"code_inline"}]},{"text":" is a ","type":"text"},{"text":"best-effort cadence target","type":"text","marks":[{"type":"strong"}]},{"text":", not an exact runtime. Complete shots may produce a shorter or longer edit; the assembler does not duplicate clips or add padding to meet the target. The output manifest retains actual ","type":"text"},{"text":"duration","type":"text","marks":[{"type":"code_inline"}]},{"text":" and adds ","type":"text"},{"text":"duration_contract","type":"text","marks":[{"type":"code_inline"}]},{"text":" with requested and actual seconds, shortfall, overrun, and the ","type":"text"},{"text":"cadence_target","type":"text","marks":[{"type":"code_inline"}]},{"text":" policy. Differences of at least one output frame are warned explicitly. No exact-duration mode is provided; when an exact runtime is required, inspect the receipt and revise or reject the cut before delivery.","type":"text"}]},{"type":"paragraph","content":[{"text":"The pipeline keeps each run's graded shot files because the editable FCPXML and EDL reference them. It refuses output collisions and reports timeline export failures. A completed render is not a saved editor project or creative approval. Preserve source assets and versioned project checkpoints before and after live edits; record the saved path and digest separately from the in-memory timeline receipt.","type":"text"}]},{"type":"paragraph","content":[{"text":"Clone-ready Fal graphs and their offline input compiler are documented in ","type":"text"},{"text":"workflows/README.md","type":"text","marks":[{"type":"code_inline"}]},{"text":". Compile brief and style direction before submission; do not assume a disconnected schema field affects a model prompt. Validate current endpoint fields against the actual provider before a paid run.","type":"text"}]},{"type":"paragraph","content":[{"text":"For Blender, use the full textured source GLB; retopology is a separately named derivative and never replaces that source. ","type":"text"},{"text":"blender_prop.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" supports explicit ","type":"text"},{"text":"--width 1920 --height 1080 --fps 30 --receipt receipt.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" and preserves pack-derived rim lighting and material textures. Saving a scene is distinct from rendering it; inspect the receipt's state and packed images.","type":"text"}]},{"type":"paragraph","content":[{"text":"For Resolve overlays, use ","type":"text"},{"text":"taste.resolve.apply_placements","type":"text","marks":[{"type":"code_inline"}]},{"text":" with injected objects from an explicitly selected, versioned target. Set ","type":"text"},{"text":"source_end_mode","type":"text","marks":[{"type":"code_inline"}]},{"text":" to the convention verified on that host. Do not assume an inclusive source end across Resolve versions. The adapter allocates overlapping effects above preserved tracks and checks every placement immediately and again after all appends. Composite integers must match the installed API; for the verified Studio 21 host, Screen is 5, not the historical builder's incorrect 22. The receipt proves in-memory placement only. Save the project, verify its checkpoint, then inspect the exact rendered output before reporting delivery.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"When to Activate","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"\"make a video in this style\" / \"apply the pack\" / \"supplement this footage\"","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Assembling generated clips into something with real edit rhythm","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Minting 3D props from a look and getting them back into the video","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Verifying that a finished piece actually matches its reference","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Division of Labour","type":"text"}]},{"type":"paragraph","content":[{"text":"The model supplies content, motion, framing and lighting structure. The pack supplies colour and rhythm.","type":"text","marks":[{"type":"strong"}]},{"text":" This is measured, not stylistic preference — see taste-distillation for the numbers. Practical consequences:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"The generation prompt contains ","type":"text"},{"text":"zero colour language","type":"text","marks":[{"type":"strong"}]},{"text":". Add ","type":"text"},{"text":"\"Colour: none. Render neutral. Grading is applied afterwards.\"","type":"text","marks":[{"type":"em"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Keep ","type":"text"},{"text":"two separate flags","type":"text","marks":[{"type":"strong"}]},{"text":": ","type":"text"},{"text":"--brief","type":"text","marks":[{"type":"code_inline"}]},{"text":" (what HAPPENS: subject, action, place) and ","type":"text"},{"text":"--style-steer","type":"text","marks":[{"type":"code_inline"}]},{"text":" (how it LOOKS). Merging them leaks style words into the scene (\"teal\" becomes a teal object) and subject words into the grade.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"The model responds to ","type":"text"},{"text":"local, checkable rules","type":"text","marks":[{"type":"strong"}]},{"text":" far better than global ones. \"Backgrounds pure black and unlit; subjects blowing toward white\" works; \"extreme contrast\" does not.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Generate TAKES, Not Shots","type":"text"}]},{"type":"paragraph","content":[{"text":"The obvious reading of \"match the cadence\" is one generation per shot. It is economically absurd. A reference averaging 0.78s/shot against an endpoint with a 4-second floor turns a 10s piece into ","type":"text"},{"text":"12 calls, 48 generated seconds for 10 used (21% efficiency)","type":"text","marks":[{"type":"strong"}]},{"text":", and twelve unrelated clips stitched into what should read as continuous.","type":"text"}]},{"type":"paragraph","content":[{"text":"Editors roll a longer take and cut inside it. Grouping shots into ~5s takes: ","type":"text"},{"text":"3 calls, 13 generated seconds, 77% efficiency","type":"text","marks":[{"type":"strong"}]},{"text":", and consecutive shots that actually belong to each other because they came from the same generation.","type":"text"}]},{"type":"paragraph","content":[{"text":"Tell the model what shape you want, or it renders a slow locked-off push and six cuts inside it read as a stutter:","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"\"Filmed as ONE continuous take with no hard cuts inside it. It will be cut into 6 pieces of roughly 0.8s in the edit, so the framing, subject and light must keep changing throughout — any 0.8s window has to stand alone as its own shot.\"","type":"text"}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Cutting Rules","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Re-encode, never stream-copy.","type":"text","marks":[{"type":"strong"}]},{"text":" Stream copy only cuts on keyframes, which at 0.78s mean shot length rounds every boundary to the nearest GOP — destroying the exact thing the pipeline exists to preserve.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"When supplementing existing footage, use its own shot boundaries.","type":"text","marks":[{"type":"strong"}]},{"text":" Slicing a base video into contiguous pieces and playing them in order just reassembles the original: every \"cut\" lands mid-shot and is invisible. Measured, a 20-shot assembly registered only 12 detected cuts. Detect real boundaries and take every Nth so consecutive picks are guaranteed discontinuous. That moved a cut measurement from 1.29s to 0.83s against a 0.78s target.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Sample shot lengths from the reference's distribution","type":"text","marks":[{"type":"strong"}]},{"text":", not from its mean, so the cut inherits rhythm variance instead of flattening to even clips.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Grading Rules","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Direct measurement beats a baked LUT","type":"text","marks":[{"type":"strong"}]},{"text":" when you have the clip: measure it, match its L* CDF, apply zone chroma. ","type":"text"},{"text":"grade_clip_direct","type":"text","marks":[{"type":"code_inline"}]},{"text":" reached MAE 1.58 and contrast 34.0 against 34.7.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Anchor, do not CDF-match, when the clip's histogram is unlike the reference's.","type":"text","marks":[{"type":"strong"}]},{"text":" Forcing a 68%-black generated clip onto a busy reference histogram lifted the entire background out of black: background preservation fell to 26.0% (CDF) versus 69.0% (anchor), while MAE and contrast both still looked excellent. Default to anchored tone.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Batch size matters.","type":"text","marks":[{"type":"strong"}]},{"text":" Grading 48 frames at once OOM-killed the process; 6 is safe.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Recovered fal Platform Behavior","type":"text"}]},{"type":"paragraph","content":[{"text":"These observations and endpoint examples came from the recovered workflow. Recheck current endpoint metadata; they are not guarantees about every future provider version. The bundled graph templates record the separately verified workflow inputs, including explicit generated-audio control.","type":"text"}]},{"type":"paragraph","content":[{"text":"These cost real time to discover. Check them before designing a graph.","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Limit","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Detail","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"No 3D renderer at all","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal has ","type":"text"},{"text":"image-to-3d","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"text-to-3d","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"3d-to-3d","type":"text","marks":[{"type":"code_inline"}]},{"text":" and nothing else. Every ","type":"text"},{"text":"3d-to-3d","type":"text","marks":[{"type":"code_inline"}]},{"text":" endpoint emits another mesh. There is no ","type":"text"},{"text":"3d-to-image","type":"text","marks":[{"type":"code_inline"}]},{"text":"/","type":"text"},{"text":"3d-to-video","type":"text","marks":[{"type":"code_inline"}]},{"text":" category, so a minted GLB ","type":"text"},{"text":"cannot","type":"text","marks":[{"type":"strong"}]},{"text":" re-enter a fal video graph. Render locally, then use ","type":"text"},{"text":"fal-ai/ffmpeg-api/images-to-video","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"compose cannot overlay","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal-ai/ffmpeg-api/compose","type":"text","marks":[{"type":"code_inline"}]},{"text":" rejects a second track with ","type":"text"},{"text":"\"Multiple video tracks are not supported\"","type":"text","marks":[{"type":"em"}]},{"text":" — and it counts an ","type":"text"},{"text":"image","type":"text","marks":[{"type":"code_inline"}]},{"text":" track as a video track. It sequences one video track only. ","type":"text"},{"text":"Composite locally with ffmpeg.","type":"text","marks":[{"type":"strong"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"compose keyframes are milliseconds","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Nothing in the response says so. A run submitted in seconds is accepted and returns a video that is 1000x too short.","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"extract-frame offers first/middle/last only","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"No arbitrary timestamp. Use the three as three distinct conditioning images.","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"No loops, no string concat in the DAG","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Per-shot fan-out has to be authored node by node, or kept local.","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Kling 3.0 has no reference-to-video","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"The v3 line is text/image/motion-control only; reference-to-video lives on the ","type":"text"},{"text":"o3","type":"text","marks":[{"type":"code_inline"}]},{"text":" line: ","type":"text"},{"text":"fal-ai/kling-video/o3/pro/reference-to-video","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Prefixes are not uniform","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"bytedance/*","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"tripo3d/*","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"meshy/*","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"minimax/*","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"openai/*","type":"text","marks":[{"type":"code_inline"}]},{"text":" carry ","type":"text"},{"text":"no","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"fal-ai/","type":"text","marks":[{"type":"code_inline"}]},{"text":" prefix. ","type":"text"},{"text":"kling-video","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"veo3.1","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"flux-*","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"hunyuan-3d","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"ffmpeg-api","type":"text","marks":[{"type":"code_inline"}]},{"text":" do.","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Endpoint picks","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Slot","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Best","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Value alternative","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"reference→video","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"bytedance/seedance-2.5/reference-to-video","type":"text","marks":[{"type":"code_inline"}]},{"text":" (~$0.473/s @720p)","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal-ai/kling-video/o3/pro/reference-to-video","type":"text","marks":[{"type":"code_inline"}]},{"text":" (~$0.112/s)","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"image→3D","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal-ai/hunyuan-3d/v3.1/pro/image-to-3d","type":"text","marks":[{"type":"code_inline"}]},{"text":" ($0.375, up to 8 views)","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"tripo3d/h3.1/image-to-3d","type":"text","marks":[{"type":"code_inline"}]},{"text":" ($0.20)","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"text→3D","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal-ai/hunyuan-3d/v3.1/pro/text-to-3d","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"tripo3d/h3.1/text-to-3d","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"retopology","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal-ai/hunyuan-3d/v3.1/smart-topology","type":"text","marks":[{"type":"code_inline"}]},{"text":" ($0.75)","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"tripo3d/tripo/remesh","type":"text","marks":[{"type":"code_inline"}]},{"text":" (~75x cheaper)","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"part split","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal-ai/hunyuan-3d/v3.1/part","type":"text","marks":[{"type":"code_inline"}]},{"text":" (FBX only)","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"tripo3d/tripo/segment","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"text→image","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal-ai/nano-banana-pro","type":"text","marks":[{"type":"code_inline"}]},{"text":" ($0.15 flat)","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"fal-ai/flux-2-pro","type":"text","marks":[{"type":"code_inline"}]},{"text":" ($0.03/MP)","type":"text"}]}]}]}]},{"type":"paragraph","content":[{"text":"Seedance is ~4x Kling o3's price for the same 5 seconds. It earns that on multi-reference fidelity (up to 50 mixed image/video/audio refs) and does ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" earn it when conditioning on a single still.","type":"text"}]},{"type":"paragraph","content":[{"text":"Model IDs and prices drift. Verify against fal.ai/models before promising any of them.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"The 3D Branch","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"python mint3d.py --genre \u003cname> --from-stills 4 --retopo --render\npython mint3d.py --genre \u003cname> --prompt \"a cracked chrome visor\" --render","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Generate a clean plate first; do not lift from reference stills.","type":"text","marks":[{"type":"strong"}]},{"text":" The endpoint's stated input requirement is simple background, single object, object >50% of frame. Reference reels are the opposite of that — collages, wide shots, several subjects, burnt-in graphics — and they produce sculpted noise. Text → single-object plate → mesh costs ~$0.15 extra and is the difference between a usable mesh and a discarded one.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Multi-view is named per-angle fields, not a list.","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"input_image_url","type":"text","marks":[{"type":"code_inline"}]},{"text":" (front, required), then ","type":"text"},{"text":"back_image_url","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"left_image_url","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"right_image_url","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"left_front_image_url","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"right_front_image_url","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"top_image_url","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"bottom_image_url","type":"text","marks":[{"type":"code_inline"}]},{"text":". There is no ","type":"text"},{"text":"input_image_urls","type":"text","marks":[{"type":"code_inline"}]},{"text":" and no ","type":"text"},{"text":"multi_view","type":"text","marks":[{"type":"code_inline"}]},{"text":" flag — inventing them degrades every mint to single-view while appearing to work. A wrong angle label is worse than omitting the view, because the model trusts it.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Address the GLB by key, not by position.","type":"text","marks":[{"type":"strong"}]},{"text":" The response carries a ","type":"text"},{"text":"thumbnail","type":"text","marks":[{"type":"code_inline"}]},{"text":" PNG and a ","type":"text"},{"text":"model_urls","type":"text","marks":[{"type":"code_inline"}]},{"text":" block alongside ","type":"text"},{"text":"model_glb","type":"text","marks":[{"type":"code_inline"}]},{"text":"; taking the first URL works only until the keys reorder, and a preview PNG downloads fine — nothing fails until Blender refuses to open it.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Request PBR maps.","type":"text","marks":[{"type":"strong"}]},{"text":" Without them the mesh lights like painted cardboard.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Retopologise","type":"text","marks":[{"type":"strong"}]},{"text":" if anyone will edit or rig it. Generated meshes are dense and chaotic.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Render locally to close the loop.","type":"text","marks":[{"type":"strong"}]},{"text":" Once a turntable is frames, it is footage, and every downstream stage already handles footage — grade it, cut it, screen it as an element, or upload it as a conditioning reference. Use Blender when a binary is on PATH; keep a dependency-light software rasteriser as the default, because a headless GL context is the single most common thing missing from a container and a renderer that only works on a workstation is not part of a pipeline.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Verify, Then Believe","type":"text"}]},{"type":"paragraph","content":[{"text":"Every other stage claims a result. Check it, and check ","type":"text"},{"text":"distribution shape","type":"text","marks":[{"type":"strong"}]},{"text":", not just moments:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"background","type":"text","marks":[{"type":"code_inline"}]},{"text":" — share of frame below L*10 vs the ","type":"text"},{"text":"pack's","type":"text","marks":[{"type":"strong"}]},{"text":" figure. Compare to the reference, ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" to the source clip: a generated source at 68% black is blacker than any reference in a 24–55% band, so \"preserve the source's blacks\" demands the wrong thing and equally excuses a lifted grade.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"chroma_mae","type":"text","marks":[{"type":"code_inline"}]},{"text":" — per-zone a*/b* error, using the ","type":"text"},{"text":"median","type":"text","marks":[{"type":"strong"}]},{"text":" (matching how the pack's targets were measured; a mean here compares a skew-sensitive statistic to a robust one and reports a definition mismatch as an error).","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"contrast","type":"text","marks":[{"type":"code_inline"}]},{"text":" / ","type":"text"},{"text":"black_point","type":"text","marks":[{"type":"code_inline"}]},{"text":" / ","type":"text"},{"text":"white_point","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"banding","type":"text","marks":[{"type":"code_inline"}]},{"text":" — empty L* histogram bins ","type":"text"},{"text":"between occupied ones","type":"text","marks":[{"type":"em"}]},{"text":". Counting total empty bins does not work: a legitimately dark clip has empty highlight bins.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"cadence","type":"text","marks":[{"type":"code_inline"}]},{"text":" — detected mean shot length vs the reference's.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Watch the ","type":"text"},{"text":"units trap","type":"text","marks":[{"type":"strong"}]},{"text":": OpenCV changes Lab convention with dtype. On float32, L* is 0–100 and a*/b* are signed; on uint8, L* is 0–255 and a*/b* are biased +128. Mixing them reports chroma errors in the hundreds.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Full Chain","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"python pipeline.py --genre \u003cname> --refs a.mov b.mov \\\n --brief \"what happens\" --duration 12 \\\n --base-video existing.mp4 --out out/FINAL.mp4","type":"text"}]},{"type":"paragraph","content":[{"text":"mint → distill → (mint3d) → apply → forge → verify. Stages 1–3 are cached, so iterating on briefs never re-measures anything. ","type":"text"},{"text":"--dry-run","type":"text","marks":[{"type":"code_inline"}]},{"text":" stubs every network call: the plan, prompts, track layout and manifest all still get exercised.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Anti-Patterns","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Don't","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Why","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"One generation per shot","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"21% efficiency, 12 unrelated clips","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Put colour in the prompt","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Measured not to work; pushes away from the neutral base the LUT wants","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Stream-copy the cuts","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Keyframe-only boundaries destroy sub-second rhythm","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Cut a base video contiguously","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Reassembles the original; every cut invisible","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Trust compose to overlay","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"It cannot; it rejects the second track","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Send seconds to compose","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Silently 1000x too short","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Ship on MAE alone","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Add the background-share check","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Borrowed Footage Carries the Capture App's UI","type":"text"}]},{"type":"paragraph","content":[{"text":"The single worst defect found in a delivered cut: the finished video shipped someone else's like button, view counter and comment bubble, because the base footage was a screen recording and nothing cropped them out.","type":"text"}]},{"type":"paragraph","content":[{"text":"content_mask","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" does not solve this and its bounding box makes it worse.","type":"text","marks":[{"type":"strong"}]},{"text":" Temporal variance keeps interface chrome, because chrome ","type":"text"},{"text":"animates","type":"text","marks":[{"type":"em"}]},{"text":" - the heart pulses, the counter ticks - so the mask marks it as moving content. Measured on three references, the mask bbox kept 100% of the width every time while the interface sat plainly in the right-hand margin.","type":"text"}]},{"type":"paragraph","content":[{"text":"The separating signal is the temporal ","type":"text"},{"text":"median","type":"text","marks":[{"type":"strong"}]},{"text":", not the variance. Footage moves, so the median of many frames averages into mush with almost no edge energy; chrome sits at fixed coordinates, so its edges survive intact. Sobel energy on the median frame lights up on chrome and goes quiet on content - measured, a right-hand column read 0.23 against an interior background of 0.03 on one reference and 0.31 against 0.15 on another.","type":"text"}]},{"type":"paragraph","content":[{"text":"Two implementation details that cost a cycle each:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Trim past the innermost outlier in each outer band, not inward from the edge.","type":"text","marks":[{"type":"strong"}]},{"text":" Walking in while the current line is hot stops immediately, because the outermost lines are letterbox - flat black, zero edge energy - and the chrome sits ","type":"text"},{"text":"inside","type":"text","marks":[{"type":"em"}]},{"text":" that at 90-95% of width. The naive version trimmed 1% of frame while the like button stayed in shot.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Scale to cover, not pad,","type":"text","marks":[{"type":"strong"}]},{"text":" when portrait source lands in a landscape cut. Padding 9:16 (narrower still after the UI crop) into 16:9 left ~60% of frame as black bars, one shot was nearly an empty rectangle, and it poisoned the background metric because bars are pure black. Covering loses the sides, which is the right trade for centre-framed material.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Overlay Plates Are Elements, Not Washes","type":"text"}]},{"type":"paragraph","content":[{"text":"A glow plate is ~4% covered by construction. Composite it at frame size and you get a small bright dot parked mid-shot - it reads as a sticker, and it was visible in a delivered cut as an unexplained coloured blob.","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Tighten each plate to its alpha bounding box first.","type":"text","marks":[{"type":"strong"}]},{"text":" That raises coverage from ~4% to 15-35% and hands size control to the caller instead of inheriting whatever fraction of the source frame the element happened to occupy.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Choose wash vs element by coverage.","type":"text","marks":[{"type":"strong"}]},{"text":" Diffuse plates (\u003c10% after tightening) work stretched full-frame at low opacity; concentrated ones want to be scaled to 35-70% of frame width and placed.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Vary placement, scale and rotation per shot","type":"text","marks":[{"type":"strong"}]},{"text":" from a seeded RNG, so the cut stays reproducible but no two stamped shots share a mark. One plate in one spot every Nth shot reads as a watermark.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Resolve element geometry in Python, not in ffmpeg expressions: ","type":"text"},{"text":"pad()","type":"text","marks":[{"type":"code_inline"}]},{"text":" rejects a negative offset and cannot pad below its input size, so an oversized or off-frame element kills the whole filtergraph.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Downstream Handoff (Resolve / Blender)","type":"text"}]},{"type":"paragraph","content":[{"text":"Always ship an editable timeline beside the mp4.","type":"text","marks":[{"type":"strong"}]},{"text":" The flattened video is a viewing copy and the one thing a colourist cannot work with — every cut is baked in and the shots are no longer separable. ","type":"text"},{"text":"forge.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" writes FCPXML 1.9 and a CMX3600 EDL referencing the individual graded shot files, so the piece lands as a timeline that can be re-cut and re-graded.","type":"text"}]},{"type":"paragraph","content":[{"text":"Validate the export, do not assume it: check that asset-clip offsets equal the running sum of prior durations (no gaps), that every ","type":"text"},{"text":"ref","type":"text","marks":[{"type":"code_inline"}]},{"text":" resolves to a declared asset, that every ","type":"text"},{"text":"media-rep src","type":"text","marks":[{"type":"code_inline"}]},{"text":" exists on disk, and that the total matches the mp4. A timeline that imports but drifts is worse than one that fails loudly.","type":"text"}]},{"type":"paragraph","content":[{"text":"Two things that silently destroy the work:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Project frame rate must be set before import.","type":"text","marks":[{"type":"strong"}]},{"text":" Resolve locks timeline fps on first timeline creation and conforms the cadence silently. At a sub-second mean shot length that conform is visible.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"The delivered shots are already graded.","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"look.cube","type":"text","marks":[{"type":"code_inline"}]},{"text":" is a ","type":"text"},{"text":"normalising","type":"text","marks":[{"type":"em"}]},{"text":" LUT for new material and for matching — applying it to the supplied shots double-grades them. Node 1, nothing before it, corrections after.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Ship a handoff doc with the measured targets in it (","type":"text"},{"text":"scripts/HANDOFF-TEMPLATE.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" is a filled example): the zone chroma table, the cadence distribution including ","type":"text"},{"text":"rhythm variance","type":"text","marks":[{"type":"strong"}]},{"text":" — an editor who matches the mean but not the variance produces something that reads completely differently — and an explicit \"do not\" list.","type":"text"}]},{"type":"paragraph","content":[{"text":"For Blender, ","type":"text"},{"text":"scripts/blender_prop.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" derives its lighting from the pack: black world with transparent film (so renders composite with no keying), key plus rim (a key alone lets the silhouette die against black), rim colour converted from the pack's peak-chroma zone via Lab→linear sRGB so the prop picks up the same cast the footage is graded to. ","type":"text"},{"text":"View transform Standard, not AgX/Filmic, and no grading in Blender","type":"text","marks":[{"type":"strong"}]},{"text":" — AgX applies its own tone curve before the LUT ever sees the pixels, and grading twice compounds.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Bundled Code","type":"text"}]},{"type":"paragraph","content":[{"text":"scripts/","type":"text","marks":[{"type":"code_inline"}]},{"text":" in this skill is a working implementation, not pseudocode. It has no project-specific assumptions: point it at any reference videos and it produces a pack.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"pip install -r scripts/requirements.txt\nexport FAL_KEY=... # only needed for the stages that call fal","type":"text"}]},{"type":"paragraph","content":[{"text":"Every network call is stubbed under ","type":"text"},{"text":"TASTE_FORGE_DRY_RUN=1","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"--dry-run","type":"text","marks":[{"type":"code_inline"}]},{"text":", so the plan, prompts, track layout and manifest can be inspected without spending.","type":"text"}]}]},"metadata":{"date":"2026-11-02","name":"taste-application","author":"@skillopedia","source":{"stars":295436,"repo_name":"ECC","origin_url":"https://github.com/affaan-m/ECC/blob/HEAD/skills/taste-application/SKILL.md","repo_owner":"affaan-m","body_sha256":"422a07f05e0dfc16a1bcc14895ff6f34891473ddec28981c4fa7986591165d2e","cluster_key":"a0d39c19b90aa883d1bd6a56463d1aeb7744806ddd194e18111ce51bb1250345","clean_bundle":{"format":"clean-skill-bundle-v1","source":"affaan-m/ECC/skills/taste-application/SKILL.md","text_bytes":744477,"source_commit":"b38be6a7f6350964d59328b52f0624e1d1d552d1","skills_sh_rank":8012,"attachment_count":77,"text_attachments":77,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/taste-application/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"media-content","category_label":"Media"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"media-content","metadata":{"origin":"ECC"},"import_tag":"skills-sh-top20k-2026-11","description":"Generate new video against a distilled style pack and cut it into a finished piece - plan takes from the reference's cut rhythm, generate on fal, grade with the pack's measured LUT, cut at the measured cadence, weave in existing footage, composite overlay plates, mint 3D props, and verify the result numerically. Use when the user wants to make a video in a captured style, supplement existing footage, or assemble generated clips into a real edit."}},"renderedAt":1793666079875}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.