| 1 | """Persist and project shot memories for the human review gate.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import os |
| 7 | import re |
| 8 | import shutil |
| 9 | import subprocess |
| 10 | import threading |
| 11 | import time |
| 12 | import urllib.request |
| 13 | import uuid |
| 14 | from copy import deepcopy |
| 15 | from datetime import datetime, timezone |
| 16 | from pathlib import Path |
| 17 | from typing import Any, Callable |
| 18 | |
| 19 | from nanobot.director.memory_review import MemoryReviewConflict |
| 20 | from nanobot.director.memory_selector import MemoryVlmSelector |
| 21 | from nanobot.director.r2v_memory_workflow import ( |
| 22 | Publisher, |
| 23 | publish_memory_record, |
| 24 | publish_memory_records, |
| 25 | ) |
| 26 | from nanobot.storage.files import configured_file_publisher |
| 27 | from nanobot.utils.helpers import write_json_atomic |
| 28 | |
| 29 | AudioExtractor = Callable[[Path, Path], None] |
| 30 | _MEMORY_REVIEW_LOCK = threading.Lock() |
| 31 | _MAX_VLM_PROMPT_IMAGES = 16 |
| 32 | _SHOT_READ_ATTEMPTS = 20 |
| 33 | _SHOT_READ_DELAY_SEC = 0.01 |
| 34 | _COMMON_MEDIA_BIN_DIRS = ( |
| 35 | Path("/opt/homebrew/bin"), |
| 36 | Path("/usr/local/bin"), |
| 37 | Path("/usr/bin"), |
| 38 | ) |
| 39 | |
| 40 | |
| 41 | def _resolve_media_binary(name: str) -> str: |
| 42 | """Resolve ffmpeg tools even when a detached Gateway has a minimal PATH.""" |
| 43 | configured = os.environ.get(f"NANOBOT_{name.upper()}_BIN", "").strip() |
| 44 | if configured: |
| 45 | return configured |
| 46 | discovered = shutil.which(name) |
| 47 | if discovered: |
| 48 | return discovered |
| 49 | for directory in _COMMON_MEDIA_BIN_DIRS: |
| 50 | candidate = directory / name |
| 51 | if candidate.is_file(): |
| 52 | return str(candidate) |
| 53 | return name |
| 54 | |
| 55 | |
| 56 | def merge_approved_memories( |
| 57 | bank: dict[str, dict[str, Any]], |
| 58 | proposed: dict[str, dict[str, Any]], |
| 59 | ) -> dict[str, dict[str, Any]]: |
| 60 | """Merge an approved proposal without degrading confirmed visual memories.""" |
| 61 | merged = deepcopy(bank) |
| 62 | for memory_id, candidate in proposed.items(): |
| 63 | existing = merged.get(memory_id) |
| 64 | if existing and existing.get("visual_status") == "confirmed": |
| 65 | if candidate.get("visual_status") != "confirmed": |
| 66 | continue |
| 67 | if float(candidate.get("confidence") or 0) <= float( |
| 68 | existing.get("confidence") or 0 |
| 69 | ): |
| 70 | continue |
| 71 | replacement = deepcopy(candidate) |
| 72 | if existing: |
| 73 | for key in ("local_audio_path", "audio_path", "audio_source_shot_id"): |
| 74 | if not replacement.get(key) and existing.get(key): |
| 75 | replacement[key] = existing[key] |
| 76 | merged[memory_id] = replacement |
| 77 | return merged |
| 78 | |
| 79 | |
| 80 | def materialize_character_memories( |
| 81 | *, |
| 82 | selections: dict[str, Any], |
| 83 | candidates: list[dict[str, Any]], |
| 84 | shot_id: int, |
| 85 | video_path: Path, |
| 86 | memory_dir: Path, |
| 87 | audio_extractor: AudioExtractor, |
| 88 | ) -> dict[str, dict[str, Any]]: |
| 89 | """Copy selected frames and bind each new visual to this shot's audio.""" |
| 90 | by_index = {int(item["candidate_index"]): item for item in candidates} |
| 91 | rows = selections.get("selections") |
| 92 | if not isinstance(rows, list): |
| 93 | return {} |
| 94 | memory_dir.mkdir(parents=True, exist_ok=True) |
| 95 | source_audio = memory_dir / f"_source_shot_{shot_id:03d}.wav" |
| 96 | if rows: |
| 97 | audio_extractor(video_path, source_audio) |
| 98 | proposed: dict[str, dict[str, Any]] = {} |
| 99 | for raw in rows: |
| 100 | if not isinstance(raw, dict): |
| 101 | continue |
| 102 | memory_id = str(raw.get("character_id") or "").strip() |
| 103 | try: |
| 104 | candidate = by_index[int(raw["candidate_index"])] |
| 105 | confidence = float(raw["confidence"]) |
| 106 | except (KeyError, TypeError, ValueError): |
| 107 | continue |
| 108 | if not memory_id: |
| 109 | continue |
| 110 | image = memory_dir / f"{memory_id}.jpg" |
| 111 | audio = memory_dir / f"{memory_id}.wav" |
| 112 | shutil.copyfile(str(candidate["path"]), image) |
| 113 | shutil.copyfile(source_audio, audio) |
| 114 | proposed[memory_id] = { |
| 115 | "memory_id": memory_id, |
| 116 | "kind": "character", |
| 117 | "candidate_index": int(candidate["candidate_index"]), |
| 118 | "frame_index": int(candidate["frame_index"]), |
| 119 | "timestamp_sec": float(candidate["timestamp_sec"]), |
| 120 | "confidence": confidence, |
| 121 | "target_only": bool(raw.get("target_only", False)), |
| 122 | "visible_character_ids": list(raw.get("visible_character_ids") or []), |
| 123 | "reasoning": str(raw.get("reasoning") or ""), |
| 124 | "visual_status": "confirmed" if raw.get("target_only") else "provisional", |
| 125 | "source_shot_id": int(shot_id), |
| 126 | "audio_source_shot_id": int(shot_id), |
| 127 | "local_image_path": str(image.resolve()), |
| 128 | "local_audio_path": str(audio.resolve()), |
| 129 | } |
| 130 | source_audio.unlink(missing_ok=True) |
| 131 | return proposed |
| 132 | |
| 133 | |
| 134 | def _review_selection(memory_id: str, record: dict[str, Any], *, kind: str) -> dict[str, Any]: |
| 135 | selection = { |
| 136 | "memory_id": memory_id, |
| 137 | "kind": kind, |
| 138 | "candidate_index": int(record.get("candidate_index") or 0), |
| 139 | "frame_index": int(record.get("frame_index") or 0), |
| 140 | "timestamp_sec": float(record.get("timestamp_sec") or 0), |
| 141 | "confidence": float(record.get("confidence") or 0), |
| 142 | "visual_status": str(record.get("visual_status") or ( |
| 143 | "representative" if kind == "previous_shot" else "provisional")), |
| 144 | "source_shot_id": int(record.get("source_shot_id") or 0), |
| 145 | "audio_source_shot_id": int(record.get("audio_source_shot_id") |
| 146 | or record.get("source_shot_id") or 0), |
| 147 | "reasoning": str(record.get("reasoning") or ""), |
| 148 | } |
| 149 | for key in ("local_image_path", "local_audio_path", "image_path", "audio_path"): |
| 150 | value = record.get(key) |
| 151 | if isinstance(value, str) and value.strip(): |
| 152 | selection[key] = value.strip() |
| 153 | return selection |
| 154 | |
| 155 | |
| 156 | def build_memory_review( |
| 157 | *, |
| 158 | shot_id: int, |
| 159 | attempt: int, |
| 160 | candidate_count: int, |
| 161 | bank: dict[str, dict[str, Any]], |
| 162 | previous_shot: dict[str, Any] | None, |
| 163 | review_id: str, |
| 164 | rejected_candidate_indices: list[int] | None = None, |
| 165 | history: list[dict[str, Any]] | None = None, |
| 166 | updated_at: str = "", |
| 167 | ) -> dict[str, Any]: |
| 168 | """Build a proposal containing every stored ID plus PREVIOUS_SHOT.""" |
| 169 | selections = [ |
| 170 | _review_selection(memory_id, record, kind="character") |
| 171 | for memory_id, record in bank.items() |
| 172 | if isinstance(record, dict) |
| 173 | and (record.get("local_image_path") or record.get("image_path")) |
| 174 | ] |
| 175 | if previous_shot and (previous_shot.get("local_image_path") |
| 176 | or previous_shot.get("image_path")): |
| 177 | selections.append(_review_selection( |
| 178 | "PREVIOUS_SHOT", previous_shot, kind="previous_shot")) |
| 179 | return { |
| 180 | "review_id": review_id, |
| 181 | "status": "awaiting_review", |
| 182 | "shot_id": int(shot_id), |
| 183 | "attempt": int(attempt), |
| 184 | "candidate_count": int(candidate_count), |
| 185 | "rejected_candidate_indices": sorted(set(rejected_candidate_indices or [])), |
| 186 | "selections": selections, |
| 187 | "history": deepcopy(history or []), |
| 188 | "error": None, |
| 189 | "updated_at": updated_at, |
| 190 | } |
| 191 | |
| 192 | |
| 193 | def _read_json(path: Path, default: Any) -> Any: |
| 194 | if not path.is_file(): |
| 195 | return deepcopy(default) |
| 196 | try: |
| 197 | return json.loads(path.read_text(encoding="utf-8")) |
| 198 | except (OSError, json.JSONDecodeError): |
| 199 | return deepcopy(default) |
| 200 | |
| 201 | |
| 202 | def _write_json(path: Path, value: Any) -> None: |
| 203 | write_json_atomic(path, value) |
| 204 | |
| 205 | |
| 206 | def _load_shot_for_memory(shot_path: Path, shot_id: int) -> dict[str, Any]: |
| 207 | """Read a shot JSON, retrying through concurrent truncated writes.""" |
| 208 | for attempt in range(_SHOT_READ_ATTEMPTS): |
| 209 | shot: dict[str, Any] | None = None |
| 210 | try: |
| 211 | if shot_path.is_file(): |
| 212 | raw = shot_path.read_text(encoding="utf-8") |
| 213 | if raw.strip(): |
| 214 | parsed = json.loads(raw) |
| 215 | if isinstance(parsed, dict): |
| 216 | shot = parsed |
| 217 | except (OSError, json.JSONDecodeError): |
| 218 | shot = None |
| 219 | if isinstance(shot, dict): |
| 220 | try: |
| 221 | loaded_id = int(shot.get("shot_id") or 0) |
| 222 | except (TypeError, ValueError): |
| 223 | loaded_id = 0 |
| 224 | if loaded_id == shot_id: |
| 225 | return shot |
| 226 | if attempt + 1 < _SHOT_READ_ATTEMPTS: |
| 227 | time.sleep(_SHOT_READ_DELAY_SEC) |
| 228 | raise ValueError(f"shot {shot_id} not found for memory selection") |
| 229 | |
| 230 | |
| 231 | def mark_memory_review_selecting( |
| 232 | *, |
| 233 | workspace: Path, |
| 234 | work_id: str, |
| 235 | shot_id: int, |
| 236 | ) -> dict[str, Any]: |
| 237 | """Persist selecting before workplace push so auto-generate does not race.""" |
| 238 | shot_path = ( |
| 239 | workspace / "director" / "works" / work_id / "shots" / f"shot_{shot_id:03d}.json" |
| 240 | ) |
| 241 | shot = _load_shot_for_memory(shot_path, shot_id) |
| 242 | review = shot.get("memory_review") |
| 243 | review = review if isinstance(review, dict) else {} |
| 244 | status = str(review.get("status") or "") |
| 245 | if status not in {"awaiting_review", "approved", "selecting", "reselecting"}: |
| 246 | review["status"] = "selecting" |
| 247 | review["shot_id"] = int(shot_id) |
| 248 | shot["memory_review"] = review |
| 249 | _write_json(shot_path, shot) |
| 250 | return shot |
| 251 | |
| 252 | |
| 253 | def _ordered_character_ids(caption: str) -> list[str]: |
| 254 | return list(dict.fromkeys(re.findall( |
| 255 | r"(?<![A-Za-z0-9_-])ID_[A-Za-z0-9_-]+(?![A-Za-z0-9_-])", |
| 256 | caption, |
| 257 | ))) |
| 258 | |
| 259 | |
| 260 | def initialize_memory_review_method_prompt( |
| 261 | *, |
| 262 | workspace: Path, |
| 263 | work_id: str, |
| 264 | shot_id: int, |
| 265 | error: str | None = None, |
| 266 | ) -> dict[str, Any]: |
| 267 | """Pause after generation and ask how memories should be selected.""" |
| 268 | work_dir = workspace / "director" / "works" / work_id |
| 269 | shot_path = work_dir / "shots" / f"shot_{shot_id:03d}.json" |
| 270 | state_path = work_dir / "state.json" |
| 271 | bank_path = work_dir / "memory" / "memory_bank.json" |
| 272 | shot = _load_shot_for_memory(shot_path, shot_id) |
| 273 | caption = str(shot.get("caption") or shot.get("summary") or "") |
| 274 | character_ids = _ordered_character_ids(caption) |
| 275 | bank = _read_json(bank_path, {}) |
| 276 | bank = bank if isinstance(bank, dict) else {} |
| 277 | missing_character_ids = [ |
| 278 | memory_id |
| 279 | for memory_id in character_ids |
| 280 | if not isinstance(bank.get(memory_id), dict) |
| 281 | or not ( |
| 282 | bank[memory_id].get("local_image_path") |
| 283 | or bank[memory_id].get("image_path") |
| 284 | ) |
| 285 | ] |
| 286 | review_character_ids = missing_character_ids or character_ids |
| 287 | review = { |
| 288 | "review_id": f"memory-review-{uuid.uuid4().hex}", |
| 289 | "status": "awaiting_method", |
| 290 | "shot_id": int(shot_id), |
| 291 | "attempt": 1, |
| 292 | "candidate_count": 0, |
| 293 | "rejected_candidate_indices": [], |
| 294 | "required_memory_ids": review_character_ids + ["PREVIOUS_SHOT"], |
| 295 | "manual_selected_ids": [], |
| 296 | "retained_memory_ids": [], |
| 297 | "selection_mode": None, |
| 298 | "selections": [], |
| 299 | "history": [], |
| 300 | "error": error.strip() if isinstance(error, str) and error.strip() else None, |
| 301 | "updated_at": datetime.now(timezone.utc).isoformat( |
| 302 | timespec="seconds" |
| 303 | ).replace("+00:00", "Z"), |
| 304 | } |
| 305 | shot["memory_review"] = review |
| 306 | _write_json(shot_path, shot) |
| 307 | state = _read_json(state_path, {}) |
| 308 | if isinstance(state, dict): |
| 309 | state["stage"] = "awaiting_memory_review" |
| 310 | _write_json(state_path, state) |
| 311 | return review |
| 312 | |
| 313 | |
| 314 | class ManualMemorySelector: |
| 315 | """Create non-final placeholders after the user chooses manual mode.""" |
| 316 | |
| 317 | @staticmethod |
| 318 | def _middle(candidates: list[dict[str, Any]]) -> dict[str, Any]: |
| 319 | if not candidates: |
| 320 | raise ValueError("manual memory selection has no source frames") |
| 321 | return candidates[len(candidates) // 2] |
| 322 | |
| 323 | def select_characters( |
| 324 | self, |
| 325 | *, |
| 326 | character_ids: list[str], |
| 327 | candidates: list[dict[str, Any]], |
| 328 | **_: Any, |
| 329 | ) -> dict[str, Any]: |
| 330 | middle = self._middle(candidates) |
| 331 | return { |
| 332 | "reasoning": "Waiting for manual identity-frame selection.", |
| 333 | "selections": [ |
| 334 | { |
| 335 | "character_id": memory_id, |
| 336 | "candidate_index": int(middle["candidate_index"]), |
| 337 | "confidence": 0.0, |
| 338 | "target_only": False, |
| 339 | "visible_character_ids": list(character_ids), |
| 340 | "reasoning": "Drag through the video and choose this identity memory.", |
| 341 | } |
| 342 | for memory_id in character_ids |
| 343 | ], |
| 344 | } |
| 345 | |
| 346 | def decide_scene_transition(self, **_: Any) -> dict[str, Any]: |
| 347 | return { |
| 348 | "scene_transition": False, |
| 349 | "reasoning": "Scene continuity will be selected manually.", |
| 350 | } |
| 351 | |
| 352 | def select_representative( |
| 353 | self, |
| 354 | *, |
| 355 | candidates: list[dict[str, Any]], |
| 356 | **_: Any, |
| 357 | ) -> dict[str, Any]: |
| 358 | middle = self._middle(candidates) |
| 359 | return { |
| 360 | "candidate_index": int(middle["candidate_index"]), |
| 361 | "confidence": 0.0, |
| 362 | "reasoning": "Drag through the video and choose the scene memory.", |
| 363 | } |
| 364 | |
| 365 | |
| 366 | def _next_planned_shot(work_dir: Path, shot_id: int) -> dict[str, Any] | None: |
| 367 | """Return the first later shot that has enough screenplay text to compare.""" |
| 368 | for path in sorted((work_dir / "shots").glob("shot_*.json")): |
| 369 | candidate = _read_json(path, {}) |
| 370 | if not isinstance(candidate, dict): |
| 371 | continue |
| 372 | try: |
| 373 | candidate_id = int(candidate.get("shot_id") or 0) |
| 374 | except (TypeError, ValueError): |
| 375 | continue |
| 376 | caption = str(candidate.get("caption") or candidate.get("summary") or "").strip() |
| 377 | if candidate_id > shot_id and caption: |
| 378 | return candidate |
| 379 | return None |
| 380 | |
| 381 | |
| 382 | def prepare_memory_review( |
| 383 | *, |
| 384 | workspace: Path, |
| 385 | work_id: str, |
| 386 | shot_id: int, |
| 387 | selector: Any, |
| 388 | video_fetcher: Callable[[str, Path], None], |
| 389 | frame_sampler: Callable[[Path, Path, int], list[dict[str, Any]]], |
| 390 | audio_extractor: AudioExtractor, |
| 391 | review_id_factory: Callable[[], str], |
| 392 | now: Callable[[], str], |
| 393 | candidate_count: int = 6, |
| 394 | target_memory_id: str | None = None, |
| 395 | publisher: Publisher | None = None, |
| 396 | ) -> dict[str, Any]: |
| 397 | """Create and persist a real review proposal for a completed R2V shot.""" |
| 398 | work_dir = workspace / "director" / "works" / work_id |
| 399 | shot_path = work_dir / "shots" / f"shot_{shot_id:03d}.json" |
| 400 | state_path = work_dir / "state.json" |
| 401 | bank_path = work_dir / "memory" / "memory_bank.json" |
| 402 | shot = _load_shot_for_memory(shot_path, shot_id) |
| 403 | state = _read_json(state_path, {}) |
| 404 | bank = _read_json(bank_path, {}) |
| 405 | if str(shot.get("status") or "") not in {"generated", "review_pass", "approved"}: |
| 406 | raise ValueError(f"shot {shot_id} is not generated") |
| 407 | if not isinstance(bank, dict): |
| 408 | bank = {} |
| 409 | |
| 410 | old_review = shot.get("memory_review") |
| 411 | old_review = old_review if isinstance(old_review, dict) else {} |
| 412 | if target_memory_id: |
| 413 | review_ids = { |
| 414 | str(item.get("memory_id") or "") |
| 415 | for item in old_review.get("selections", []) |
| 416 | if isinstance(item, dict) |
| 417 | } |
| 418 | if target_memory_id not in review_ids: |
| 419 | raise ValueError( |
| 420 | f"memory {target_memory_id} is not in the current review" |
| 421 | ) |
| 422 | attempt = int(old_review.get("attempt") or 0) + 1 |
| 423 | rejected = { |
| 424 | int(value) for value in old_review.get("rejected_candidate_indices", []) |
| 425 | if isinstance(value, int) or (isinstance(value, str) and value.isdigit()) |
| 426 | } |
| 427 | history = old_review.get("history") |
| 428 | history = history if isinstance(history, list) else [] |
| 429 | |
| 430 | memory_root = work_dir / "memory" |
| 431 | video_path = memory_root / "videos" / f"shot_{shot_id:03d}.mp4" |
| 432 | artifact = str(shot.get("artifact_url") or shot.get("artifact_path") or "").strip() |
| 433 | if not artifact: |
| 434 | raise ValueError(f"shot {shot_id} has no video artifact") |
| 435 | if not video_path.is_file(): |
| 436 | video_fetcher(artifact, video_path) |
| 437 | candidate_dir = memory_root / "candidates" / f"shot_{shot_id:03d}" |
| 438 | candidates = frame_sampler(video_path, candidate_dir, candidate_count) |
| 439 | if not candidates: |
| 440 | raise ValueError(f"shot {shot_id} produced no memory candidates") |
| 441 | |
| 442 | caption = str(shot.get("caption") or shot.get("summary") or "") |
| 443 | character_ids = _ordered_character_ids(caption) |
| 444 | proposal_base = ( |
| 445 | deepcopy(old_review.get("proposed_bank")) |
| 446 | if target_memory_id and isinstance(old_review.get("proposed_bank"), dict) |
| 447 | else deepcopy(bank) |
| 448 | ) |
| 449 | missing_character_ids = [ |
| 450 | memory_id |
| 451 | for memory_id in character_ids |
| 452 | if not isinstance(bank.get(memory_id), dict) |
| 453 | or not ( |
| 454 | bank[memory_id].get("local_image_path") |
| 455 | or bank[memory_id].get("image_path") |
| 456 | ) |
| 457 | ] |
| 458 | if target_memory_id == "PREVIOUS_SHOT": |
| 459 | selection_character_ids = [] |
| 460 | elif target_memory_id: |
| 461 | selection_character_ids = [target_memory_id] |
| 462 | else: |
| 463 | selection_character_ids = missing_character_ids or character_ids |
| 464 | if selection_character_ids: |
| 465 | try: |
| 466 | character_selection = selector.select_characters( |
| 467 | shot_id=shot_id, |
| 468 | caption=caption, |
| 469 | character_ids=selection_character_ids, |
| 470 | candidates=candidates, |
| 471 | rejected_candidate_indices=rejected, |
| 472 | ) |
| 473 | except Exception as exc: |
| 474 | raise RuntimeError( |
| 475 | f"memory character selection failed for shot {shot_id}: {exc}" |
| 476 | ) from exc |
| 477 | else: |
| 478 | character_selection = {"reasoning": "no character IDs", "selections": []} |
| 479 | selection_rows = character_selection.get("selections") |
| 480 | if not isinstance(selection_rows, list): |
| 481 | selection_rows = [] |
| 482 | character_selection["selections"] = selection_rows |
| 483 | selected_ids = { |
| 484 | str(item.get("character_id") or "") |
| 485 | for item in selection_rows |
| 486 | if isinstance(item, dict) |
| 487 | } |
| 488 | unselected_ids = [ |
| 489 | memory_id |
| 490 | for memory_id in selection_character_ids |
| 491 | if memory_id not in selected_ids |
| 492 | ] |
| 493 | if unselected_ids: |
| 494 | raise RuntimeError( |
| 495 | "memory character selection returned no frame for required IDs: " |
| 496 | + ", ".join(unselected_ids) |
| 497 | ) |
| 498 | proposal_dir = memory_root / "proposals" / f"shot_{shot_id:03d}_attempt_{attempt:02d}" |
| 499 | proposed = materialize_character_memories( |
| 500 | selections=character_selection, |
| 501 | candidates=candidates, |
| 502 | shot_id=shot_id, |
| 503 | video_path=video_path, |
| 504 | memory_dir=proposal_dir, |
| 505 | audio_extractor=audio_extractor, |
| 506 | ) |
| 507 | proposed_bank = merge_approved_memories(proposal_base, proposed) |
| 508 | changed_memory_ids = { |
| 509 | memory_id: record |
| 510 | for memory_id, record in proposed_bank.items() |
| 511 | if memory_id not in bank or record != bank[memory_id] |
| 512 | } |
| 513 | |
| 514 | next_shot = _next_planned_shot(work_dir, shot_id) |
| 515 | transition = False |
| 516 | transition_reasoning = "next shot unavailable; preserve continuity by default" |
| 517 | if next_shot is not None: |
| 518 | next_shot_id = int(next_shot["shot_id"]) |
| 519 | next_caption = str(next_shot.get("caption") or next_shot.get("summary") or "") |
| 520 | try: |
| 521 | decision = selector.decide_scene_transition( |
| 522 | previous_shot_id=shot_id, |
| 523 | previous_caption=caption, |
| 524 | next_shot_id=next_shot_id, |
| 525 | next_caption=next_caption, |
| 526 | ) |
| 527 | transition = bool(decision["scene_transition"]) |
| 528 | transition_reasoning = str(decision["reasoning"]) |
| 529 | except Exception as exc: |
| 530 | raise RuntimeError( |
| 531 | f"memory scene-transition selection failed for shot {shot_id}: {exc}" |
| 532 | ) from exc |
| 533 | |
| 534 | # The active mode always carries one representative continuity frame. |
| 535 | # Scene-transition analysis remains metadata and does not remove this slot. |
| 536 | previous = ( |
| 537 | deepcopy(old_review.get("previous_shot")) |
| 538 | if target_memory_id and target_memory_id != "PREVIOUS_SHOT" |
| 539 | and isinstance(old_review.get("previous_shot"), dict) |
| 540 | else None |
| 541 | ) |
| 542 | if candidates and ( |
| 543 | target_memory_id is None or target_memory_id == "PREVIOUS_SHOT" |
| 544 | ): |
| 545 | try: |
| 546 | representative_selection = selector.select_representative( |
| 547 | shot_id=shot_id, |
| 548 | caption=caption, |
| 549 | candidates=candidates, |
| 550 | rejected_candidate_indices=rejected, |
| 551 | ) |
| 552 | except Exception as exc: |
| 553 | raise RuntimeError( |
| 554 | f"memory representative selection failed for shot {shot_id}: {exc}" |
| 555 | ) from exc |
| 556 | by_index = {int(item["candidate_index"]): item for item in candidates} |
| 557 | representative_index = int(representative_selection["candidate_index"]) |
| 558 | if representative_index not in by_index: |
| 559 | raise RuntimeError( |
| 560 | "memory representative selection returned invalid candidate index " |
| 561 | f"{representative_index} for shot {shot_id}" |
| 562 | ) |
| 563 | representative_candidate = by_index[representative_index] |
| 564 | representative_image = proposal_dir / "PREVIOUS_SHOT.jpg" |
| 565 | representative_audio = proposal_dir / "PREVIOUS_SHOT.wav" |
| 566 | shutil.copyfile(str(representative_candidate["path"]), representative_image) |
| 567 | audio_extractor(video_path, representative_audio) |
| 568 | previous = { |
| 569 | "memory_id": "PREVIOUS_SHOT", |
| 570 | "kind": "previous_shot", |
| 571 | "candidate_index": int(representative_candidate["candidate_index"]), |
| 572 | "frame_index": int(representative_candidate["frame_index"]), |
| 573 | "timestamp_sec": float(representative_candidate["timestamp_sec"]), |
| 574 | "confidence": float(representative_selection["confidence"]), |
| 575 | "reasoning": str(representative_selection["reasoning"]), |
| 576 | "visual_status": "representative", |
| 577 | "source_shot_id": shot_id, |
| 578 | "audio_source_shot_id": shot_id, |
| 579 | "local_image_path": str(representative_image.resolve()), |
| 580 | "local_audio_path": str(representative_audio.resolve()), |
| 581 | } |
| 582 | if publisher is not None: |
| 583 | proposed_bank = publish_memory_records(proposed_bank, publisher=publisher) |
| 584 | changed_bank = { |
| 585 | memory_id: proposed_bank[memory_id] |
| 586 | for memory_id in changed_memory_ids |
| 587 | if memory_id in proposed_bank |
| 588 | } |
| 589 | if previous is not None: |
| 590 | previous = publish_memory_record( |
| 591 | previous, |
| 592 | prefix=f"shot_{shot_id:03d}_PREVIOUS_SHOT", |
| 593 | publisher=publisher, |
| 594 | ) |
| 595 | else: |
| 596 | changed_bank = changed_memory_ids |
| 597 | review = build_memory_review( |
| 598 | shot_id=shot_id, attempt=attempt, candidate_count=len(candidates), |
| 599 | bank=changed_bank, previous_shot=previous, review_id=review_id_factory(), |
| 600 | rejected_candidate_indices=sorted(rejected), history=history, updated_at=now()) |
| 601 | review["proposed_bank"] = proposed_bank |
| 602 | review["previous_shot"] = previous |
| 603 | review["scene_transition"] = transition |
| 604 | review["scene_transition_reasoning"] = transition_reasoning |
| 605 | review["selection_warnings"] = [] |
| 606 | review["candidate_paths"] = candidates |
| 607 | review["selection_mode"] = ( |
| 608 | "manual" if isinstance(selector, ManualMemorySelector) else "vlm" |
| 609 | ) |
| 610 | review["retained_memory_ids"] = ( |
| 611 | [] |
| 612 | if isinstance(selector, ManualMemorySelector) |
| 613 | else [ |
| 614 | str(item.get("memory_id") or "") |
| 615 | for item in review.get("selections", []) |
| 616 | if isinstance(item, dict) and item.get("memory_id") |
| 617 | ] |
| 618 | ) |
| 619 | # Re-read latest shot/state from disk before writing to avoid overwriting |
| 620 | # concurrent changes (e.g. user accept during a reselect VLM window). |
| 621 | try: |
| 622 | latest_shot = _load_shot_for_memory(shot_path, shot_id) |
| 623 | except ValueError: |
| 624 | latest_shot = shot |
| 625 | latest_state = _read_json(state_path, {}) |
| 626 | if isinstance(latest_shot, dict): |
| 627 | latest_shot["memory_review"] = review |
| 628 | _write_json(shot_path, latest_shot) |
| 629 | else: |
| 630 | shot["memory_review"] = review |
| 631 | _write_json(shot_path, shot) |
| 632 | if isinstance(latest_state, dict): |
| 633 | latest_state["stage"] = "awaiting_memory_review" |
| 634 | _write_json(state_path, latest_state) |
| 635 | else: |
| 636 | _write_json(state_path, state) |
| 637 | return review |
| 638 | |
| 639 | |
| 640 | def download_video(locator: str, target: Path) -> None: |
| 641 | target.parent.mkdir(parents=True, exist_ok=True) |
| 642 | source = Path(locator) |
| 643 | if source.is_file(): |
| 644 | shutil.copyfile(source, target) |
| 645 | return |
| 646 | request = urllib.request.Request( |
| 647 | locator, |
| 648 | headers={"Accept": "video/mp4,*/*", "User-Agent": "EchoMemoryAgent/1.0"}, |
| 649 | ) |
| 650 | with urllib.request.urlopen(request, timeout=300) as response: |
| 651 | with target.open("wb") as output: |
| 652 | shutil.copyfileobj(response, output) |
| 653 | if not target.is_file() or target.stat().st_size <= 0: |
| 654 | raise RuntimeError(f"downloaded empty video: {locator}") |
| 655 | |
| 656 | |
| 657 | def sample_video_frames(video: Path, output: Path, count: int) -> list[dict[str, Any]]: |
| 658 | probe = subprocess.run( |
| 659 | [ |
| 660 | _resolve_media_binary("ffprobe"), |
| 661 | "-v", "error", "-select_streams", "v:0", |
| 662 | "-show_entries", "stream=r_frame_rate,duration:format=duration", |
| 663 | "-of", "json", str(video), |
| 664 | ], |
| 665 | capture_output=True, |
| 666 | text=True, |
| 667 | check=True, |
| 668 | ) |
| 669 | metadata = json.loads(probe.stdout) |
| 670 | stream = (metadata.get("streams") or [{}])[0] |
| 671 | duration = float( |
| 672 | stream.get("duration") or (metadata.get("format") or {}).get("duration") or 0 |
| 673 | ) |
| 674 | numerator, _, denominator = str(stream.get("r_frame_rate") or "25/1").partition("/") |
| 675 | fps = float(numerator) / max(float(denominator or 1), 1.0) |
| 676 | if duration <= 0 or fps <= 0: |
| 677 | raise RuntimeError("invalid generated video metadata") |
| 678 | output.mkdir(parents=True, exist_ok=True) |
| 679 | rows: list[dict[str, Any]] = [] |
| 680 | for index in range(count): |
| 681 | timestamp = duration * (0.05 + 0.90 * index / max(count - 1, 1)) |
| 682 | image = output / f"candidate_{index:02d}.jpg" |
| 683 | result = subprocess.run( |
| 684 | [ |
| 685 | _resolve_media_binary("ffmpeg"), |
| 686 | "-loglevel", "error", "-y", "-ss", f"{timestamp:.6f}", |
| 687 | "-i", str(video), "-frames:v", "1", "-vf", "scale=640:-2", |
| 688 | "-q:v", "3", str(image), |
| 689 | ], |
| 690 | capture_output=True, |
| 691 | text=True, |
| 692 | ) |
| 693 | if result.returncode == 0 and image.is_file(): |
| 694 | rows.append({ |
| 695 | "candidate_index": index, |
| 696 | "frame_index": max(0, int(round(timestamp * fps))), |
| 697 | "timestamp_sec": round(timestamp, 6), |
| 698 | "path": str(image.resolve()), |
| 699 | }) |
| 700 | return rows |
| 701 | |
| 702 | |
| 703 | def extract_video_frame( |
| 704 | video: Path, |
| 705 | target: Path, |
| 706 | timestamp_sec: float, |
| 707 | ) -> tuple[float, int]: |
| 708 | """Extract one exact review frame and return its clamped time and index.""" |
| 709 | probe = subprocess.run( |
| 710 | [ |
| 711 | _resolve_media_binary("ffprobe"), |
| 712 | "-v", "error", "-select_streams", "v:0", |
| 713 | "-show_entries", "stream=r_frame_rate,duration:format=duration", |
| 714 | "-of", "json", str(video), |
| 715 | ], |
| 716 | capture_output=True, |
| 717 | text=True, |
| 718 | check=True, |
| 719 | ) |
| 720 | metadata = json.loads(probe.stdout) |
| 721 | stream = (metadata.get("streams") or [{}])[0] |
| 722 | duration = float( |
| 723 | stream.get("duration") or (metadata.get("format") or {}).get("duration") or 0 |
| 724 | ) |
| 725 | numerator, _, denominator = str(stream.get("r_frame_rate") or "25/1").partition("/") |
| 726 | fps = float(numerator) / max(float(denominator or 1), 1.0) |
| 727 | if duration <= 0 or fps <= 0: |
| 728 | raise RuntimeError("invalid generated video metadata") |
| 729 | timestamp = min(max(float(timestamp_sec), 0.0), max(duration - (0.5 / fps), 0.0)) |
| 730 | target.parent.mkdir(parents=True, exist_ok=True) |
| 731 | result = subprocess.run( |
| 732 | [ |
| 733 | _resolve_media_binary("ffmpeg"), |
| 734 | "-loglevel", "error", "-y", "-ss", f"{timestamp:.6f}", |
| 735 | "-i", str(video), "-frames:v", "1", "-vf", "scale=640:-2", |
| 736 | "-q:v", "3", str(target), |
| 737 | ], |
| 738 | capture_output=True, |
| 739 | text=True, |
| 740 | ) |
| 741 | if result.returncode != 0 or not target.is_file() or target.stat().st_size <= 0: |
| 742 | raise RuntimeError(f"failed to extract manual memory frame: {result.stderr[:1000]}") |
| 743 | return round(timestamp, 6), max(0, int(round(timestamp * fps))) |
| 744 | |
| 745 | |
| 746 | def select_manual_memory_frame( |
| 747 | *, |
| 748 | workspace: Path, |
| 749 | work_id: str, |
| 750 | shot_id: int, |
| 751 | review_id: str, |
| 752 | attempt: int, |
| 753 | memory_id: str, |
| 754 | timestamp_sec: float, |
| 755 | video_fetcher: Callable[[str, Path], None] = download_video, |
| 756 | frame_extractor: Callable[[Path, Path, float], tuple[float, int]] = extract_video_frame, |
| 757 | now: Callable[[], str] | None = None, |
| 758 | ) -> dict[str, Any]: |
| 759 | """Replace one VLM proposal image with a user-selected source-video frame.""" |
| 760 | with _MEMORY_REVIEW_LOCK: |
| 761 | work_dir = workspace / "director" / "works" / work_id |
| 762 | shot_path = work_dir / "shots" / f"shot_{shot_id:03d}.json" |
| 763 | shot = _load_shot_for_memory(shot_path, shot_id) |
| 764 | review = shot.get("memory_review") |
| 765 | if not isinstance(review, dict): |
| 766 | raise ValueError(f"shot {shot_id} has no memory review") |
| 767 | if ( |
| 768 | str(review.get("review_id") or "") != review_id |
| 769 | or int(review.get("attempt") or 0) != int(attempt) |
| 770 | ): |
| 771 | raise MemoryReviewConflict("stale memory review attempt") |
| 772 | status = str(review.get("status") or "") |
| 773 | if status not in {"awaiting_review", "error"}: |
| 774 | raise MemoryReviewConflict( |
| 775 | f"memory frame cannot be selected from {status}" |
| 776 | ) |
| 777 | selections = [ |
| 778 | item for item in review.get("selections", []) if isinstance(item, dict) |
| 779 | ] |
| 780 | if not any(str(item.get("memory_id") or "") == memory_id for item in selections): |
| 781 | raise MemoryReviewConflict(f"memory {memory_id} is not in this review") |
| 782 | |
| 783 | video_path = work_dir / "memory" / "videos" / f"shot_{shot_id:03d}.mp4" |
| 784 | if not video_path.is_file(): |
| 785 | artifact = str( |
| 786 | shot.get("artifact_url") or shot.get("artifact_path") or "" |
| 787 | ).strip() |
| 788 | if not artifact: |
| 789 | raise ValueError(f"shot {shot_id} has no video artifact") |
| 790 | video_fetcher(artifact, video_path) |
| 791 | |
| 792 | safe_memory_id = re.sub(r"[^A-Za-z0-9._-]+", "_", memory_id).strip("._") |
| 793 | safe_memory_id = safe_memory_id or "memory" |
| 794 | image_path = ( |
| 795 | work_dir |
| 796 | / "memory" |
| 797 | / "proposals" |
| 798 | / f"shot_{shot_id:03d}_attempt_{attempt:02d}" |
| 799 | / f"{safe_memory_id}_manual.jpg" |
| 800 | ) |
| 801 | selected_time, frame_index = frame_extractor( |
| 802 | video_path, image_path, float(timestamp_sec) |
| 803 | ) |
| 804 | |
| 805 | if memory_id == "PREVIOUS_SHOT": |
| 806 | source = review.get("previous_shot") |
| 807 | kind = "previous_shot" |
| 808 | else: |
| 809 | proposed_bank = review.get("proposed_bank") |
| 810 | source = proposed_bank.get(memory_id) if isinstance(proposed_bank, dict) else None |
| 811 | kind = "character" |
| 812 | if not isinstance(source, dict): |
| 813 | raise MemoryReviewConflict(f"memory {memory_id} has no proposal record") |
| 814 | |
| 815 | replacement = deepcopy(source) |
| 816 | for key in ("local_image_path", "image_path", "image_url", "image"): |
| 817 | replacement.pop(key, None) |
| 818 | replacement.update( |
| 819 | { |
| 820 | "memory_id": memory_id, |
| 821 | "kind": kind, |
| 822 | "candidate_index": -1, |
| 823 | "frame_index": frame_index, |
| 824 | "timestamp_sec": selected_time, |
| 825 | "confidence": 1.0, |
| 826 | "reasoning": "Selected manually from the source video.", |
| 827 | "visual_status": ( |
| 828 | "representative" if kind == "previous_shot" else "confirmed" |
| 829 | ), |
| 830 | "source_shot_id": shot_id, |
| 831 | "local_image_path": str(image_path.resolve()), |
| 832 | } |
| 833 | ) |
| 834 | if memory_id == "PREVIOUS_SHOT": |
| 835 | review["previous_shot"] = replacement |
| 836 | else: |
| 837 | proposed_bank = review.get("proposed_bank") |
| 838 | if not isinstance(proposed_bank, dict): |
| 839 | proposed_bank = {} |
| 840 | review["proposed_bank"] = proposed_bank |
| 841 | proposed_bank[memory_id] = replacement |
| 842 | |
| 843 | review["selections"] = [ |
| 844 | _review_selection(memory_id, replacement, kind=kind) |
| 845 | if str(item.get("memory_id") or "") == memory_id |
| 846 | else item |
| 847 | for item in selections |
| 848 | ] |
| 849 | selected_ids = review.setdefault("manual_selected_ids", []) |
| 850 | if not isinstance(selected_ids, list): |
| 851 | selected_ids = [] |
| 852 | review["manual_selected_ids"] = selected_ids |
| 853 | if memory_id not in selected_ids: |
| 854 | selected_ids.append(memory_id) |
| 855 | retained_ids = review.setdefault("retained_memory_ids", []) |
| 856 | if not isinstance(retained_ids, list): |
| 857 | retained_ids = [] |
| 858 | review["retained_memory_ids"] = retained_ids |
| 859 | if memory_id not in retained_ids: |
| 860 | retained_ids.append(memory_id) |
| 861 | review["status"] = "awaiting_review" |
| 862 | review["error"] = None |
| 863 | review["updated_at"] = ( |
| 864 | now() |
| 865 | if now is not None |
| 866 | else datetime.now(timezone.utc).isoformat(timespec="seconds").replace( |
| 867 | "+00:00", "Z" |
| 868 | ) |
| 869 | ) |
| 870 | shot["memory_review"] = review |
| 871 | _write_json(shot_path, shot) |
| 872 | return review |
| 873 | |
| 874 | |
| 875 | def extract_video_audio(video: Path, target: Path) -> None: |
| 876 | target.parent.mkdir(parents=True, exist_ok=True) |
| 877 | result = subprocess.run( |
| 878 | [ |
| 879 | _resolve_media_binary("ffmpeg"), |
| 880 | "-loglevel", "error", "-y", "-i", str(video), "-vn", |
| 881 | "-acodec", "pcm_s16le", "-ar", "48000", str(target), |
| 882 | ], |
| 883 | capture_output=True, |
| 884 | text=True, |
| 885 | ) |
| 886 | if result.returncode != 0 or not target.is_file() or target.stat().st_size <= 44: |
| 887 | raise RuntimeError(f"failed to extract memory audio: {result.stderr[:1000]}") |
| 888 | |
| 889 | |
| 890 | def run_memory_review_from_config( |
| 891 | *, |
| 892 | workspace: Path, |
| 893 | work_id: str, |
| 894 | shot_id: int, |
| 895 | target_memory_id: str | None = None, |
| 896 | api_base: str | None = None, |
| 897 | api_key: str | None = None, |
| 898 | vlm_model: str | None = None, |
| 899 | candidate_count: int = 24, |
| 900 | selection_mode: str = "vlm", |
| 901 | ) -> dict[str, Any]: |
| 902 | """Production callback entrypoint using an explicitly resolved VLM route.""" |
| 903 | resolved_api_base = (api_base or "").strip() |
| 904 | resolved_api_key = (api_key or "").strip() |
| 905 | resolved_model = (vlm_model or "").strip() |
| 906 | if selection_mode not in {"manual", "vlm"}: |
| 907 | raise ValueError("selection_mode must be manual or vlm") |
| 908 | if selection_mode == "vlm" and not all( |
| 909 | (resolved_api_base, resolved_api_key, resolved_model) |
| 910 | ): |
| 911 | raise RuntimeError( |
| 912 | "memory review requires tools.memoryReview.provider/model and " |
| 913 | "the referenced provider apiKey/apiBase" |
| 914 | ) |
| 915 | with _MEMORY_REVIEW_LOCK: |
| 916 | # JD's OpenAI-compatible Qwen VLM route rejects prompts containing |
| 917 | # more than 16 images. Keep the richer 24-frame strip for manual |
| 918 | # scrubbing, but cap automatic review before frames are sampled and |
| 919 | # embedded into the VLM prompt. |
| 920 | effective_candidate_count = ( |
| 921 | min(candidate_count, _MAX_VLM_PROMPT_IMAGES) |
| 922 | if selection_mode == "vlm" |
| 923 | else candidate_count |
| 924 | ) |
| 925 | selector = ( |
| 926 | ManualMemorySelector() |
| 927 | if selection_mode == "manual" |
| 928 | else MemoryVlmSelector( |
| 929 | api_base=resolved_api_base, |
| 930 | api_key=resolved_api_key, |
| 931 | model=resolved_model, |
| 932 | ) |
| 933 | ) |
| 934 | review = prepare_memory_review( |
| 935 | workspace=workspace, |
| 936 | work_id=work_id, |
| 937 | shot_id=shot_id, |
| 938 | selector=selector, |
| 939 | video_fetcher=download_video, |
| 940 | frame_sampler=sample_video_frames, |
| 941 | audio_extractor=extract_video_audio, |
| 942 | review_id_factory=lambda: f"memory-review-{uuid.uuid4().hex}", |
| 943 | now=lambda: datetime.now(timezone.utc).isoformat( |
| 944 | timespec="seconds" |
| 945 | ).replace("+00:00", "Z"), |
| 946 | candidate_count=effective_candidate_count, |
| 947 | target_memory_id=target_memory_id, |
| 948 | publisher=( |
| 949 | configured_file_publisher(work_id) |
| 950 | if selection_mode == "vlm" |
| 951 | else None |
| 952 | ), |
| 953 | ) |
| 954 | if selection_mode == "manual": |
| 955 | review["selection_mode"] = selection_mode |
| 956 | review["required_memory_ids"] = [ |
| 957 | str(item.get("memory_id") or "") |
| 958 | for item in review.get("selections", []) |
| 959 | if isinstance(item, dict) and item.get("memory_id") |
| 960 | ] |
| 961 | review["manual_selected_ids"] = [] |
| 962 | review["retained_memory_ids"] = [] |
| 963 | shot_path = ( |
| 964 | workspace / "director" / "works" / work_id / "shots" |
| 965 | / f"shot_{shot_id:03d}.json" |
| 966 | ) |
| 967 | shot = _load_shot_for_memory(shot_path, shot_id) |
| 968 | shot["memory_review"] = review |
| 969 | _write_json(shot_path, shot) |
| 970 | return review |
| 971 |