#!/usr/bin/env python3 from __future__ import annotations import json import errno import os import re import signal import shutil import subprocess import threading import time import uuid import urllib.request from pathlib import Path from typing import Any HOT_DISK_MOUNT = Path("/Volumes/Hot Disk") ROOT = HOT_DISK_MOUNT / "The Object" / "produktion" CAPTURES = ROOT / "captures" INGEST_JOBS = ROOT / "jobs" KAMERA_STATE = Path("/Users/victorholland/Vibe Coding/dispatcher/kameramotor/state") KAMERA_DONE = Path("/Users/victorholland/Vibe Coding/dispatcher/kameramotor/done") KAMERA_FAILED = Path("/Users/victorholland/Vibe Coding/dispatcher/kameramotor/failed") KAMERA_STATE_MAGNIFIC = Path("/Users/victorholland/Vibe Coding/dispatcher/kameramotor/state_magnific") KAMERA_DONE_MAGNIFIC = Path("/Users/victorholland/Vibe Coding/dispatcher/kameramotor/done_magnific") KAMERA_FAILED_MAGNIFIC = Path("/Users/victorholland/Vibe Coding/dispatcher/kameramotor/failed_magnific") MAGNIFIC_VIDEO_JOBS = Path("/Users/victorholland/Vibe Coding/dispatcher/kameramotor/jobs_magnific") KLING_JOBS = Path("/Users/victorholland/Vibe Coding/dispatcher/kling-kameramotor/jobs") KLING_STATE = Path("/Users/victorholland/Vibe Coding/dispatcher/kling-kameramotor/state") KLING_DONE = Path("/Users/victorholland/Vibe Coding/dispatcher/kling-kameramotor/done") KLING_FAILED = Path("/Users/victorholland/Vibe Coding/dispatcher/kling-kameramotor/failed") KLING_PAUSE_FLAGS = [ Path("/Users/victorholland/Vibe Coding/dispatcher/kling-kameramotor/.kameramotor_kling_paused"), Path.home() / ".kameramotor_paused", ] PROMPT_DIR = HOT_DISK_MOUNT / "The Object" / "produktion" / "prompts" / "prompts_motor_safe" CANONICAL_PROMPT_ROOT = HOT_DISK_MOUNT / "The Object" / "produktion" / "prompts" RUNTIME_PROMPT_CACHE = Path("/Users/victorholland/Vibe Coding/The Object/runtime_prompt_cache_20260610") POLL_SECONDS = 12 AUDIO_SOFTENING_VERSION = 2 PIPELINE_VERSION = "deterministic_6_images_8_films_v1" VIDEO_PROVIDER_VERSION = "magnific_seedance15pro_only_20260611" VIDEO_PROVIDER_CONFIGS = { "S": { "badge": "S", "provider_name": "Seedance 1.5 Pro", "provider": "magnific", "route": "magnific_cdp_unlimited", "api": "bytedance", "model": "seedance", "mode": "pro-1.5", "slug": "bytedance-seedance-pro-1.5", "resolution": "Draft", "duration": 5, "withSoundEffects": True, "requires_end_image": True, "output_dir": "seedance_video_outputs", "output_suffix": "seedance15pro_draft", }, # K (Kling 2.5) removed per Victor-Grundsatzentscheidung 2026-06-11. # Existing manifests with K-jobs are harmless — video_variant() falls back # to "S" for any variant not in this dict (see provider_config fallback below). } VIDEO_PROVIDER_CONFIG = VIDEO_PROVIDER_CONFIGS["S"] FOLLOWUP_LOG = ROOT / "diagnostics" / "autopilot_followup_20260610.jsonl" FILE_RETRY_ATTEMPTS = 18 FILE_RETRY_BASE_SECONDS = 0.08 FILE_RETRY_MAX_SECONDS = 0.5 FOLLOWUP_IDS = { "13C10CEF-7C3C-4C8C-95AE-5DF0C7285ABF", "31F22DE8-5083-436A-A21A-6B817A149EBD", "E9698C8B-C6DA-4A22-A7FA-6768FDFDFF5B", "58015F6D-03FB-4EFE-BEE3-73D1FB5C7518", "4F87C3CA-0EAF-4424-A5AC-0FF6477A83BC", "0CA368B1-1891-4ED5-AAE6-F2E07FE1732A", "7EFFAB8C-D17A-4020-9256-7479714AE303", "66630B04-8B75-4AAA-BAE4-B0C425E49B3E", } PROMPTS = { "a_to_b": "1_a_to_b_reality_to_hero_motor.txt", "real_explosion": "2_b_to_z_real_explosion_motor.txt", "unreal_explosion": "3_b_to_z_unreal_explosion_motor.txt", "hero_360": "4a_hero_360_clockwork_asmr_motor.txt", "real_360": "4b_exploded_real_360_clockwork_asmr_motor.txt", "unreal_360": "4c_exploded_unreal_360_clockwork_asmr_motor.txt", } IMAGE_PROMPTS = { "C": "C_exploded_real.txt", "D": "D_exploded_secret.txt", "E": "E_neatify.txt", "F": "F_secret_detail.txt", } VIDEO_PROMPTS = { "G": "G_original_to_hero.txt", "H": "H_hero_360.txt", "I": "I_hero_to_exploded_real.txt", "J": "J_exploded_real_360.txt", "K": "K_exploded_real_to_neatify.txt", "L": "L_hero_to_exploded_secret.txt", "M": "M_exploded_secret_360.txt", "N": "N_exploded_secret_to_secret_detail.txt", } IMAGE_SPECS = [ {"slot": "C", "label": "exploded_real", "source": "B", "filename": "C_exploded_real.png"}, {"slot": "D", "label": "exploded_secret", "source": "B", "filename": "D_exploded_secret.png"}, {"slot": "E", "label": "neatify", "source": "C", "filename": "E_neatify.png"}, {"slot": "F", "label": "secret_detail", "source": "D", "filename": "F_secret_detail.png"}, ] VIDEO_SPECS = [ {"slot": "G", "order": 1, "label": "original_to_hero", "filter": "film_01_original_to_hero", "start": "A", "end": "B", "filename": "01_G_original_to_hero.mp4"}, {"slot": "H", "order": 2, "label": "hero_360", "filter": "film_02_hero_orbit_360", "start": "B", "end": "B", "filename": "02_H_hero_360.mp4"}, {"slot": "I", "order": 3, "label": "hero_to_exploded_real", "filter": "film_03_hero_to_exploded_real", "start": "B", "end": "C", "filename": "03_I_hero_to_exploded_real.mp4"}, {"slot": "J", "order": 4, "label": "exploded_real_360", "filter": "film_04_exploded_real_orbit_360", "start": "C", "end": "C", "filename": "04_J_exploded_real_360.mp4"}, {"slot": "K", "order": 5, "label": "exploded_real_to_neatify", "filter": "film_05_exploded_real_to_neatify", "start": "C", "end": "E", "filename": "05_K_exploded_real_to_neatify.mp4"}, {"slot": "L", "order": 6, "label": "hero_to_exploded_secret", "filter": "film_06_hero_to_exploded_secret", "start": "B", "end": "D", "filename": "06_L_hero_to_exploded_secret.mp4"}, {"slot": "M", "order": 7, "label": "exploded_secret_360", "filter": "film_07_exploded_secret_orbit_360", "start": "D", "end": "D", "filename": "07_M_exploded_secret_360.mp4"}, {"slot": "N", "order": 8, "label": "exploded_secret_to_secret_detail", "filter": "film_08_exploded_secret_to_secret_detail", "start": "D", "end": "F", "filename": "08_N_exploded_secret_to_secret_detail.mp4"}, ] NEGATIVE = ( "melting, morphing, random anatomy, gore, blood, sexualized body, duplicate walls, " "impossible holes in background, generic treasure, diamond, ring, heart, tiny library, " "visible creature, face, person, animal, unrelated object, camera drift, zoom, pan, tilt, " "background change, harsh sound, music, melody, beat, voice, drone, silence" ) def _is_retryable_file_error(exc: OSError) -> bool: return exc.errno in {errno.EDEADLK, errno.EAGAIN, errno.EBUSY, errno.EPERM, 11} def with_file_retry(label: str, fn, *, attempts: int = FILE_RETRY_ATTEMPTS): last_exc = None for attempt in range(1, attempts + 1): try: return fn() except OSError as exc: last_exc = exc if not _is_retryable_file_error(exc) or attempt == attempts: raise delay = min(FILE_RETRY_BASE_SECONDS * attempt, FILE_RETRY_MAX_SECONDS) time.sleep(delay) if last_exc is not None: raise OSError(last_exc.errno, f"{label}: {last_exc}") # pragma: no cover raise RuntimeError(f"{label}: unknown file retry failure") # pragma: no cover def safe_read_text(path: Path) -> str: label = f"read:{path}" try: return with_file_retry(label, lambda: path.read_text(encoding="utf-8")) except OSError as exc: if not _is_retryable_file_error(exc): raise def _cat() -> str: proc = subprocess.run( ["/bin/cat", str(path)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) return proc.stdout.decode("utf-8") return with_file_retry(f"cat:{path}", _cat, attempts=6) def safe_write_text(path: Path, text: str) -> None: with_file_retry(f"write:{path}", lambda: path.write_text(text, encoding="utf-8")) def safe_copy2(src: str | Path, dst: str | Path) -> None: src_path = Path(src) dst_path = Path(dst) dst_path.parent.mkdir(parents=True, exist_ok=True) def _copy() -> None: shutil.copy2(src_path, dst_path) try: with_file_retry(f"copy:{src_path}->{dst_path}", _copy) except Exception as exc: if isinstance(exc, OSError) and not _is_retryable_file_error(exc): raise tmp = dst_path.with_name(f".{dst_path.name}.{os.getpid()}.{threading.get_ident()}.{uuid.uuid4().hex}.copytmp") with_file_retry( f"copyfile:{src_path}->{tmp}", lambda: shutil.copyfile(src_path, tmp), ) with_file_retry(f"replace:{tmp}->{dst_path}", lambda: os.replace(tmp, dst_path)) def log_followup(capture_id: str, stage: str, **extra: Any) -> None: FOLLOWUP_LOG.parent.mkdir(parents=True, exist_ok=True) event = {"ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "captureId": capture_id, "stage": stage, **extra} line = json.dumps(event, ensure_ascii=False) + "\n" def _append() -> None: with FOLLOWUP_LOG.open("a", encoding="utf-8") as handle: handle.write(line) with_file_retry(f"append:{FOLLOWUP_LOG}", _append) def read_json(path: Path) -> dict[str, Any] | None: if not path.exists(): return None try: return json.loads(safe_read_text(path)) except OSError as exc: if _is_retryable_file_error(exc): raise return None except Exception: return None def kling_pause_status() -> dict[str, Any] | None: for flag in KLING_PAUSE_FLAGS: if not flag.exists(): continue try: text = safe_read_text(flag) except Exception as exc: text = f"pause flag exists, read failed: {exc}" reason = "Kling paused" timestamp = None for line in text.splitlines(): if line.startswith("reason="): reason = line.removeprefix("reason=").strip() or reason elif line.startswith("ts="): timestamp = line.removeprefix("ts=").strip() or None return { "status": "needs_victor", "blocked_by": "kling_paused", "flag": str(flag), "reason": reason, "paused_at": timestamp, "note": "Only Victor may remove the pause flag.", } return None def write_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) # Clamp tmp-name to 250 chars to avoid ENAMETOOLONG on deep/long filenames (Fix 2026-06-12) _suffix = f".{os.getpid()}.{threading.get_ident()}.{uuid.uuid4().hex}.tmp" _prefix = f".{path.name}" _max_prefix = 250 - len(_suffix) _tmp_name = (_prefix[:_max_prefix] if len(_prefix) > _max_prefix else _prefix) + _suffix tmp = path.with_name(_tmp_name) safe_write_text(tmp, json.dumps(payload, indent=2, ensure_ascii=False)) try: with_file_retry(f"replace:{tmp}->{path}", lambda: os.replace(tmp, path)) finally: if tmp.exists(): try: tmp.unlink() except OSError: pass def one_line(text: str) -> str: return re.sub(r"\s+", " ", text).strip() def safe_slug(value: str) -> str: value = value.lower() value = re.sub(r"[^a-z0-9_-]+", "_", value) return value.strip("_")[:64] or f"live_{int(time.time())}" def job_id(slug: str, filter_name: str) -> str: return f"{safe_slug(slug)}_{safe_slug(filter_name)}" def post_json(url: str, payload: dict[str, Any]) -> dict[str, Any]: req = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), method="POST", headers={"content-type": "application/json"}, ) with urllib.request.urlopen(req, timeout=25) as response: return json.loads(response.read().decode("utf-8")) def duration(path: Path) -> float: out = subprocess.run( ["/opt/homebrew/bin/ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nk=1:nw=1", str(path)], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True, ).stdout.strip() return float(out) def has_audio(path: Path) -> bool: return subprocess.run( [ "/opt/homebrew/bin/ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "stream=index", "-of", "csv=p=0", str(path), ], text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False, ).stdout.strip() != "" def audio_softening_filter(path: Path) -> str: dur = max(duration(path), 0.4) fade_in = min(0.32, max(dur * 0.08, 0.18)) fade_out = min(0.42, max(dur * 0.10, 0.24)) fade_out_start = max(dur - fade_out, 0.02) return ( "loudnorm=I=-20:TP=-2:LRA=11," f"afade=t=in:st=0:d={fade_in:.3f}," f"afade=t=out:st={fade_out_start:.3f}:d={fade_out:.3f}" ) def frame_count(path: Path) -> int: out = subprocess.run( [ "/opt/homebrew/bin/ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=nb_frames", "-of", "default=nk=1:nw=1", str(path), ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ).stdout.strip() try: return int(out) except ValueError: return 0 def freezes(path: Path) -> list[tuple[float, float, float]]: proc = subprocess.run( ["/opt/homebrew/bin/ffmpeg", "-hide_banner", "-nostats", "-i", str(path), "-vf", "freezedetect=n=-45dB:d=0.18", "-an", "-f", "null", "-"], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) found: list[tuple[float, float, float]] = [] start = None freeze_duration = None for line in proc.stderr.splitlines(): if "freeze_start:" in line: start = float(line.split("freeze_start:")[-1].strip()) elif "freeze_duration:" in line: match = re.search(r"freeze_duration:\s*([0-9.]+)", line) freeze_duration = float(match.group(1)) if match else None elif "freeze_end:" in line: match = re.search(r"freeze_end:\s*([0-9.]+)", line) end = float(match.group(1)) if match else None if start is not None and end is not None: found.append((start, end, freeze_duration or (end - start))) start = None freeze_duration = None return found def trim_points(path: Path) -> tuple[float, float, float]: dur = duration(path) start_trim = 0.0 end_trim = dur for start, end, _ in freezes(path): if start <= 0.18 and end > start_trim: start_trim = end if dur - end <= 0.25 and start < end_trim: end_trim = start elif dur - start <= 0.85 and start < end_trim: end_trim = start if end_trim - start_trim < 2.4: return 0.0, dur, dur return start_trim, end_trim, dur def trim_video_in_place(path: Path) -> dict[str, Any]: marker = path.with_suffix(path.suffix + ".freezetrim.json") if marker.exists(): return read_json(marker) or {"already_trimmed": True} start, end, dur = trim_points(path) result = {"duration": dur, "trim_start": start, "trim_end_removed": dur - end, "final_duration": end - start} if start <= 0.001 and dur - end <= 0.001: write_json(marker, result) return result backup = path.with_name(path.stem + "__pre_freezetrim" + path.suffix) if not backup.exists(): shutil.copy2(path, backup) tmp = path.with_suffix(".trimtmp.mp4") subprocess.run( [ "/opt/homebrew/bin/ffmpeg", "-y", "-ss", f"{start:.6f}", "-to", f"{end:.6f}", "-i", str(path), "-map", "0:v:0", "-map", "0:a?", "-vf", "fps=24,scale=trunc(iw/2)*2:trunc(ih/2)*2", "-c:v", "libx264", "-preset", "veryfast", "-crf", "18", "-g", "1", "-keyint_min", "1", "-sc_threshold", "0", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "160k", "-movflags", "+faststart", str(tmp), ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, ) tmp.replace(path) write_json(marker, result) normalize_video_for_app(path) return result def normalize_video_for_app(path: Path) -> dict[str, Any]: marker = path.with_suffix(path.suffix + ".appvideo.json") if marker.exists(): result = read_json(marker) or {"already_normalized": True} if result.get("audio_softening_version") == AUDIO_SOFTENING_VERSION and result.get("source_mtime", 0) >= path.stat().st_mtime - 0.001: ensure_reverse_video_for_app(path) return result backup = path.with_name(path.stem + "__pre_appvideo" + path.suffix) if not backup.exists(): shutil.copy2(path, backup) tmp = path.with_suffix(".appvideo.tmp.mp4") command = [ "/opt/homebrew/bin/ffmpeg", "-y", "-i", str(path), "-map", "0:v:0", "-map", "0:a?", "-vf", "fps=24,scale=trunc(iw/2)*2:trunc(ih/2)*2", ] audio_present = has_audio(path) if audio_present: command += ["-af", audio_softening_filter(path)] command += [ "-c:v", "libx264", "-preset", "veryfast", "-crf", "18", "-g", "1", "-keyint_min", "1", "-sc_threshold", "0", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "160k", "-movflags", "+faststart", str(tmp), ] subprocess.run( command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, ) tmp.replace(path) result = { "normalized": True, "codec": "h264", "fps": 24, "all_keyframes": True, "faststart": True, "audio_softening_version": AUDIO_SOFTENING_VERSION, "audio_loudnorm": audio_present, "audio_fades": audio_present, "source_mtime": path.stat().st_mtime, } write_json(marker, result) ensure_reverse_video_for_app(path) return result def reverse_video_path(path: Path) -> Path: return path.with_name(path.stem + "__reverse" + path.suffix) def ensure_reverse_video_for_app(path: Path) -> Path: reverse = reverse_video_path(path) marker = reverse.with_suffix(reverse.suffix + ".appvideo.json") if reverse.exists() and marker.exists() and reverse.stat().st_mtime >= path.stat().st_mtime: return reverse tmp = reverse.with_suffix(".tmp.mp4") audio_present = has_audio(path) command = [ "/opt/homebrew/bin/ffmpeg", "-y", "-i", str(path), "-map", "0:v:0", ] if audio_present: command += ["-map", "0:a:0", "-vf", "reverse", "-af", "areverse"] else: command += ["-vf", "reverse", "-an"] command += [ "-c:v", "libx264", "-preset", "veryfast", "-crf", "18", "-g", "1", "-keyint_min", "1", "-sc_threshold", "0", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "160k", "-movflags", "+faststart", str(tmp), ] subprocess.run( command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, ) tmp.replace(reverse) write_json(marker, { "normalized": True, "reverse": True, "codec": "h264", "fps": 24, "all_keyframes": True, "faststart": True, "audio_softening_version": AUDIO_SOFTENING_VERSION, "source_mtime": path.stat().st_mtime, }) return reverse def final_image_from_state(state: dict[str, Any]) -> Path | None: files = state.get("outputFiles") or [] for item in files: p = Path(item) if p.exists() and p.suffix.lower() in {".png", ".jpg", ".jpeg"}: return p downloaded = state.get("downloaded") or {} if isinstance(downloaded, dict): for item in downloaded.values(): if isinstance(item, dict): p = Path(item.get("outPath") or "") if p.exists() and p.suffix.lower() in {".png", ".jpg", ".jpeg"}: return p return None def final_video_from_state(state: dict[str, Any]) -> Path | None: downloaded = state.get("downloaded") or {} if isinstance(downloaded, dict): p = Path(downloaded.get("outPath") or "") if p.exists() and p.suffix.lower() == ".mp4": return p files = state.get("outputFiles") or [] for item in files: p = Path(item) if p.exists() and p.suffix.lower() == ".mp4": return p return None def hero_job_id(manifest: dict[str, Any]) -> str | None: response = manifest.get("kameramotor_response") or {} return response.get("job_id") or response.get("id") def camera_job_path(job_id_value: str, kind: str) -> Path: dirs = { "state": [KAMERA_STATE, KAMERA_STATE_MAGNIFIC], "done": [KAMERA_DONE, KAMERA_DONE_MAGNIFIC], "failed": [KAMERA_FAILED, KAMERA_FAILED_MAGNIFIC], }[kind] for directory in dirs: path = directory / f"{job_id_value}.json" if path.exists(): return path return dirs[0] / f"{job_id_value}.json" def existing_file(value: str | None) -> Path | None: if not value: return None path = Path(value) if path.exists() and path.is_file(): return path return None def ensure_prompts(out_root: Path) -> dict[str, str]: out = out_root / "prompts_motor_safe" out.mkdir(parents=True, exist_ok=True) prompt_files: dict[str, str] = {} for key, filename in PROMPTS.items(): src = PROMPT_DIR / filename dst = out / filename safe_write_text(dst, one_line(safe_read_text(src))) prompt_files[key] = str(dst) return prompt_files def runtime_prompt(kind: str, filename: str) -> Path: src = CANONICAL_PROMPT_ROOT / kind / filename dst = RUNTIME_PROMPT_CACHE / kind / filename if dst.exists(): return dst dst.parent.mkdir(parents=True, exist_ok=True) safe_write_text(dst, safe_read_text(src)) return dst def ensure_canonical_prompts(out_root: Path) -> tuple[dict[str, str], dict[str, str]]: image_out = out_root / "prompts_snapshot" video_out = out_root / "video_prompts_snapshot" image_out.mkdir(parents=True, exist_ok=True) video_out.mkdir(parents=True, exist_ok=True) image_files: dict[str, str] = {} for slot, filename in IMAGE_PROMPTS.items(): src = runtime_prompt("prompts_snapshot", filename) dst = image_out / filename if not dst.exists(): try: safe_write_text(dst, safe_read_text(src)) except OSError: pass image_files[slot] = str(src) video_files: dict[str, str] = {} for slot, filename in VIDEO_PROMPTS.items(): src = runtime_prompt("video_prompts_snapshot", filename) dst = video_out / filename if not dst.exists(): try: safe_write_text(dst, safe_read_text(src)) except OSError: pass video_files[slot] = str(src) return image_files, video_files def queue_kling(job: dict[str, Any]) -> str: KLING_JOBS.mkdir(parents=True, exist_ok=True) job_file = KLING_JOBS / f"{job['id']}.json" if not job_file.exists(): write_json(job_file, job) return str(job_file) def queue_magnific_video_job(job: dict[str, Any]) -> str: MAGNIFIC_VIDEO_JOBS.mkdir(parents=True, exist_ok=True) job_file = MAGNIFIC_VIDEO_JOBS / f"{job['id']}.json" if not job_file.exists(): write_json(job_file, job) return str(job_file) def queue_magnific_image_job(job: dict[str, Any]) -> str: response = post_json("http://127.0.0.1:8089/api/kameramotor/job", job) return str(response.get("job_id") or response.get("id") or "") def image_job_state(job_id_value: str) -> dict[str, Any] | None: if not job_id_value: return None if camera_job_path(job_id_value, "failed").exists(): return {"failed": True, "job_id": job_id_value} state = read_json(camera_job_path(job_id_value, "state")) if state and camera_job_path(job_id_value, "done").exists(): return state return None def queue_image_stages(manifest_path: Path, manifest: dict[str, Any], anchors: dict[str, str], out_root: Path, image_prompts: dict[str, str]) -> bool: image_manifest_path = out_root / "manifest_images_cdef.json" image_manifest = read_json(image_manifest_path) or { "status": "waiting", "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "capture_manifest": str(manifest_path), "jobs": [], } queued_slots = {job.get("slot") for job in image_manifest.get("jobs") or []} changed = False ratio = (manifest.get("payload") or {}).get("ratio") or "16:9" slug = safe_slug(manifest.get("capture_id") or manifest_path.parent.name) for spec in IMAGE_SPECS: if spec["slot"] in queued_slots: continue source_path = existing_file(anchors.get(spec["source"])) if not source_path: continue output_dir = out_root / "generated_images" / f"{spec['slot']}_{spec['label']}" payload = { "image": str(source_path), "prompt_file": image_prompts[spec["slot"]], "ratio": ratio, "aspectRatio": ratio, "num_images": 1, "resolution": "4k", "thinking_level": "high", "generator": "nano", "provider": "magnific", "provider_route": "magnific_mcp_m084", "mode": "imagen-nano-banana-2-flash", "stem": f"The Object {slug} {spec['slot']} {spec['label']}", "filter": f"image_{spec['slot']}_{spec['label']}", "source": "THE OBJECT deterministic live image pipeline", "output_dir": str(output_dir), "group_id": f"theobject-live:{slug}:image:{spec['slot']}", } jid = queue_magnific_image_job(payload) image_manifest["jobs"].append({"slot": spec["slot"], "label": spec["label"], "id": jid, "source_slot": spec["source"], "prompt_file": image_prompts[spec["slot"]], "output_dir": str(output_dir), "status": "queued"}) queued_slots.add(spec["slot"]) changed = True time.sleep(0.01) for job in image_manifest.get("jobs") or []: slot = job.get("slot") if not slot or slot in anchors and Path(anchors[slot]).exists(): continue state = image_job_state(job.get("id", "")) if not state: continue if state.get("failed"): job["status"] = "failed" changed = True continue image = final_image_from_state(state) if not image: continue final_path = out_root / "images" / next(spec["filename"] for spec in IMAGE_SPECS if spec["slot"] == slot) final_path.parent.mkdir(parents=True, exist_ok=True) if not final_path.exists(): safe_copy2(image, final_path) anchors[slot] = str(final_path) job["status"] = "done" job["image"] = str(final_path) log_followup(manifest.get("capture_id") or manifest_path.parent.name, "image_slot_done", slot=slot, path=str(final_path)) changed = True image_manifest["anchors"] = anchors image_manifest["status"] = "ready" if all(existing_file(anchors.get(slot)) for slot in ["C", "D", "E", "F"]) else "running" image_manifest["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z") if changed or not image_manifest_path.exists(): write_json(image_manifest_path, image_manifest) return image_manifest["status"] == "ready" def magnific_video_job_path(job_id_value: str, kind: str) -> Path: dirs = { "state": KAMERA_STATE_MAGNIFIC, "done": KAMERA_DONE_MAGNIFIC, "failed": KAMERA_FAILED_MAGNIFIC, } return dirs[kind] / f"{job_id_value}.json" def video_job_state(job_id_value: str) -> dict[str, Any] | None: if not job_id_value: return None if magnific_video_job_path(job_id_value, "failed").exists(): return {"failed": True, "job_id": job_id_value} state = read_json(magnific_video_job_path(job_id_value, "state")) if state and magnific_video_job_path(job_id_value, "done").exists(): return state return None def video_variant(job: dict[str, Any] | None) -> str: variant = str((job or {}).get("variant") or (job or {}).get("badge") or "").upper() if variant in VIDEO_PROVIDER_CONFIGS: return variant slug = str((job or {}).get("slug") or "") api = str((job or {}).get("api") or "") if "kling" in slug.lower() or api.lower() == "kling": return "K" return "S" def variant_ready_filename(spec: dict[str, Any], variant: str) -> str: return f"{variant}_{spec['filename']}" def queue_video_stages(manifest_path: Path, manifest: dict[str, Any], anchors: dict[str, str], out_root: Path, video_prompts: dict[str, str]) -> bool: video_manifest_path = out_root / "manifest_videos_gn.json" video_manifest = read_json(video_manifest_path) or { "status": "waiting", "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "capture_manifest": str(manifest_path), "jobs": [], } queued_variants = {(job.get("slot"), video_variant(job)) for job in video_manifest.get("jobs") or []} changed = False ratio = (manifest.get("payload") or {}).get("ratio") or "16:9" slug = safe_slug(manifest.get("capture_id") or manifest_path.parent.name) for spec in VIDEO_SPECS: for variant, provider_config in VIDEO_PROVIDER_CONFIGS.items(): if (spec["slot"], variant) in queued_variants: continue start_image = existing_file(anchors.get(spec["start"])) end_image = existing_file(anchors.get(spec["end"])) if not start_image or (provider_config.get("requires_end_image") and not end_image): continue output_dir = out_root / provider_config["output_dir"] / f"{variant}_{spec['order']:02d}_{spec['slot']}_{spec['label']}" suffix = provider_config["output_suffix"] payload = { "id": job_id(slug, f"{spec['slot']}_{spec['label']}_{suffix}_{variant}"), "type": "video", "video": True, "variant": variant, "badge": provider_config["badge"], "provider_name": provider_config["provider_name"], "provider": provider_config["provider"], "provider_route": provider_config["route"], "api": provider_config["api"], "model": provider_config["model"], "mode": provider_config["mode"], "slug": provider_config["slug"], "model_mode": provider_config["mode"], "resolution": provider_config["resolution"], "duration": provider_config["duration"], "ratio": ratio, "aspectRatio": ratio, "imageCount": 1, "enable_audio": bool(provider_config["withSoundEffects"]), "withSoundEffects": provider_config["withSoundEffects"], "requires_end_image": provider_config["requires_end_image"], "disable_end_keyframe": not provider_config["requires_end_image"], "prompt_file": video_prompts[spec["slot"]], "start_image": str(start_image), "negative_prompt": NEGATIVE, "output_dir": str(output_dir), "output_name_prefix": f"{slug}_{variant}_{spec['order']:02d}_{spec['slot']}_{spec['label']}_{suffix}", "filter": f"{slug}_{variant}_{spec['order']:02d}_{spec['slot']}_{spec['label']}_{suffix}", "source": f"THE OBJECT deterministic live Magnific {provider_config['provider_name']} pipeline", "video_provider_version": VIDEO_PROVIDER_VERSION, "cost_note": f"Victor Go: {provider_config['provider_name']} via Magnific CDP Unlimited route. Generation is guarded by simulation before submit.", "app_note": "Raw clip is not final app material. App-side follow-up: loudnorm, fades, GOP6, no B-frames, cacheable MP4.", "object_slug": slug, "object_label": "Live Object", "order_index": spec["order"], "slot_letter": spec["slot"], "start_slot": spec["start"], "end_slot": spec["end"], } if end_image and provider_config.get("requires_end_image"): payload["end_image"] = str(end_image) job_file = queue_magnific_video_job(payload) video_manifest["jobs"].append({ "slot": spec["slot"], "variant": variant, "badge": provider_config["badge"], "provider_name": provider_config["provider_name"], "label": spec["label"], "id": payload["id"], "filter": spec["filter"], "start_slot": spec["start"], "end_slot": spec["end"], "prompt_file": video_prompts[spec["slot"]], "output_dir": str(output_dir), "job_file": job_file, "provider": payload["provider"], "provider_route": payload["provider_route"], "api": payload["api"], "slug": payload["slug"], "resolution": payload["resolution"], "withSoundEffects": payload["withSoundEffects"], "status": "queued", }) queued_variants.add((spec["slot"], variant)) changed = True time.sleep(0.01) app_ready_dir = out_root / "app_ready_gop6_loudnorm" app_ready_dir.mkdir(parents=True, exist_ok=True) app_ready_videos: list[dict[str, Any]] = [] for job in video_manifest.get("jobs") or []: slot = job.get("slot") spec = next((item for item in VIDEO_SPECS if item["slot"] == slot), None) if not spec: continue variant = video_variant(job) # Bestandsschutz: K-Jobs aus alten Manifesten liefern variant="K", # das nach Victor-Grundsatz 2026-06-11 nicht mehr in VIDEO_PROVIDER_CONFIGS # ist. K-Jobs werden übersprungen — sie dürfen nie als S umetikettiert werden # (Fix 2026-06-12: variant='S' würde S-Plätze mit K-Inhalt überschreiben). if variant not in VIDEO_PROVIDER_CONFIGS: continue provider_config = VIDEO_PROVIDER_CONFIGS[variant] ready_path = app_ready_dir / variant_ready_filename(spec, variant) legacy_ready_path = app_ready_dir / spec["filename"] if variant == "S" and not ready_path.exists() and legacy_ready_path.exists(): safe_copy2(legacy_ready_path, ready_path) if ready_path.exists(): app_ready_videos.append({"slot": slot, "variant": variant, "badge": provider_config["badge"], "provider_name": provider_config["provider_name"], "label": spec["label"], "order_index": spec["order"], "path": str(ready_path), "start_slot": spec["start"], "end_slot": spec["end"]}) continue state = video_job_state(job.get("id", "")) if not state: continue if state.get("failed"): job["status"] = "failed" changed = True continue video = final_video_from_state(state) if not video: continue trim_video_in_place(video) normalize_video_for_app(video) if not ready_path.exists(): safe_copy2(video, ready_path) normalize_video_for_app(ready_path) job["status"] = "done" job["variant"] = variant job["badge"] = provider_config["badge"] job["provider_name"] = provider_config["provider_name"] job["video"] = str(ready_path) app_ready_videos.append({"slot": slot, "variant": variant, "badge": provider_config["badge"], "provider_name": provider_config["provider_name"], "label": spec["label"], "order_index": spec["order"], "path": str(ready_path), "start_slot": spec["start"], "end_slot": spec["end"]}) log_followup(manifest.get("capture_id") or manifest_path.parent.name, "video_slot_done", slot=slot, variant=variant, path=str(ready_path)) changed = True app_ready_keys = {(item.get("slot"), item.get("variant") or "S") for item in app_ready_videos} expected_keys = {(spec["slot"], variant) for spec in VIDEO_SPECS for variant in VIDEO_PROVIDER_CONFIGS} video_manifest["app_ready_videos"] = sorted(app_ready_videos, key=lambda item: (item["order_index"], item.get("variant") or "S")) if app_ready_keys >= expected_keys: video_manifest["status"] = "ready" video_manifest.pop("pause", None) else: video_manifest["status"] = "running" video_manifest.pop("pause", None) video_manifest["provider"] = "magnific" video_manifest["provider_route"] = "magnific_cdp_unlimited" video_manifest["variants"] = VIDEO_PROVIDER_CONFIGS video_manifest["video_provider_version"] = VIDEO_PROVIDER_VERSION video_manifest["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z") if changed or not video_manifest_path.exists(): write_json(video_manifest_path, video_manifest) return video_manifest["status"] == "ready" def queue_phase1(manifest_path: Path, manifest: dict[str, Any], hero_path: Path) -> None: capture_dir = manifest_path.parent capture_id = manifest.get("capture_id") or capture_dir.name slug = safe_slug(capture_id) ratio = (manifest.get("payload") or {}).get("ratio") or "16:9" out_root = capture_dir / "fullrun" anchors = out_root / "anchors" anchors.mkdir(parents=True, exist_ok=True) a_anchor = anchors / f"{slug}__A_original.png" b_anchor = anchors / f"{slug}__B_hero.png" if not a_anchor.exists(): safe_copy2(manifest["image"], a_anchor) if not b_anchor.exists(): safe_copy2(hero_path, b_anchor) prompt_files = ensure_prompts(out_root) base = { "type": "video", "video": True, "provider": "kling", "engine": "kling", "kling": True, "kling_version": "3.0", "model_mode": "720p", "resolution": "720p", "duration": 5, "ratio": ratio, "aspectRatio": ratio, "imageCount": 1, "enable_audio": True, "withSoundEffects": True, "negative_prompt": NEGATIVE, "source": "THE OBJECT live automatic run via Kling-Kameramotor", "object_slug": slug, "object_label": "Live Object", "batch_output_root": str(out_root), } specs = [ ("01_a_to_b_reality_to_hero", prompt_files["a_to_b"], a_anchor, b_anchor, False), ("02_b_to_z_real_explosion", prompt_files["real_explosion"], b_anchor, None, True), ("03_b_to_z_unreal_explosion", prompt_files["unreal_explosion"], b_anchor, None, True), ("04a_hero_360_clockwork_asmr", prompt_files["hero_360"], b_anchor, b_anchor, False), ] jobs = [] for filter_name, prompt_file, start_image, end_image, start_only in specs: jid = job_id(slug, filter_name) output_dir = out_root / "videos" / slug / filter_name output_dir.mkdir(parents=True, exist_ok=True) job = { "id": jid, **base, "filter": filter_name, "prompt_file": prompt_file, "start_image": str(start_image), "output_dir": str(output_dir), "output_name_prefix": f"{slug}_{filter_name}", } if end_image is not None: job["end_image"] = str(end_image) if start_only: job["start_only"] = True job["end_optional"] = True job["end_frame_required"] = False job_file = queue_kling(job) jobs.append({"id": jid, "filter": filter_name, "job_file": job_file, "output_dir": str(output_dir)}) time.sleep(0.01) write_json( out_root / "manifest_phase1.json", { "status": "queued_phase1", "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "capture_manifest": str(manifest_path), "output_root": str(out_root), "anchors": {"A": str(a_anchor), "B": str(b_anchor)}, "prompts": prompt_files, "jobs": jobs, }, ) def extract_last_frame(video: Path, out_path: Path, force: bool = False) -> None: if out_path.exists() and not force and out_path.stat().st_mtime >= video.stat().st_mtime: return out_path.parent.mkdir(parents=True, exist_ok=True) frames = frame_count(video) if frames > 1: vf = f"select=eq(n\\,{frames - 1})" command = ["/opt/homebrew/bin/ffmpeg", "-y", "-i", str(video), "-vf", vf, "-frames:v", "1", "-update", "1", str(out_path)] else: command = ["/opt/homebrew/bin/ffmpeg", "-y", "-sseof", "-0.04", "-i", str(video), "-frames:v", "1", "-update", "1", str(out_path)] subprocess.run(command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) def refresh_app_poster(video: Path, out_path: Path) -> None: extract_last_frame(video, out_path, force=True) def queue_phase2(phase1_path: Path) -> None: phase1 = read_json(phase1_path) if not phase1: return out_root = Path(phase1["output_root"]) slug = Path(phase1["anchors"]["B"]).name.split("__B_")[0] ratio = None jobs = phase1.get("jobs") or [] phase2_path = out_root / "manifest_phase2_exploded_360.json" phase2 = read_json(phase2_path) or {"jobs": []} queued = {j.get("filter") for j in phase2.get("jobs", [])} prompt_files = phase1.get("prompts") or ensure_prompts(out_root) for kind, source_filter, phase2_filter, prompt_key in [ ("real", "02_b_to_z_real_explosion", "04b_exploded_real_360_clockwork_asmr", "real_360"), ("unreal", "03_b_to_z_unreal_explosion", "04c_exploded_unreal_360_clockwork_asmr", "unreal_360"), ]: if phase2_filter in queued: continue source = next((j for j in jobs if j.get("filter") == source_filter), None) if not source: continue source_id = source["id"] if (KLING_FAILED / f"{source_id}.json").exists(): phase2["jobs"].append({"filter": phase2_filter, "status": "blocked_source_failed", "source_job_id": source_id}) continue state = read_json(KLING_STATE / f"{source_id}.json") if not state or not (KLING_DONE / f"{source_id}.json").exists(): continue video = final_video_from_state(state) if not video: continue trim_video_in_place(video) z_path = out_root / "anchors_z_from_kling" / f"{slug}__Z_{kind}_from_kling_final_frame.png" extract_last_frame(video, z_path, force=True) if ratio is None: done_job = read_json(KLING_DONE / f"{source_id}.json") or {} ratio = done_job.get("ratio") or done_job.get("aspectRatio") or "16:9" jid = job_id(slug, phase2_filter) output_dir = out_root / "videos" / slug / phase2_filter output_dir.mkdir(parents=True, exist_ok=True) job = { "id": jid, "type": "video", "video": True, "provider": "kling", "engine": "kling", "kling": True, "kling_version": "3.0", "model_mode": "720p", "resolution": "720p", "duration": 5, "ratio": ratio or "16:9", "aspectRatio": ratio or "16:9", "imageCount": 1, "enable_audio": True, "withSoundEffects": True, "negative_prompt": NEGATIVE, "prompt_file": prompt_files[prompt_key], "start_image": str(z_path), "end_image": str(z_path), "output_dir": str(output_dir), "output_name_prefix": f"{slug}_{phase2_filter}", "filter": phase2_filter, "source": "THE OBJECT live automatic phase 2 via Kling-Kameramotor", "object_slug": slug, "object_label": "Live Object", } job_file = queue_kling(job) phase2["jobs"].append({"id": jid, "filter": phase2_filter, "source_job_id": source_id, "z_frame": str(z_path), "job_file": job_file, "status": "queued"}) queued.add(phase2_filter) phase2["status"] = "queued_all_phase2" if len([j for j in phase2["jobs"] if j.get("status") == "queued"]) >= 2 else "waiting_for_z_frames" phase2["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z") write_json(phase2_path, phase2) def trim_finished_phase1_and_phase2(phase1_path: Path) -> None: phase1 = read_json(phase1_path) if not phase1: return all_jobs = list(phase1.get("jobs") or []) phase2 = read_json(Path(phase1["output_root"]) / "manifest_phase2_exploded_360.json") or {} all_jobs.extend(phase2.get("jobs") or []) for job in all_jobs: jid = job.get("id") if not jid: continue state = read_json(KLING_STATE / f"{jid}.json") if not state or not (KLING_DONE / f"{jid}.json").exists(): continue video = final_video_from_state(state) if video: trim_video_in_place(video) normalize_video_for_app(video) if job.get("filter") == "01_a_to_b_reality_to_hero": anchors = phase1.get("anchors") or {} slug = Path(anchors.get("B", "")).name.split("__B_")[0] if slug: poster = Path(phase1["output_root"]) / "app_posters" / f"{slug}__B_from_a_to_b_final_frame.png" refresh_app_poster(video, poster) def process_manifest(manifest_path: Path) -> None: manifest = read_json(manifest_path) if not manifest or not manifest.get("ok"): return payload = manifest.get("payload") or {} if manifest.get("pipeline_version") != PIPELINE_VERSION and payload.get("pipeline_version") != PIPELINE_VERSION: return capture_dir = manifest_path.parent out_root = capture_dir / "fullrun" if manifest.get("autopilot_status") == "app_ready" and (out_root / "RUN_LEDGER.json").exists(): video_manifest = read_json(out_root / "manifest_videos_gn.json") or {} app_ready_videos = video_manifest.get("app_ready_videos") or [] ready_keys = {(item.get("slot"), item.get("variant") or "S") for item in app_ready_videos} expected_keys = {(spec["slot"], variant) for spec in VIDEO_SPECS for variant in VIDEO_PROVIDER_CONFIGS} if ready_keys >= expected_keys: return images_dir = out_root / "images" images_dir.mkdir(parents=True, exist_ok=True) image_prompts, video_prompts = ensure_canonical_prompts(out_root) anchors: dict[str, str] = {} a_anchor = images_dir / "A_original_from_iphone.jpg" if not a_anchor.exists() and Path(manifest.get("image", "")).exists(): safe_copy2(manifest["image"], a_anchor) if a_anchor.exists(): anchors["A"] = str(a_anchor) hid = hero_job_id(manifest) if not hid: return if camera_job_path(hid, "failed").exists(): manifest["autopilot_status"] = "hero_failed" write_json(manifest_path, manifest) return state = read_json(camera_job_path(hid, "state")) if not state or not camera_job_path(hid, "done").exists(): manifest["autopilot_status"] = "waiting_for_B_hero" write_json(manifest_path, manifest) return hero = final_image_from_state(state) if not hero: return b_anchor = images_dir / "B_hero.png" if not b_anchor.exists(): safe_copy2(hero, b_anchor) anchors["B"] = str(b_anchor) manifest["hero_image"] = str(b_anchor) capture_id = str(manifest.get("capture_id") or manifest_path.parent.name) log_followup(capture_id, "hero_ready", hero=str(hero), anchor=str(b_anchor)) image_manifest = read_json(out_root / "manifest_images_cdef.json") or {} anchors.update({k: v for k, v in (image_manifest.get("anchors") or {}).items() if isinstance(v, str)}) for spec in IMAGE_SPECS: existing = images_dir / spec["filename"] if existing.exists(): anchors[spec["slot"]] = str(existing) if not queue_image_stages(manifest_path, manifest, anchors, out_root, image_prompts): manifest["autopilot_status"] = "running_images_CDEF" log_followup(capture_id, "waiting_images", anchors=sorted(anchors.keys())) write_json(manifest_path, manifest) return image_manifest = read_json(out_root / "manifest_images_cdef.json") or {} anchors.update({k: v for k, v in (image_manifest.get("anchors") or {}).items() if isinstance(v, str)}) for spec in IMAGE_SPECS: existing = images_dir / spec["filename"] if existing.exists(): anchors[spec["slot"]] = str(existing) if not queue_video_stages(manifest_path, manifest, anchors, out_root, video_prompts): video_manifest = read_json(out_root / "manifest_videos_gn.json") or {} pause = video_manifest.get("pause") if pause: manifest["autopilot_status"] = "needs_victor_video_paused" manifest["autopilot_pause"] = pause log_followup(capture_id, "needs_victor_video_paused", reason=pause.get("reason")) else: manifest["autopilot_status"] = "running_videos_GN" manifest.pop("autopilot_pause", None) log_followup(capture_id, "waiting_videos", anchors=sorted(anchors.keys())) write_json(manifest_path, manifest) return video_manifest = read_json(out_root / "manifest_videos_gn.json") or {} write_json(out_root / "RUN_LEDGER.json", { "status": "app_ready", "capture_id": manifest.get("capture_id"), "ratio": (manifest.get("payload") or {}).get("ratio") or "16:9", "A": anchors.get("A"), "B": anchors.get("B"), "C": anchors.get("C"), "D": anchors.get("D"), "E": anchors.get("E"), "F": anchors.get("F"), "image_jobs": image_manifest.get("jobs") or [], "video_jobs": video_manifest.get("jobs") or [], "app_ready_videos": video_manifest.get("app_ready_videos") or [], "updated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), }) manifest["autopilot_status"] = "app_ready" log_followup(capture_id, "app_ready", videos=len(video_manifest.get("app_ready_videos") or [])) write_json(manifest_path, manifest) _mount_missing_logged = False def loop() -> None: global _mount_missing_logged if threading.current_thread() is threading.main_thread(): signal.signal(signal.SIGHUP, signal.SIG_IGN) while True: if not os.path.ismount(str(HOT_DISK_MOUNT)): if not _mount_missing_logged: print(json.dumps({"ok": False, "error": f"Hot Disk nicht gemountet: {HOT_DISK_MOUNT}", "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z")}), flush=True) _mount_missing_logged = True time.sleep(30) continue _mount_missing_logged = False ROOT.mkdir(parents=True, exist_ok=True) manifest_paths = sorted( CAPTURES.glob("*/manifest.json"), key=lambda path: path.stat().st_mtime if path.exists() else 0, reverse=True, ) for manifest_path in manifest_paths: try: process_manifest(manifest_path) except OSError as exc: # Watson-Schubs 2026-06-11: Fehlerbehandlung darf den Loop-Thread # NIE toeten (Incident 2026-06-10 23:14 — log_followup warf selbst # EDEADLK und der Daemon-Thread starb still). Alles defensiv. try: if not _is_retryable_file_error(exc): err_path = manifest_path.parent / "autopilot_error.json" write_json(err_path, {"error": str(exc), "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z")}) continue wait_path = manifest_path.parent / "autopilot_waiting.json" write_json(wait_path, { "status": "waiting_for_file_lock", "error": str(exc), "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), }) try: manifest = read_json(manifest_path) or {} except OSError: manifest = {} capture_id = str(manifest.get("capture_id") or manifest_path.parent.name) log_followup(capture_id, "waiting_for_file_lock", error=str(exc)) except Exception: pass except Exception as exc: try: err_path = manifest_path.parent / "autopilot_error.json" write_json(err_path, {"error": str(exc), "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z")}) manifest = read_json(manifest_path) or {} capture_id = str(manifest.get("capture_id") or manifest_path.parent.name) log_followup(capture_id, "error", error=str(exc)) except Exception: pass time.sleep(POLL_SECONDS) if __name__ == "__main__": loop()