ingest: index on dedup too — files without rows must repair

An interrupted mirror leaves bytes on disk with no library or worklog row;
the dedup branch skipped indexing, so re-ingesting repaired nothing. The
library upserts by path; the worklog now upserts by output. Recovery is a
re-POST of the same bytes.

middleware/studio_home.py

@@ -728,10 +728,14 @@ def _index_upload(org: str, root: Path, rel: str, *, design, kind, role, node) -
     else:
         ex.update(design=design, kind=kind, role=role, updatedAt=now_iso)
     _save_library(root, lib)
-    worklog.record(org, kind="render", prompt="", refs=[], uploads=[], parents=[],
-                   output_prefix=rel, node=node, status="done",
-                   lane=("gpu" if node not in ("upload", "studio pod") else "local"),
-                   pid="up-" + uuid.uuid4().hex[:12])
+    # The worklog upserts by output too: re-ingesting identical bytes (an
+    # interrupted mirror, a restored store) must repair a missing row, never
+    # stack a second one.
+    if not worklog.has_output(org, rel):
+        worklog.record(org, kind="render", prompt="", refs=[], uploads=[], parents=[],
+                       output_prefix=rel, node=node, status="done",
+                       lane=("gpu" if node not in ("upload", "studio pod") else "local"),
+                       pid="up-" + uuid.uuid4().hex[:12])
 
 
 def add_studio_home_routes(routes: web.RouteTableDef, server) -> None:
@@ -1023,6 +1027,14 @@ def add_studio_home_routes(routes: web.RouteTableDef, server) -> None:
             return web.json_response({"error": "invalid path"}, status=400)
         rel = f"{sub}/{name}"
         if dest.is_file() and dest.stat().st_size == len(data):
+            # Identical bytes: skip the write but STILL index — files can exist
+            # without rows (an interrupted mirror, a restored store), and the
+            # index upsert is idempotent by path. Recovery depends on this.
+            _index_upload(org, out_root, rel,
+                          design=request.query.get("design") or None,
+                          kind=request.query.get("kind") or "render",
+                          role=request.query.get("role") or None,
+                          node=request.query.get("node") or "upload")
             return web.json_response({"ok": True, "existed": True, "path": rel})
         dest.parent.mkdir(parents=True, exist_ok=True)
         tmp = dest.with_name(dest.name + ".tmp")

middleware/worklog.py

@@ -59,6 +59,13 @@ def _save(org: str, rows: list[dict]) -> None:
     tmp.replace(p)
 
 
+def has_output(org: str, output_prefix: str) -> bool:
+    """Whether any row already carries this landed output — the ingest upsert
+    key (re-ingesting identical bytes must repair, never duplicate)."""
+    return any(r.get("output_prefix") == output_prefix or r.get("output") == output_prefix
+               for r in load(org))
+
+
 def record(org: str, *, kind: str, prompt: str, refs: list, uploads: list,
            parents: list, output_prefix: str, params: dict | None = None,
            pid: str | None = None, lane: str | None = None, node: str | None = None,