From 58a8e3b900f34c892aed593a4ca392526bee2677 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 06:28:45 +0000 Subject: [PATCH 01/26] feat: backport tool, command hub, restored gate APIs and memory fixes - scripts/canonical_backport.py (+ self-test): plan public-to-canonical backports with three-way merges, style-preserving writes and rollback - scripts/hao_autopilot.py: one command hub for the kit's entry points - shorts_gate: restore DEFAULT_RULES, merge_rules and rules= so example 04 runs again; reading speed counts half-width characters as half units - teardown: restore rhythm_stats, pace_profile and ocr_status for example 06 - knowledge_lifecycle: learning lives in data/knowledge_state.json, seeded once from the shipped state; support counts distinct sources - context_router: find references in the public layout and route only to references that exist in the install - shorts_autopilot: --platform for scan; usage errors and --help have no side effects Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XLikBpKWqhMahFDuvVj6wh --- examples/04_shorts_gate.py | 23 +- scripts/canonical_backport.py | 471 +++++++++++++++++++ scripts/canonical_backport_selftest.py | 217 +++++++++ scripts/hao_autopilot.py | 111 +++++ src/context_router.py | 108 ++++- src/knowledge_lifecycle.py | 189 +++++++- src/longform_maker/shorts_gate.py | 146 +++++- src/longform_maker/shorts_gate_validation.py | 4 +- src/shorts_autopilot.py | 29 +- src/teardown.py | 103 +++- 10 files changed, 1314 insertions(+), 87 deletions(-) create mode 100644 scripts/canonical_backport.py create mode 100644 scripts/canonical_backport_selftest.py create mode 100644 scripts/hao_autopilot.py diff --git a/examples/04_shorts_gate.py b/examples/04_shorts_gate.py index 8bdf6a0..1e50cb9 100644 --- a/examples/04_shorts_gate.py +++ b/examples/04_shorts_gate.py @@ -7,8 +7,8 @@ that fire at build time, so nobody has to remember them. This demo runs the gate three times: - 1) a BROKEN spec → 3 blocking failures (duration dead zone / slow first cut / - missing opening ID) + 1) a BROKEN spec → 2 blocking failures (duration dead zone / slow first cut) + plus an S-A warning (the "what is this" line is missing) 2) the FIXED spec → passes, and the gate hands back caption times it computed from segment indexes (you never hand-type timecodes) 3) YOUR OWN band → the same 31s spec that the default rules reject, accepted @@ -22,6 +22,11 @@ Note the S-O line in the reports: it is a **warning, not a failure**. Warnings tell you something and let the build through; only `[FAIL]` stops it. +S-R (reading speed) counts full-width CJK characters as 1 unit and half-width +letters/digits as 0.5, so the same thresholds work for English captions. Each +caption still needs enough screen time to be read - which is why the fixed spec +gives the opening hook its own segment instead of squeezing two lines into 2s. + Run: python examples/04_shorts_gate.py @@ -65,7 +70,7 @@ def show(title, spec, rules=None): print("\n" + "=" * 64) print("%s -> %s" % (title, "PASS" if ok else "FAIL")) print("=" * 64) - print(" platform : %s" % rep.get("platform", "?")) + print(" platform : %s" % spec.get("platform", "yt_shorts (default)")) print(" duration : %.1fs" % rep.get("dur", 0.0)) if rep.get("cap_rate") is not None: print(" captions : %.1f lines/min, median dwell %.2fs" @@ -85,7 +90,7 @@ def show(title, spec, rules=None): def broken(): - """Three rule breaks at once — each one alone is enough to block the build.""" + """Two blocking breaks plus a missing "what" line (S-A warns, S-B/S-C block).""" return dict( name="short_demo_broken", place="Riverside Market", @@ -110,7 +115,7 @@ def broken(): def fixed(): - """Same footage plan, all three breaks repaired.""" + """Same footage plan with the breaks repaired and readable caption timing.""" return dict( name="short_demo_fixed", place="Riverside Market", @@ -124,10 +129,10 @@ def fixed(): (CLIP, 2.4, 1.6), # loop: 2.4 + 1.6 = 4.0 == first segment's in-point ], # 13.4s total — inside the band caps_by_seg=[ - (0, [("Riverside Market", "gold")], "hook"), # who/where - (0, [("a 40-year-old noodle stall", "white")], "sub"), # what is this - (1, [("hand-pulled every morning", "white")], "sub"), - (2, [("the broth simmers 8 hours", "white")], "sub"), + (0, [("Riverside Market", "gold")], "hook"), # who/where, alone in the 2s opener + (1, [("a 40-year-old noodle stall", "white")], "sub"), # what is this + (2, [("hand-pulled every morning", "white")], "sub"), + (3, [("8-hour broth", "white")], "sub"), # two short lines share 3.4s (3, [("USD 3 a bowl", "white")], "sub"), ], bgm_folder="", diff --git a/scripts/canonical_backport.py b/scripts/canonical_backport.py new file mode 100644 index 0000000..7ee1103 --- /dev/null +++ b/scripts/canonical_backport.py @@ -0,0 +1,471 @@ +#!/usr/bin/env python3 +"""Backport public-kit edits into the private canonical Video Autopilot skill. + +``sync_canonical.py`` publishes the canonical skill through privacy transforms, +so an edit made only in the public kit is overwritten by the next sync. This +tool replays the public delta between two commits onto the canonical files with +a three-way merge (``git merge-file``). Canonical-only text that the public +transforms removed or rewrote is kept; overlapping edits become conflicts +instead of silent replacements. + +Dry-run is the default. ``--apply`` is all-or-nothing unless ``--partial`` is +given; it backs up every touched canonical file, writes atomically, keeps the +canonical BOM and line endings, and never writes a conflicted merge. After an +apply, re-run ``sync_canonical.py --canonical ``: when the backport is +exact, ``git status`` shows only ``sync-receipt.json`` (new canonical hashes). +""" +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time +from dataclasses import dataclass, field +from pathlib import Path + +from public_sync_inventory import ( + CLEANUP_HELPER_FILES, + DRAMA_MODULES, + KNOWLEDGE_FILES, + LONGFORM_MODULES, + PUBLIC_OWNED_PATHS, + REFERENCE_FILES, + ROOT_MODULES, + SANITIZED_KNOWLEDGE_FILES, + SILENT_VLOG_MODULES, + SYNC_RECEIPT_PATH, + WORKFLOW_SKILL_FILES, +) + +SKILL_DIR = "codex-skill/video-autopilot" +SKILL_ENTRY = SKILL_DIR + "/SKILL.md" +# (public prefix, managed leaf names, canonical root key, canonical prefix) +GROUPS = ( + ("src/", ROOT_MODULES, "canonical", ""), + ("src/longform_maker/", LONGFORM_MODULES, "canonical", "longform_maker/"), + ("src/silent_vlog_maker/", SILENT_VLOG_MODULES, "canonical", "silent_vlog_maker/"), + ("src/drama_pipeline/", DRAMA_MODULES, "canonical", "drama_pipeline/"), + ("knowledge/runtime/", KNOWLEDGE_FILES, "canonical", "knowledge/"), + (SKILL_DIR + "/references/", REFERENCE_FILES, "canonical", "references/"), + (SKILL_DIR + "/", WORKFLOW_SKILL_FILES, "canonical", ""), + (SKILL_DIR + "/agents/", ("openai.yaml",), "canonical", "agents/"), + ("", ("audit.config.json",), "canonical", ""), + ("tools/code-cleanup-helper/", CLEANUP_HELPER_FILES, "cleanup", ""), + ("docs/", ("AUTOPILOT_ARCHITECTURE_V6.md",), "workspace", ""), +) +GENERATED = { + SYNC_RECEIPT_PATH: "regenerated by sync_canonical.py", + "AUTOPILOT_MANIFEST.json": "generated by sync_canonical.py", + SKILL_ENTRY: "rendered from scripts/public_skill_skeleton.py; merge the " + "exported delta into the private canonical SKILL.md by hand", +} +PROJECTIONS = { + "knowledge/runtime/state.json": "public empty-ledger projection; never backport", +} +REVIEW_NOTES = { + "src/broll_qa.py": "public form is rendered by public_sync_renderers.py", + "src/editorial_templates.py": "public form is rendered by public_sync_renderers.py", + "audit.config.json": "public form prefixes architecture paths with src/", + "tools/code-cleanup-helper/audit.config.json": "public form drops privacy tokens", + **{ + "knowledge/runtime/" + name: "public form is a sanitized JSON projection" + for name in SANITIZED_KNOWLEDGE_FILES + }, +} +ROOT_FLAGS = {"canonical": "--canonical", "cleanup": "--cleanup-helper", "workspace": "--workspace"} +WRITES = ("FAST_FORWARD", "MERGED", "CREATE", "DELETE") +# Public-only rewrites live in these scripts. When they change between base and +# head, the public delta contains rewrites that must never reach the canonical. +TRANSFORM_SCRIPTS = frozenset({ + "scripts/sync_canonical.py", "scripts/public_sync_transforms.py", + "scripts/public_sync_support.py", "scripts/public_sync_renderers.py", + "scripts/public_privacy_sanitizer.py", "scripts/public_privacy_legacy.py", + "scripts/public_privacy_profiles.py", "scripts/public_privacy_references.py", + "scripts/public_privacy_sources.py", +}) +TRANSFORM_WORKER = r""" +import json, sys +from pathlib import Path +from sync_canonical import _transform +from public_privacy_sanitizer import sanitize_public_text +result = {} +for relative, text in json.load(sys.stdin).items(): + try: + result[relative] = sanitize_public_text(relative, _transform(text, Path(relative).name)) + except Exception: # noqa: BLE001 - the caller falls back to the raw base text + result[relative] = None +json.dump(result, sys.stdout, ensure_ascii=False) +""" +BLOCKERS = ("CONFLICT", "ERROR") + + +@dataclass(frozen=True) +class Target: + kind: str + root: str = "" + relative: str = "" + note: str = "" + + +@dataclass(frozen=True) +class TextStyle: + bom: bool = False + crlf: bool = False + + +@dataclass +class Item: + target: Target + sources: list = field(default_factory=list) + status: str = "SKIPPED" + detail: str = "" + result: object = None + style: TextStyle = TextStyle() + conflicts: int = 0 + + def row(self) -> dict: + return { + "status": self.status, "kind": self.target.kind, + "public": self.sources, "root": self.target.root, + "canonical": self.target.relative, "detail": self.detail, + "conflicts": self.conflicts, + } + + +def classify(relative: str) -> Target: + """Map one public path to its canonical source, or explain why none.""" + if relative in GENERATED: + return Target("generated", note=GENERATED[relative]) + if relative in PROJECTIONS: + return Target("projection", note=PROJECTIONS[relative]) + if relative in PUBLIC_OWNED_PATHS: + return Target("public-owned", note="owned by the public kit; nothing to backport") + for prefix, names, root, canonical_prefix in GROUPS: + if relative.startswith(prefix) and relative[len(prefix):] in names: + leaf = relative[len(prefix):] + return Target("derived", root, canonical_prefix + leaf, REVIEW_NOTES.get(relative, "")) + return Target("repo-native", note="lives only in the public kit") + + +def decode(payload: bytes) -> tuple: + bom = payload.startswith(b"\xef\xbb\xbf") + text = payload.decode("utf-8-sig") + crlf = "\r\n" in text + return text.replace("\r\n", "\n").replace("\r", "\n"), TextStyle(bom, crlf) + + +def encode(text: str, style: TextStyle) -> bytes: + if style.crlf: + text = text.replace("\n", "\r\n") + data = text.encode("utf-8") + return b"\xef\xbb\xbf" + data if style.bom else data + + +def _git(repository: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess: + return subprocess.run(["git", *args], cwd=str(repository), capture_output=True, check=check) + + +def changed_paths(repository: Path, base: str, head: str) -> list: + raw = _git(repository, "diff", "--name-only", "--no-renames", "-z", base, head).stdout + return sorted(part for part in raw.decode("utf-8").split("\0") if part) + + +def blob(repository: Path, ref: str, relative: str): + result = _git(repository, "show", "%s:%s" % (ref, relative), check=False) + return decode(result.stdout)[0] if result.returncode == 0 else None + + +def _transform_with(scripts: Path, texts: dict) -> dict: + """Run the public transform pipeline of one scripts/ tree over ``texts``.""" + env = dict(os.environ, PYTHONUTF8="1", PYTHONDONTWRITEBYTECODE="1") + result = subprocess.run([sys.executable, "-c", TRANSFORM_WORKER], cwd=str(scripts), + input=json.dumps(texts).encode("utf-8"), capture_output=True, env=env) + if result.returncode: + return {} + return json.loads(result.stdout.decode("utf-8")) + + +def transform_adjusted_bases(repository: Path, base: str, head: str, bases: dict) -> dict: + """Fold transform-rule changes into the merge base so they stay public-only. + + For each base text B the base and head pipelines give Tb and Th; merging + B with base Tb and other Th applies exactly the rewrites the head rules + add. Non-idempotent legacy rules cancel out because both sides share them. + """ + if not TRANSFORM_SCRIPTS & set(changed_paths(repository, base, head)) or not bases: + return dict(bases) + with tempfile.TemporaryDirectory(prefix="canonical-backport-scripts-") as temp: + names = _git(repository, "ls-tree", "--name-only", base, "scripts/").stdout.decode("utf-8").split() + for name in names: + if name.endswith(".py"): + Path(temp, Path(name).name).write_bytes(_git(repository, "show", "%s:%s" % (base, name)).stdout) + before = _transform_with(Path(temp), bases) + after = _transform_with(repository / "scripts", bases) + adjusted = {} + for relative, text in bases.items(): + old, new = before.get(relative), after.get(relative) + if old is None or new is None or old == new: + adjusted[relative] = text + continue + merged, conflicts = merge3(text, old, new) + adjusted[relative] = text if conflicts else merged + return adjusted + + +def merge3(current: str, base: str, other: str) -> tuple: + """Return (merged LF text, conflict count) from ``git merge-file``.""" + with tempfile.TemporaryDirectory(prefix="canonical-backport-") as temp: + paths = [] + for name, text in (("canonical", current), ("public-base", base), ("public-head", other)): + path = Path(temp) / name + path.write_bytes(text.encode("utf-8")) + paths.append(str(path)) + labels = ("-L", "canonical", "-L", "public-base", "-L", "public-head") + result = subprocess.run( + ["git", "merge-file", "-p", "--diff3", *labels, *paths], capture_output=True + ) + if result.returncode > 127: + raise RuntimeError("git merge-file failed: " + result.stderr.decode("utf-8", "replace")) + return result.stdout.decode("utf-8"), result.returncode + + +def decide(item: Item, base, head, current) -> None: + """Fill status/result for one canonical file from public base/head texts.""" + if current is None: + if base is None and head is not None: + item.status, item.result = "CREATE", head + elif head is None: + item.status = "ALREADY_APPLIED" + else: + item.status, item.detail = "ERROR", "canonical file is missing" + return + text, item.style = decode(current) + if head is None: + item.status = "DELETE" if text == base else "CONFLICT" + item.detail = "" if text == base else "canonical diverged from the deleted public base" + elif text == head: + item.status = "ALREADY_APPLIED" + elif base is None: + item.status, item.detail = "CONFLICT", "canonical already has a different file" + elif text == base: + item.status, item.result = "FAST_FORWARD", head + else: + item.result, item.conflicts = merge3(text, base, head) + if item.conflicts: + item.status = "CONFLICT" + else: + item.status = "ALREADY_APPLIED" if item.result == text else "MERGED" + + +def public_paths_for(root_key: str, canonical_relative: str) -> list: + """Return every public output generated from one canonical file.""" + paths = [] + for prefix, names, root, canonical_prefix in GROUPS: + if root == root_key and canonical_relative.startswith(canonical_prefix): + if canonical_relative[len(canonical_prefix):] in names: + paths.append(prefix + canonical_relative[len(canonical_prefix):]) + return paths + + +def _plan_target(repository: Path, base: str, head: str, item: Item, root, bases: dict) -> None: + members = public_paths_for(item.target.root, item.target.relative) + versions = {(bases.get(name, blob(repository, base, name)), blob(repository, head, name)) + for name in members} + if len(versions) != 1: + item.status, item.detail = "ERROR", "mirrored public copies differ: " + ", ".join(members) + return + if root is None: + item.detail = "pass %s to backport this file" % ROOT_FLAGS[item.target.root] + return + (base_text, head_text), = versions + path = Path(root) / item.target.relative + try: + decide(item, base_text, head_text, path.read_bytes() if path.is_file() else None) + except (UnicodeDecodeError, RuntimeError) as exc: + item.status, item.detail = "ERROR", str(exc) + return + if item.target.note and item.status in WRITES: + item.detail = "review: " + item.target.note + + +def plan(repository: Path, base: str, head: str, roots: dict) -> list: + """Classify every changed public path and plan its canonical effect.""" + grouped: dict = {} + items = [] + for relative in changed_paths(repository, base, head): + target = classify(relative) + if target.kind != "derived": + items.append(Item(target, [relative], "SKIPPED", target.note)) + else: + grouped.setdefault((target.root, target.relative), Item(target)).sources.append(relative) + raw = {name: blob(repository, base, name) for item in grouped.values() + for name in public_paths_for(item.target.root, item.target.relative)} + bases = transform_adjusted_bases(repository, base, head, + {name: text for name, text in raw.items() if text is not None}) + for key in sorted(grouped): + item = grouped[key] + _plan_target(repository, base, head, item, roots.get(item.target.root), bases) + items.append(item) + return items + + +def _atomic_write(path: Path, payload: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + handle, temp = tempfile.mkstemp(prefix="." + path.name + ".", suffix=".tmp", dir=str(path.parent)) + try: + with os.fdopen(handle, "wb") as stream: + stream.write(payload) + os.replace(temp, path) + except BaseException: + if os.path.exists(temp): + os.unlink(temp) + raise + + +def _restore(written: list, backup: Path) -> None: + for path, key, relative in reversed(written): + saved = backup / key / relative + if saved.is_file(): + _atomic_write(path, saved.read_bytes()) + elif path.is_file(): + path.unlink() + + +def apply(items: list, roots: dict, backup: Path, partial: bool) -> list: + """Write planned changes; restore every touched file on any failure.""" + blocked = [item for item in items if item.status in BLOCKERS] + if blocked and not partial: + raise RuntimeError("%d conflict/error item(s); resolve them or pass --partial" % len(blocked)) + written = [] + try: + for item in (item for item in items if item.status in WRITES): + path = Path(roots[item.target.root]) / item.target.relative + if path.is_file(): + saved = backup / item.target.root / item.target.relative + saved.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(str(path), str(saved)) + written.append((path, item.target.root, item.target.relative)) + if item.status == "DELETE": + path.unlink() + else: + _atomic_write(path, encode(item.result, item.style)) + except BaseException: + _restore(written, backup) + raise + for item in blocked: + if isinstance(item.result, str): + preview = backup / "conflicts" / item.target.root / (item.target.relative + ".conflict") + _atomic_write(preview, item.result.encode("utf-8")) + return [str(path) for path, _, _ in written] + + +def export_skill_delta(repository: Path, base: str, head: str, backup: Path) -> None: + """Export the rendered public SKILL.md delta for manual canonical review.""" + diff = _git(repository, "diff", base, head, "--", SKILL_ENTRY).stdout + if diff: + _atomic_write(backup / "SKILL.public-delta.diff", diff) + current = blob(repository, head, SKILL_ENTRY) or "" + _atomic_write(backup / "SKILL.public-head.md", current.encode("utf-8")) + + +def print_report(items: list) -> None: + order = {name: index for index, name in enumerate(BLOCKERS + WRITES + ("ALREADY_APPLIED", "SKIPPED"))} + for item in sorted(items, key=lambda value: (order.get(value.status, 99), value.sources)): + destination = "%s:%s" % (item.target.root, item.target.relative) if item.target.root else "-" + print("%-15s %-58s -> %s" % (item.status, ", ".join(item.sources), destination)) + if item.detail: + print(" " * 16 + item.detail) + counts: dict = {} + for item in items: + counts[item.status] = counts.get(item.status, 0) + 1 + print("SUMMARY " + ", ".join("%s=%d" % pair for pair in sorted(counts.items()))) + + +def _preflight(repository: Path, canonical: Path, refs: tuple) -> str: + """Return a blocking problem, or an empty string when inputs are sane.""" + if not (canonical / "SKILL.md").is_file() or not (canonical / "workflow_contract.json").is_file(): + return "not a canonical video-autopilot skill (SKILL.md + workflow_contract.json): %s" % canonical + try: + canonical.relative_to(repository) + except ValueError: + pass + else: + return "--canonical points inside the public repository; use the private skill copy" + for ref in refs: + if _git(repository, "rev-parse", "--verify", "--quiet", ref + "^{commit}", check=False).returncode: + return "unknown commit: %s" % ref + return "" + + +def _roots(args) -> dict: + roots = {"canonical": args.canonical} + if args.cleanup_helper: + roots["cleanup"] = args.cleanup_helper + if args.workspace: + roots["workspace"] = args.workspace + return {key: Path(value).expanduser().resolve() for key, value in roots.items()} + + +def run(args) -> int: + repository = Path(args.repository).resolve() + roots = _roots(args) + problem = _preflight(repository, roots["canonical"], (args.base, args.head)) + if problem: + print("BACKPORT BLOCKED: " + problem) + return 2 + items = plan(repository, args.base, args.head, roots) + print_report(items) + if args.report: + payload = {"base": args.base, "head": args.head, "items": [item.row() for item in items]} + _atomic_write(Path(args.report), (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8")) + blocked = any(item.status in BLOCKERS for item in items) + if not args.apply: + print("DRY RUN: nothing written. Re-run with --apply to write the canonical files.") + return 1 if blocked else 0 + stamp = time.strftime("%Y%m%d-%H%M%S") + backup = Path(args.backup_dir or repository / ".video-autopilot" / "backport" / stamp).resolve() + try: + written = apply(items, roots, backup, args.partial) + except RuntimeError as exc: + print("APPLY REFUSED: %s" % exc) + return 1 + export_skill_delta(repository, args.base, args.head, backup) + print("APPLIED %d file(s); backups and conflict previews: %s" % (len(written), backup)) + print("NEXT: python scripts/sync_canonical.py --canonical \"%s\" then git status" % roots["canonical"]) + print(" An exact backport leaves only sync-receipt.json modified.") + return 1 if blocked else 0 + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--canonical", help="private canonical skill root (contains SKILL.md)") + parser.add_argument("--base", help="public commit that matches the canonical (last sync)") + parser.add_argument("--head", default="HEAD", help="public commit with the edits (default HEAD)") + parser.add_argument("--repository", default=str(Path(__file__).resolve().parents[1])) + parser.add_argument("--cleanup-helper", help="canonical code-cleanup-helper skill root") + parser.add_argument("--workspace", help="private workspace root holding AUTOPILOT_ARCHITECTURE_V6.md") + parser.add_argument("--apply", action="store_true", help="write the canonical files") + parser.add_argument("--partial", action="store_true", help="apply clean items even if others conflict") + parser.add_argument("--backup-dir", help="backup/report directory for --apply") + parser.add_argument("--report", help="write the plan as JSON to this path") + parser.add_argument("--self-test", action="store_true", help="run deterministic fixtures") + args = parser.parse_args(argv) + if args.self_test: + if __package__: + from .canonical_backport_selftest import run_self_test + else: + from canonical_backport_selftest import run_self_test + + run_self_test(sys.modules[__name__]) + return 0 + if not args.canonical or not args.base: + parser.error("--canonical and --base are required (see --help)") + return run(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/canonical_backport_selftest.py b/scripts/canonical_backport_selftest.py new file mode 100644 index 0000000..b4e82b3 --- /dev/null +++ b/scripts/canonical_backport_selftest.py @@ -0,0 +1,217 @@ +#!/usr/bin/env python3 +"""Deterministic fixtures for ``canonical_backport.py`` (needs only git). + +The module under test is injected by ``run_self_test`` so the static import +graph stays acyclic, mirroring ``src/release_manager_selftest.py``. +""" +from __future__ import annotations + +import importlib +import subprocess +import tempfile +from pathlib import Path +from unittest.mock import patch + +backport = None # injected by run_self_test + +BOM = b"\xef\xbb\xbf" +SKILL = "codex-skill/video-autopilot/" +GIT = ("git", "-c", "user.name=fixture", "-c", "user.email=fixture", + "-c", "commit.gpgsign=false", "-c", "core.autocrlf=false") +BASE_FILES = { + "src/aesthetic_score.py": "alpha\nbeta\ngamma\ndelta\nepsilon\n", + SKILL + "references/craft-index.md": "x\n", + "knowledge/runtime/filter_library.json": '{"v": 1}\n', + "src/workflow_json.py": "w = 1\n", + SKILL + "workflow_json.py": "w = 1\n", + "src/workflow_state.py": "s = 1\n", + SKILL + "workflow_state.py": "s = 1\n", + "src/visual_master.py": "k = 1\n", + "README.md": "readme\n", + "knowledge/runtime/state.json": "{}\n", + SKILL + "references/knowledge-lifecycle.md": "old\n", + SKILL + "SKILL.md": "skill\n", + "tools/code-cleanup-helper/SKILL.md": "cleanup\n", + SKILL + "references/quality-95-system.md": "private wording approves\nkeep\n", + "scripts/sync_canonical.py": "def _transform(text, name):\n return text\n", + "scripts/public_privacy_sanitizer.py": "def sanitize_public_text(relative, text):\n return text\n", +} +HEAD_CHANGES = { + "src/aesthetic_score.py": "alpha\nbeta\ngamma\nDELTA\nepsilon\n", + SKILL + "references/craft-index.md": "y\n", + "knowledge/runtime/filter_library.json": '{"v": 2}\n', + "src/workflow_json.py": "w = 2\n", + SKILL + "workflow_json.py": "w = 2\n", + "src/workflow_state.py": "s = 2\n", + SKILL + "workflow_state.py": "s = 3\n", + "src/visual_master.py": "k = 2\n", + "README.md": "readme 2\n", + "knowledge/runtime/state.json": '{"records": []}\n', + SKILL + "references/knowledge-lifecycle.md": None, + SKILL + "references/token-budget-system.md": "new\n", + SKILL + "SKILL.md": "skill 2\n", + "tools/code-cleanup-helper/SKILL.md": "cleanup 2\n", + # A new public-only rewrite rule plus one real edit in the same file. + "scripts/sync_canonical.py": ("def _transform(text, name):\n" + " return text.replace('private wording', 'public wording')\n"), + SKILL + "references/quality-95-system.md": "public wording approves\nkeep\nnew line\n", +} +CANONICAL = { + "SKILL.md": b"---\nname: video-autopilot\n---\nprivate body\n", + "workflow_contract.json": b"{}\n", + "aesthetic_score.py": BOM + b"PRIVATE\r\nalpha\r\nbeta\r\ngamma\r\ndelta\r\nepsilon\r\n", + "references/craft-index.md": b"x\n", + "knowledge/filter_library.json": b'{"v": 2}\n', + "workflow_json.py": b"w = 1\r\n", + "workflow_state.py": b"s = 1\n", + "visual_master.py": b"k = 3\n", + "references/knowledge-lifecycle.md": b"old\r\n", + "references/quality-95-system.md": b"private wording approves\r\nkeep\r\n", +} +EXPECTED = { + "aesthetic_score.py": "MERGED", "references/craft-index.md": "FAST_FORWARD", + "knowledge/filter_library.json": "ALREADY_APPLIED", "workflow_json.py": "FAST_FORWARD", + "workflow_state.py": "ERROR", "visual_master.py": "CONFLICT", + "references/knowledge-lifecycle.md": "DELETE", "references/token-budget-system.md": "CREATE", + "references/quality-95-system.md": "MERGED", +} + + +def _write_tree(root: Path, files: dict) -> None: + for relative, payload in files.items(): + path = root / relative + if payload is None: + if path.exists(): + path.unlink() + continue + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(payload if isinstance(payload, bytes) else payload.encode("utf-8")) + + +def _commit(repository: Path, message: str) -> str: + subprocess.run([*GIT, "add", "-A"], cwd=str(repository), check=True, capture_output=True) + subprocess.run([*GIT, "commit", "-q", "-m", message], cwd=str(repository), check=True, capture_output=True) + return subprocess.run([*GIT, "rev-parse", "HEAD"], cwd=str(repository), check=True, + capture_output=True).stdout.decode("ascii").strip() + + +def _snapshot(root: Path) -> dict: + return {path.relative_to(root).as_posix(): path.read_bytes() + for path in sorted(root.rglob("*")) if path.is_file()} + + +def _fixture(temp: Path) -> tuple: + repository, canonical = temp / "public", temp / "canonical" + repository.mkdir() + subprocess.run([*GIT, "init", "-q"], cwd=str(repository), check=True, capture_output=True) + _write_tree(repository, BASE_FILES) + base = _commit(repository, "base") + _write_tree(repository, HEAD_CHANGES) + head = _commit(repository, "head") + _write_tree(canonical, CANONICAL) + return repository, canonical, base, head + + +def _check_classification() -> None: + assert backport.classify("src/system_health.py").kind == "public-owned" + assert backport.classify(SKILL + "SKILL.md").kind == "generated" + assert backport.classify("knowledge/runtime/state.json").kind == "projection" + assert backport.classify("docs/OPEN_SOURCE_SUITE.md").kind == "repo-native" + longform = backport.classify("src/longform_maker/shorts_gate.py") + assert (longform.kind, longform.relative) == ("derived", "longform_maker/shorts_gate.py") + cleanup = backport.classify("tools/code-cleanup-helper/scripts/audit.py") + assert (cleanup.root, cleanup.relative) == ("cleanup", "scripts/audit.py") + assert backport.public_paths_for("canonical", "workflow_json.py") == [ + "src/workflow_json.py", SKILL + "workflow_json.py"] + + +def _check_plan(repository: Path, canonical: Path, base: str, head: str) -> None: + items = backport.plan(repository, base, head, {"canonical": canonical}) + derived = {item.target.relative: item.status for item in items if item.target.kind == "derived"} + cleanup = [item for item in items if item.target.root == "cleanup"] + assert {key: derived[key] for key in EXPECTED} == EXPECTED, derived + assert cleanup and cleanup[0].status == "SKIPPED" and "--cleanup-helper" in cleanup[0].detail + skipped = {item.sources[0]: item.target.kind for item in items if item.status == "SKIPPED"} + assert skipped["README.md"] == "repo-native" and skipped[SKILL + "SKILL.md"] == "generated" + assert skipped["knowledge/runtime/state.json"] == "projection" + + +def _check_apply(repository: Path, canonical: Path, base: str, temp: Path) -> None: + items = backport.plan(repository, base, "HEAD", {"canonical": canonical}) + before = _snapshot(canonical) + try: + backport.apply(items, {"canonical": canonical}, temp / "refused", partial=False) + except RuntimeError as exc: + assert "--partial" in str(exc) + else: + raise AssertionError("conflicting backport applied without --partial") + assert _snapshot(canonical) == before + backup = temp / "backup" + backport.apply(items, {"canonical": canonical}, backup, partial=True) + after = _snapshot(canonical) + assert after["aesthetic_score.py"] == BOM + b"PRIVATE\r\nalpha\r\nbeta\r\ngamma\r\nDELTA\r\nepsilon\r\n" + assert after["references/craft-index.md"] == b"y\n" and after["workflow_json.py"] == b"w = 2\r\n" + assert after["references/token-budget-system.md"] == b"new\n" + assert after["references/quality-95-system.md"] == b"private wording approves\r\nkeep\r\nnew line\r\n", \ + "a public-only rewrite rule must not be written back into the canonical" + assert "references/knowledge-lifecycle.md" not in after + assert after["visual_master.py"] == b"k = 3\n" and after["workflow_state.py"] == b"s = 1\n" + assert (backup / "canonical/references/knowledge-lifecycle.md").read_bytes() == b"old\r\n" + assert b"<<<<<<<" in (backup / "conflicts/canonical/visual_master.py.conflict").read_bytes() + replay = backport.plan(repository, base, "HEAD", {"canonical": canonical}) + statuses = {item.target.relative: item.status for item in replay if item.target.kind == "derived"} + assert statuses["aesthetic_score.py"] == statuses["references/token-budget-system.md"] == "ALREADY_APPLIED" + assert statuses["references/knowledge-lifecycle.md"] == "ALREADY_APPLIED" + + +def _check_rollback(repository: Path, temp: Path, base: str, head: str) -> None: + canonical = temp / "rollback" + _write_tree(canonical, CANONICAL) + before = _snapshot(canonical) + items = backport.plan(repository, base, head, {"canonical": canonical}) + real_write, calls = backport._atomic_write, [] + + def failing_write(path, payload): + calls.append(path) + if len(calls) == 3: + raise OSError("injected write failure") + real_write(path, payload) + + with patch.object(backport, "_atomic_write", failing_write): + try: + backport.apply(items, {"canonical": canonical}, temp / "rollback-backup", partial=True) + except OSError: + pass + else: + raise AssertionError("injected failure did not abort the backport") + assert _snapshot(canonical) == before, "failed backport was not rolled back" + + +def _check_cli(repository: Path, canonical: Path, base: str, temp: Path) -> None: + report = temp / "plan.json" + assert backport.main(["--canonical", str(canonical), "--base", base, + "--repository", str(repository), "--report", str(report)]) == 1 + assert '"CONFLICT"' in report.read_text(encoding="utf-8") + inside = repository / "codex-skill" + _write_tree(inside, {"SKILL.md": b"x\n", "workflow_contract.json": b"{}\n"}) + assert backport.main(["--canonical", str(inside), "--base", base, "--repository", str(repository)]) == 2 + assert backport.main(["--canonical", str(canonical), "--base", "0" * 40, + "--repository", str(repository)]) == 2 + + +def run_self_test(module) -> None: + global backport + backport = module + _check_classification() + with tempfile.TemporaryDirectory(prefix="canonical-backport-test-") as raw: + temp = Path(raw) + repository, canonical, base, head = _fixture(temp) + _check_plan(repository, canonical, base, head) + _check_rollback(repository, temp, base, head) + _check_cli(repository, canonical, base, temp) + _check_apply(repository, canonical, base, temp) + print("canonical_backport self-test GREEN") + + +if __name__ == "__main__": + run_self_test(importlib.import_module("canonical_backport")) diff --git a/scripts/hao_autopilot.py b/scripts/hao_autopilot.py new file mode 100644 index 0000000..eadd83a --- /dev/null +++ b/scripts/hao_autopilot.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +"""Public command hub for the Video Autopilot kit. + +The README, TROUBLESHOOTING and the Codex Skill references document commands as +``python scripts/hao_autopilot.py ...``. Each command forwards to the +public module that owns it, so behavior, exit codes and JSON output are the same +as calling that module directly. The hub adds no state, network access or file +writes of its own. + + python scripts/hao_autopilot.py # list commands + python scripts/hao_autopilot.py workflow next + python scripts/hao_autopilot.py learn --scope shorts/food --rule "..." --evidence "..." +""" +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +# command -> (module relative to the kit root, leading arguments, summary) +COMMANDS = { + "workflow": ("src/workflow_contract.py", (), + "Editkin v4 run: create/status/next/claim/complete/fail/resume/verify/reject-render"), + "thumbnail-score": ("src/thumbnail_algorithm_score.py", ("score",), + "score long-form thumbnail evidence (+ 168x94 preview)"), + "dedupe-archive": ("src/storage_lifecycle.py", ("dedupe-archive",), + "preview, then --apply, verified archive deduplication"), + "storage": ("src/storage_lifecycle.py", (), + "storage audit/check/prune/deliver/snapshot"), + "knowledge": ("src/knowledge_lifecycle.py", ("audit",), + "audit learning memory; --digest prints the runtime card; " + "--rotate-video-log [--apply] rotates the video log"), + "learn": ("src/knowledge_lifecycle.py", ("record",), + "record one feedback rule with its evidence (soft until repeated)"), + "project95": ("src/project_quality_95.py", ("audit",), + "whole-project quality-95 acceptance report"), + "health": ("src/system_health.py", (), + "kit health (--quick for the fast subset)"), +} +KNOWLEDGE_FLAGS = {"--rotate-video-log": "rotate-video-log", "--digest": "digest"} + + +def resolve(argv: list) -> list: + """Translate hub arguments into the module command line (no execution).""" + if not argv or argv[0] not in COMMANDS: + raise KeyError(argv[0] if argv else "") + module, lead, _summary = COMMANDS[argv[0]] + rest = list(argv[1:]) + if argv[0] == "knowledge": + flags = [flag for flag in KNOWLEDGE_FLAGS if flag in rest] + if len(flags) > 1: + raise ValueError("choose one of " + ", ".join(KNOWLEDGE_FLAGS)) + if flags: + rest.remove(flags[0]) + lead = (KNOWLEDGE_FLAGS[flags[0]],) + return [sys.executable, str(ROOT / module), *lead, *rest] + + +def usage() -> str: + rows = ["usage: python scripts/hao_autopilot.py [args...]", "", "commands:"] + rows += [" %-16s %s" % (name, spec[2]) for name, spec in COMMANDS.items()] + rows += [" %-16s %s" % ("selftest", "verify every command maps to a shipped module")] + return "\n".join(rows) + + +def self_test() -> int: + missing = [spec[0] for spec in COMMANDS.values() if not (ROOT / spec[0]).is_file()] + assert not missing, "hub points at missing modules: %s" % missing + assert resolve(["knowledge"])[2:] == ["audit"] + assert resolve(["knowledge", "--rotate-video-log", "--apply"])[2:] == ["rotate-video-log", "--apply"] + assert resolve(["learn", "--rule", "r", "--evidence", "e"])[2:] == ["record", "--rule", "r", "--evidence", "e"] + assert resolve(["thumbnail-score", "e.json", "--output-dir", "o"])[2:] == ["score", "e.json", "--output-dir", "o"] + assert resolve(["workflow", "next", "run"])[1].endswith("workflow_contract.py") + for bad in ([], ["unknown"]): + try: + resolve(bad) + except KeyError: + pass + else: + raise AssertionError("unknown hub command accepted: %r" % bad) + try: + resolve(["knowledge", "--digest", "--rotate-video-log"]) + except ValueError: + pass + else: + raise AssertionError("conflicting knowledge flags accepted") + print("hao_autopilot command hub self-test GREEN") + return 0 + + +def main(argv=None) -> int: + args = list(sys.argv[1:] if argv is None else argv) + if not args or args[0] in {"-h", "--help", "help"}: + print(usage()) + return 0 + if args[0] == "selftest": + return self_test() + try: + command = resolve(args) + except KeyError: + print("unknown command: %s\n\n%s" % (args[0], usage()), file=sys.stderr) + return 2 + except ValueError as exc: + print("invalid arguments: %s" % exc, file=sys.stderr) + return 2 + return subprocess.call(command) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/context_router.py b/src/context_router.py index 6ce8571..f18742f 100644 --- a/src/context_router.py +++ b/src/context_router.py @@ -25,7 +25,14 @@ ROOT = Path(__file__).resolve().parent -REFS = ROOT / "references" +# Skill layout keeps references beside this module; the public kit keeps them +# under codex-skill/. A release-managed kit has release-manifest.json above src/. +REFS = next((path for path in (ROOT / "references", + ROOT.parent / "codex-skill" / "video-autopilot" / "references") + if path.is_dir()), ROOT / "references") +KIT_LAYOUT = (ROOT.parent / "release-manifest.json").is_file() +PROFILE_DIR = ROOT.parent / "profiles" +PROFILE_NAMES = ("voice", "brand", "style", "show", "your_context") PACKET_NAME = "current_context_packet.json" DEFAULT_MAX_TOKENS = 900 MODE_TOKEN_LIMITS = { @@ -51,6 +58,13 @@ "interview": "No-Face Documentary：最強一句＋真 proof 冷開；H/G 形狀＋文字辨識講者，以操作、手部／環境、文件與圖解承載畫面。禁人物臉、AI 頭像、剪影、reaction face；純波形滿版不超過 3 秒。", } +# The interview card follows the creator's face policy (profiles/brand.md). +INTERVIEW_CARDS = { + "no_face": FORMAT_CARDS["interview"], + "on_camera": "訪談：最強一句＋真 proof 冷開；主持人出鏡依 profiles/brand.md，來賓的臉需本人同意；操作、文件與圖解承載證據，純波形滿版不超過 3 秒。", + "unset": "訪談：最強一句＋真 proof 冷開；出鏡政策未設定——先讀 profiles/brand.md 或詢問創作者，不預設露臉或不露臉；來賓的臉需本人同意。", +} + DOMAIN_CARDS = { "ai": "結果→真操作→局部放大→錯誤修正→證據；畫面跟著操作意圖。", "food": "環境→製作→質地極近→入口反應；滋滋、切、脆聲高於裝飾 BGM。", @@ -72,12 +86,19 @@ "general": "結果→脈絡→行動→轉折→呼吸→payoff；每顆畫面回答當下語意。", } +# Ordered preferences; only references that exist in this install are routed, +# so a private or sibling-skill source falls back to its public counterpart. REFERENCE_ROUTES = { "shorts": ("shorts-mastery-2026.md", "competitor-vertical-teardown-2026.md"), - "longform": ("hao-teaching-longform-method.md", "script-retention-2026.md"), + "longform": ("hao-teaching-longform-method.md", "script-retention-2026.md", + "editing-master-techniques.md"), "vlog": ("genre-editing-craft-2026.md", "cinematic-wave-and-domain-grammar-2026.md"), - "interview": ("../../interview-show/references/format-bible.md", "script-retention-2026.md"), + "interview": ("../../interview-show/references/format-bible.md", "script-retention-2026.md", + "genre-editing-craft-2026.md"), } +ROUTE_SOURCE_LIMIT = 2 +REFERENCE_ROUTES_FLAT = frozenset(name for names in REFERENCE_ROUTES.values() for name in names) +LAST_RESORT_SOURCES = ("meta-lessons-canon.md", "craft-index.md") TOPIC_ROUTES = ( ( @@ -127,6 +148,36 @@ def _fingerprint(path: Path) -> dict: "sha256_12": hashlib.sha256(data).hexdigest()[:12]} +def creator_face_policy(profile: Path | None = None) -> str: + """Return no_face / on_camera / unset from the brand profile checkbox.""" + path = profile or PROFILE_DIR / "brand.md" + try: + text = path.read_text(encoding="utf-8") + except OSError: + text = "" + match = re.search(r"露臉嗎？\s*`?\[([ xX])\]\s*會\s*\[([ xX])\]\s*不會", text) + if match and (match.group(1) != " ") != (match.group(2) != " "): + return "on_camera" if match.group(1) != " " else "no_face" + # Private skill installs keep their established no-face card; a public kit + # must not inherit a maintainer default it never configured. + return "unset" if KIT_LAYOUT else "no_face" + + +def _route_paths(route_names: list) -> tuple: + """Existing sources only, most relevant first, capped per format route.""" + topic_first = [name for name in route_names if name not in REFERENCE_ROUTES_FLAT] + route = [name for name in route_names if name in REFERENCE_ROUTES_FLAT] + existing = [(REFS / name).resolve() for name in route if (REFS / name).resolve().is_file()] + topical = [(REFS / name).resolve() for name in topic_first if (REFS / name).resolve().is_file()] + return tuple(dict.fromkeys([*topical, *existing[:ROUTE_SOURCE_LIMIT]])) + + +def _creator_profiles() -> dict: + present = [name for name in PROFILE_NAMES if (PROFILE_DIR / (name + ".md")).is_file()] + return {"present": ["profiles/%s.md" % name for name in present], + "missing": [name for name in PROFILE_NAMES if name not in present]} + + def _all_reference_cost() -> dict: paths = sorted(REFS.glob("*.md")) total_bytes = sum(p.stat().st_size for p in paths) @@ -141,13 +192,16 @@ def build_context_packet(mode: str = "build", format: str = "shorts", domain: st domain = infer_domain(topic, domain, SUPPORTED_DOMAINS) max_tokens = int(max_tokens if max_tokens is not None else MODE_TOKEN_LIMITS[mode]) route_names = list(REFERENCE_ROUTES[format]) + format_card = INTERVIEW_CARDS[creator_face_policy()] if format == "interview" else FORMAT_CARDS[format] if re.search(r"字幕|字卡|文字效果|浮空字|巨字|caption|kinetic|Mr\.?\s*Beast", str(topic), re.I): route_names.insert(0, "caption-art-direction.md") for pattern, reference in TOPIC_ROUTES: if re.search(pattern, str(topic), re.I): route_names.insert(0, reference) - # 同一來源只保留一次，並維持「本題最相關」在第一順位。 - route_paths = tuple((REFS / p).resolve() for p in dict.fromkeys(route_names)) + # 同一來源只保留一次，並維持「本題最相關」在第一順位；不存在的來源不路由。 + route_paths = _route_paths(route_names) + last_resort = next((REFS / name for name in LAST_RESORT_SOURCES if (REFS / name).is_file()), + REFS / LAST_RESORT_SOURCES[-1]) # Four is the normal runtime cap; the memory layer itself supports up to six # for explicit deep/learn escalation without pushing build packets over 800. rule_limit = 6 if mode == "learn" else 4 @@ -161,7 +215,7 @@ def build_context_packet(mode: str = "build", format: str = "shorts", domain: st "topic": str(topic or "")[:topic_limit]}, "instruction": "只讀這個 packet 執行；遇到列出的 escalation trigger 才加讀一份 source。", "mode_card": MODE_CARDS[mode], - "format_card": FORMAT_CARDS[format], + "format_card": format_card, "domain_card": DOMAIN_CARDS[domain], "knowledge_card": knowledge_card, "hard_guardrails": list(HARD_GUARDRAILS), @@ -171,27 +225,39 @@ def build_context_packet(mode: str = "build", format: str = "shorts", domain: st }, "quality_95": { "target": 95, - "rule": "已知負面案例可機械封鎖；審美必須由 Hao 完成時間碼審片才可認證，裝置不限。", + "rule": "已知負面案例可機械封鎖；審美必須由創作者（或其指定審片者）完成時間碼審片才可認證，裝置不限。", "artifacts": ["_qa/QUALITY_95.json", "_review/review.html"], }, "escalation": { "triggers": ["規則衝突", "新題材沒有 domain card", "QA 連續失敗", "使用者要求深度研究／學習"], "read_at_most_one_first": True, "sources": [_path_label(p) for p in route_paths], - "full_canon_last_resort": _path_label(REFS / "meta-lessons-canon.md"), + "full_canon_last_resort": _path_label(last_resort), }, "source_fingerprints": [_fingerprint(p) for p in route_paths], "budget": {"max_tokens": max_tokens}, } + if KIT_LAYOUT: + packet["creator_profiles"] = _creator_profiles() + packet["instruction"] = ("先讀 creator_profiles.present（聲音、品牌、風格）再執行；缺少的不得自行假設。" + "其餘只讀這個 packet；遇到 escalation trigger 才加讀一份 source。") selected_rules = len(packet["knowledge_card"]) # 不讓新增記憶把固定 packet 撐爆；依排序只從尾端移除全域低優先規則， # 任務精確 scoped memory 位於前端，因此不會再次「明明 pinned 卻載不到」。 while True: packet["budget"]["knowledge_rules_selected"] = len(packet["knowledge_card"]) packet["budget"]["knowledge_rules_trimmed"] = selected_rules - len(packet["knowledge_card"]) - if estimate_tokens(packet) <= max_tokens or not packet["knowledge_card"]: + if estimate_tokens(packet) <= max_tokens: + break + if packet["knowledge_card"]: + packet["knowledge_card"].pop() + elif len(packet["escalation"]["sources"]) > 1: + # Multi-route topics may not crash a small packet: drop the least + # relevant escalation source (with its fingerprint) before failing. + packet["escalation"]["sources"].pop() + packet["source_fingerprints"].pop() + else: break - packet["knowledge_card"].pop() packet["budget"]["estimated_tokens"] = estimate_tokens(packet) packet["budget"]["within_budget"] = packet["budget"]["estimated_tokens"] <= max_tokens if not packet["budget"]["within_budget"]: @@ -243,13 +309,33 @@ def self_test() -> None: assert any(source.endswith("caption-art-direction.md") for source in caption["escalation"]["sources"]) assert any(source.endswith("mrbeast-and-yingshi-benchmark.md") for source in caption["escalation"]["sources"]) interview = build_context_packet("build", "interview", "interview", "全程不露臉創業訪談") - assert interview["route"]["format"] == "interview" and "reaction face" in interview["format_card"] + assert interview["route"]["format"] == "interview" + assert interview["format_card"] == INTERVIEW_CARDS[creator_face_policy()] assert interview["budget"]["within_budget"] verbose = build_context_packet("build", "shorts", "toy", "戰鬥陀螺對戰與完整字幕規劃" * 40) assert len(verbose["route"]["topic"]) <= 64 and verbose["budget"]["within_budget"] + for fmt in FORMAT_CARDS: + routed = build_context_packet("plan", fmt, "general", "") + assert routed["escalation"]["sources"], "every format must route at least one real source" + assert not any(item.get("missing") for item in routed["source_fingerprints"]), fmt + crowded = build_context_packet("outcome", "longform", "ai", "MrBeast 字幕 3D 設計系統 美感") + assert crowded["budget"]["within_budget"] and crowded["escalation"]["sources"] + assert "Hao" not in json.dumps(crowded, ensure_ascii=False) + _self_test_face_policy() print("context_router self-test GREEN") +def _self_test_face_policy() -> None: + with tempfile.TemporaryDirectory(prefix="context-profile-") as temp: + brand = Path(temp) / "brand.md" + brand.write_text("- 你會錄 talking-head / 露臉嗎？ `[ ] 會 [x] 不會`\n", encoding="utf-8") + assert creator_face_policy(brand) == "no_face" + brand.write_text("- 你會錄 talking-head / 露臉嗎？ `[x] 會 [ ] 不會`\n", encoding="utf-8") + assert creator_face_policy(brand) == "on_camera" + brand.write_text("- 你會錄 talking-head / 露臉嗎？ `[ ] 會 [ ] 不會`\n", encoding="utf-8") + assert creator_face_policy(brand) == ("unset" if KIT_LAYOUT else "no_face") + + def main() -> int: parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="command") diff --git a/src/knowledge_lifecycle.py b/src/knowledge_lifecycle.py index 6fe65f6..d9fc5e1 100644 --- a/src/knowledge_lifecycle.py +++ b/src/knowledge_lifecycle.py @@ -23,10 +23,48 @@ DEFAULT_VIDEO_ARCHIVE = SKILL_ROOT.parent / "data" / "video_log_archive.md" +LEDGER_ENV = "VIDEO_AUTOPILOT_KNOWLEDGE_STATE" +SOURCE_KEY_LIMIT = 64 + + def _now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") +def default_ledger_path() -> Path: + """Return where learned rules are written. + + A release-managed kit (``release-manifest.json`` beside the code root) + ships ``DEFAULT_STATE`` as a seed that updates may replace, so learning + goes to the protected ``data/`` folder instead. Other layouts keep + learning in ``DEFAULT_STATE``. ``VIDEO_AUTOPILOT_KNOWLEDGE_STATE`` + overrides both. + """ + explicit = os.environ.get(LEDGER_ENV, "").strip() + if explicit: + return Path(explicit).expanduser() + kit_root = SKILL_ROOT.parent + if (kit_root / "release-manifest.json").is_file(): + return kit_root / "data" / "knowledge_state.json" + return Path(DEFAULT_STATE) + + +def _resolve_state(state_path: str | os.PathLike | None) -> Path: + """Explicit paths win; otherwise read the ledger, migrating a learned seed once.""" + if state_path is not None: + return Path(state_path) + ledger, seed = default_ledger_path(), Path(DEFAULT_STATE) + if ledger.is_file() or ledger == seed or not seed.is_file(): + return ledger + seeded = load_state(seed) + if not seeded["records"]: + return seed + # Rules learned before the ledger moved must not vanish when an update + # replaces the managed seed: copy them into protected data first. + _atomic_write(ledger, seeded) + return ledger + + def estimate_tokens(value) -> int: text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, separators=(",", ":")) cjk = len(re.findall(r"[\u3400-\u9fff\uf900-\ufaff]", text)) @@ -79,6 +117,40 @@ def _scope_values(values: list[str] | None) -> list[str]: return ["*"] if normalized & {"*", "all"} else sorted(normalized) +def _source_key(source: str | None, evidence: str) -> str: + """One observation = one source (a video/version id); else its evidence text.""" + basis = _normalize(source) if source and str(source).strip() else _normalize(evidence) + return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:12] + + +def _count_source(record: dict, key: str) -> None: + """Support counts distinct observations, so a retried review cannot promote a rule.""" + if "sources" not in record: + # Legacy rows: keep their support, and treat their evidence as already seen. + record["sources"] = [_source_key(None, item) for item in record.get("evidence", [])] + if key in record["sources"]: + return + record["sources"] = (record["sources"] + [key])[-SOURCE_KEY_LIMIT:] + record["support"] = int(record.get("support", 0)) + 1 + + +def _resolve_supersede(state: dict, record: dict, old_id: str) -> None: + other = next((row for row in state["records"] if row.get("id") == old_id), None) + if not other: + raise ValueError(f"Unknown superseded rule: {old_id}") + if other["id"] == record["id"]: + raise ValueError("a rule cannot supersede itself") + other["pinned"] = False + other["pinned_explicit"] = False + other["superseded_by"] = record["id"] + # Superseding is the human resolution of a conflict between the two rules. + record["contradictions"] = [item for item in record.get("contradictions", []) if item != other["id"]] + other["contradictions"] = [item for item in other.get("contradictions", []) if item != record["id"]] + record.setdefault("supersedes", []) + if other["id"] not in record["supersedes"]: + record["supersedes"].append(other["id"]) + + def record_feedback( rule: str, evidence: str, @@ -91,16 +163,24 @@ def record_feedback( contradicts: str | None = None, supersedes: str | None = None, pin: bool = False, - state_path: str | os.PathLike = DEFAULT_STATE, + source: str | None = None, + state_path: str | os.PathLike | None = None, ) -> dict: + """Record one observation of a rule. + + ``source`` identifies the observed video/version (for example + ``content_id@sha12``); repeats of the same source never add support. + ``pin`` is an explicit creator instruction and survives later + observations until the rule is contradicted or superseded. + """ if not str(rule).strip() or not str(evidence).strip(): raise ValueError("rule and evidence are required") if kind not in {"hard", "soft"}: raise ValueError("kind must be hard or soft") if contradicts and supersedes: raise ValueError("use either contradicts or supersedes, not both") - path = Path(state_path) - state = load_state(path) + state = load_state(_resolve_state(state_path)) + path = Path(state_path) if state_path is not None else default_ledger_path() fingerprint = _fingerprint(scope, rule) record = next((row for row in state["records"] if row.get("fingerprint") == fingerprint), None) created = record is None @@ -121,7 +201,7 @@ def record_feedback( "created_at": _now(), } state["records"].append(record) - record["support"] = int(record.get("support", 0)) + 1 + _count_source(record, _source_key(source, evidence)) record["formats"] = _scope_values(record.get("formats")) record["domains"] = _scope_values(record.get("domains")) if triggers is not None: @@ -144,21 +224,14 @@ def record_feedback( if record["id"] not in other["contradictions"]: other["contradictions"].append(record["id"]) other["pinned"] = False + other["pinned_explicit"] = False if supersedes: - other = next((row for row in state["records"] if row.get("id") == supersedes), None) - if not other: - raise ValueError(f"Unknown superseded rule: {supersedes}") - if other["id"] == record["id"]: - raise ValueError("a rule cannot supersede itself") - other["pinned"] = False - other["superseded_by"] = record["id"] - record.setdefault("supersedes", []) - if other["id"] not in record["supersedes"]: - record["supersedes"].append(other["id"]) + _resolve_supersede(state, record, supersedes) minimum = int(state.get("promotion_policy", {}).get("minimum_support", 3)) - record["pinned"] = bool(pin or ( - record["support"] >= minimum and not record.get("contradictions") - )) + record["pinned_explicit"] = bool(pin or record.get("pinned_explicit")) + record["pinned"] = not record.get("contradictions") and bool( + record["pinned_explicit"] or record["support"] >= minimum + ) state["records"] = sorted(state["records"], key=lambda row: row["id"]) state["revision"] = int(state.get("revision", 0)) + 1 state["updated_at"] = _now() @@ -173,9 +246,9 @@ def select_rules( topic: str = "", *, limit: int = 6, - state_path: str | os.PathLike = DEFAULT_STATE, + state_path: str | os.PathLike | None = None, ) -> list[dict]: - state = load_state(state_path) + state = load_state(_resolve_state(state_path)) topic_normalized = _normalize(topic) candidates = [] for record in state["records"]: @@ -213,8 +286,8 @@ def select_rules( ] -def build_digest(state_path: str | os.PathLike = DEFAULT_STATE) -> dict: - state = load_state(state_path) +def build_digest(state_path: str | os.PathLike | None = None) -> dict: + state = load_state(_resolve_state(state_path)) pinned = [row for row in state["records"] if row.get("pinned") and not row.get("contradictions") and not row.get("superseded_by")] conflicts = [row for row in state["records"] if row.get("contradictions")] @@ -240,8 +313,8 @@ def build_digest(state_path: str | os.PathLike = DEFAULT_STATE) -> dict: return digest -def audit(state_path: str | os.PathLike = DEFAULT_STATE) -> dict: - state = load_state(state_path) +def audit(state_path: str | os.PathLike | None = None) -> dict: + state = load_state(_resolve_state(state_path)) ids = [row["id"] for row in state["records"]] fingerprints = [row["fingerprint"] for row in state["records"]] conflicts = [row["id"] for row in state["records"] if row.get("contradictions")] @@ -372,9 +445,76 @@ def self_test() -> None: assert applied["changed"] and "old evidence" not in log.read_text(encoding="utf-8") assert "old evidence" in archive.read_text(encoding="utf-8") assert rotate_numbered_video_log(log, archive, apply=True)["sections"] == 0 + _self_test_memory_guards(Path(temp)) + _self_test_ledger_location(Path(temp)) print("knowledge_lifecycle self-test GREEN") +def _empty_state(path: Path, records: list | None = None) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({ + "schema_version": 2, "revision": 0, + "promotion_policy": {"minimum_support": 3}, "records": records or [], + }, ensure_ascii=False), encoding="utf-8") + return path + + +def _self_test_memory_guards(temp: Path) -> None: + path = _empty_state(temp / "guards.json") + for _ in range(3): + retried = record_feedback("同一支片的意見", "cut-1 @ 3.0s", state_path=path) + assert retried["record"]["support"] == 1 and not retried["record"]["pinned"], \ + "a retried review of one cut must not promote a rule" + for note in ("first wording", "second wording", "third wording"): + same_video = record_feedback("同一來源", note, source="video-a@abc123", state_path=path) + assert same_video["record"]["support"] == 1 and not same_video["record"]["pinned"] + for video in ("video-a@1", "video-b@2", "video-c@3"): + repeated = record_feedback("三支片都出現", "note", source=video, state_path=path) + assert repeated["record"]["support"] == 3 and repeated["record"]["pinned"] + hard = record_feedback("安全硬規則", "creator instruction", kind="hard", pin=True, state_path=path) + again = record_feedback("安全硬規則", "seen once more", kind="hard", state_path=path) + assert again["record"]["pinned"] and again["record"]["pinned_explicit"], \ + "more evidence must not demote an explicit pin" + rival = record_feedback("相反規則", "creator note", pin=True, contradicts=hard["record"]["id"], + state_path=path) + blocked = {row["id"] for row in select_rules(format="shorts", limit=10, state_path=path)} + assert hard["record"]["id"] not in blocked and rival["record"]["id"] not in blocked + record_feedback("相反規則", "creator resolved it", supersedes=hard["record"]["id"], state_path=path) + resolved = {row["id"] for row in select_rules(format="shorts", limit=10, state_path=path)} + assert rival["record"]["id"] in resolved and hard["record"]["id"] not in resolved, \ + "superseding must resolve the conflict it settles" + assert audit(path)["status"] == "GREEN" + + +def _self_test_ledger_location(temp: Path) -> None: + global SKILL_ROOT, DEFAULT_STATE + kit = temp / "kit" + (kit / "src").mkdir(parents=True) + (kit / "release-manifest.json").write_text("{}", encoding="utf-8") + learned = {"id": "K-legacy", "fingerprint": "legacy", "scope": "project", "formats": ["*"], + "domains": ["*"], "kind": "soft", "rule": "舊規則", "support": 3, + "evidence": ["a", "b", "c"], "contradictions": [], "pinned": True} + seed = _empty_state(kit / "knowledge" / "runtime" / "state.json", [learned]) + saved = SKILL_ROOT, DEFAULT_STATE, os.environ.pop(LEDGER_ENV, None) + SKILL_ROOT, DEFAULT_STATE = kit / "src", seed + try: + ledger = kit / "data" / "knowledge_state.json" + assert default_ledger_path() == ledger + assert [row["rule"] for row in select_rules(format="longform")] == ["舊規則"] + assert ledger.is_file(), "rules learned in the managed seed must migrate to data/" + _empty_state(seed) # an update replaces the seed with the empty public ledger + record_feedback("新規則", "after update") + rules = {row["rule"] for row in load_state(ledger)["records"]} + assert rules == {"舊規則", "新規則"} and not load_state(seed)["records"] + os.environ[LEDGER_ENV] = str(temp / "override.json") + assert default_ledger_path() == temp / "override.json" + finally: + SKILL_ROOT, DEFAULT_STATE = saved[0], saved[1] + os.environ.pop(LEDGER_ENV, None) + if saved[2] is not None: + os.environ[LEDGER_ENV] = saved[2] + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Bounded learning memory") subs = parser.add_subparsers(dest="command", required=True) @@ -391,6 +531,8 @@ def main(argv: list[str] | None = None) -> int: rec.add_argument("--contradicts", default=None) rec.add_argument("--supersedes", default=None) rec.add_argument("--pin", action="store_true") + rec.add_argument("--source", default=None, + help="observed video/version id; repeats of one source never add support") rotate = subs.add_parser("rotate-video-log") rotate.add_argument("--apply", action="store_true") subs.add_parser("selftest") @@ -410,6 +552,7 @@ def main(argv: list[str] | None = None) -> int: contradicts=args.contradicts, supersedes=args.supersedes, pin=args.pin, + source=args.source, ) elif args.command == "rotate-video-log": result = rotate_numbered_video_log(apply=args.apply) diff --git a/src/longform_maker/shorts_gate.py b/src/longform_maker/shorts_gate.py index 08ebc26..66f1d57 100644 --- a/src/longform_maker/shorts_gate.py +++ b/src/longform_maker/shorts_gate.py @@ -10,9 +10,14 @@ 【內容】S-J 讀畫面文字（品名+價格）/ S-K 運鏡不移開主體 / S-L 品名取該主體正上方那張牌 API: - expand_caps(spec) -> [(start, end, blocks, kind)] # 由 segment 索引算時間 - gate_shorts(spec) -> (ok, report) # 全規則檢查 - assert_shorts(spec) -> spec（含展開後 caps） # build 前呼叫，不過直接 raise + expand_caps(spec) -> [(start, end, blocks, kind)] # 由 segment 索引算時間 + gate_shorts(spec, rules=None) -> (ok, report) # 全規則檢查 + assert_shorts(spec, rules=None)-> spec（含展開後 caps） # build 前呼叫，不過直接 raise + merge_rules(rules=None, platform=None) -> dict # DEFAULT_RULES ← 平台 ← 你的覆寫 + +門檻是【範例校準值】：用自己表現最好／最差的 3-5 支片重算後，以 `rules=` 逐鍵覆寫， +不必改本檔。鍵名打錯直接 raise（靜默沿用預設＝你以為改了其實沒有）。 +片長三鍵（dur_min / dur_max / dur_deadzone）覆寫的是 spec 所選平台的片長帶。 spec 結構（一支 Short）: { @@ -29,9 +34,11 @@ """ from __future__ import annotations +import dataclasses import json import math import re +import unicodedata import os import sys @@ -111,8 +118,14 @@ SR_FAIL = 7.0 # 字/秒 > 7 → fail（讀不完=白寫） -def _nchars(txt: str) -> int: - return sum(1 for ch in txt if not ch.isspace()) +def _nchars(txt: str) -> float: + """閱讀單位：全形（CJK／全形標點／emoji）計 1、半形英數與符號計 0.5、空白不計。 + + 同東亞排版的全形／半形寬度；英文與中英混排字幕因此能用同一組 S-R／S-S + 門檻，不會把 15 個英文字母當成 15 個中文字而誤擋。純中文字幕的結果不變。 + """ + return sum(1.0 if unicodedata.east_asian_width(ch) in ("W", "F") else 0.5 + for ch in txt if not ch.isspace()) def _is_official_go_shoot(txt: str) -> bool: @@ -165,6 +178,57 @@ def _is_official_go_shoot(txt: str) -> bool: first_frame_ratio=SQ_RATIO, ) +# ── rules= 可覆寫門檻：公開鍵名 → GatePolicy 欄位（片長三鍵另外覆寫平台片長帶） +RULE_FIELDS = { + "first_cut_max": "first_cut_max", + "loop_tol": "loop_tolerance", + "tail_clear": "tail_clear", + "nonwhite_max_ratio": "nonwhite_max_ratio", + "nonwhite_max_colors": "nonwhite_max_colors", + "kinetic_max_ratio": "kinetic_max_ratio", + "reading_warn": "reading_warn", + "reading_fail": "reading_fail", + "cap_dwell_warn": "caption_dwell_warn", + "cap_rate_warn": "caption_rate_warn", + "first_frame_ratio": "first_frame_ratio", +} +DURATION_FIELDS = {"dur_min": "dur_min", "dur_max": "dur_max", "dur_deadzone": "deadzone"} +DEFAULT_RULES = { + **{key: PLATFORM_RULES[DEFAULT_PLATFORM][field] for key, field in DURATION_FIELDS.items()}, + **{key: getattr(_GATE_POLICY, field) for key, field in RULE_FIELDS.items()}, +} + + +def merge_rules(rules: dict | None = None, platform: str | None = None) -> dict: + """DEFAULT_RULES ← platform 片長帶 ← 你的 rules（後者永遠贏）；未知平台／鍵名直接 raise。""" + merged = dict(DEFAULT_RULES) + if platform is not None: + if platform not in PLATFORM_RULES: + raise AssertionError("unknown platform %r; valid: %s" + % (platform, ", ".join(sorted(PLATFORM_RULES)))) + band = PLATFORM_RULES[platform] + merged.update({key: band[field] for key, field in DURATION_FIELDS.items()}) + for key, value in (rules or {}).items(): + if key not in DEFAULT_RULES: + raise AssertionError("unknown rule key %r; valid: %s" + % (key, ", ".join(sorted(DEFAULT_RULES)))) + merged[key] = value + return merged + + +def policy_for(rules: dict | None = None) -> GatePolicy: + """回傳套用 rules 的 GatePolicy；None／空 dict = 出廠校準值。""" + if not rules: + return _GATE_POLICY + merge_rules(rules) + changes = {RULE_FIELDS[key]: value for key, value in rules.items() if key in RULE_FIELDS} + band = {DURATION_FIELDS[key]: value for key, value in rules.items() if key in DURATION_FIELDS} + if band: + changes["platform_rules"] = { + name: dict(rule, **band) for name, rule in PLATFORM_RULES.items() + } + return dataclasses.replace(_GATE_POLICY, **changes) + def _first_frame_quality(spec: dict): """回 {chosen, best, top} 或 None（無 _scan.json 時靜默跳過=向後相容）。""" @@ -381,26 +445,27 @@ def _battle_result_findings(spec: dict) -> tuple[list[str], list[str]]: # ────────────────────────────────────────────── 總閘門 -def gate_shorts(spec: dict): - """回傳 (ok, report)。report["fails"] 非空 = 不准出片。""" +def gate_shorts(spec: dict, rules: dict | None = None): + """回傳 (ok, report)。report["fails"] 非空 = 不准出片；rules 見 merge_rules。""" + policy = policy_for(rules) fails, warns = [], [] validate_required_fields(spec, fails) if fails: return False, _report(fails, warns) - timeline = validate_timeline(spec, _GATE_POLICY, fails) + timeline = validate_timeline(spec, policy, fails) captions = validate_opening_and_battle( spec, fails, warns, _battle_matchup_failures, _battle_result_findings, ) validate_caption_plan( - spec, timeline, captions, _GATE_POLICY, fails, warns, + spec, timeline, captions, policy, fails, warns, _nchars, _is_official_go_shoot, ) if fails: return False, _report(fails, warns, dur=timeline.duration) return finalize_expanded_report( - spec, timeline, _GATE_POLICY, fails, warns, + spec, timeline, policy, fails, warns, expand_caps=expand_caps, seg_bounds=seg_bounds, report=_report, @@ -424,11 +489,13 @@ def _attach_addr(spec: dict, rep: dict) -> dict: return dict(spec, caps=sorted(caps), _dur=rep["dur"], _warns=rep["warns"]) -# build 前呼叫：不過直接 raise；過了回傳含展開 caps 的 spec（附地址常駐條）。 -assert_shorts = make_assert(gate_shorts, - lambda spec: spec.get("name", "?"), - "Shorts gate FAIL", - post=_attach_addr) +def assert_shorts(spec: dict, rules: dict | None = None) -> dict: + """build 前呼叫：不過直接 raise；過了回傳含展開 caps 的 spec（附地址常駐條）。""" + policy_for(rules) # 鍵名打錯當場 raise，不必等跑完 gate + return make_assert(lambda item: gate_shorts(item, rules), + lambda item: item.get("name", "?"), + "Shorts gate FAIL", + post=_attach_addr)(spec) # ────────────────────────────────────────────── self-test @@ -783,8 +850,57 @@ def _selftest_reading_rules(check, mk, good): check("assert_shorts raises on fail", "Shorts gate FAIL" in str(e)) +def _selftest_reading_units(check, mk, good): + check("CJK counts one unit per character", _nchars("測試說明") == 4.0) + check("half-width letters count half a unit", _nchars("Riverside Market") == 7.5) + check("mixed caption sums both widths", _nchars("只要 557") == 3.5) + english = mk(caps_by_seg=[(0, [("Riverside Market", "gold")], "hook"), + (1, [("a 40-year-old noodle stall", "white")], "sub"), + (2, [("hand-pulled every morning", "white")], "sub"), + (3, [("the broth simmers 8 hours", "white")], "sub")], + what="a 40-year-old noodle stall", place="Riverside Market") + ok, rep = gate_shorts(english) + check("readable English captions are not S-R blocked", + not any(item.startswith("S-R") for item in rep["fails"])) + rushed = dict(english, caps_by_seg=[(0, [("Riverside Market", "gold")], "hook"), + (0, [("a 40-year-old noodle stall since 1985", "white")], "sub"), + *english["caps_by_seg"][2:]]) + rushed_fails = gate_shorts(rushed)[1]["fails"] + check("rushed English line is still S-R blocked", + any(item.startswith("S-R") for item in rushed_fails)) + + +def _selftest_rule_overrides(check, mk, good): + first, *middle, loop = good["segs"] + long_cut = mk(segs=[first, *[(src, start, 9.0) for src, start, _ in middle], loop]) + ok_default, rep_default = gate_shorts(long_cut) + check("31s spec fails the shipped YT band", not ok_default + and any(item.startswith("S-B") for item in rep_default["fails"])) + own = {"dur_min": 26.0, "dur_max": 60.0, "dur_deadzone": None} + check("rules= widens the duration band", gate_shorts(long_cut, own)[0]) + check("platform ig_reels needs no override", gate_shorts(dict(long_cut, platform="ig_reels"))[0]) + check("rules= beats the platform band", + not gate_shorts(dict(long_cut, platform="ig_reels"), {"dur_max": 25.0})[0]) + check("merge_rules(None) is the shipped calibration", merge_rules() == DEFAULT_RULES) + check("merge_rules applies the platform band", + merge_rules(None, "ig_reels")["dur_deadzone"] is None) + for bad_call in (lambda: merge_rules({"dur_mni": 1.0}), lambda: merge_rules(None, "tiktok"), + lambda: gate_shorts(good, {"typo": 1}), lambda: assert_shorts(good, {"typo": 1})): + try: + bad_call() + check("unknown rule/platform names raise", False) + except AssertionError as exc: + check("unknown rule/platform names raise", "unknown" in str(exc)) + strict = gate_shorts(good, {"first_cut_max": 0.5}) + check("rules= tightens S-C first cut", not strict[0] + and any(item.startswith("S-C") for item in strict[1]["fails"])) + check("assert_shorts honours rules=", assert_shorts(long_cut, own)["_dur"] > 30.0) + + def _selftest_body(check): dummy, mk, good = _selftest_fixture() + _selftest_rule_overrides(check, mk, good) + _selftest_reading_units(check, mk, good) _selftest_battle_rules(check, mk, good) _selftest_gate_rules(check, dummy, mk, good) _selftest_caption_rendering(check, mk, good) diff --git a/src/longform_maker/shorts_gate_validation.py b/src/longform_maker/shorts_gate_validation.py index bf556da..b816741 100644 --- a/src/longform_maker/shorts_gate_validation.py +++ b/src/longform_maker/shorts_gate_validation.py @@ -228,7 +228,7 @@ def _validate_caption_render_contract(spec: dict, captions: list, policy: GatePo official = bool(spec.get("battle_matchup")) and is_official_go_shoot(text) if kind in policy.kind_char_limit and not official \ and nchars(text) > policy.kind_char_limit[kind]: - fails.append("S-S %s 字幕 %d 字 > %d 字；強效果必須短" + fails.append("S-S %s 字幕 %g 字 > %d 字；強效果必須短" % (kind, nchars(text), policy.kind_char_limit[kind])) if kind in policy.kinetic_kinds: kinetic.append(kind) @@ -289,7 +289,7 @@ def _validate_reading_speed(spec: dict, content: list, policy: GatePolicy, if bool(spec.get("battle_matchup")) and is_official_go_shoot(full_text): continue if chars_per_second > policy.reading_fail: - fails.append("S-R 讀不完：%r %d 字只停 %.2fs = %.1f 字/秒（上限 %.0f）" + fails.append("S-R 讀不完：%r %g 字只停 %.2fs = %.1f 字/秒（上限 %.0f）" % (blocks[0][0].replace("\n", "/")[:12], chars, dwell, chars_per_second, policy.reading_fail)) elif chars_per_second > policy.reading_warn: diff --git a/src/shorts_autopilot.py b/src/shorts_autopilot.py index 3b6a872..b1efe05 100644 --- a/src/shorts_autopilot.py +++ b/src/shorts_autopilot.py @@ -55,8 +55,11 @@ # ───────────────────────────────────────────── 步驟1：scan -def scan(folder_id: str) -> dict: - """正規化 + GPS + 接觸表 + 標價牌放大圖 + _plan.py 骨架。""" +def scan(folder_id: str, platform: str | None = None) -> dict: + """正規化 + GPS + 接觸表 + 標價牌放大圖 + _plan.py 骨架。 + + platform 寫進 _plan.py 的 SPEC；build 的 shorts_gate 依它選片長帶（未指定＝yt_shorts）。 + """ from silent_vlog_maker import normalize_to_portrait, extract_gps src_dir = os.path.join(INBOX, folder_id) @@ -151,6 +154,7 @@ def scan(folder_id: str) -> dict: "SPEC = dict(", ' name="s%s_TODO",' % folder_id, ' niche="auto", # auto / food / travel / cafe / toy / product / game / diy / ai / documentary / interview / automotive / fitness / fashion / architecture / business / nature / music', + ' platform="%s", # yt_shorts（26-44s 死區）/ ig_reels / fb_reels；決定 gate 的片長帶' % (platform or "yt_shorts"), ' place="TODO 地名/店名",', ' what="TODO 一句這是什麼",', ' addr="TODO 📍 名稱｜完整地址",', @@ -737,21 +741,24 @@ def build(folder_id: str) -> dict: def main(): + ap = argparse.ArgumentParser() + ap.add_argument("cmd", choices=["scan", "snapshot", "build", "selftest"]) + ap.add_argument("folders", nargs="*") + ap.add_argument("--platform", choices=("yt_shorts", "ig_reels", "fb_reels"), default=None, + help="scan：寫進 _plan.py 的目標平台（片長帶依平台；預設 yt_shorts）") + a = ap.parse_args() # Public-kit installs include a bounded, compatibility-gated updater. The # private canonical workspace intentionally has no startup_update module, - # so development runs remain local and deterministic. - raw_args = sys.argv[1:] - if raw_args and raw_args[0] != "selftest": + # so development runs remain local and deterministic. It runs only for a + # real production command: --help, a usage error or selftest must not reach + # the network or migrate the workspace. + if a.cmd != "selftest" and a.folders: try: from startup_update import ensure_current # type: ignore except ImportError: pass else: ensure_current() - ap = argparse.ArgumentParser() - ap.add_argument("cmd", choices=["scan", "snapshot", "build", "selftest"]) - ap.add_argument("folders", nargs="*") - a = ap.parse_args() if a.cmd == "selftest": six = _wave_segment_durations(6, 12.4) assert len(six) == 6 and abs(sum(six) - 12.4) < .08 @@ -764,6 +771,8 @@ def main(): "chosen-folder.wav", ) assert locked == "chosen-folder.wav" and source == "folder_lock" + from shorts_gate import PLATFORM_RULES + assert set(PLATFORM_RULES) == {"yt_shorts", "ig_reels", "fb_reels"}, "sync --platform choices" assert LATEST_VISUAL_PLAN_VERSION == 9 assert LATEST_COMPOSITION_SCHEMA == "hao.motion-composition/v1" print("shorts_autopilot self-test OK") @@ -772,7 +781,7 @@ def main(): ap.error("scan/snapshot/build requires at least one folder id") for fid in a.folders: if a.cmd == "scan": - scan(fid) + scan(fid, a.platform) elif a.cmd == "snapshot": rec = snapshot_editorial_baseline(fid) print("[snapshot] %s %s" % (fid, rec["sha256"][:12])) diff --git a/src/teardown.py b/src/teardown.py index c9eada0..fd44d41 100644 --- a/src/teardown.py +++ b/src/teardown.py @@ -26,16 +26,85 @@ "mid": (0.42, 0.24, 0.18), } +# 換句÷剪點 高於這個倍數 → 判定「節奏主體是字幕」（knowledge/vertical-teardown-method.md §3）。 +CAPTION_DRIVEN_RATIO = 1.15 + +# 選配套件（缺了只降級，節奏量測照跑） +OPTIONAL_PKGS = ("rapidocr-onnxruntime", "opencc-python-reimplemented") + def _run(argv): return subprocess.run(argv, capture_output=True, text=True, errors="replace") +def rhythm_stats(timestamps, duration): + """Pure-Python rhythm statistics for one event track (cuts or caption changes). + + Returns n, per_min, gaps, gap_median, gap_stdev, gap_min and gap_max. The + standard deviation separates a metronomic 1.1s cut pattern from a 0.4s/3s + alternation that has the same median. Non-positive duration yields + per_min 0.0 instead of a division error. + """ + ts = sorted(float(t) for t in timestamps) + dur = float(duration) + gaps = [round(b - a, 3) for a, b in zip(ts, ts[1:])] + out = {"n": len(ts), "per_min": round(len(ts) / dur * 60, 2) if dur > 0 else 0.0, + "gaps": gaps, "gap_median": None, "gap_stdev": None, + "gap_min": None, "gap_max": None} + if gaps: + out["gap_median"] = round(st.median(gaps), 2) + out["gap_stdev"] = round(st.pstdev(gaps), 2) if len(gaps) > 1 else 0.0 + out["gap_min"] = round(min(gaps), 2) + out["gap_max"] = round(max(gaps), 2) + return out + + +def pace_profile(cut_times, caption_times, duration): + """Compare the cut track with the caption-change track; report the rhythm driver. + + driver is 'captions', 'cuts' or 'unknown'. The evidence level is PLAUSIBLE + (successful samples only, no failure control group), so this only reports + numbers and never blocks a build. + """ + cuts_s = rhythm_stats(cut_times, duration) + caps_s = rhythm_stats(caption_times, duration) + if cuts_s["per_min"] <= 0: + ratio, driver = None, "unknown" + verdict = "no cuts detected -> cannot compare (single take, or lower --thresh)" + else: + ratio = round(caps_s["per_min"] / cuts_s["per_min"], 2) + driver = "captions" if ratio > CAPTION_DRIVEN_RATIO else "cuts" + verdict = "captions/cuts = %.2fx -> rhythm is %s-driven" % ( + ratio, "CAPTION" if driver == "captions" else "CUT") + return {"cuts": cuts_s, "captions": caps_s, "ratio": ratio, + "driver": driver, "verdict": verdict} + + +def ocr_status(): + """Return (ocr_ok, s2t_ok, install_hint) by import probing only (no model load).""" + missing = [] + try: + import rapidocr_onnxruntime # noqa: F401 + ocr_ok = True + except Exception: # noqa: BLE001 - absent package or failed onnx load both degrade + ocr_ok = False + missing.append(OPTIONAL_PKGS[0]) + try: + import opencc # noqa: F401 + s2t_ok = True + except Exception: # noqa: BLE001 + s2t_ok = False + missing.append(OPTIONAL_PKGS[1]) + return ocr_ok, s2t_ok, ("pip install " + " ".join(missing)) if missing else "" + + def probe(path): o = _run(["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,r_frame_rate", "-show_entries", "format=duration", "-of", "csv=p=0", path]).stdout vals = [x for x in o.replace("\n", ",").split(",") if x] + if len(vals) < 4: + raise RuntimeError("ffprobe gave no usable video stream info for %r" % path) w, h = int(vals[0]), int(vals[1]) fr = vals[2] dur = float(vals[-1]) @@ -59,15 +128,14 @@ def loudness(path): def _stats(ts, dur, label): - iv = [round(b - a, 2) for a, b in zip(ts, ts[1:])] - print(" %-10s %3d 次 / %5.1f 次每分" % (label, len(ts), len(ts) / dur * 60), end="") - if iv: + stats = rhythm_stats(ts, dur) + print(" %-10s %3d 次 / %5.1f 次每分" % (label, stats["n"], stats["per_min"]), end="") + if stats["gaps"]: print(" 中位間隔 %.2fs 標準差 %.2f 最短 %.2f" % - (st.median(iv), st.pstdev(iv) if len(iv) > 1 else 0, - min(iv))) + (stats["gap_median"], stats["gap_stdev"], stats["gap_min"])) else: print() - return iv + return stats["gaps"] def caption_track(path, band, dur, w, h, step=0.5, min_conf=0.75): @@ -148,28 +216,27 @@ def _near_dup(a, b, ratio=0.86): return SequenceMatcher(None, a, b).ratio() >= ratio -def teardown(path, band="wide"): +def teardown(path, band="wide", thresh=0.25): w, h, fps, dur = probe(path) print("=" * 62) print("%s %dx%d %gfps %.1fs LUFS %s" % (os.path.basename(path), w, h, fps, dur, loudness(path))) print("=" * 62) - ct = cuts(path) - cut_iv = _stats(ct, dur, "剪點") + ct = cuts(path, thresh) + _stats(ct, dur, "剪點") caps = caption_track(path, band, dur, w, h) if caps is None: - print(" [SKIP] OCR 套件未安裝 -> pip install rapidocr-onnxruntime " - "opencc-python-reimplemented") + hint = ocr_status()[2] or "pip install " + " ".join(OPTIONAL_PKGS) + print(" [SKIP] OCR 套件未安裝 -> %s（上方節奏數字完整不受影響）" % hint) return cap_ts = [t for t, _x in caps] - cap_iv = _stats(cap_ts, dur, "換句") - - if cut_iv and cap_iv: - r = (len(cap_ts) / dur) / max(len(ct) / dur, 1e-9) + _stats(cap_ts, dur, "換句") + profile = pace_profile(ct, cap_ts, dur) + if profile["ratio"] is not None: print(" 換句/剪點 = %.1fx -> %s" % - (r, "節奏主體是字幕" if r > 1.15 else "節奏主體是剪點")) + (profile["ratio"], "節奏主體是字幕" if profile["driver"] == "captions" else "節奏主體是剪點")) print("\n -- 自動抽出的字幕腳本（OCR + s2twp；**需人工複核**）--") for t, txt in caps: @@ -182,6 +249,8 @@ def main(): ap.add_argument("target", help="影片檔或含影片的資料夾") ap.add_argument("--band", default="wide", choices=sorted(BANDS), help="字幕帶（預設 wide：字幕會在片內移動，窄帶會靜默漏掉）") + ap.add_argument("--thresh", type=float, default=0.25, + help="剪點偵測的 scene 門檻（預設 0.25；先用人工數 10 秒校準）") a = ap.parse_args() vids = ([a.target] if os.path.isfile(a.target) @@ -190,7 +259,7 @@ def main(): print("找不到 mp4：%s" % a.target) return 1 for v in vids: - teardown(v, a.band) + teardown(v, a.band, a.thresh) print() return 0 From 86ce54a989e3b878780350776e155781728e033a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 06:40:45 +0000 Subject: [PATCH 02/26] fix: bind human review to the reviewed render and refuse self-certification - review server stamps saved reviews with an HMAC over the exact video sha256 - finalize recomputes the render hash, requires a submitted stamped review, archives stale reviews on rebuild and learns each issue once per render - quality_95 certifies only from a verified review; missing evidence becomes an explicit REVIEW gap instead of a silent full score; NaN never scores - aesthetic/thumbnail scores reject non-finite ratings; flat or low-res thumbnails raise review signals; shorts rule overrides must be finite - silent_vlog_maker imports again on Python 3.9 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XLikBpKWqhMahFDuvVj6wh --- .../design_runtime/src/aesthetic_score.py | 21 +- src/aesthetic_score.py | 21 +- src/longform_maker/shorts_gate.py | 24 +++ src/longform_maker/shorts_gate_validation.py | 4 +- src/quality_95.py | 186 +++++++++++++++--- src/review_loop.py | 152 ++++++++++++-- src/silent_vlog_maker/asset_scanner.py | 2 + src/thumbnail_algorithm_score.py | 36 +++- 8 files changed, 399 insertions(+), 47 deletions(-) diff --git a/codex-skill/video-autopilot/design_runtime/src/aesthetic_score.py b/codex-skill/video-autopilot/design_runtime/src/aesthetic_score.py index 3939f5e..7659c9c 100644 --- a/codex-skill/video-autopilot/design_runtime/src/aesthetic_score.py +++ b/codex-skill/video-autopilot/design_runtime/src/aesthetic_score.py @@ -9,6 +9,7 @@ import argparse import json +import math from pathlib import Path from typing import Any @@ -138,6 +139,13 @@ def _signal_blockers(signals: dict[str, Any]) -> list[str]: return blockers +def _rating(raw: Any) -> float | None: + if isinstance(raw, bool) or not isinstance(raw, (int, float)): + return None + value = float(raw) + return value if math.isfinite(value) else None + + def score_review(review: dict[str, Any], *, format: str = "shorts", signals: dict[str, Any] | None = None, standard: dict[str, Any] | None = None) -> dict[str, Any]: @@ -148,12 +156,13 @@ def score_review(review: dict[str, Any], *, format: str = "shorts", low = [] minimum = float(schema["contract"]["minimum_dimension_rating"]) for key, dim in schema["dimensions"].items(): - raw = ratings.get(key) + raw = _rating(ratings.get(key)) if raw is None: + # Absent, NaN, boolean or non-numeric ratings are unrated, never a 5. missing.append(key) rating = 0.0 else: - rating = max(1.0, min(5.0, float(raw))) + rating = max(1.0, min(5.0, raw)) points = rating / 5.0 * float(dim["weight"]) earned += points if raw is not None and rating < minimum: @@ -163,7 +172,8 @@ def score_review(review: dict[str, Any], *, format: str = "shorts", "question": dim["question"]}) score = round(earned, 1) blockers = _signal_blockers(signals or {}) - completed = not missing and bool(review.get("completed_at") or review.get("review_id")) + # Only an explicit submission completes a review; an id alone is bookkeeping. + completed = not missing and bool(review.get("completed_at")) contract = schema["contract"] if blockers or (completed and score < float(contract["block_below"])): status = "BLOCKED" @@ -217,6 +227,11 @@ def self_test() -> None: assert score_review(bad, signals={"grid_used_as_default_opener": True})["status"] == "BLOCKED" mismatch = score_review(bad, signals={"style_domain_match": False}) assert mismatch["status"] == "BLOCKED" + for invalid in (float("nan"), float("inf"), True, "5"): + forged = score_review({"ratings": {key: invalid for key in dims}, "completed_at": "now"}) + assert forged["status"] == "REVIEW" and not forged["completed"], invalid + unsubmitted = score_review({"ratings": {key: 5 for key in dims}, "review_id": "auto-id"}) + assert unsubmitted["status"] == "REVIEW", "a review id is not a submission" print("aesthetic score self-test GREEN") diff --git a/src/aesthetic_score.py b/src/aesthetic_score.py index 3939f5e..7659c9c 100644 --- a/src/aesthetic_score.py +++ b/src/aesthetic_score.py @@ -9,6 +9,7 @@ import argparse import json +import math from pathlib import Path from typing import Any @@ -138,6 +139,13 @@ def _signal_blockers(signals: dict[str, Any]) -> list[str]: return blockers +def _rating(raw: Any) -> float | None: + if isinstance(raw, bool) or not isinstance(raw, (int, float)): + return None + value = float(raw) + return value if math.isfinite(value) else None + + def score_review(review: dict[str, Any], *, format: str = "shorts", signals: dict[str, Any] | None = None, standard: dict[str, Any] | None = None) -> dict[str, Any]: @@ -148,12 +156,13 @@ def score_review(review: dict[str, Any], *, format: str = "shorts", low = [] minimum = float(schema["contract"]["minimum_dimension_rating"]) for key, dim in schema["dimensions"].items(): - raw = ratings.get(key) + raw = _rating(ratings.get(key)) if raw is None: + # Absent, NaN, boolean or non-numeric ratings are unrated, never a 5. missing.append(key) rating = 0.0 else: - rating = max(1.0, min(5.0, float(raw))) + rating = max(1.0, min(5.0, raw)) points = rating / 5.0 * float(dim["weight"]) earned += points if raw is not None and rating < minimum: @@ -163,7 +172,8 @@ def score_review(review: dict[str, Any], *, format: str = "shorts", "question": dim["question"]}) score = round(earned, 1) blockers = _signal_blockers(signals or {}) - completed = not missing and bool(review.get("completed_at") or review.get("review_id")) + # Only an explicit submission completes a review; an id alone is bookkeeping. + completed = not missing and bool(review.get("completed_at")) contract = schema["contract"] if blockers or (completed and score < float(contract["block_below"])): status = "BLOCKED" @@ -217,6 +227,11 @@ def self_test() -> None: assert score_review(bad, signals={"grid_used_as_default_opener": True})["status"] == "BLOCKED" mismatch = score_review(bad, signals={"style_domain_match": False}) assert mismatch["status"] == "BLOCKED" + for invalid in (float("nan"), float("inf"), True, "5"): + forged = score_review({"ratings": {key: invalid for key in dims}, "completed_at": "now"}) + assert forged["status"] == "REVIEW" and not forged["completed"], invalid + unsubmitted = score_review({"ratings": {key: 5 for key in dims}, "review_id": "auto-id"}) + assert unsubmitted["status"] == "REVIEW", "a review id is not a submission" print("aesthetic score self-test GREEN") diff --git a/src/longform_maker/shorts_gate.py b/src/longform_maker/shorts_gate.py index 66f1d57..6e93b18 100644 --- a/src/longform_maker/shorts_gate.py +++ b/src/longform_maker/shorts_gate.py @@ -212,10 +212,27 @@ def merge_rules(rules: dict | None = None, platform: str | None = None) -> dict: if key not in DEFAULT_RULES: raise AssertionError("unknown rule key %r; valid: %s" % (key, ", ".join(sorted(DEFAULT_RULES)))) + if not _valid_rule_value(key, value): + # NaN/inf/text would silently disable a rule instead of calibrating it. + raise AssertionError("invalid value for rule %r: %r" % (key, value)) merged[key] = value + if merged["dur_min"] >= merged["dur_max"]: + raise AssertionError("dur_min must be below dur_max: %r" % merged) return merged +def _finite_number(value) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) + + +def _valid_rule_value(key: str, value) -> bool: + if key == "dur_deadzone": + return value is None or ( + isinstance(value, (tuple, list)) and len(value) == 2 + and all(_finite_number(item) for item in value) and value[0] < value[1]) + return _finite_number(value) and value >= 0 + + def policy_for(rules: dict | None = None) -> GatePolicy: """回傳套用 rules 的 GatePolicy；None／空 dict = 出廠校準值。""" if not rules: @@ -891,6 +908,13 @@ def _selftest_rule_overrides(check, mk, good): check("unknown rule/platform names raise", False) except AssertionError as exc: check("unknown rule/platform names raise", "unknown" in str(exc)) + for bad_value in ({"first_cut_max": float("nan")}, {"reading_fail": "7"}, {"dur_min": 30.0}, + {"dur_deadzone": (40.0, 30.0)}, {"nonwhite_max_ratio": True}): + try: + gate_shorts(good, bad_value) + check("non-finite or inverted thresholds raise", False) + except AssertionError as exc: + check("non-finite or inverted thresholds raise", "rule" in str(exc) or "dur_min" in str(exc)) strict = gate_shorts(good, {"first_cut_max": 0.5}) check("rules= tightens S-C first cut", not strict[0] and any(item.startswith("S-C") for item in strict[1]["fails"])) diff --git a/src/longform_maker/shorts_gate_validation.py b/src/longform_maker/shorts_gate_validation.py index b816741..98d66a1 100644 --- a/src/longform_maker/shorts_gate_validation.py +++ b/src/longform_maker/shorts_gate_validation.py @@ -209,7 +209,9 @@ def _validate_caption_render_contract(spec: dict, captions: list, policy: GatePo fails: list[str], warns: list[str], nchars, is_official_go_shoot) -> None: resolve_color = _color_resolver() - if resolve_color: + if not resolve_color: + warns.append("顏色鍵未驗證：silent_vlog_maker 無法載入（缺 Pillow／numpy 時屬正常降級）") + else: for _index, blocks, _kind in captions: for _text, color in blocks: try: diff --git a/src/quality_95.py b/src/quality_95.py index f1dd748..a82c374 100644 --- a/src/quality_95.py +++ b/src/quality_95.py @@ -8,14 +8,19 @@ from __future__ import annotations import argparse +import hashlib +import hmac import json +import math import os +import secrets import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Any from asset_memory import asset_fatigue_report +from aesthetic_score import review_schema from aesthetic_score import score_review as score_aesthetic_review from project_paths import discover_project_root from quality_corpus import detect_negative, load_corpus, narrative_similarity @@ -24,6 +29,67 @@ ROOT = Path(__file__).resolve().parent PROJECT_ROOT = discover_project_root(ROOT) NARRATIVE_STATE = PROJECT_ROOT / "videos" / "_state" / "quality_narratives.json" +REVIEW_KEY_ENV = "VIDEO_AUTOPILOT_REVIEW_KEY" +# Signals a spec may only make stricter: a good signal can be lowered, never raised. +GOOD_SIGNALS = frozenset({"text_safe_area_pass", "style_domain_match", "design_dna_compiled", + "tracked_label_has_evidence"}) + + +def review_key_path() -> Path: + explicit = os.environ.get(REVIEW_KEY_ENV, "").strip() + return Path(explicit).expanduser() if explicit else Path.home() / ".video-autopilot" / "review-signing.key" + + +def _review_key(create: bool = False) -> bytes | None: + """Per-machine HMAC key held by the review server, never inside a project.""" + path = review_key_path() + if create and not path.is_file(): + path.parent.mkdir(parents=True, exist_ok=True) + try: + handle = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError: + pass + else: + with os.fdopen(handle, "w", encoding="ascii") as stream: + stream.write(secrets.token_hex(32)) + try: + key = path.read_text(encoding="ascii").strip() + except (OSError, UnicodeDecodeError): + return None + return key.encode("ascii") if len(key) >= 32 else None + + +def _review_mac(key: bytes, review: dict) -> str: + body = {name: value for name, value in review.items() if name != "receipt"} + canonical = json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hmac.new(key, canonical.encode("utf-8"), hashlib.sha256).hexdigest() + + +def stamp_review(review: dict, artifact_sha256: str) -> dict: + """Review-server stamp binding a saved review to the exact reviewed bytes.""" + key = _review_key(create=True) + if key is None: + raise RuntimeError("review signing key is unavailable") + stamped = {name: value for name, value in review.items() if name != "receipt"} + stamped["artifact_sha256"] = str(artifact_sha256 or "") + stamped["receipt"] = _review_mac(key, stamped) + return stamped + + +def verify_review(review: Any, artifact_sha256: str | None = None) -> str: + """Return '' for a submitted, server-stamped review of ``artifact_sha256``.""" + if not isinstance(review, dict) or not review.get("completed_at"): + return "review was not submitted from the review page (completed_at missing)" + key = _review_key() + receipt = review.get("receipt") + if key is None or not isinstance(receipt, str): + return "review is not stamped by the review server" + if not hmac.compare_digest(receipt, _review_mac(key, review)): + return "review stamp does not verify (edited after submission)" + if not review.get("artifact_sha256") or ( + artifact_sha256 is not None and review.get("artifact_sha256") != artifact_sha256): + return "review belongs to a different render of this video" + return "" def _atomic_json(path: Path, payload: dict) -> None: @@ -70,15 +136,26 @@ def record_narrative(content_id: str, signature: str, _atomic_json(path, state) +def _finite(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + number = float(value) + return number if math.isfinite(number) else None + + def _value(evidence: dict, name: str) -> tuple[float, str]: raw = evidence.get("dimensions", {}).get(name) if isinstance(raw, dict): - return max(0.0, min(1.0, float(raw.get("value", 0)))), str(raw.get("evidence", "")) + number = _finite(raw.get("value", 0)) + if number is None: + return 0.0, "invalid value" + return max(0.0, min(1.0, number)), str(raw.get("evidence", "")) if isinstance(raw, bool): return (1.0 if raw else 0.0), "boolean evidence" - if isinstance(raw, (int, float)): - return max(0.0, min(1.0, float(raw))), "numeric evidence" - return 0.0, "missing" + number = _finite(raw) + if number is not None: + return max(0.0, min(1.0, number)), "numeric evidence" + return 0.0, "missing" if raw is None else "invalid value" def score_quality(evidence: dict[str, Any], corpus: dict[str, Any] | None = None) -> dict: @@ -94,15 +171,32 @@ def score_quality(evidence: dict[str, Any], corpus: dict[str, Any] | None = None score = round(earned, 1) signals = dict(evidence.get("signals") or {}) hits = detect_negative(signals, corpus) + gaps = [row["id"] for row in rows if row["evidence"] in {"missing", "invalid value"} + or str(row["evidence"]).startswith("GAP:")] + if gaps: + # Missing evidence is not a failure, but it can never certify. + hits.append({"id": "evidence-gap", "severity": "REVIEW", + "message": "Evidence missing or invalid for: " + ", ".join(gaps)}) blocks = [row for row in hits if row["severity"] == "BLOCK"] reviews = [row for row in hits if row["severity"] == "REVIEW"] human_value, _ = _value(evidence, "human_aesthetic_review") - human_complete = bool(evidence.get("human_review", {}).get("completed")) and human_value > 0 + human = dict(evidence.get("human_review") or {}) + review_problem = verify_review(human.get("review"), human.get("artifact_sha256")) + human_complete = bool(human.get("completed")) and human_value > 0 and not review_problem aesthetic = dict(evidence.get("aesthetic_review") or {}) + if human.get("completed") and not review_problem: + # Certification follows the verified review itself, not a copied status field. + aesthetic = score_aesthetic_review(human["review"], format=evidence.get("format", "shorts"), + signals=signals) + elif human.get("completed"): + unverified = {"id": "human-review-unverified", "severity": "REVIEW", + "message": "Human review cannot certify: " + review_problem} + reviews.append(unverified) + hits.append(unverified) aesthetic_status = aesthetic.get("status", "REVIEW") if aesthetic_status == "BLOCKED": aesthetic_hit = {"id": "aesthetic-review-blocked", "severity": "BLOCK", - "message": "Hao aesthetic review contains a blocker or scores below 70."} + "message": "Creator aesthetic review contains a blocker or scores below 70."} blocks.append(aesthetic_hit) hits.append(aesthetic_hit) if blocks or score < 85: @@ -122,7 +216,8 @@ def score_quality(evidence: dict[str, Any], corpus: dict[str, Any] | None = None "aesthetic_review": aesthetic, "dimensions": rows, "negative_regressions": hits, - "certification_rule": "score>=95 + Hao aesthetic PASSED + human review complete + no BLOCK/REVIEW regression", + "certification_rule": ("score>=95 + server-stamped creator review of this exact render " + "PASSED + no BLOCK/REVIEW regression"), } @@ -206,7 +301,12 @@ def _short_signals( "fake_3d_claim": False, "3d_light_or_shadow_mismatch": False, } - signals.update(dict(spec.get("quality_signals") or {})) + for key, value in dict(spec.get("quality_signals") or {}).items(): + # A spec may report extra problems or withdraw a pass, never self-certify one. + if key in GOOD_SIGNALS: + signals[key] = bool(signals.get(key, False)) and value is True + else: + signals[key] = bool(signals.get(key, False)) or value is True return signals @@ -240,7 +340,8 @@ def _short_dimensions( "evidence": qa.get("caption_sheet", "missing"), }, "transition_motivation": {"value": 1.0 if transitions_ok else .45, - "evidence": "sequence cut_motivation"}, + "evidence": "sequence cut_motivation" if sequences + else "GAP: no sequence plan recorded"}, "rhythm_and_flow": {"value": 1.0 if visual_plan.get("energy_curve") else .6, "evidence": "energy curve + duration gate"}, "asset_novelty": {"value": max(0.0, 1.0 - fatigue["peak"]), @@ -251,7 +352,7 @@ def _short_dimensions( }, "tracking_and_compositing": {"value": 1.0 if compositing_ok else 0.0, "evidence": compositing_evidence}, - "human_aesthetic_review": {"value": 0.0, "evidence": "pending Hao review"}, + "human_aesthetic_review": {"value": 0.0, "evidence": "pending creator review"}, } @@ -312,10 +413,9 @@ def longform_evidence(*, content_id: str, delivery_qa: dict, "style_domain_match": True, "exact_reference_layout_copy": False, "visual_density_overload": False, - "design_dna_compiled": bool( - (visual_plan.get("design_system_v6") or {}).get("compiler") == - "hao-design-system-v6" - ), + "design_dna_compiled": str( + (visual_plan.get("design_system_v6") or {}).get("compiler") or "" + ).endswith("design-system-v6"), "subject_sheen_leaks_outside_matte": False, "subject_sheen_used_as_transition": False, "subject_reveal_not_full_black_at_start": False, @@ -346,7 +446,8 @@ def longform_evidence(*, content_id: str, delivery_qa: dict, }, "typography_and_safe_area": { "value": 1.0 if linebreaks is not False else 0.0, - "evidence": "caption linebreak gate", + "evidence": "caption linebreak gate" if linebreaks is not None + else "GAP: caption linebreak gate not run", }, "transition_motivation": { "value": .9 if not delivery_qa.get("flash_flag") else .4, @@ -354,7 +455,7 @@ def longform_evidence(*, content_id: str, delivery_qa: dict, }, "rhythm_and_flow": { "value": 1.0 if pacing.get("ok") is True else .9, - "evidence": "scene pacing + Hao review", + "evidence": "scene pacing + creator review", }, "asset_novelty": { "value": 1.0, "evidence": "no fatigue signal in delivery report"}, @@ -362,10 +463,11 @@ def longform_evidence(*, content_id: str, delivery_qa: dict, "value": .8, "evidence": "requires editorial comparison"}, "tracking_and_compositing": { "value": 1.0 if grade.get("ok", True) else .5, - "evidence": "grade/composite consistency", + "evidence": "grade/composite consistency" if grade + else "GAP: no grade report", }, "human_aesthetic_review": { - "value": 0.0, "evidence": "pending Hao review"}, + "value": 0.0, "evidence": "pending creator review"}, }, "human_review": {"completed": False}, "signature": str(content_id), @@ -390,7 +492,7 @@ def write_report(output_dir: str | Path, evidence: dict, lines += ["", "## 回歸命中"] + [ "- [%s] %s：%s" % (row["severity"], row["id"], row["message"]) for row in report["negative_regressions"]] - lines += ["", "> 機械檢查只能擋已知爛法；由 Hao 完成時間碼審片後才能取得 CERTIFIED_95，審片裝置不限。", ""] + lines += ["", "> 機械檢查只能擋已知爛法；由創作者（或其指定審片者）在審片頁完成這一版的時間碼審片後，才能取得 CERTIFIED_95，審片裝置不限。", ""] (output_dir / "QUALITY_95.md").write_text("\n".join(lines), encoding="utf-8") if remember_narrative and report["status"] != "BLOCKED" and evidence.get("signature"): record_narrative(evidence["content_id"], evidence["signature"]) @@ -407,24 +509,60 @@ def apply_human_review(evidence: dict, review: dict) -> dict: ) evidence["dimensions"]["human_aesthetic_review"] = { "value": max(0.0, min(1.0, float(aesthetic["score"]) / 100.0)), - "evidence": "Hao aesthetic standard %.1f/100 (%s)" % + "evidence": "Creator aesthetic standard %.1f/100 (%s)" % (aesthetic["score"], aesthetic["status"]), } evidence["human_review"] = { "completed": bool(aesthetic["completed"]), "average": round(float(aesthetic["score"]) / 20.0, 2), "review_id": review.get("review_id"), + "artifact_sha256": review.get("artifact_sha256"), + "review": review, } evidence["aesthetic_review"] = aesthetic return evidence +def _self_test_review_receipts(clean: dict) -> dict: + """Only a server-stamped, submitted review of the same render can certify.""" + ratings = {key: 5 for key in review_schema("shorts")["dimensions"]} + review = {"review_id": "selftest", "ratings": ratings, "issues": [], + "completed_at": "selftest-submitted"} + forged = score_quality(clean) + assert forged["status"] == "REVIEW", "self-written evidence must not certify" + assert any(hit["id"] == "human-review-unverified" for hit in forged["negative_regressions"]) + stamped = stamp_review(review, "a" * 64) + certified = apply_human_review({key: value for key, value in clean.items() + if key not in {"human_review", "aesthetic_review"}}, stamped) + assert score_quality(certified)["status"] == "CERTIFIED_95" + tampered = json.loads(json.dumps(certified)) + tampered["human_review"]["review"]["ratings"][next(iter(ratings))] = 1 + assert score_quality(tampered)["status"] != "CERTIFIED_95", "edited review must not certify" + draft = apply_human_review(clean, stamp_review(dict(review, completed_at=""), "a" * 64)) + assert score_quality(draft)["status"] != "CERTIFIED_95", "autosaved draft must not certify" + assert verify_review(stamped, "b" * 64), "a review of another render must not verify" + nan = json.loads(json.dumps(certified)) + nan["dimensions"]["technical_delivery"] = float("nan") + assert score_quality(nan)["status"] != "CERTIFIED_95" and _value(nan, "technical_delivery")[0] == 0.0 + return certified + + def self_test() -> None: dimensions = {name: 1.0 for name in load_corpus()["dimensions"]} - clean = {"dimensions": dimensions, "signals": {"text_safe_area_pass": True}, - "human_review": {"completed": True}, - "aesthetic_review": {"status": "PASSED", "score": 100}} - assert score_quality(clean)["status"] == "CERTIFIED_95" + forged = {"format": "shorts", "dimensions": dimensions, "signals": {"text_safe_area_pass": True}, + "human_review": {"completed": True}, + "aesthetic_review": {"status": "PASSED", "score": 100}} + previous_key = os.environ.get(REVIEW_KEY_ENV) + key_dir = tempfile.TemporaryDirectory(prefix="quality-95-key-") + os.environ[REVIEW_KEY_ENV] = str(Path(key_dir.name) / "review-signing.key") + try: + clean = _self_test_review_receipts(forged) + finally: + if previous_key is None: + os.environ.pop(REVIEW_KEY_ENV, None) + else: + os.environ[REVIEW_KEY_ENV] = previous_key + key_dir.cleanup() pending = json.loads(json.dumps(clean)) pending["human_review"] = {"completed": False} pending["dimensions"]["human_aesthetic_review"] = 0 diff --git a/src/review_loop.py b/src/review_loop.py index 4910580..10d916a 100644 --- a/src/review_loop.py +++ b/src/review_loop.py @@ -42,7 +42,7 @@ from knowledge_lifecycle import record_feedback from aesthetic_score import review_schema -from quality_95 import apply_human_review, write_report +from quality_95 import apply_human_review, stamp_review, verify_review, write_report VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".webm", ".mkv"} @@ -319,8 +319,12 @@ def create_bundle(video: str | Path, content_id: str, can_finalize=can_finalize) (bundle / "review.html").write_text(page, encoding="utf-8") first_video = next((item["path"] for item in media if item["kind"] == "video"), "") + video_sha256 = _file_sha256(Path(first_video)) if can_finalize else "" + if can_finalize: + _archive_stale_review(bundle, video_sha256) manifest = {"schema_version": 2, "content_id": content_id, "source": str(source), "media": media, "media_count": len(media), "video": first_video, + "video_sha256": video_sha256, "quality_json": str(Path(quality_json).resolve()) if quality_json else "", "created_at": datetime.now(timezone.utc).isoformat(timespec="seconds")} _atomic_json(bundle / "manifest.json", manifest) @@ -329,33 +333,93 @@ def create_bundle(video: str | Path, content_id: str, "serve_command": "python review_loop.py serve %s" % json.dumps(str(bundle))} +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _archive_stale_review(bundle: Path, video_sha256: str) -> None: + """A rebuilt video starts a fresh review; an older review is kept, never reused.""" + review = bundle / "review.json" + if not review.is_file(): + return + try: + reviewed = json.loads(review.read_text(encoding="utf-8")).get("artifact_sha256") + except (OSError, ValueError, AttributeError): + reviewed = None + if reviewed and reviewed == video_sha256: + return + os.replace(review, bundle / ("review.stale-%s-%d.json" % ((reviewed or "unbound")[:12], time.time()))) + + +def _stamp_review_payload(payload: dict[str, Any], artifact_sha256: str) -> dict[str, Any]: + """Stamp a saved review server-side; a draft still saves if the key is unavailable.""" + try: + return stamp_review(payload, artifact_sha256) + except (RuntimeError, OSError): + unsigned = {key: value for key, value in payload.items() + if key not in {"receipt", "artifact_sha256"}} + return dict(unsigned, receipt_error="review signing key unavailable") + + +def _unique_issues(issues: Any) -> list[dict[str, Any]]: + seen, unique = set(), [] + for issue in issues if isinstance(issues, list) else []: + if not isinstance(issue, dict): + continue + key = (str(issue.get("category", "review")), " ".join(str(issue.get("comment", "")).split())) + if key not in seen: + seen.add(key) + unique.append(issue) + return unique + + def finalize(bundle_dir: str | Path, *, learn: bool = True) -> dict[str, Any]: bundle = Path(bundle_dir).resolve() manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8")) review = json.loads((bundle / "review.json").read_text(encoding="utf-8")) if not manifest.get("quality_json") or not manifest.get("video"): raise ValueError("finalize is only for one QA-linked delivery video; material reviews stay in review.json") + current = _file_sha256(Path(manifest["video"])) + if current != manifest.get("video_sha256"): + raise ValueError("影片在建立審片包後已改變：請重新 create 審片包並重新審片，舊評分不得轉移到新版本") + problem = verify_review(review, current) + if problem: + raise ValueError("無法以此審片認證：" + problem) quality_path = Path(manifest["quality_json"]) quality = json.loads(quality_path.read_text(encoding="utf-8")) evidence = quality.get("evidence") or {} - for issue in review.get("issues", []): + previous = _read_json_or_empty(bundle / "finalized.json") + learned = (previous.get("review") or {}).get("receipt") == review.get("receipt") + source = "%s@%s" % (manifest["content_id"], current[:12]) + for issue in _unique_issues(review.get("issues", [])): mapping = CATEGORY_SIGNALS.get(issue.get("category")) if mapping: evidence.setdefault("signals", {})[mapping[0]] = mapping[1] - if learn: + if learn and not learned: record_feedback( rule="%s：%s" % (issue.get("category", "review"), issue.get("comment", "")), - evidence="%s @ %.2fs / %s" % (manifest["content_id"], float(issue.get("time", 0)), - manifest["video"]), + evidence="%s @ %.2fs" % (source, float(issue.get("time", 0) or 0)), scope="project", formats=[evidence.get("format", "*")], - domains=["*"], kind="soft", + domains=["*"], kind="soft", source=source, ) - review["review_id"] = review.get("review_id") or (manifest["content_id"] + "-mobile") evidence = apply_human_review(evidence, review) + evidence["human_review"]["review_id"] = review.get("review_id") or (manifest["content_id"] + "-mobile") result = write_report(quality_path.parent, evidence, remember_narrative=True) _atomic_json(bundle / "finalized.json", {"review": review, "quality": result}) return result + +def _read_json_or_empty(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + return value if isinstance(value, dict) else {} + def _parse_byte_range(value: str, size: int) -> tuple[int, int] | None: if not value: return None @@ -645,6 +709,7 @@ class _MediaReviewHandler(BaseHTTPRequestHandler): media: list[dict[str, str]] auth: _RemoteReviewAuth | None can_finalize: bool + artifact_sha256: str = "" protocol_version = "HTTP/1.1" def _send_bytes(self, payload: bytes, content_type: str, *, status: int = 200, @@ -856,7 +921,8 @@ def do_POST(self) -> None: # noqa: N802 if not _review_payload_valid(payload): self._fail(400, "invalid_review_payload") return - _atomic_json(self.bundle / "review.json", payload) + _atomic_json(self.bundle / "review.json", + _stamp_review_payload(payload, self.artifact_sha256)) self._send_json({"ok": True}) def log_message(self, _format: str, *_args: object) -> None: @@ -865,11 +931,12 @@ def log_message(self, _format: str, *_args: object) -> None: def _handler_factory(bundle: Path, media: list[dict[str, str]], auth: _RemoteReviewAuth | None, - can_finalize: bool = False): + can_finalize: bool = False, artifact_sha256: str = ""): class Bound(_MediaReviewHandler): pass Bound.bundle, Bound.media, Bound.auth = bundle, media, auth Bound.can_finalize = can_finalize + Bound.artifact_sha256 = artifact_sha256 return Bound @@ -1442,7 +1509,8 @@ def serve_remote(bundle_dir: str | Path, port: int = 0, cloudflared: str = "", idle_timeout_seconds=idle_timeout_seconds, ) server = ThreadingHTTPServer(("127.0.0.1", port), - _handler_factory(bundle, media, auth, can_finalize)) + _handler_factory(bundle, media, auth, can_finalize, + str(manifest.get("video_sha256") or ""))) server.daemon_threads = True local_port = int(server.server_address[1]) server_thread = threading.Thread(target=server.serve_forever, daemon=True) @@ -1551,7 +1619,8 @@ def serve(bundle_dir: str | Path, port: int = 8765) -> None: media[0]["kind"] == "video") print("僅限本機開啟：http://127.0.0.1:%d/review.html" % port) ThreadingHTTPServer(("127.0.0.1", port), - _handler_factory(bundle, media, None, can_finalize)).serve_forever() + _handler_factory(bundle, media, None, can_finalize, + str(manifest.get("video_sha256") or ""))).serve_forever() def _self_test_bundle(temp: str) -> tuple[dict[str, Any], list[dict[str, Any]], Path]: @@ -1992,17 +2061,78 @@ def _self_test_secret_surfaces(temp: str, image: Path, media: list[dict[str, Any finally: globals()["start_remote_background"] = original_start +def _expect_refused(bundle: Path, fragment: str) -> None: + try: + finalize(bundle) + except ValueError as exc: + assert fragment in str(exc), exc + else: + raise AssertionError("finalize accepted an unverifiable review: " + fragment) + + +def _self_test_review_binding(temp: str) -> None: + """Certification needs a submitted, server-stamped review of these exact bytes.""" + from quality_95 import REVIEW_KEY_ENV, load_corpus + from knowledge_lifecycle import LEDGER_ENV, load_state + + root = Path(temp) / "binding" + root.mkdir() + video, quality = root / "current.mp4", root / "QUALITY_95.json" + video.write_bytes(b"render-one") + quality.write_text(json.dumps({"evidence": { + "content_id": "bind", "format": "shorts", "signals": {"text_safe_area_pass": True}, + "dimensions": {name: 1.0 for name in load_corpus()["dimensions"]}}}), encoding="utf-8") + saved = {name: os.environ.get(name) for name in (REVIEW_KEY_ENV, LEDGER_ENV)} + os.environ[REVIEW_KEY_ENV] = str(Path(temp) / "review-signing.key") + os.environ[LEDGER_ENV] = str(Path(temp) / "ledger.json") + try: + bundle = Path(create_bundle(video, "bind", quality)["bundle"]) + artifact = json.loads((bundle / "manifest.json").read_text(encoding="utf-8"))["video_sha256"] + ratings = {key: 5 for key in review_schema("shorts")["dimensions"]} + issue = {"category": "note", "comment": "節奏太慢", "time": 1.0} + review = {"ratings": ratings, "issues": [issue, dict(issue, time=2.0), dict(issue, time=3.0)]} + _atomic_json(bundle / "review.json", dict(review, completed_at="now")) + _expect_refused(bundle, "not stamped") + _atomic_json(bundle / "review.json", _stamp_review_payload(review, artifact)) + _expect_refused(bundle, "not submitted") + _atomic_json(bundle / "review.json", _stamp_review_payload(dict(review, completed_at="now"), artifact)) + assert finalize(bundle)["human_review_complete"] is True + finalize(bundle) + rules = [row for row in load_state(os.environ[LEDGER_ENV])["records"] if "節奏太慢" in row["rule"]] + assert len(rules) == 1 and rules[0]["support"] == 1 and not rules[0]["pinned"], rules + video.write_bytes(b"render-two") + _expect_refused(bundle, "重新 create") + create_bundle(video, "bind", quality) + assert not (bundle / "review.json").exists() and list(bundle.glob("review.stale-*.json")) + finally: + for name, value in saved.items(): + if value is None: + os.environ.pop(name, None) + else: + os.environ[name] = value + + def self_test() -> None: assert '.viewer{display:block' in HTML_TEMPLATE assert 'placeholder="未評"' in HTML_TEMPLATE assert "data.ratings[k]||5" not in HTML_TEMPLATE assert "input.checkValidity()" in HTML_TEMPLATE + from quality_95 import REVIEW_KEY_ENV + + previous_key = os.environ.get(REVIEW_KEY_ENV) with tempfile.TemporaryDirectory(prefix="review-loop-") as temp: + # Stamping during the HTTP tests must never create a key in the real home. + os.environ[REVIEW_KEY_ENV] = str(Path(temp) / "http-review-signing.key") bundle, media, image = _self_test_bundle(temp) _self_test_http(bundle, media) _self_test_pairing_ui() identity, forbidden_url_json = _self_test_remote_lifecycle(temp, bundle, media) _self_test_secret_surfaces(temp, image, media, identity, forbidden_url_json) + _self_test_review_binding(temp) + if previous_key is None: + os.environ.pop(REVIEW_KEY_ENV, None) + else: + os.environ[REVIEW_KEY_ENV] = previous_key print("review_loop self-test GREEN") diff --git a/src/silent_vlog_maker/asset_scanner.py b/src/silent_vlog_maker/asset_scanner.py index bc3e77a..1dc3f04 100644 --- a/src/silent_vlog_maker/asset_scanner.py +++ b/src/silent_vlog_maker/asset_scanner.py @@ -13,6 +13,8 @@ scan_all_assets 在 v0.3 模式只補缺不清洗人工語意層。 self-test：python -m silent_vlog_maker.asset_scanner """ +from __future__ import annotations + import json import subprocess import sys diff --git a/src/thumbnail_algorithm_score.py b/src/thumbnail_algorithm_score.py index 10aa234..b67e950 100644 --- a/src/thumbnail_algorithm_score.py +++ b/src/thumbnail_algorithm_score.py @@ -9,6 +9,7 @@ import argparse import json +import math import os import tempfile from datetime import datetime, timezone @@ -89,10 +90,10 @@ def _value(raw: Any) -> tuple[float, str]: proof = str(raw.get("evidence", "")) else: value, proof = raw, "numeric evidence" - try: - return max(0.0, min(1.0, float(value))), proof - except (TypeError, ValueError): - return 0.0, "missing" + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + # NaN/inf/text never clamp to a full score; absent and malformed are both unscored. + return 0.0, "missing" if value is None else "invalid value" + return max(0.0, min(1.0, float(value))), proof def _signal_hits(signals: dict[str, Any], standard: dict[str, Any]) -> list[dict[str, Any]]: @@ -104,6 +105,22 @@ def _signal_hits(signals: dict[str, Any], standard: dict[str, Any]) -> list[dict return hits +FLAT_LUMA_STDDEV = 12.0 + + +def _image_metric_hits(metrics: dict[str, Any]) -> list[dict[str, Any]]: + """Objective pixel facts that a self-rated evidence file cannot override.""" + hits = [] + if metrics.get("resolution_pass") is False: + hits.append({"id": "low-resolution", "severity": "REVIEW", + "message": "封面解析度低於 1280×720，上傳後會被放大而發糊。"}) + stddev = metrics.get("luma_contrast_stddev") + if isinstance(stddev, (int, float)) and not isinstance(stddev, bool) and stddev < FLAT_LUMA_STDDEV: + hits.append({"id": "flat-low-contrast", "severity": "REVIEW", + "message": "封面明暗幾乎沒有層次，手機縮圖下主體與文字難以分離。"}) + return hits + + def score_thumbnail(evidence: dict[str, Any], standard: dict[str, Any] | None = None) -> dict[str, Any]: standard = standard or load_standard() @@ -135,7 +152,7 @@ def score_thumbnail(evidence: dict[str, Any], image_metrics = dict(evidence.get("image_metrics") or {}) if "aspect_16_9" in image_metrics: signals["aspect_16_9"] = bool(image_metrics["aspect_16_9"]) - hits = _signal_hits(signals, standard) + hits = _signal_hits(signals, standard) + _image_metric_hits(image_metrics) blocks = [row for row in hits if row["severity"] == "BLOCK"] reviews = [row for row in hits if row["severity"] == "REVIEW"] completed = not missing @@ -223,6 +240,15 @@ def self_test() -> None: revise = json.loads(json.dumps(clean)) revise["dimensions"]["promise_payoff_alignment"] = 0.5 assert score_thumbnail(revise)["status"] == "REVISE" + nan = {"dimensions": {name: float("nan") for name in standard["dimensions"]}, + "signals": {"aspect_16_9": True}} + assert score_thumbnail(nan)["status"] != "PACKAGING_READY", "NaN must never score as 1.0" + flat = json.loads(json.dumps(clean)) + flat["image_metrics"] = {"aspect_16_9": True, "resolution_pass": False, + "luma_contrast_stddev": 0.0} + flat_report = score_thumbnail(flat) + assert flat_report["status"] == "REVISE" + assert {"low-resolution", "flat-low-contrast"} <= {hit["id"] for hit in flat_report["negative_signals"]} print("thumbnail algorithm score self-test GREEN") From c3ae06678a26392a9efd204824f090862e43615b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 06:48:32 +0000 Subject: [PATCH 03/26] feat: route every reference from the public skill and fix sanitizer fixtures - public SKILL.md: bilingual trigger description, start-of-task memory and context-packet steps, a task router reaching all 56 references (large files by section), definition of done, memory rules and an opt-in remote review - public agent card rendered from a fixed template (no automatic remote link) - skeleton validator rejects unlisted, anchored or ./references links - metric sanitizer keeps fixtures short ('12K') so length-checked self-tests and the caption showcase run again; the marker moves to a trailing comment - canonical_backport treats sanitizer/transform-only changes as already applied by projecting the canonical through the head pipeline Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XLikBpKWqhMahFDuvVj6wh --- codex-skill/video-autopilot/SKILL.md | 267 +++++++------ .../video-autopilot/agents/openai.yaml | 4 +- scripts/canonical_backport.py | 27 +- scripts/canonical_backport_selftest.py | 9 +- scripts/public_privacy_profiles.py | 12 +- scripts/public_skill_skeleton.py | 357 +++++++++++------- scripts/sync_canonical.py | 13 +- src/longform_maker/emphasis_overlays.py | 4 +- src/render_caption_showcase.py | 4 +- src/visual_director.py | 4 +- 10 files changed, 439 insertions(+), 262 deletions(-) diff --git a/codex-skill/video-autopilot/SKILL.md b/codex-skill/video-autopilot/SKILL.md index cdc3f2b..93470bc 100644 --- a/codex-skill/video-autopilot/SKILL.md +++ b/codex-skill/video-autopilot/SKILL.md @@ -1,130 +1,171 @@ --- name: video-autopilot -description: Plan, build, review, package, release, and improve long-form videos and vertical shorts with evidence-gated, creator-configurable workflows. +description: Plan, edit, caption, color-grade, review, package and publish YouTube long-form videos, Shorts/Reels and illustrated music videos with evidence-gated Editkin v4 workflows, creator memory and QA. Use for 剪片/剪輯、上字幕、調色、配樂/音效、做Shorts/Reels、做封面/縮圖、審片/品質檢查、發布文案、從成效學習. --- # Hao Video Autopilot Turn a content brief and creator-owned source material into reviewable video -artifacts and publishing packages. The workflow supports long-form video, -YouTube Shorts, Instagram Reels, illustrated music videos, and reusable image -or motion assets. It does not ship a maintainer profile: voice, face policy, -captions, palette, outro, -channel links, performance baselines, and aesthetic choices come from the -active creator configuration or an explicit brief. - -## Modes - -- **Plan** — clarify the audience promise, evidence, format, platform, - constraints, and packaging hypotheses; produce a script and edit plan. -- **Build** — inspect authoritative source material, bind every decision to - evidence, apply an audited edit plan atomically, render, and run delivery QA. -- **Log Outcome** — record human review and comparable platform outcomes in - the creator's protected local state; never treat missing values as zero. -- **Optimize Patterns** — propose reversible changes from repeated comparable - evidence. A single preference or result cannot become a universal default. +artifacts and publishing packages: YouTube long-form, Shorts, Instagram Reels, +illustrated music videos and reusable image or motion assets. Nothing creative +is hard-coded for a maintainer: voice, face policy, captions, palette, outro, +channel links, performance baselines and taste come from the active creator's +profiles, learned memory or an explicit brief. + +## Start every task + +1. **Project root.** Find it from `AUTOPILOT_MANIFEST.json` or + `release-manifest.json` (or `VIDEO_AUTOPILOT_ROOT`) and keep every output + inside it; never invent sibling project folders. Commands below run from that + root; `references/` sits beside this file. +2. **Creator memory first.** Read `profiles/voice.md`, `profiles/brand.md` and + `profiles/style.md` (plus `profiles/show.md` for interviews) when they exist. + If one is missing, say so and stay neutral; never invent a creator's voice, + face policy, palette, outro or numbers. +3. **Context packet.** Run + `python src/context_router.py route --mode --format --topic ""` + and work from that packet: it carries the learned rules for this format and + topic, the creator-profile status and a short list of escalation sources. + Read one more source only when a listed trigger fires. +4. **Route.** Open only the references in the matching row below. For large + references read only the named section (`grep -n "^## "` lists them). +5. **Evidence before plans.** Inspect the actual footage, audio and transcript + before planning. Claims about products, places, prices, results, licenses or + identities need verifiable evidence; otherwise mark `ASSET_REQUIRED` and plan + the capture. + +## Route by task + +| Task (typical requests) | Read | Run | +|---|---|---| +| Any edit, 剪片／剪輯 | [Editorial intelligence contract](references/editorial-intelligence-contract.md); [Workflow execution](references/editkin-workflow-execution.md) | `python scripts/hao_autopilot.py workflow create`, then `next`, `claim`, `complete`, `verify` | +| Batch runs, plugins, phone pairing | [Batch workflow](references/editkin-batch-workflow.md); [Plugin automation](references/editkin-plugin-automation.md); [Mobile device binding](references/editkin-mobile-device-binding.md) | Editkin MCP tools | +| Script, hook, 開場, 留存 | [Script and retention calibration](references/script-retention-2026.md); `knowledge/script-retention-craft.md` | `src/longform_maker/script_gate.py` | +| Shot choice, 選鏡 | [Shot selection language](references/shot-selection-language.md) | `python src/visual_director.py --shot-style