From 262feffe1f9a91c893c68877be0dff6fa442dbbf Mon Sep 17 00:00:00 2001 From: Affaan Mustafa Date: Mon, 7 Sep 2026 14:17:02 +0300 Subject: [PATCH] fix: report best-effort duration targets and shortfalls --- skills/taste-application/SKILL.md | 9 +++ skills/taste-application/scripts/apply.py | 2 +- skills/taste-application/scripts/forge.py | 19 ++++- skills/taste-application/scripts/pipeline.py | 3 +- tests/test_taste_overlays.py | 81 ++++++++++++++++++++ 5 files changed, 110 insertions(+), 4 deletions(-) diff --git a/skills/taste-application/SKILL.md b/skills/taste-application/SKILL.md index 70553db58..2e080c2fb 100644 --- a/skills/taste-application/SKILL.md +++ b/skills/taste-application/SKILL.md @@ -29,6 +29,15 @@ python scripts/pipeline.py --genre example --root stylepacks \ --out out/review-v1.mp4 ``` +`--duration` is a **best-effort cadence target**, 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 +`duration` and adds `duration_contract` with requested and actual seconds, +shortfall, overrun, and the `cadence_target` 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. + 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 diff --git a/skills/taste-application/scripts/apply.py b/skills/taste-application/scripts/apply.py index 742a6af47..16c785143 100644 --- a/skills/taste-application/scripts/apply.py +++ b/skills/taste-application/scripts/apply.py @@ -566,7 +566,7 @@ def main() -> None: help="WHAT HAPPENS: subject and action, e.g. 'a courier weaves " "through night traffic'") ap.add_argument("--duration", type=float, default=20.0, - help="target total seconds; shot lengths are drawn from the pack's cadence") + help="best-effort target seconds, not exact; shot lengths follow the pack's cadence and actual assembly duration is reported") ap.add_argument("--base-video", help="optional existing footage; each shot is conditioned on the frame " "at its own timecode so generated shots supplement the edit") diff --git a/skills/taste-application/scripts/forge.py b/skills/taste-application/scripts/forge.py index 09cd98f63..718664d21 100644 --- a/skills/taste-application/scripts/forge.py +++ b/skills/taste-application/scripts/forge.py @@ -240,12 +240,25 @@ def forge( break kept.append(c); acc += d if kept: - print(f" trimmed {len(order)} -> {len(kept)} shots to hit {duration:.1f}s") + print(f" trimmed {len(order)} -> {len(kept)} shots toward target {duration:.1f}s") order = kept out_path = Path(out) asm.concat(order, out_path, fps=FPS) final = frame_mod.probe(out_path) + duration_delta = final.duration - duration if duration is not None else 0.0 + duration_contract = { + "policy": "cadence_target", + "requested_seconds": duration, + "actual_seconds": round(final.duration, 6), + "shortfall_seconds": round(max(0.0, -duration_delta), 6), + "overrun_seconds": round(max(0.0, duration_delta), 6), + } + if duration is not None and abs(duration_delta) + 1e-9 >= 1.0 / FPS: + print(f" WARNING: cadence target {duration:.3f}s produced {final.duration:.3f}s " + f"(shortfall {duration_contract['shortfall_seconds']:.3f}s, " + f"overrun {duration_contract['overrun_seconds']:.3f}s); " + "whole cadence shots are preserved without padding or duplication") # Ship an EDITABLE timeline beside the flattened mp4. # @@ -288,6 +301,7 @@ def forge( "generated_shots": len(gen_shots), "base_shots": len(base_shots), "duration": round(final.duration, 3), + "duration_contract": duration_contract, "grade_strength": strength, "timelines": timelines, "shot_files": [str(Path(c).resolve()) for c in order], @@ -311,7 +325,8 @@ def main() -> None: ap.add_argument("--overlays", nargs="*", default=None, help="overlay image paths") ap.add_argument("--overlay-every", type=int, default=4) ap.add_argument("--overlay-opacity", type=float, default=0.3) - ap.add_argument("--duration", type=float, default=None) + ap.add_argument("--duration", type=float, default=None, + help="best-effort cadence target in seconds, not an exact output duration") ap.add_argument("--strength", type=float, default=1.0, help="0-1 grade intensity") ap.add_argument("--width", type=int, default=None) ap.add_argument("--take-len", type=float, default=5.0) diff --git a/skills/taste-application/scripts/pipeline.py b/skills/taste-application/scripts/pipeline.py index 6bf801ffd..ea4a2c9b2 100644 --- a/skills/taste-application/scripts/pipeline.py +++ b/skills/taste-application/scripts/pipeline.py @@ -58,7 +58,8 @@ def main() -> None: ap.add_argument("--fps", type=float, default=None, help="output frame rate") ap.add_argument("--brief", default="", help="WHAT HAPPENS in the new piece") ap.add_argument("--style-steer", default="", help="HOW IT LOOKS, per-run nudge") - ap.add_argument("--duration", type=float, default=12.0) + ap.add_argument("--duration", type=float, default=12.0, + help="best-effort cadence target in seconds, not an exact duration; actual result is reported") ap.add_argument("--base-video", default=None, help="existing footage to supplement") ap.add_argument("--base-ratio", type=float, default=0.35) ap.add_argument("--out", default=None) diff --git a/tests/test_taste_overlays.py b/tests/test_taste_overlays.py index 8c790bb73..bf7ebd067 100644 --- a/tests/test_taste_overlays.py +++ b/tests/test_taste_overlays.py @@ -1,11 +1,13 @@ """Requested image overlays must fail closed if compositing fails.""" import importlib.util +import io import shutil import subprocess import sys import tempfile import unittest +from contextlib import redirect_stdout from pathlib import Path from types import SimpleNamespace from unittest.mock import patch @@ -174,5 +176,84 @@ class OverlayFailureTests(unittest.TestCase): self.assertEqual(plate.read_bytes(), b"original image") +class DurationContractTests(unittest.TestCase): + def test_cadence_target_records_actual_duration_and_warns_on_frame_difference(self): + for requested, shortfall, overrun, warning in ( + (2.0, 0.7, 0.0, True), + (1.3, 0.0, 0.0, False), + (1.3 + 1 / 30, 0.033333, 0.0, True), + (1.31, 0.01, 0.0, False), + (1.0, 0.0, 0.3, True), + (None, 0.0, 0.0, False), + ): + with ( + self.subTest(requested=requested), + tempfile.TemporaryDirectory() as directory, + ): + root = Path(directory) + take, out = root / "take.mp4", root / "out.mp4" + take.write_bytes(b"original") + info = SimpleNamespace(width=320, height=180, fps=30, duration=1.3) + stats = SimpleNamespace(contrast=1, black_point=0, white_point=1) + stdout = io.StringIO() + + def timeline(*args, **kwargs): + path = kwargs["out_path"] + path.touch() + return path + + with ( + patch.object( + forge.pack_mod, + "load", + return_value=SimpleNamespace( + grade_path="grade", cadence_path="cadence" + ), + ), + patch.object(forge.grade_mod, "load_stats", return_value=stats), + patch.object( + forge.cad_mod, + "load", + return_value=SimpleNamespace( + mean_shot=1, cuts_per_min=60, rhythm_variance=0 + ), + ), + patch.object(forge.frame_mod, "probe", return_value=info), + patch.object(forge.asm, "normalize", return_value=take), + patch.object(forge.grade_mod, "grade_clip_direct"), + patch.object(forge.asm, "cut_take", return_value=[take]), + patch.object(forge.asm, "concat"), + patch.object(forge.tl_mod, "write_timeline", side_effect=timeline), + patch.object(forge.asm, "write_manifest") as manifest, + redirect_stdout(stdout), + ): + forge.forge( + "look", + [str(take)], + str(out), + duration=requested, + work=str(root / "work"), + fps=30, + plan=[{"shots": [{"start": 0, "duration": 1.3}]}], + ) + receipt = manifest.call_args.args[1] + self.assertEqual(receipt["duration"], 1.3) + self.assertEqual( + receipt["duration_contract"], + { + "policy": "cadence_target", + "requested_seconds": requested, + "actual_seconds": 1.3, + "shortfall_seconds": shortfall, + "overrun_seconds": overrun, + }, + ) + self.assertEqual( + "WARNING: cadence target" in stdout.getvalue(), warning + ) + self.assertNotIn("to hit", stdout.getvalue()) + self.assertEqual(take.read_bytes(), b"original") + + if __name__ == "__main__": unittest.main()