feat: publish asset-manifest v1 from alternative-song sidecars

app/services/alternative_songs.py

@@ -15,6 +15,7 @@
 import uuid
 from typing import Any, Callable
 
+from services.asset_manifest import SCHEMA_NAME, publish_generation_sidecar, read_asset_manifest
 from services.mix_concat import concatenate_multi_clip_videos, probe_duration_seconds
 from services.output_result_kind import classify_output_result_kind
 from services.video_editor import probe_audio
@@ -45,15 +46,66 @@ def load_sidecar(video_path: str) -> dict[str, Any]:
     return payload
 
 
-def save_sidecar(video_path: str, sidecar: dict[str, Any]) -> None:
-    path = sidecar_path(video_path)
-    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
-    tmp = path + ".tmp"
-    with open(tmp, "w", encoding="utf-8") as handle:
-        json.dump(sidecar, handle, indent=2, ensure_ascii=False)
-        handle.flush()
-        os.fsync(handle.fileno())
-    os.replace(tmp, path)
+def _clean_text(value: Any) -> str | None:
+    text = str(value or "").strip()
+    return text or None
+
+
+def _v1_origin(document: Any) -> dict[str, Any] | None:
+    if not isinstance(document, dict) or document.get("schema") != SCHEMA_NAME:
+        return None
+    origin = document.get("origin")
+    return origin if isinstance(origin, dict) else None
+
+
+def _has_director_pipeline(payload: dict[str, Any]) -> bool:
+    params = payload.get("params") if isinstance(payload.get("params"), dict) else {}
+    return any(
+        _clean_text(value)
+        for value in (
+            payload.get("pipeline_id"),
+            payload.get("director_pipeline_id"),
+            params.get("director_pipeline_id"),
+            params.get("_director_pipeline_id"),
+        )
+    )
+
+
+def save_sidecar(video_path: str, sidecar: dict[str, Any], *, tool: str | None = None) -> None:
+    """Publish gallery sidecar fields through the v1 asset-manifest writer.
+
+    An explicit ``tool`` wins. Otherwise a canonical ``origin.tool`` on the
+    payload or existing v1 file is preserved so series-assembly/director/studio
+    are not rewritten. If only a director pipeline id is present, tool is
+    omitted so publish can attribute director. Actor is never invented as
+    ``user``; missing workspace stays unset instead of ``default``.
+    """
+    payload = sidecar if isinstance(sidecar, dict) else {}
+    payload_origin = _v1_origin(payload)
+    disk_origin = None
+    existing = load_sidecar(video_path)
+    if _v1_origin(existing) is not None:
+        manifest = read_asset_manifest(video_path)
+        origin = None if manifest is None else manifest.get("origin")
+        disk_origin = origin if isinstance(origin, dict) else None
+    resolved_tool = (
+        _clean_text(tool)
+        or _clean_text((payload_origin or {}).get("tool"))
+        or _clean_text((disk_origin or {}).get("tool"))
+    )
+    if resolved_tool is None and _has_director_pipeline(payload):
+        resolved_tool = None
+    workspace_id = (
+        _clean_text(payload.get("workspace"))
+        or _clean_text((payload_origin or {}).get("workspace_id"))
+        or _clean_text((disk_origin or {}).get("workspace_id"))
+    )
+    publish_generation_sidecar(
+        video_path,
+        payload,
+        workspace_id=workspace_id,
+        tool=resolved_tool,
+    )
 
 
 def _song_list(sidecar: dict[str, Any]) -> list[dict[str, Any]]:
@@ -323,7 +375,7 @@ def write_mounted_sidecar(
         "created_at": time.time(),
         "parent_output": parent_name,
     }
-    save_sidecar(output_path, payload)
+    save_sidecar(output_path, payload, tool="alternative-songs")
     song["status"] = "mounted"
     song["mounted_output"] = os.path.basename(output_path)
     song["extra_clip_count"] = extra_count
@@ -335,7 +387,7 @@ def write_mounted_sidecar(
     )
     if classify != "music_video":
         payload["params"]["result_kind"] = "music_video"
-        save_sidecar(output_path, payload)
+        save_sidecar(output_path, payload, tool="alternative-songs")
 
 
 def remount_clips(

tests/test_alternative_songs.py

@@ -1,3 +1,4 @@
+import json
 import os
 import random
 import sys
@@ -20,8 +21,15 @@
     remove_song,
     resolve_remount_sources,
     save_sidecar,
+    sidecar_path,
     source_clip_names,
     unique_mounted_name,
+    write_mounted_sidecar,
+)
+from services.asset_manifest import (  # noqa: E402
+    SCHEMA_NAME,
+    publish_generation_sidecar,
+    read_asset_manifest,
 )
 from services.output_result_kind import classify_output_result_kind  # noqa: E402
 
@@ -149,5 +157,109 @@ def test_sidecar_roundtrip_keeps_songs(self):
             self.assertTrue(songs[0]["id"].startswith("song-"))
 
 
+class AlternativeSongSidecarTests(unittest.TestCase):
+    def test_save_sidecar_publishes_v1_and_roundtrips_songs(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            video = os.path.join(tmp, "clip.mp4")
+            Path(video).write_bytes(b"not-a-real-mp4")
+            sidecar = load_sidecar(video)
+            attach_song(sidecar, audio_name="en.mp3", duration_seconds=9)
+            sidecar["params"]["api_key"] = "secret-key"
+            save_sidecar(video, sidecar)
+            raw = json.loads(Path(sidecar_path(video)).read_text(encoding="utf-8"))
+            loaded = read_asset_manifest(video)
+            text = Path(sidecar_path(video)).read_text(encoding="utf-8")
+            self.assertEqual(raw["schema"], SCHEMA_NAME)
+            self.assertIsNotNone(loaded)
+            self.assertEqual(loaded["origin"]["actor"], "unknown")
+            self.assertEqual(raw["params"]["alternative_songs"][0]["audio_name"], "en.mp3")
+            self.assertTrue(raw["params"]["alternative_songs"][0]["id"].startswith("song-"))
+            self.assertNotIn("secret-key", text)
+            self.assertNotEqual(loaded["origin"].get("actor"), "user")
+
+    def test_save_sidecar_keeps_existing_v1_asset_id(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            video = os.path.join(tmp, "clip.mp4")
+            Path(video).write_bytes(b"not-a-real-mp4")
+            sidecar = load_sidecar(video)
+            attach_song(sidecar, audio_name="en.mp3", duration_seconds=9)
+            save_sidecar(video, sidecar)
+            first = read_asset_manifest(video)
+            self.assertIsNotNone(first)
+            asset_id = first["asset"]["id"]
+            mutated = load_sidecar(video)
+            attach_song(mutated, audio_name="pt.mp3", duration_seconds=8)
+            save_sidecar(video, mutated)
+            again = read_asset_manifest(video)
+            raw = json.loads(Path(sidecar_path(video)).read_text(encoding="utf-8"))
+            self.assertIsNotNone(again)
+            self.assertEqual(again["asset"]["id"], asset_id)
+            self.assertEqual(len(raw["params"]["alternative_songs"]), 2)
+
+    def test_save_sidecar_keeps_series_assembly_origin_tool(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            video = os.path.join(tmp, "episode.mp4")
+            Path(video).write_bytes(b"video")
+            publish_generation_sidecar(
+                video,
+                {
+                    "generation_mode": "video",
+                    "params": {"pipeline_type": "series_episode"},
+                },
+                workspace_id="lab",
+                tool="series-assembly",
+            )
+            before = read_asset_manifest(video)
+            self.assertIsNotNone(before)
+            sidecar = load_sidecar(video)
+            attach_song(sidecar, audio_name="en.mp3", duration_seconds=9)
+            save_sidecar(video, sidecar)
+            loaded = read_asset_manifest(video)
+            raw = json.loads(Path(sidecar_path(video)).read_text(encoding="utf-8"))
+            self.assertIsNotNone(loaded)
+            self.assertEqual(loaded["origin"]["tool"], "series-assembly")
+            self.assertEqual(loaded["asset"]["id"], before["asset"]["id"])
+            self.assertEqual(raw["params"]["alternative_songs"][0]["audio_name"], "en.mp3")
+
+    def test_write_mounted_sidecar_publishes_alternative_songs_origin(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            output = os.path.join(tmp, "clip_en_mv.mp4")
+            Path(output).write_bytes(b"video")
+            parent_sidecar = {
+                "params": {
+                    "pipeline_type": "music_video",
+                    "model_type": "ffmpeg_remount",
+                    "resolution": "720p",
+                },
+            }
+            song = attach_song({"params": {}}, audio_name="en.mp3", duration_seconds=9)
+            planned = [{
+                "name": "shot.mp4",
+                "path": os.path.join(tmp, "shot.mp4"),
+                "duration": 4,
+                "used": 4,
+                "extra": False,
+            }]
+            write_mounted_sidecar(
+                output_path=output,
+                parent_name="clip.mp4",
+                parent_sidecar=parent_sidecar,
+                song=song,
+                planned=planned,
+                job_id="alt-song-1",
+                workspace="night-shift",
+            )
+            raw = json.loads(Path(sidecar_path(output)).read_text(encoding="utf-8"))
+            loaded = read_asset_manifest(output, workspace_id="night-shift")
+            self.assertEqual(raw["schema"], SCHEMA_NAME)
+            self.assertEqual(raw["parent_output"], "clip.mp4")
+            self.assertEqual(raw["params"]["parent_output"], "clip.mp4")
+            self.assertEqual(song["status"], "mounted")
+            self.assertIsNotNone(loaded)
+            self.assertEqual(loaded["origin"]["tool"], "alternative-songs")
+            self.assertEqual(loaded["origin"]["workspace_id"], "night-shift")
+            self.assertEqual(loaded["origin"]["actor"], "unknown")
+
+
 if __name__ == "__main__":
     unittest.main()