merge: W2 ingest

# Conflicts:
#	apps/desktop/src/preload.test.ts

packages/adapters/src/media-ingest-analyze.test.ts

@@ -0,0 +1,154 @@
+import type { MediaCapabilities } from "@rakazo/contracts";
+import { defaultTeaserWeights } from "@rakazo/media";
+import { describe, expect, it } from "vitest";
+import type { AnalysisBundle } from "./media-ingest-analyze.js";
+import {
+  analysisModelVersion,
+  buildSegments,
+  isTransientInferenceError,
+  keyframeTimes,
+} from "./media-ingest-analyze.js";
+import { MediaSidecarError } from "./media-sidecar.js";
+
+const empty: AnalysisBundle = {
+  scenes: null,
+  transcript: null,
+  faceSpans: null,
+  embeddings: null,
+  audio: null,
+  saliency: null,
+};
+const weights = defaultTeaserWeights("teaser");
+
+describe("keyframeTimes", () => {
+  it("samples once per second, a still once, and spreads long clips over 60 frames", () => {
+    expect(keyframeTimes(null, true)).toEqual([0]);
+    expect(keyframeTimes(800, false)).toEqual([0]);
+    expect(keyframeTimes(4000, false)).toEqual([0, 1000, 2000, 3000]);
+    const long = keyframeTimes(600_000, false);
+    expect(long).toHaveLength(60);
+    expect(long[1]).toBe(10_000);
+  });
+});
+
+describe("buildSegments", () => {
+  it("uses detected shots and scores each from the analyses that exist", () => {
+    const unit = (x: number) => Float32Array.from([x, Math.sqrt(1 - x * x)]);
+    const segments = buildSegments({
+      durationMs: 4000,
+      still: false,
+      width: 1920,
+      height: 1080,
+      weights,
+      bundle: {
+        ...empty,
+        scenes: {
+          shots: [
+            { startMs: 0, endMs: 2000, score: 1 },
+            { startMs: 2000, endMs: 4000, score: 1 },
+          ],
+          scenes: [{ startMs: 0, endMs: 4000 }],
+        },
+        transcript: {
+          language: "en",
+          words: [
+            { startMs: 2100, endMs: 2400, text: "Wait", confidence: 0.9 },
+            { startMs: 2500, endMs: 2800, text: "what?", confidence: 0.9 },
+          ],
+        },
+        faceSpans: [{ startMs: 0, endMs: 1000 }],
+        embeddings: {
+          timesMs: [0, 1000, 2000, 3000],
+          vectors: [unit(1), unit(1), unit(0), unit(0.1)],
+        },
+        audio: {
+          beatsMs: [0, 2100],
+          energy: [
+            { tMs: 500, value: 0.2 },
+            { tMs: 2500, value: 0.8 },
+          ],
+        },
+        saliency: null,
+      },
+    });
+    expect(segments.map((segment) => [segment.kind, segment.startMs, segment.endMs])).toEqual([
+      ["shot", 0, 2000],
+      ["shot", 2000, 4000],
+    ]);
+    const [first, second] = segments;
+    expect(first?.scores?.faces).toBe(0.5);
+    expect(second?.scores?.faces).toBe(0);
+    expect(first?.scores?.sharpness).toBe(1);
+    expect(second?.scores?.audioPeak).toBe(1);
+    expect(first?.scores?.audioPeak).toBe(0.25);
+    expect(first?.scores?.beatProximity).toBe(1);
+    expect(second?.transcriptExcerpt).toBe("Wait what?");
+    expect(first?.transcriptExcerpt).toBeNull();
+    expect(first?.scores?.novelty).toBe(0.5);
+    expect(second?.scores?.novelty).toBeGreaterThan(0.5);
+    expect(first?.scores?.aesthetics).toBeUndefined();
+    for (const segment of segments) {
+      expect(segment.teaserScore).toBeGreaterThanOrEqual(0);
+      expect(segment.teaserScore).toBeLessThanOrEqual(1);
+    }
+  });
+
+  it("falls back to 3 s windows without shots and to one span for a still", () => {
+    const windows = buildSegments({
+      durationMs: 7000,
+      still: false,
+      width: null,
+      height: null,
+      weights,
+      bundle: empty,
+    });
+    expect(windows.map((segment) => [segment.kind, segment.startMs, segment.endMs])).toEqual([
+      ["scene", 0, 3000],
+      ["scene", 3000, 6000],
+      ["scene", 6000, 7000],
+    ]);
+    expect(windows[0]?.scores).toEqual({});
+    const still = buildSegments({
+      durationMs: null,
+      still: true,
+      width: 640,
+      height: 480,
+      weights,
+      bundle: { ...empty, faceSpans: [{ startMs: 0, endMs: 1 }] },
+    });
+    expect(still).toHaveLength(1);
+    expect(still[0]?.scores?.faces).toBe(1);
+  });
+});
+
+describe("analysis versions", () => {
+  it("derives the version from the provider id and the operation's model", () => {
+    const capabilities = {
+      configured: true,
+      accelerator: "cpu",
+      hardwareEncoders: [],
+      operations: {
+        embedFrames: {
+          available: true,
+          experimental: false,
+          model: "siglip-b16",
+          license: "Apache-2.0",
+          commercialUse: true,
+        },
+      },
+    } as unknown as MediaCapabilities;
+    expect(analysisModelVersion("sidecar", capabilities, "embeddings")).toBe("sidecar:siglip-b16");
+    expect(analysisModelVersion("sidecar", capabilities, "scenes")).toBe("sidecar:default");
+    expect(analysisModelVersion("emulator", null, "faces")).toBe("emulator:default");
+  });
+
+  it("treats an unreachable or saturated sidecar as transient", () => {
+    const error = (code: "unreachable" | "busy" | "timeout" | "internal") =>
+      new MediaSidecarError(code, "x", null, "detectScenes");
+    expect(isTransientInferenceError(error("unreachable"))).toBe(true);
+    expect(isTransientInferenceError(error("busy"))).toBe(true);
+    expect(isTransientInferenceError(error("timeout"))).toBe(true);
+    expect(isTransientInferenceError(error("internal"))).toBe(false);
+    expect(isTransientInferenceError(new Error("unreachable"))).toBe(false);
+  });
+});

packages/adapters/src/media-ingest-analyze.ts

@@ -0,0 +1,692 @@
+import { readFile, rm } from "node:fs/promises";
+import type {
+  AudioFeaturesResult,
+  DetectScenesResult,
+  MediaInferenceProvider,
+  SaliencyResult,
+  TranscribeResult,
+  TranscriptWord,
+} from "@rakazo/adapter-kit";
+import { UnsupportedMediaOperationError } from "@rakazo/adapter-kit";
+import type {
+  MediaAnalysisKind,
+  MediaCapabilities,
+  MediaInferenceOperation,
+  TeaserFeature,
+  TeaserWeights,
+} from "@rakazo/contracts";
+import {
+  BoxKeyframeSchema,
+  MediaAnalysisKindSchema,
+  TEASER_LEARNING_MIN_SAMPLES,
+} from "@rakazo/contracts";
+import type { NewMediaFaceTrack, NewMediaSegment, Prisma } from "@rakazo/db";
+import {
+  cosineSimilarity,
+  decodeVectors,
+  defaultTeaserWeights,
+  encodeVectors,
+  meanVector,
+  scoreSegment,
+  speechHookScore,
+} from "@rakazo/media";
+import { assertInside, resolveInside } from "@rakazo/media/node";
+import { z } from "zod";
+import type { MediaServiceDeps } from "./media-context.js";
+import { assetDir, writeFileAtomic } from "./media-ingest-files.js";
+import { MediaSidecarError } from "./media-sidecar.js";
+
+/*
+ * Sidecar analyses for one asset and the scored segments built from them.
+ *
+ * Biometric rule: face identity embeddings are stored only in
+ * MediaFaceTrack.embedding. The "faces" analysis row holds track spans and
+ * counts, never an embedding; box keyframes live in a JSON file per track.
+ *
+ * Each analysis is stored under modelVersion "<provider id>:<model>", so a
+ * model change adds rows instead of overwriting earlier results. An operation
+ * the provider reports as unsupported is skipped and records nothing.
+ */
+
+/** Version of the segment pipeline written to MediaAsset.analysisVersion. */
+export const MEDIA_ANALYSIS_VERSION = 1;
+
+const KEYFRAME_LIMIT = 60;
+const WINDOW_MS = 3000;
+const STILL_SEGMENT_MS = 3000;
+const FACE_SAMPLE_FPS = 2;
+const CLASSIFIER_SAMPLE_FPS = 2;
+const SAME_FACE_SIMILARITY = 0.6;
+const EXCERPT_LIMIT = 500;
+
+const ANALYSIS_OPERATION: Record<MediaAnalysisKind, MediaInferenceOperation> = {
+  scenes: "detectScenes",
+  transcript: "transcribe",
+  faces: "trackFaces",
+  embeddings: "embedFrames",
+  classifier: "classifyContent",
+  audio: "audioFeatures",
+  saliency: "saliency",
+};
+
+/** Kinds that need a time axis and audio; stills skip them. */
+const VIDEO_ONLY: ReadonlySet<MediaAnalysisKind> = new Set(["scenes", "transcript", "audio"]);
+
+/** Keyframe times: one per second, at most 60 spread evenly; a still is sampled once. */
+export function keyframeTimes(durationMs: number | null, still: boolean): number[] {
+  if (still || !durationMs || durationMs <= 0) return [0];
+  const seconds = Math.floor(durationMs / 1000);
+  if (seconds <= 1) return [0];
+  const count = Math.min(KEYFRAME_LIMIT, seconds);
+  const step = count === seconds ? 1000 : durationMs / count;
+  return Array.from({ length: count }, (_, index) => Math.round(index * step));
+}
+
+// ---------------------------------------------------------------------------
+// Stored analysis shapes (written here, read back for partial re-runs)
+// ---------------------------------------------------------------------------
+
+const Span = z.object({ startMs: z.number(), endMs: z.number() });
+const ScenesData = z.object({
+  shots: z.array(Span.extend({ score: z.number() })),
+  scenes: z.array(Span),
+});
+const TranscriptData = z.object({
+  language: z.string().nullable(),
+  words: z.array(Span.extend({ text: z.string(), confidence: z.number() })),
+});
+const FacesData = z.object({
+  sampleFps: z.number(),
+  trackCount: z.number(),
+  tracks: z.array(Span.extend({ keyframeCount: z.number() })),
+});
+const EmbeddingsData = z.object({ dim: z.number().int().positive(), timesMs: z.array(z.number()) });
+const AudioData = z.object({
+  beatsMs: z.array(z.number()),
+  energy: z.array(z.object({ tMs: z.number(), value: z.number() })),
+});
+const SaliencyData = z.object({
+  points: z.array(
+    z.object({ tMs: z.number(), cx: z.number(), cy: z.number(), strength: z.number() }),
+  ),
+});
+
+/** Box keyframes of one face track as written to `faces/<track>.json`. */
+export const FaceKeyframesFileSchema = z.object({
+  sampleFps: z.number().positive(),
+  keyframes: z.array(BoxKeyframeSchema.extend({ confidence: z.number() })),
+});
+export type FaceKeyframesFile = z.infer<typeof FaceKeyframesFileSchema>;
+
+/** Read a face track's keyframes file (path relative to the media root). */
+export async function readFaceKeyframes(
+  mediaRoot: string,
+  keyframesPath: string,
+): Promise<FaceKeyframesFile> {
+  const file = await resolveInside(mediaRoot, keyframesPath);
+  return FaceKeyframesFileSchema.parse(JSON.parse(await readFile(file, "utf8")));
+}
+
+/** Everything segment scoring reads; null where the analysis is missing or unsupported. */
+export interface AnalysisBundle {
+  scenes: DetectScenesResult | null;
+  transcript: Pick<TranscribeResult, "language" | "words"> | null;
+  faceSpans: Array<{ startMs: number; endMs: number }> | null;
+  embeddings: { timesMs: number[]; vectors: Float32Array[] } | null;
+  audio: Pick<AudioFeaturesResult, "beatsMs" | "energy"> | null;
+  saliency: SaliencyResult | null;
+}
+
+const EMPTY_BUNDLE: AnalysisBundle = {
+  scenes: null,
+  transcript: null,
+  faceSpans: null,
+  embeddings: null,
+  audio: null,
+  saliency: null,
+};
+
+// ---------------------------------------------------------------------------
+// Segments
+// ---------------------------------------------------------------------------
+
+const round3 = (value: number) => Math.round(Math.min(1, Math.max(0, value)) * 1000) / 1000;
+
+function inSpan(tMs: number, span: { startMs: number; endMs: number }): boolean {
+  return tMs >= span.startMs && tMs < span.endMs;
+}
+
+/** Union length of `spans` clipped to `window`. */
+function coveredMs(
+  spans: Array<{ startMs: number; endMs: number }>,
+  window: { startMs: number; endMs: number },
+): number {
+  const clipped = spans
+    .map((span) => ({
+      startMs: Math.max(span.startMs, window.startMs),
+      endMs: Math.min(span.endMs, window.endMs),
+    }))
+    .filter((span) => span.endMs > span.startMs)
+    .sort((a, b) => a.startMs - b.startMs);
+  let total = 0;
+  let end = Number.NEGATIVE_INFINITY;
+  for (const span of clipped) {
+    const start = Math.max(span.startMs, end);
+    if (span.endMs > start) total += span.endMs - start;
+    end = Math.max(end, span.endMs);
+  }
+  return total;
+}
+
+function segmentSpans(
+  durationMs: number | null,
+  still: boolean,
+  scenes: DetectScenesResult | null,
+): Array<{ startMs: number; endMs: number; kind: "shot" | "scene" }> {
+  if (still) return [{ startMs: 0, endMs: STILL_SEGMENT_MS, kind: "shot" }];
+  const total = durationMs && durationMs > 0 ? durationMs : WINDOW_MS;
+  const shots = (scenes?.shots ?? [])
+    .map((shot) => ({
+      startMs: Math.max(0, Math.round(shot.startMs)),
+      endMs: Math.min(total, Math.round(shot.endMs)),
+    }))
+    .filter((shot) => shot.endMs > shot.startMs)
+    .sort((a, b) => a.startMs - b.startMs);
+  if (shots.length > 0) return shots.map((shot) => ({ ...shot, kind: "shot" as const }));
+  const windows: Array<{ startMs: number; endMs: number; kind: "scene" }> = [];
+  for (let start = 0; start < total; start += WINDOW_MS) {
+    windows.push({ startMs: start, endMs: Math.min(total, start + WINDOW_MS), kind: "scene" });
+  }
+  return windows;
+}
+
+/**
+ * Scored segments: detected shots, else fixed 3 s windows (one span for a still).
+ * Features are 0..1; a feature without its analysis is omitted (scores as 0).
+ *   motion        mean visual change between consecutive keyframe embeddings
+ *   sharpness     resolution proxy (shorter side / 1080); no blur measure yet
+ *   faces         share of the span covered by face tracks
+ *   audioPeak     loudest energy sample in the span relative to the asset maximum
+ *   speechHook    speechHookScore of the words inside the span
+ *   novelty       embedding distance to the previous span (first span 0.5)
+ *   beatProximity closeness of the span start to a beat (within 500 ms)
+ *   aesthetics    saliency strength proxy
+ * gaze stays unset: no landmarks are computed during ingest.
+ */
+export function buildSegments(input: {
+  durationMs: number | null;
+  still: boolean;
+  width: number | null;
+  height: number | null;
+  bundle: AnalysisBundle;
+  weights: TeaserWeights;
+}): NewMediaSegment[] {
+  const { bundle } = input;
+  const spans = segmentSpans(input.durationMs, input.still, bundle.scenes);
+  const shortSide = Math.min(input.width ?? 0, input.height ?? 0);
+  const energyMax = Math.max(0, ...(bundle.audio?.energy.map((sample) => sample.value) ?? []));
+  const language = bundle.transcript?.language ?? undefined;
+  let previousMean: Float32Array | null = null;
+
+  return spans.map((span, index) => {
+    const scores: Partial<Record<TeaserFeature, number>> = {};
+    const embeddings = bundle.embeddings;
+    const frames = embeddings
+      ? embeddings.timesMs.flatMap((tMs, i) => {
+          const vector = embeddings.vectors[i];
+          return vector && (input.still || inSpan(tMs, span)) ? [vector] : [];
+        })
+      : [];
+    if (frames.length >= 2) {
+      let change = 0;
+      for (let i = 1; i < frames.length; i++) {
+        change += 1 - cosineSimilarity(frames[i - 1] as Float32Array, frames[i] as Float32Array);
+      }
+      scores.motion = round3((change / (frames.length - 1)) * 4);
+    } else if (bundle.scenes && !input.still) {
+      const cuts = bundle.scenes.shots.filter(
+        (shot) => shot.startMs > span.startMs && shot.startMs < span.endMs,
+      ).length;
+      scores.motion = round3(cuts / Math.max(1, (span.endMs - span.startMs) / 1000));
+    }
+    if (shortSide > 0) scores.sharpness = round3(shortSide / 1080);
+    if (bundle.faceSpans) {
+      scores.faces = input.still
+        ? bundle.faceSpans.length > 0
+          ? 1
+          : 0
+        : round3(coveredMs(bundle.faceSpans, span) / (span.endMs - span.startMs));
+    }
+    if (bundle.audio && energyMax > 0) {
+      const inside = bundle.audio.energy.filter((sample) => inSpan(sample.tMs, span));
+      scores.audioPeak = round3(Math.max(0, ...inside.map((sample) => sample.value)) / energyMax);
+    }
+    let excerpt: string | null = null;
+    if (bundle.transcript) {
+      const words = bundle.transcript.words.filter((word) =>
+        inSpan((word.startMs + word.endMs) / 2, span),
+      );
+      excerpt =
+        words
+          .map((word) => word.text.trim())
+          .filter(Boolean)
+          .join(" ") || null;
+      scores.speechHook = excerpt ? round3(speechHookScore(excerpt, language)) : 0;
+    }
+    const mean = frames.length > 0 ? meanVector(frames) : null;
+    if (mean) {
+      scores.novelty =
+        index === 0 || !previousMean || previousMean.length !== mean.length
+          ? 0.5
+          : round3(1 - cosineSimilarity(previousMean, mean));
+      previousMean = mean;
+    }
+    if (bundle.audio && bundle.audio.beatsMs.length > 0 && !input.still) {
+      const distance = Math.min(
+        ...bundle.audio.beatsMs.map((beat) => Math.abs(beat - span.startMs)),
+      );
+      scores.beatProximity = round3(1 - distance / 500);
+    }
+    if (bundle.saliency && bundle.saliency.points.length > 0) {
+      const inside = bundle.saliency.points.filter(
+        (point) => input.still || inSpan(point.tMs, span),
+      );
+      const points = inside.length > 0 ? inside : bundle.saliency.points;
+      scores.aesthetics = round3(
+        points.reduce((sum, point) => sum + point.strength, 0) / points.length,
+      );
+    }
+    return {
+      kind: span.kind,
+      startMs: span.startMs,
+      endMs: span.endMs,
+      scores,
+      teaserScore: Math.round(scoreSegment(scores, input.weights) * 1000) / 1000,
+      transcriptExcerpt: excerpt ? excerpt.slice(0, EXCERPT_LIMIT) : null,
+      labels: [],
+    };
+  });
+}
+
+// ---------------------------------------------------------------------------
+// Running analyses
+// ---------------------------------------------------------------------------
+
+type AssetRow = Awaited<ReturnType<MediaServiceDeps["repos"]["getAssetRecord"]>>;
+
+/** Sidecar errors that say nothing about the asset (down, saturated, timed out). */
+export function isTransientInferenceError(error: unknown): boolean {
+  return (
+    error instanceof MediaSidecarError &&
+    (error.code === "unreachable" || error.code === "busy" || error.code === "timeout")
+  );
+}
+
+/** JSON-safe copy of a provider result for a Prisma Json column. */
+function json(value: unknown): Prisma.InputJsonValue {
+  return JSON.parse(JSON.stringify(value)) as Prisma.InputJsonValue;
+}
+
+/** Run `fn`, or return null when the provider does not support the operation. */
+async function unlessUnsupported<T>(fn: () => Promise<T>): Promise<T | null> {
+  try {
+    return await fn();
+  } catch (error) {
+    if (error instanceof UnsupportedMediaOperationError) return null;
+    throw error;
+  }
+}
+
+function safeName(value: string): string {
+  return value.replace(/[^A-Za-z0-9_-]/g, "_").slice(0, 64) || "track";
+}
+
+/**
+ * Model version of an analysis kind: "<provider id>:<model>" from the
+ * provider's capabilities, so a model change writes new rows instead of
+ * overwriting, and the skip check sees the same key the write uses.
+ */
+export function analysisModelVersion(
+  providerId: string,
+  capabilities: MediaCapabilities | null,
+  kind: MediaAnalysisKind,
+): string {
+  const model = capabilities?.operations[ANALYSIS_OPERATION[kind]]?.model ?? "default";
+  return `${providerId}:${model}`;
+}
+
+/**
+ * Re-analysis keeps the person a track was assigned to when the new track of the
+ * same asset carries the same identity; tracks of other assets are never matched here.
+ */
+async function previousAssignments(deps: MediaServiceDeps, spaceId: string, assetId: string) {
+  const rows = await deps.prisma.mediaFaceTrack.findMany({
+    where: { spaceId, assetId, personId: { not: null }, embedding: { not: null } },
+    select: { personId: true, embedding: true },
+  });
+  return rows.flatMap((row) => {
+    const bytes = row.embedding as Uint8Array;
+    if (!row.personId || bytes.byteLength === 0 || bytes.byteLength % 4 !== 0) return [];
+    const [vector] = decodeVectors(bytes, bytes.byteLength / 4);
+    return vector ? [{ personId: row.personId, vector }] : [];
+  });
+}
+
+async function storedBundle(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  assetId: string,
+  kinds: MediaAnalysisKind[],
+): Promise<AnalysisBundle> {
+  const bundle: AnalysisBundle = { ...EMPTY_BUNDLE };
+  for (const kind of kinds) {
+    const row = await deps.repos.latestAnalysis(spaceId, assetId, kind);
+    if (!row) continue;
+    if (kind === "scenes") {
+      const parsed = ScenesData.safeParse(row.data);
+      if (parsed.success) bundle.scenes = parsed.data;
+    } else if (kind === "transcript") {
+      const parsed = TranscriptData.safeParse(row.data);
+      if (parsed.success) bundle.transcript = parsed.data;
+    } else if (kind === "faces") {
+      const parsed = FacesData.safeParse(row.data);
+      if (parsed.success) bundle.faceSpans = parsed.data.tracks;
+    } else if (kind === "embeddings") {
+      const parsed = EmbeddingsData.safeParse(row.data);
+      if (parsed.success && row.vectors) {
+        bundle.embeddings = {
+          timesMs: parsed.data.timesMs,
+          vectors: decodeVectors(row.vectors, parsed.data.dim),
+        };
+      }
+    } else if (kind === "audio") {
+      const parsed = AudioData.safeParse(row.data);
+      if (parsed.success) bundle.audio = parsed.data;
+    } else if (kind === "saliency") {
+      const parsed = SaliencyData.safeParse(row.data);
+      if (parsed.success) bundle.saliency = parsed.data;
+    }
+  }
+  return bundle;
+}
+
+export interface AnalyzeOutcome {
+  ran: MediaAnalysisKind[];
+  /** Requested but unsupported by the provider or not applicable to a still. */
+  skipped: MediaAnalysisKind[];
+  /** Already stored at the current model version, so not run again. */
+  reused: MediaAnalysisKind[];
+  segments: number;
+  qualityScore: number | null;
+}
+
+/**
+ * Run analyses on `inputPath`, store them, and rebuild the asset's segments.
+ * Explicit `kinds` always run; without them only kinds that have no result at
+ * the provider's current model version run. The caller owns ingest state.
+ */
+export async function analyzeAsset(input: {
+  deps: MediaServiceDeps;
+  inference: MediaInferenceProvider;
+  spaceId: string;
+  asset: AssetRow;
+  inputPath: string;
+  kinds?: MediaAnalysisKind[];
+  signal?: AbortSignal;
+}): Promise<AnalyzeOutcome> {
+  const { deps, inference, spaceId, asset, inputPath, signal } = input;
+  const still = asset.kind === "photo" || !asset.durationMs;
+  const capabilities = await inference.capabilities(signal).catch((error: unknown) => {
+    if (signal?.aborted || isTransientInferenceError(error)) throw error;
+    return null;
+  });
+  const times = keyframeTimes(asset.durationMs, still);
+  const bundle: AnalysisBundle = { ...EMPTY_BUNDLE };
+  const ran: MediaAnalysisKind[] = [];
+  const skipped: MediaAnalysisKind[] = [];
+  const reused: MediaAnalysisKind[] = [];
+  const request = { path: inputPath, ...(signal ? { signal } : {}) };
+
+  const versionOf = (kind: MediaAnalysisKind) =>
+    analysisModelVersion(inference.id, capabilities, kind);
+
+  let requested: Set<MediaAnalysisKind>;
+  if (input.kinds) {
+    requested = new Set(input.kinds);
+  } else {
+    const existing = await deps.prisma.mediaAnalysis.findMany({
+      where: { spaceId, assetId: asset.id },
+      select: { kind: true, modelVersion: true },
+    });
+    const done = new Set(existing.map((row) => `${row.kind}\n${row.modelVersion}`));
+    requested = new Set();
+    for (const kind of MediaAnalysisKindSchema.options) {
+      if (done.has(`${kind}\n${versionOf(kind)}`)) reused.push(kind);
+      else requested.add(kind);
+    }
+  }
+
+  const store = async (
+    kind: MediaAnalysisKind,
+    data: Prisma.InputJsonValue,
+    extra: { vectors?: Uint8Array; resultModel?: string } = {},
+  ) => {
+    const modelVersion = versionOf(kind);
+    await deps.repos.upsertAnalysis(spaceId, asset.id, {
+      kind,
+      modelId: extra.resultModel ?? modelVersion.slice(inference.id.length + 1),
+      modelVersion,
+      data,
+      vectors: extra.vectors ?? null,
+    });
+    ran.push(kind);
+  };
+  const runnable = (kind: MediaAnalysisKind) => {
+    if (!requested.has(kind)) return false;
+    const unavailable = capabilities?.operations[ANALYSIS_OPERATION[kind]]?.available === false;
+    if ((still && VIDEO_ONLY.has(kind)) || unavailable) {
+      skipped.push(kind);
+      return false;
+    }
+    return true;
+  };
+  const attempt = async <T>(kind: MediaAnalysisKind, fn: () => Promise<T>) => {
+    signal?.throwIfAborted();
+    const result = await unlessUnsupported(fn);
+    if (result === null) skipped.push(kind);
+    return result;
+  };
+
+  if (runnable("scenes")) {
+    const scenes = await attempt("scenes", () => inference.detectScenes(request));
+    if (scenes) {
+      bundle.scenes = scenes;
+      await store("scenes", json({ shots: scenes.shots, scenes: scenes.scenes }));
+    }
+  }
+
+  if (runnable("transcript")) {
+    const transcript = await attempt("transcript", () => inference.transcribe(request));
+    if (transcript) {
+      bundle.transcript = transcript;
+      const words: TranscriptWord[] = transcript.words;
+      await store(
+        "transcript",
+        json({
+          language: transcript.language,
+          words,
+          text: words
+            .map((word) => word.text.trim())
+            .filter(Boolean)
+            .join(" "),
+        }),
+      );
+    }
+  }
+
+  if (runnable("faces")) {
+    const sampleFps = still ? 1 : FACE_SAMPLE_FPS;
+    const tracked = await attempt("faces", () => inference.trackFaces({ ...request, sampleFps }));
+    if (tracked) {
+      const facesDir = `${assetDir(spaceId, asset.id)}/faces`;
+      await rm(await resolveInside(deps.mediaRoot, facesDir), { recursive: true, force: true });
+      const previous = await previousAssignments(deps, spaceId, asset.id);
+      const stepMs = Math.round(1000 / tracked.sampleFps);
+      const tracks: NewMediaFaceTrack[] = [];
+      const spans: Array<{ startMs: number; endMs: number; keyframeCount: number }> = [];
+      for (const [index, track] of tracked.tracks.entries()) {
+        const keyframes = [...track.keyframes].sort((a, b) => a.tMs - b.tMs);
+        const first = keyframes[0];
+        const last = keyframes.at(-1);
+        if (!first || !last) continue;
+        const keyframesPath = `${facesDir}/${index}-${safeName(track.trackId)}.json`;
+        const file: FaceKeyframesFile = { sampleFps: tracked.sampleFps, keyframes };
+        await writeFileAtomic(
+          await resolveInside(deps.mediaRoot, keyframesPath),
+          JSON.stringify(file),
+        );
+        const embedding = track.embedding && track.embedding.length > 0 ? track.embedding : null;
+        let personId: string | null = null;
+        if (embedding) {
+          let best = SAME_FACE_SIMILARITY;
+          for (const candidate of previous) {
+            if (candidate.vector.length !== embedding.length) continue;
+            const similarity = cosineSimilarity(candidate.vector, embedding);
+            if (similarity >= best) {
+              best = similarity;
+              personId = candidate.personId;
+            }
+          }
+        }
+        const startMs = Math.max(0, Math.round(first.tMs));
+        const endMs = Math.max(startMs + 1, Math.round(last.tMs) + stepMs);
+        tracks.push({
+          startMs,
+          endMs,
+          confidenceMin: Math.min(...keyframes.map((keyframe) => keyframe.confidence)),
+          keyframesPath,
+          embedding: embedding ? encodeVectors([embedding]) : null,
+          personId,
+        });
+        spans.push({ startMs, endMs, keyframeCount: keyframes.length });
+      }
+      await deps.repos.replaceFaceTracks(spaceId, asset.id, versionOf("faces"), tracks);
+      bundle.faceSpans = spans;
+      // Spans and counts only: identity embeddings never enter an analysis row.
+      await store(
+        "faces",
+        json({ sampleFps: tracked.sampleFps, trackCount: spans.length, tracks: spans }),
+      );
+    }
+  }
+
+  if (runnable("embeddings")) {
+    const embedded = await attempt("embeddings", () =>
+      inference.embedFrames({ ...request, timesMs: times }),
+    );
+    if (embedded && embedded.vectors.length > 0) {
+      const timesMs = times.slice(0, embedded.vectors.length);
+      bundle.embeddings = {
+        timesMs,
+        vectors: embedded.vectors.map((vector) => Float32Array.from(vector)),
+      };
+      await store("embeddings", json({ model: embedded.model, dim: embedded.dim, timesMs }), {
+        vectors: encodeVectors(embedded.vectors),
+        resultModel: embedded.model,
+      });
+    }
+  }
+
+  if (runnable("classifier")) {
+    // Every shot boundary on top of the regular samples.
+    const boundaries = [
+      ...new Set(
+        (bundle.scenes?.shots ?? []).flatMap((shot) => [shot.startMs, Math.max(0, shot.endMs - 1)]),
+      ),
+    ];
+    const classified = await attempt("classifier", () =>
+      inference.classifyContent({
+        ...request,
+        sampleFps: CLASSIFIER_SAMPLE_FPS,
+        ...(boundaries.length > 0 ? { extraTimesMs: boundaries } : {}),
+      }),
+    );
+    if (classified) {
+      await store("classifier", json({ model: classified.model, frames: classified.frames }), {
+        resultModel: classified.model,
+      });
+    }
+  }
+
+  if (runnable("audio")) {
+    const audio = await attempt("audio", () => inference.audioFeatures(request));
+    if (audio) {
+      bundle.audio = audio;
+      await store("audio", json(audio));
+    }
+  }
+
+  if (runnable("saliency")) {
+    const saliency = await attempt("saliency", () =>
+      inference.saliency({ ...request, timesMs: times }),
+    );
+    if (saliency) {
+      bundle.saliency = saliency;
+      await store("saliency", json({ points: saliency.points }));
+    }
+  }
+
+  // Kinds not run this time come from their latest stored result.
+  const missing = MediaAnalysisKindSchema.options.filter(
+    (kind) => !ran.includes(kind) && kind !== "classifier",
+  );
+  if (missing.length > 0) {
+    const stored = await storedBundle(deps, spaceId, asset.id, missing);
+    bundle.scenes ??= stored.scenes;
+    bundle.transcript ??= stored.transcript;
+    bundle.faceSpans ??= stored.faceSpans;
+    bundle.embeddings ??= stored.embeddings;
+    bundle.audio ??= stored.audio;
+    bundle.saliency ??= stored.saliency;
+  }
+
+  const learned = await deps.repos.getTeaserModel(spaceId);
+  const weights =
+    learned && learned.sampleCount >= TEASER_LEARNING_MIN_SAMPLES
+      ? learned.weights
+      : defaultTeaserWeights("teaser");
+  const segments = buildSegments({
+    durationMs: asset.durationMs,
+    still,
+    width: asset.width,
+    height: asset.height,
+    bundle,
+    weights,
+  });
+  await deps.repos.replaceSegments(spaceId, asset.id, MEDIA_ANALYSIS_VERSION, segments);
+  const teaserScores = segments.flatMap((segment) =>
+    typeof segment.teaserScore === "number" ? [segment.teaserScore] : [],
+  );
+  const qualityScore = teaserScores.length > 0 ? Math.max(...teaserScores) : null;
+  return { ran, skipped, reused, segments: segments.length, qualityScore };
+}
+
+/**
+ * The file the sidecar analyses. A referenced original inside the media root
+ * (the sidecar's only mount) is analysed directly; everything else uses the
+ * proxy, which shares the original's timeline and applies its rotation.
+ */
+export async function analysisInputPath(input: {
+  mediaRoot: string;
+  proxyPath: string;
+  referencedOriginal: string | null;
+}): Promise<string> {
+  if (input.referencedOriginal) {
+    try {
+      return await assertInside(input.mediaRoot, input.referencedOriginal);
+    } catch {
+      // Outside the media root: the sidecar cannot read it.
+    }
+  }
+  return resolveInside(input.mediaRoot, input.proxyPath);
+}

packages/adapters/src/media-ingest-cluster.test.ts

@@ -0,0 +1,77 @@
+import { describe, expect, it } from "vitest";
+import { reconcileUserEdits } from "./media-ingest-cluster.js";
+
+const computed = [
+  { key: "session:a1", assetIds: ["a1", "a2", "a3"] },
+  { key: "session:b1", assetIds: ["b1", "b2"] },
+  { key: "session:c1", assetIds: ["c1", "c2", "c3"] },
+];
+
+const keys = (clusters: Array<{ key: string; assetIds: string[] }>) =>
+  clusters.map((cluster) => [cluster.key, cluster.assetIds]);
+
+describe("reconcileUserEdits", () => {
+  it("passes computed clusters through when the user edited nothing", () => {
+    expect(reconcileUserEdits(computed, [], "session")).toEqual(computed);
+  });
+
+  it("never re-adds the assets of a merged collection to computed clusters", () => {
+    // The user merged a1.. and b1.. into the a1 collection.
+    const merged = { clusterKey: "session:a1", assetIds: ["a1", "a2", "a3", "b1", "b2"] };
+    expect(keys(reconcileUserEdits(computed, [merged], "session"))).toEqual([
+      ["session:c1", ["c1", "c2", "c3"]],
+    ]);
+  });
+
+  it("keeps both halves of a split and places new assets outside them", () => {
+    // Split c1 | c2 c3; a later sync adds c4 to the computed session.
+    const grown = [
+      ...computed.slice(0, 2),
+      { key: "session:c1", assetIds: ["c1", "c2", "c3", "c4"] },
+    ];
+    const kept = { clusterKey: "session:c1", assetIds: ["c1"] };
+    const splitOff = { clusterKey: null, assetIds: ["c2", "c3"] };
+    expect(keys(reconcileUserEdits(grown, [kept, splitOff], "session"))).toEqual([
+      ["session:a1", ["a1", "a2", "a3"]],
+      ["session:b1", ["b1", "b2"]],
+      ["session:c1", ["c1", "c4"]],
+    ]);
+    // When the kept half is not a subset any more it freezes; c4 gets its own key.
+    const moved = [...computed.slice(0, 2), { key: "session:c0", assetIds: ["c0", "c4"] }];
+    expect(
+      keys(reconcileUserEdits(moved, [{ ...kept, assetIds: ["c1", "c9"] }], "session")),
+    ).toEqual([
+      ["session:a1", ["a1", "a2", "a3"]],
+      ["session:b1", ["b1", "b2"]],
+      ["session:c0", ["c0", "c4"]],
+    ]);
+  });
+
+  it("lets a renamed collection follow its cluster, even when the cluster key changes", () => {
+    const renamed = { clusterKey: "session:a1", assetIds: ["a1", "a2"] };
+    const earlier = [
+      { key: "session:a0", assetIds: ["a0", "a1", "a2", "a3"] },
+      ...computed.slice(1),
+    ];
+    expect(keys(reconcileUserEdits(earlier, [renamed], "session"))).toEqual([
+      ["session:a1", ["a0", "a1", "a2", "a3"]],
+      ["session:b1", ["b1", "b2"]],
+      ["session:c1", ["c1", "c2", "c3"]],
+    ]);
+  });
+
+  it("re-keys a cluster that lost its first asset to a split-off collection", () => {
+    const splitOff = { clusterKey: null, assetIds: ["a1"] };
+    expect(keys(reconcileUserEdits(computed, [splitOff], "theme"))).toEqual([
+      ["theme:a2", ["a2", "a3"]],
+      ["session:b1", ["b1", "b2"]],
+      ["session:c1", ["c1", "c2", "c3"]],
+    ]);
+  });
+
+  it("ignores empty user collections", () => {
+    expect(
+      reconcileUserEdits(computed, [{ clusterKey: "session:a1", assetIds: [] }], "session"),
+    ).toEqual(computed);
+  });
+});

packages/adapters/src/media-ingest-cluster.ts

@@ -0,0 +1,451 @@
+import { readFile } from "node:fs/promises";
+import type { AgentInputImage } from "@rakazo/adapter-kit";
+import type { MediaClusterInput } from "@rakazo/db";
+import { getLogger } from "@rakazo/logging";
+import type { ClusterAsset, SessionOptions, VectorItem } from "@rakazo/media";
+import {
+  buildSessions,
+  clusterFaces,
+  clusterThemes,
+  cosineSimilarity,
+  decodeVectors,
+  encodeVectors,
+  meanVector,
+  ruleLabel,
+} from "@rakazo/media";
+import { resolveInside } from "@rakazo/media/node";
+import { z } from "zod";
+import type { MediaServiceDeps } from "./media-context.js";
+import { errorMessage, publishSafe } from "./media-ingest-files.js";
+
+/*
+ * Sessions, themes, and person suggestions for one space.
+ *
+ * Bounds: themes use average-linkage clustering (O(n²) time and memory), so
+ * each run clusters at most THEME_BATCH_LIMIT of the most recently captured
+ * analysed assets; older assets keep no theme until they fall into the window.
+ * Face clustering is bounded the same way per run (FACE_BATCH_LIMIT unassigned
+ * tracks); the rest are picked up by the next run.
+ *
+ * Theme titles come from the model only when the media model client returns
+ * one (it checks AI consent for the space's library owner); it receives cover
+ * thumbnails and nothing else. Face embeddings and person data never leave
+ * this process.
+ *
+ * Collections the user merged or split keep their membership; re-clustering
+ * never re-adds their assets elsewhere (see reconcileUserEdits).
+ */
+
+export const THEME_BATCH_LIMIT = 2000;
+export const FACE_BATCH_LIMIT = 2000;
+const THEME_MIN_SIZE = 3;
+const THEME_SIMILARITY = 0.8;
+const FACE_SIMILARITY = 0.6;
+const COVER_IMAGES = 9;
+const MODEL_LABELS_PER_RUN = 12;
+const EMBEDDING_BATCH = 500;
+
+const ThemeTitleSchema = z.object({ title: z.string().trim().min(1).max(80) });
+
+const THEME_INSTRUCTIONS =
+  'You name groups of photos and videos. Reply with one JSON object {"title": string}: ' +
+  "a short, neutral title of at most five words for what the images have in common " +
+  "(place type, activity, or subject). Never name or describe people's identities.";
+
+interface UserCollection {
+  clusterKey: string | null;
+  assetIds: string[];
+}
+
+/**
+ * Apply the user's edits to computed clusters of one kind:
+ * - A user collection without a cluster key (split off) keeps its assets; they
+ *   are withheld from every computed cluster.
+ * - A user collection whose assets all still fall into one computed cluster
+ *   (renamed only) follows that cluster: the cluster takes the collection's key
+ *   so the collection keeps updating, also when an earlier asset changes the
+ *   cluster's own key.
+ * - Any other user collection (merged, or split and kept) is frozen: its key
+ *   and assets are withheld, so re-clustering never re-adds them elsewhere.
+ * Clusters that lose their first asset are re-keyed by the first remaining one
+ * (the clustering's own rule); empty clusters and key collisions are dropped.
+ */
+export function reconcileUserEdits<T extends { key: string; assetIds: string[] }>(
+  clusters: T[],
+  userCollections: UserCollection[],
+  prefix: "session" | "theme",
+): T[] {
+  const splitOff = new Set(
+    userCollections.flatMap((collection) => (collection.clusterKey ? [] : collection.assetIds)),
+  );
+  const hostOf = new Map<string, T>();
+  for (const cluster of clusters) {
+    for (const id of cluster.assetIds) if (!splitOff.has(id)) hostOf.set(id, cluster);
+  }
+  const followedBy = new Map<string, string>();
+  const frozenKeys = new Set<string>();
+  const frozenAssets = new Set<string>();
+  for (const collection of userCollections) {
+    const [first] = collection.assetIds;
+    if (!collection.clusterKey || first === undefined) continue;
+    const host = hostOf.get(first);
+    const contained = host && collection.assetIds.every((id) => hostOf.get(id) === host);
+    if (host && contained && !followedBy.has(host.key)) {
+      followedBy.set(host.key, collection.clusterKey);
+    } else {
+      frozenKeys.add(collection.clusterKey);
+      for (const id of collection.assetIds) frozenAssets.add(id);
+    }
+  }
+  const seen = new Set<string>(frozenKeys);
+  const result: T[] = [];
+  for (const cluster of clusters) {
+    const assetIds = cluster.assetIds.filter((id) => !splitOff.has(id) && !frozenAssets.has(id));
+    const first = assetIds[0];
+    if (first === undefined) continue;
+    const key =
+      followedBy.get(cluster.key) ??
+      (first === cluster.assetIds[0] ? cluster.key : `${prefix}:${first}`);
+    if (seen.has(key)) continue;
+    seen.add(key);
+    result.push({ ...cluster, key, assetIds });
+  }
+  return result;
+}
+
+type ClusterRow = Awaited<ReturnType<MediaServiceDeps["repos"]["listAssetsForClustering"]>>[number];
+
+/** Mean frame embedding per asset from the latest embeddings analysis, read in batches. */
+async function meanEmbeddings(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  assetIds: string[],
+): Promise<Map<string, Float32Array>> {
+  const means = new Map<string, Float32Array>();
+  for (let offset = 0; offset < assetIds.length; offset += EMBEDDING_BATCH) {
+    const batch = assetIds.slice(offset, offset + EMBEDDING_BATCH);
+    const rows = await deps.prisma.mediaAnalysis.findMany({
+      where: { spaceId, kind: "embeddings", assetId: { in: batch }, vectors: { not: null } },
+      select: { assetId: true, data: true, vectors: true },
+      orderBy: [{ createdAt: "desc" }, { id: "desc" }],
+    });
+    for (const row of rows) {
+      if (means.has(row.assetId) || !row.vectors) continue;
+      const data = row.data as { dim?: unknown } | null;
+      const dim = typeof data?.dim === "number" ? data.dim : 0;
+      if (!Number.isInteger(dim) || dim <= 0 || row.vectors.byteLength % (dim * 4) !== 0) continue;
+      const mean = meanVector(decodeVectors(row.vectors, dim));
+      if (mean) means.set(row.assetId, mean);
+    }
+  }
+  return means;
+}
+
+/** The vectors of the most common dimension (a model change mixes dimensions). */
+function sameDimension<T extends { vector: Float32Array }>(items: T[]): T[] {
+  const counts = new Map<number, number>();
+  for (const item of items)
+    counts.set(item.vector.length, (counts.get(item.vector.length) ?? 0) + 1);
+  const dim = [...counts.entries()].sort((a, b) => b[1] - a[1] || b[0] - a[0])[0]?.[0];
+  return items.filter((item) => item.vector.length === dim);
+}
+
+async function userCollections(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  kind: "session" | "theme",
+): Promise<UserCollection[]> {
+  const rows = await deps.prisma.mediaCollection.findMany({
+    where: { spaceId, kind, labelSource: "user" },
+    select: {
+      clusterKey: true,
+      items: { select: { assetId: true }, orderBy: { position: "asc" } },
+    },
+  });
+  return rows.map((row) => ({
+    clusterKey: row.clusterKey,
+    assetIds: row.items.map((item) => item.assetId),
+  }));
+}
+
+/** Whose AI consent governs model labels: the owner of the space's first media source. */
+async function libraryOwner(deps: MediaServiceDeps, spaceId: string): Promise<string | null> {
+  const source = await deps.prisma.mediaSource.findFirst({
+    where: { spaceId },
+    orderBy: [{ createdAt: "asc" }, { id: "asc" }],
+    select: { ownerUserId: true },
+  });
+  if (source) return source.ownerUserId;
+  const space = await deps.prisma.space.findUnique({
+    where: { id: spaceId },
+    select: { createdByUserId: true },
+  });
+  return space?.createdByUserId ?? null;
+}
+
+function validTimeZone(zone: string | null | undefined): string | null {
+  if (!zone) return null;
+  try {
+    new Intl.DateTimeFormat("en", { timeZone: zone });
+    return zone;
+  } catch {
+    return null;
+  }
+}
+
+export interface ClusterOutcome {
+  sessions: number;
+  themes: number;
+  modelLabels: number;
+  assignedTracks: number;
+  suggestedPeople: number;
+}
+
+export async function clusterSpace(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  options: { signal?: AbortSignal; sessionOptions?: Partial<SessionOptions> } = {},
+): Promise<ClusterOutcome> {
+  const userId = await libraryOwner(deps, spaceId);
+  const profile = userId
+    ? await deps.prisma.userProfile.findUnique({
+        where: { userId },
+        select: { timeZone: true, locale: true },
+      })
+    : null;
+  const timeZone = validTimeZone(profile?.timeZone) ?? "UTC";
+  const locale = profile?.locale ?? undefined;
+  const label = (range: { startAt: Date | null; endAt: Date | null }, assetIds: string[]) =>
+    ruleLabel({ ...range, albumNames: albums(assetIds) }, locale, timeZone);
+  const rows = await deps.repos.listAssetsForClustering(spaceId);
+  const extras = await deps.prisma.mediaAsset.findMany({
+    where: { spaceId },
+    select: { id: true, thumbPath: true, qualityScore: true },
+  });
+  const extra = new Map(extras.map((row) => [row.id, row]));
+  const embeddings = await meanEmbeddings(
+    deps,
+    spaceId,
+    rows.map((row) => row.id),
+  );
+  const byId = new Map(rows.map((row) => [row.id, row]));
+
+  const cover = (assetIds: string[]): string | null => {
+    const ranked = assetIds
+      .map((id) => extra.get(id))
+      .filter((row) => row?.thumbPath)
+      .sort((a, b) => (b?.qualityScore ?? 0) - (a?.qualityScore ?? 0));
+    return ranked[0]?.id ?? assetIds[0] ?? null;
+  };
+  const span = (assetIds: string[]) => {
+    const dates = assetIds
+      .flatMap((id) => {
+        const at = byId.get(id)?.capturedAt;
+        return at ? [at.getTime()] : [];
+      })
+      .sort((a, b) => a - b);
+    return {
+      startAt: dates.length > 0 ? new Date(dates[0] as number) : null,
+      endAt: dates.length > 0 ? new Date(dates.at(-1) as number) : null,
+    };
+  };
+  const albums = (assetIds: string[]) => assetIds.flatMap((id) => byId.get(id)?.albumNames ?? []);
+
+  // Sessions -----------------------------------------------------------------
+  const clusterAssets: ClusterAsset[] = rows.map((row: ClusterRow) => ({
+    id: row.id,
+    kind: row.kind as ClusterAsset["kind"],
+    capturedAt: row.capturedAt,
+    gpsLat: row.gpsLat,
+    gpsLon: row.gpsLon,
+    deviceModel: row.deviceModel,
+    albumNames: row.albumNames,
+    embedding: embeddings.get(row.id) ?? null,
+  }));
+  const computedSessions = buildSessions(clusterAssets, { timeZone, ...options.sessionOptions });
+  const sessions: MediaClusterInput[] = reconcileUserEdits(
+    computedSessions,
+    await userCollections(deps, spaceId, "session"),
+    "session",
+  ).map((session) => {
+    const range = span(session.assetIds);
+    return {
+      key: session.key,
+      title: label(range, session.assetIds),
+      labelSource: "rule",
+      ...range,
+      centroid: session.centroid,
+      coverAssetId: cover(session.assetIds),
+      assetIds: session.assetIds,
+    };
+  });
+  options.signal?.throwIfAborted();
+  await deps.repos.replaceClusters(spaceId, "session", sessions);
+  await publishSafe(deps, spaceId, { type: "media.collections", kind: "session" });
+
+  // Themes -------------------------------------------------------------------
+  const recent = rows
+    .flatMap((row) => {
+      const vector = embeddings.get(row.id);
+      return vector ? [{ id: row.id, vector, at: row.capturedAt?.getTime() ?? 0 }] : [];
+    })
+    .sort((a, b) => b.at - a.at || a.id.localeCompare(b.id))
+    .slice(0, THEME_BATCH_LIMIT);
+  const items: VectorItem[] = sameDimension(recent).map(({ id, vector }) => ({ id, vector }));
+  const computedThemes = reconcileUserEdits(
+    clusterThemes(items, { similarity: THEME_SIMILARITY, minSize: THEME_MIN_SIZE }).map(
+      (theme) => ({ key: theme.key, assetIds: theme.ids }),
+    ),
+    await userCollections(deps, spaceId, "theme"),
+    "theme",
+  );
+  const existing = await deps.prisma.mediaCollection.findMany({
+    where: { spaceId, kind: "theme", clusterKey: { in: computedThemes.map((theme) => theme.key) } },
+    select: { clusterKey: true, title: true, labelSource: true },
+  });
+  const modelTitles = new Map(
+    existing.flatMap((row) =>
+      row.labelSource === "model" && row.clusterKey ? [[row.clusterKey, row.title] as const] : [],
+    ),
+  );
+  let modelCalls = 0;
+  let modelLabels = 0;
+  const themes: MediaClusterInput[] = [];
+  for (const theme of computedThemes) {
+    options.signal?.throwIfAborted();
+    const range = span(theme.assetIds);
+    const coverAssetId = cover(theme.assetIds);
+    let title = modelTitles.get(theme.key) ?? null;
+    if (!title && userId && modelCalls < MODEL_LABELS_PER_RUN) {
+      modelCalls += 1;
+      title = await themeTitle(deps, spaceId, userId, theme.assetIds, extra, options.signal);
+    }
+    if (title) modelLabels += 1;
+    themes.push({
+      key: theme.key,
+      title: title ?? label(range, theme.assetIds),
+      labelSource: title ? "model" : "rule",
+      ...range,
+      centroid: null,
+      coverAssetId,
+      assetIds: theme.assetIds,
+    });
+  }
+  await deps.repos.replaceClusters(spaceId, "theme", themes);
+  await publishSafe(deps, spaceId, { type: "media.collections", kind: "theme" });
+
+  // People -------------------------------------------------------------------
+  const people = await suggestPeople(deps, spaceId);
+  if (people.assignedTracks + people.suggestedPeople > 0) {
+    await publishSafe(deps, spaceId, { type: "media.people" });
+  }
+  return { sessions: sessions.length, themes: themes.length, modelLabels, ...people };
+}
+
+/** Ask the model for a theme title from cover thumbnails only; null falls back to rules. */
+async function themeTitle(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  userId: string,
+  assetIds: string[],
+  extra: Map<string, { id: string; thumbPath: string | null; qualityScore: number | null }>,
+  signal?: AbortSignal,
+): Promise<string | null> {
+  const covers = assetIds
+    .flatMap((id) => {
+      const row = extra.get(id);
+      return row?.thumbPath ? [row] : [];
+    })
+    .sort((a, b) => (b.qualityScore ?? 0) - (a.qualityScore ?? 0) || a.id.localeCompare(b.id))
+    .slice(0, COVER_IMAGES);
+  const images: AgentInputImage[] = [];
+  for (const [index, row] of covers.entries()) {
+    try {
+      const file = await resolveInside(deps.mediaRoot, row.thumbPath as string);
+      images.push({
+        name: `cover-${index + 1}.jpg`,
+        mimeType: "image/jpeg",
+        data: await readFile(file),
+      });
+    } catch {
+      // A missing thumbnail only shrinks the set.
+    }
+  }
+  if (images.length === 0) return null;
+  try {
+    const reply = await deps.model.generate({
+      spaceId,
+      userId,
+      instructions: THEME_INSTRUCTIONS,
+      prompt: `Name what these ${images.length} images have in common.`,
+      images,
+      schema: ThemeTitleSchema,
+      tier: "fast",
+      ...(signal ? { signal } : {}),
+    });
+    return reply?.title ?? null;
+  } catch (error) {
+    getLogger().warn("media theme label failed", { reason: errorMessage(error) });
+    return null;
+  }
+}
+
+/**
+ * Unassigned face tracks join an unconfirmed suggestion with the same identity;
+ * the rest are clustered into new suggestions. Confirmed people only ever gain
+ * tracks through the user.
+ */
+async function suggestPeople(
+  deps: MediaServiceDeps,
+  spaceId: string,
+): Promise<{ assignedTracks: number; suggestedPeople: number }> {
+  const unassigned = (await deps.repos.faceTrackEmbeddings(spaceId, { unassignedOnly: true }))
+    .slice(0, FACE_BATCH_LIMIT)
+    .flatMap((track) => {
+      const bytes = track.embedding;
+      if (bytes.byteLength === 0 || bytes.byteLength % 4 !== 0) return [];
+      const [vector] = decodeVectors(bytes, bytes.byteLength / 4);
+      return vector ? [{ id: track.trackId, vector }] : [];
+    });
+  if (unassigned.length === 0) return { assignedTracks: 0, suggestedPeople: 0 };
+
+  const confirmed = new Set(
+    (await deps.repos.listPersonRecords(spaceId))
+      .filter((person) => person.confirmed)
+      .map((person) => person.id),
+  );
+  const suggestions = (await deps.repos.personCentroids(spaceId)).flatMap((person) => {
+    if (confirmed.has(person.personId)) return [];
+    const bytes = person.centroid;
+    if (bytes.byteLength === 0 || bytes.byteLength % 4 !== 0) return [];
+    const [vector] = decodeVectors(bytes, bytes.byteLength / 4);
+    return vector ? [{ personId: person.personId, vector }] : [];
+  });
+
+  let assignedTracks = 0;
+  const remaining: VectorItem[] = [];
+  for (const track of unassigned) {
+    let best: string | null = null;
+    let bestSimilarity = FACE_SIMILARITY;
+    for (const suggestion of suggestions) {
+      if (suggestion.vector.length !== track.vector.length) continue;
+      const similarity = cosineSimilarity(suggestion.vector, track.vector);
+      if (similarity >= bestSimilarity) {
+        best = suggestion.personId;
+        bestSimilarity = similarity;
+      }
+    }
+    if (best) {
+      await deps.repos.assignTrack(spaceId, track.id, best);
+      assignedTracks += 1;
+    } else {
+      remaining.push(track);
+    }
+  }
+  const groups = clusterFaces(sameDimension(remaining), { similarity: FACE_SIMILARITY });
+  const created = await deps.repos.createPersonSuggestions(
+    spaceId,
+    groups.map((group) => ({ centroid: encodeVectors([group.centroid]), trackIds: group.ids })),
+  );
+  return { assignedTracks, suggestedPeople: created.length };
+}

packages/adapters/src/media-ingest-files.ts

@@ -0,0 +1,110 @@
+import { createHash, randomUUID } from "node:crypto";
+import { createReadStream } from "node:fs";
+import { copyFile, lstat, mkdir, rename, rm, unlink, writeFile } from "node:fs/promises";
+import path from "node:path";
+import { pipeline } from "node:stream/promises";
+import { getLogger } from "@rakazo/logging";
+import { resolveInside } from "@rakazo/media/node";
+import type { MediaRealtimeEvent, MediaServiceDeps } from "./media-context.js";
+import { FOLDER_MEDIA_EXTENSIONS } from "./media-folder-source.js";
+
+/*
+ * File helpers for media ingest. Every stored path is relative to the media
+ * root and resolved through resolveInside; writes go to a temp name in the
+ * target directory and are renamed into place.
+ */
+
+/** Relative directory of one asset's derived files. */
+export function assetDir(spaceId: string, assetId: string): string {
+  return `${spaceId}/assets/${assetId}`;
+}
+
+/** Lower-case original extension for an uploaded file name, or null when it is not a media type. */
+export function originalExtension(fileName: string): string | null {
+  const ext = path.extname(fileName).slice(1).toLowerCase();
+  return Object.hasOwn(FOLDER_MEDIA_EXTENSIONS, ext) ? ext : null;
+}
+
+/** Streamed SHA-256 (hex) of a file. */
+export async function sha256File(file: string, signal?: AbortSignal): Promise<string> {
+  const hash = createHash("sha256");
+  await pipeline(createReadStream(file), hash, { signal });
+  return hash.digest("hex");
+}
+
+/** Delete media-root-relative files or directories; missing ones are fine. */
+export async function deleteMediaPaths(mediaRoot: string, paths: Iterable<string>): Promise<void> {
+  for (const relative of new Set(paths)) {
+    try {
+      await rm(await resolveInside(mediaRoot, relative), { recursive: true, force: true });
+    } catch (error) {
+      getLogger().warn("media file cleanup failed", {
+        reason: error instanceof Error ? error.message : String(error),
+      });
+    }
+  }
+}
+
+/** A temp path next to `target` that a later rename moves into place atomically. */
+export function tempSibling(target: string, ext: string): string {
+  return path.join(path.dirname(target), `.tmp-${randomUUID()}${ext}`);
+}
+
+/** Write bytes atomically: temp file in the same directory, then rename. */
+export async function writeFileAtomic(target: string, data: string | Uint8Array): Promise<void> {
+  await mkdir(path.dirname(target), { recursive: true });
+  const temp = tempSibling(target, path.extname(target));
+  try {
+    await writeFile(temp, data);
+    await rename(temp, target);
+  } finally {
+    await rm(temp, { force: true });
+  }
+}
+
+/** Rename, falling back to copy + unlink across devices. */
+export async function moveFile(from: string, to: string): Promise<void> {
+  await mkdir(path.dirname(to), { recursive: true });
+  try {
+    await rename(from, to);
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code !== "EXDEV") throw error;
+    const temp = tempSibling(to, path.extname(to));
+    try {
+      await copyFile(from, temp);
+      await rename(temp, to);
+    } finally {
+      await rm(temp, { force: true });
+    }
+    await unlink(from);
+  }
+}
+
+/** Realtime is best effort: a failed notification never fails ingest work. */
+export async function publishSafe(
+  deps: Pick<MediaServiceDeps, "publish">,
+  spaceId: string,
+  event: MediaRealtimeEvent,
+): Promise<void> {
+  try {
+    await deps.publish(spaceId, event);
+  } catch (error) {
+    getLogger().warn("media realtime publish failed", {
+      type: event.type,
+      reason: error instanceof Error ? error.message : String(error),
+    });
+  }
+}
+
+export function errorMessage(error: unknown): string {
+  return (error instanceof Error ? error.message : String(error)).slice(0, 500);
+}
+
+/** Whether a media-root-relative path is an existing regular file. */
+export async function fileExists(mediaRoot: string, relative: string): Promise<boolean> {
+  try {
+    return (await lstat(await resolveInside(mediaRoot, relative))).isFile();
+  } catch {
+    return false;
+  }
+}

packages/adapters/src/media-ingest-proxy.test.ts

@@ -0,0 +1,185 @@
+import { mkdir, mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { ffmpegAvailable } from "./media-emulator-clips.js";
+import { runMediaTool } from "./media-folder-source-probe.js";
+import { buildProxy, probeOriginal } from "./media-ingest-proxy.js";
+
+/*
+ * Real ffmpeg runs on synthetic lavfi fixtures (test patterns and tones only).
+ * Skipped where ffmpeg is not installed.
+ */
+
+const describeFfmpeg = ffmpegAvailable() ? describe : describe.skip;
+
+interface Probe {
+  streams: Array<{
+    codec_type: string;
+    codec_name: string;
+    width?: number;
+    height?: number;
+    r_frame_rate?: string;
+    avg_frame_rate?: string;
+    color_transfer?: string;
+    side_data_list?: Array<{ rotation?: number }>;
+  }>;
+  format: { tags?: Record<string, string> };
+}
+
+async function ffprobe(file: string): Promise<Probe> {
+  const { stdout } = await runMediaTool("ffprobe", [
+    ...["-v", "error", "-print_format", "json", "-show_format", "-show_streams", file],
+  ]);
+  return JSON.parse(stdout.toString("utf8")) as Probe;
+}
+
+async function keyframeTimes(file: string): Promise<number[]> {
+  const { stdout } = await runMediaTool("ffprobe", [
+    ...["-v", "error", "-select_streams", "v:0", "-skip_frame", "nokey"],
+    ...["-show_entries", "frame=pts_time", "-of", "csv=p=0", file],
+  ]);
+  return stdout
+    .toString("utf8")
+    .trim()
+    .split("\n")
+    .map((line) => Number(line.split(",")[0]));
+}
+
+describeFfmpeg("buildProxy (ffmpeg)", () => {
+  let root: string;
+  let originals: string;
+
+  async function ffmpeg(args: string[]): Promise<void> {
+    await runMediaTool("ffmpeg", ["-v", "error", "-y", ...args]);
+  }
+
+  beforeAll(async () => {
+    root = await mkdtemp(path.join(tmpdir(), "media-proxy-"));
+    originals = path.join(root, "originals");
+    await mkdir(originals, { recursive: true });
+  });
+
+  afterAll(async () => {
+    await rm(root, { recursive: true, force: true });
+  });
+
+  it("builds a 720p CFR proxy with 1 s keyframes, AAC, poster, and sprite without metadata", async () => {
+    const source = path.join(originals, "landscape.mp4");
+    // Variable frame rate: every third frame is held a little longer.
+    await ffmpeg([
+      ...["-f", "lavfi", "-i", "testsrc2=s=1920x1080:r=30:d=3"],
+      ...["-f", "lavfi", "-i", "sine=f=440:sample_rate=44100:d=3"],
+      ...["-vf", "setpts='PTS+if(eq(mod(N,3),0),0.02/TB,0)'", "-fps_mode", "vfr"],
+      ...["-c:v", "libx264", "-preset", "ultrafast", "-g", "90", "-pix_fmt", "yuv420p"],
+      ...["-c:a", "aac", "-shortest"],
+      ...["-metadata", "location=+52.5200+013.4050/", "-metadata", "make=TestCam", source],
+    ]);
+    const outputs = await buildProxy({
+      mediaRoot: root,
+      spaceId: "space",
+      assetId: "landscape",
+      original: source,
+      still: false,
+    });
+    expect(outputs.proxyPath).toBe("space/assets/landscape/proxy.mp4");
+    expect(outputs.spritePath).toBe("space/assets/landscape/sprite.jpg");
+    expect(outputs.hdr).toBe(false);
+
+    const proxy = await ffprobe(path.join(root, outputs.proxyPath));
+    const video = proxy.streams.find((stream) => stream.codec_type === "video");
+    const audio = proxy.streams.find((stream) => stream.codec_type === "audio");
+    expect([video?.width, video?.height]).toEqual([1280, 720]);
+    expect(video?.codec_name).toBe("h264");
+    expect(video?.r_frame_rate).toBe("30/1");
+    expect(audio?.codec_name).toBe("aac");
+    expect(proxy.format.tags?.location).toBeUndefined();
+    expect(proxy.format.tags?.make).toBeUndefined();
+
+    const keyframes = await keyframeTimes(path.join(root, outputs.proxyPath));
+    expect(keyframes.length).toBeGreaterThanOrEqual(3);
+    for (let i = 1; i < keyframes.length; i++) {
+      expect((keyframes[i] ?? 0) - (keyframes[i - 1] ?? 0)).toBeLessThanOrEqual(1.05);
+    }
+
+    const sprite = await ffprobe(path.join(root, outputs.spritePath as string));
+    expect(sprite.streams[0]?.width).toBe(1600);
+    const thumb = await ffprobe(path.join(root, outputs.thumbPath));
+    expect(thumb.streams[0]?.width).toBe(480);
+    expect(thumb.format.tags?.location).toBeUndefined();
+  });
+
+  it("applies display rotation without upscaling", async () => {
+    const plain = path.join(originals, "plain.mp4");
+    const rotated = path.join(originals, "rotated.mp4");
+    await ffmpeg([
+      ...["-f", "lavfi", "-i", "testsrc2=s=640x360:r=25:d=2"],
+      ...["-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p", plain],
+    ]);
+    await ffmpeg(["-display_rotation", "90", "-i", plain, "-c", "copy", rotated]);
+    const { probe } = await probeOriginal(rotated, {});
+    expect(Math.abs(probe.rotation)).toBe(90);
+
+    const outputs = await buildProxy({
+      mediaRoot: root,
+      spaceId: "space",
+      assetId: "rotated",
+      original: rotated,
+      still: false,
+    });
+    const proxy = await ffprobe(path.join(root, outputs.proxyPath));
+    const video = proxy.streams.find((stream) => stream.codec_type === "video");
+    expect([video?.width, video?.height]).toEqual([360, 640]);
+    expect(video?.side_data_list?.some((data) => data.rotation)).toBeFalsy();
+    expect(proxy.streams.some((stream) => stream.codec_type === "audio")).toBe(false);
+  });
+
+  it("tone-maps HDR sources to SDR", async () => {
+    const hdr = path.join(originals, "hdr.mp4");
+    await ffmpeg([
+      ...["-f", "lavfi", "-i", "testsrc2=s=1280x720:r=25:d=1"],
+      ...["-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p10le"],
+      ...["-color_primaries", "bt2020", "-color_trc", "smpte2084", "-colorspace", "bt2020nc"],
+      hdr,
+    ]);
+    const outputs = await buildProxy({
+      mediaRoot: root,
+      spaceId: "space",
+      assetId: "hdr",
+      original: hdr,
+      still: false,
+    });
+    expect(outputs.hdr).toBe(true);
+    expect(outputs.tonemapped).toBe(true);
+    const proxy = await ffprobe(path.join(root, outputs.proxyPath));
+    expect(proxy.streams[0]?.color_transfer).not.toBe("smpte2084");
+  });
+
+  it("gives a photo a resized JPEG proxy and poster but no sprite", async () => {
+    const photo = path.join(originals, "photo.jpg");
+    await ffmpeg(["-f", "lavfi", "-i", "smptehdbars=s=1600x1200", "-frames:v", "1", photo]);
+    const outputs = await buildProxy({
+      mediaRoot: root,
+      spaceId: "space",
+      assetId: "photo",
+      original: photo,
+      still: true,
+    });
+    expect(outputs.proxyPath).toBe("space/assets/photo/proxy.jpg");
+    expect(outputs.spritePath).toBeNull();
+    const proxy = await ffprobe(path.join(root, outputs.proxyPath));
+    expect([proxy.streams[0]?.width, proxy.streams[0]?.height]).toEqual([960, 720]);
+  });
+
+  it("refuses to write outside the media root", async () => {
+    await expect(
+      buildProxy({
+        mediaRoot: root,
+        spaceId: "../escape",
+        assetId: "x",
+        original: path.join(originals, "plain.mp4"),
+        still: false,
+      }),
+    ).rejects.toThrow(/outside/);
+  });
+});

packages/adapters/src/media-ingest-proxy.ts

@@ -0,0 +1,223 @@
+import { mkdir, rename, rm, stat } from "node:fs/promises";
+import path from "node:path";
+import { getLogger } from "@rakazo/logging";
+import { resolveInside } from "@rakazo/media/node";
+import type { MediaToolPaths, ProbedMedia } from "./media-folder-source-probe.js";
+import { parseFfprobe, runMediaTool } from "./media-folder-source-probe.js";
+import { assetDir, tempSibling } from "./media-ingest-files.js";
+
+/*
+ * Proxy, poster and sprite generation with ffmpeg (argv only, never a shell).
+ *
+ * Videos: 720p (shorter side) H.264 with a keyframe every second, constant
+ * frame rate (variable-rate sources are normalized for frame-accurate cuts),
+ * stereo AAC, display rotation applied by ffmpeg's autorotate, HDR (PQ/HLG)
+ * tone-mapped to SDR BT.709 when zscale and tonemap are available. Photos get
+ * a resized JPEG proxy. Posters and sprites are JPEGs written with
+ * `-map_metadata -1`, so no capture date, GPS, or device tags survive.
+ * Every output is written to a temp name and renamed into place.
+ */
+
+const PROXY_SHORT_SIDE = 720;
+const THUMB_WIDTH = 480;
+const SPRITE_TILES = 10;
+const SPRITE_TILE_WIDTH = 160;
+const MAX_PROXY_FPS = 60;
+const HDR_TRANSFERS = new Set(["smpte2084", "arib-std-b67"]);
+
+export interface ProxyOutputs {
+  proxyPath: string;
+  thumbPath: string;
+  spritePath: string | null;
+  probe: ProbedMedia;
+  hdr: boolean;
+  tonemapped: boolean;
+}
+
+interface ProbeStream {
+  codec_type?: string;
+  color_transfer?: string;
+}
+
+/** Even-dimension scale so the shorter side is at most 720 px, never upscaled. */
+const SCALE_720 =
+  `scale=w='if(gte(iw,ih),-2,trunc(min(${PROXY_SHORT_SIDE},iw)/2)*2)'` +
+  `:h='if(gte(iw,ih),trunc(min(${PROXY_SHORT_SIDE},ih)/2)*2,-2)'`;
+
+const TONEMAP =
+  "zscale=t=linear:npl=100,format=gbrpf32le,zscale=p=bt709," +
+  "tonemap=tonemap=hable:desat=0,zscale=t=bt709:m=bt709:r=tv";
+
+let tonemapSupport: Promise<boolean> | null = null;
+
+/** Whether this ffmpeg build has zscale and tonemap (cached per process). */
+function hasTonemapFilters(ffmpeg: string): Promise<boolean> {
+  tonemapSupport ??= runMediaTool(ffmpeg, ["-hide_banner", "-filters"])
+    .then(({ stdout }) => {
+      const text = stdout.toString("utf8");
+      return /\szscale\s/.test(text) && /\stonemap\s/.test(text);
+    })
+    .catch(() => false);
+  return tonemapSupport;
+}
+
+export async function probeOriginal(
+  file: string,
+  tools: MediaToolPaths,
+  signal?: AbortSignal,
+): Promise<{ probe: ProbedMedia; hdr: boolean }> {
+  const { stdout } = await runMediaTool(
+    tools.ffprobePath ?? "ffprobe",
+    ["-v", "error", "-print_format", "json", "-show_format", "-show_streams", "--", file],
+    { signal },
+  );
+  const json = JSON.parse(stdout.toString("utf8")) as { streams?: ProbeStream[] };
+  const video = json.streams?.find((stream) => stream.codec_type === "video");
+  return {
+    probe: parseFfprobe(json as Parameters<typeof parseFfprobe>[0]),
+    hdr: HDR_TRANSFERS.has(video?.color_transfer ?? ""),
+  };
+}
+
+async function ffmpegTo(
+  target: string,
+  ext: string,
+  build: (out: string) => string[],
+  tools: MediaToolPaths,
+  signal?: AbortSignal,
+): Promise<void> {
+  await mkdir(path.dirname(target), { recursive: true });
+  const temp = tempSibling(target, ext);
+  try {
+    await runMediaTool(tools.ffmpegPath ?? "ffmpeg", build(temp), { signal });
+    if ((await stat(temp)).size === 0) throw new Error("ffmpeg produced an empty file");
+    await rename(temp, target);
+  } finally {
+    await rm(temp, { force: true });
+  }
+}
+
+function proxyFps(probe: ProbedMedia): number {
+  const fps = probe.fps && probe.fps > 0 ? probe.fps : 30;
+  return Math.min(MAX_PROXY_FPS, Math.max(1, Math.round(fps)));
+}
+
+/** Build proxy, poster, and sprite for one original. Returns media-root-relative paths. */
+export async function buildProxy(input: {
+  mediaRoot: string;
+  spaceId: string;
+  assetId: string;
+  original: string;
+  still: boolean;
+  tools?: MediaToolPaths;
+  signal?: AbortSignal;
+}): Promise<ProxyOutputs> {
+  const tools = input.tools ?? {};
+  const signal = input.signal;
+  const dir = assetDir(input.spaceId, input.assetId);
+  const { probe, hdr } = await probeOriginal(input.original, tools, signal);
+  const still = input.still || !probe.hasVideo || !probe.durationMs;
+  const ffmpeg = tools.ffmpegPath ?? "ffmpeg";
+
+  let tonemapped = false;
+  const filters: string[] = [];
+  if (hdr) {
+    if (await hasTonemapFilters(ffmpeg)) {
+      filters.push(TONEMAP);
+      tonemapped = true;
+    } else {
+      getLogger().warn("media proxy: HDR source kept without tone mapping", {
+        reason: "ffmpeg lacks zscale or tonemap",
+      });
+    }
+  }
+  filters.push(SCALE_720);
+
+  const thumbRel = `${dir}/thumb.jpg`;
+  const thumb = await resolveInside(input.mediaRoot, thumbRel);
+
+  if (still) {
+    const proxyRel = `${dir}/proxy.jpg`;
+    const proxy = await resolveInside(input.mediaRoot, proxyRel);
+    await ffmpegTo(
+      proxy,
+      ".jpg",
+      (out) => [
+        ...["-v", "error", "-y", "-i", input.original, "-map", "0:v:0"],
+        ...["-map_metadata", "-1", "-frames:v", "1", "-update", "1"],
+        ...["-vf", [...filters, "format=yuvj420p"].join(","), "-q:v", "3", out],
+      ],
+      tools,
+      signal,
+    );
+    await writeThumb(proxy, thumb, 0, tools, signal);
+    return { proxyPath: proxyRel, thumbPath: thumbRel, spritePath: null, probe, hdr, tonemapped };
+  }
+
+  const proxyRel = `${dir}/proxy.mp4`;
+  const proxy = await resolveInside(input.mediaRoot, proxyRel);
+  filters.push(`fps=${proxyFps(probe)}`, "format=yuv420p");
+  await ffmpegTo(
+    proxy,
+    ".mp4",
+    (out) => [
+      ...["-v", "error", "-y", "-i", input.original],
+      ...["-map", "0:v:0", "-map", "0:a:0?", "-map_metadata", "-1", "-map_chapters", "-1"],
+      ...["-vf", filters.join(",")],
+      ...["-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-profile:v", "high"],
+      ...["-force_key_frames", "expr:gte(t,n_forced*1)", "-sc_threshold", "0"],
+      ...["-c:a", "aac", "-b:a", "128k", "-ac", "2", "-ar", "48000"],
+      ...["-movflags", "+faststart", "-f", "mp4", out],
+    ],
+    tools,
+    signal,
+  );
+
+  const durationMs = probe.durationMs ?? 1000;
+  await writeThumb(proxy, thumb, Math.min(1000, Math.floor(durationMs / 2)), tools, signal);
+
+  const spriteRel = `${dir}/sprite.jpg`;
+  const sprite = await resolveInside(input.mediaRoot, spriteRel);
+  const sampleFps = (SPRITE_TILES / Math.max(0.1, durationMs / 1000)).toFixed(6);
+  await ffmpegTo(
+    sprite,
+    ".jpg",
+    (out) => [
+      ...["-v", "error", "-y", "-i", proxy, "-map", "0:v:0", "-map_metadata", "-1"],
+      ...["-vf", `fps=${sampleFps},scale=${SPRITE_TILE_WIDTH}:-2,tile=${SPRITE_TILES}x1`],
+      ...["-frames:v", "1", "-update", "1", "-q:v", "4", out],
+    ],
+    tools,
+    signal,
+  );
+  return {
+    proxyPath: proxyRel,
+    thumbPath: thumbRel,
+    spritePath: spriteRel,
+    probe,
+    hdr,
+    tonemapped,
+  };
+}
+
+async function writeThumb(
+  source: string,
+  target: string,
+  atMs: number,
+  tools: MediaToolPaths,
+  signal?: AbortSignal,
+): Promise<void> {
+  await ffmpegTo(
+    target,
+    ".jpg",
+    (out) => [
+      // Seeking a single image, even to 0, yields no frame.
+      ...["-v", "error", "-y", ...(atMs > 0 ? ["-ss", (atMs / 1000).toFixed(3)] : [])],
+      ...["-i", source],
+      ...["-map", "0:v:0", "-map_metadata", "-1", "-frames:v", "1", "-update", "1"],
+      ...["-vf", `scale='min(${THUMB_WIDTH},iw)':-2`, "-q:v", "3", out],
+    ],
+    tools,
+    signal,
+  );
+}

packages/adapters/src/media-ingest.postgres.test.ts

@@ -0,0 +1,697 @@
+import { createHash, randomUUID } from "node:crypto";
+import { mkdir, mkdtemp, readdir, rm, stat, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import type {
+  BackgroundJob,
+  BackgroundJobHandlers,
+  JobPublisher,
+  MediaInferenceProvider,
+} from "@rakazo/adapter-kit";
+import { dispatchBackgroundJob } from "@rakazo/adapter-kit";
+import type { MediaCapabilities } from "@rakazo/contracts";
+import type { MediaRepos, PrismaClient } from "@rakazo/db";
+import { bootstrapUserSpace, createDb, createMediaRepos } from "@rakazo/db";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import type { MediaModelClient, MediaRealtimeEvent, MediaServiceDeps } from "./media-context.js";
+import { EmulatorMediaInferenceProvider } from "./media-emulator.js";
+import { ffmpegAvailable } from "./media-emulator-clips.js";
+import { EmulatorMediaSourceProvider } from "./media-emulator-source.js";
+import { FolderMediaSourceProvider } from "./media-folder-source.js";
+import type { MediaIngestHandlerOptions } from "./media-ingest.js";
+import {
+  createMediaIngestHandlers,
+  createMediaIngestService,
+  MEDIA_BUSY_RETRY_MS,
+  MEDIA_PAUSE_RETRY_MS,
+  MEDIA_STALE_CLAIM_MS,
+  resumeMediaIngest,
+} from "./media-ingest.js";
+import { MediaSidecarError } from "./media-sidecar.js";
+
+/*
+ * The ingest job chain against PostgreSQL, the deterministic media emulator,
+ * and real ffmpeg on synthetic clips. Jobs are recorded and drained by hand.
+ */
+
+const databaseUrl = process.env.DATABASE_URL;
+const describePostgres =
+  process.env.VERIFY_DATABASE && databaseUrl && ffmpegAvailable()
+    ? describe.sequential
+    : describe.skip;
+
+const signupEnv = { signupsEnabled: "true", signupAllowlist: "" };
+const LIMITS = { gpu: 1, cpu: 2 };
+
+/** Records jobs keyed like the real queue: a replace key replaces the pending job. */
+class RecordingJobs implements JobPublisher {
+  readonly pending = new Map<string, BackgroundJob>();
+  readonly log: BackgroundJob[] = [];
+
+  async enqueue(job: BackgroundJob): Promise<void> {
+    this.log.push(job);
+    this.pending.set(job.replaceKey ?? `${job.name}:${randomUUID()}`, job);
+  }
+
+  async cancel(key: string): Promise<void> {
+    this.pending.delete(key);
+  }
+
+  async close(): Promise<void> {}
+
+  reset(): void {
+    this.pending.clear();
+    this.log.length = 0;
+  }
+
+  named(name: BackgroundJob["name"]): BackgroundJob[] {
+    return this.log.filter((job) => job.name === name);
+  }
+}
+
+/** Run pending jobs (delays ignored) until the queue is empty. */
+async function drain(
+  handlers: ReturnType<typeof createMediaIngestHandlers>,
+  jobs: RecordingJobs,
+  limit = 1000,
+): Promise<number> {
+  for (let ran = 0; ran < limit; ran++) {
+    const next = jobs.pending.entries().next();
+    if (next.done) return ran;
+    const [key, job] = next.value;
+    jobs.pending.delete(key);
+    await dispatchBackgroundJob(handlers as BackgroundJobHandlers, job.name, job.payload);
+  }
+  throw new Error("media job queue did not drain");
+}
+
+describePostgres("media ingest jobs (PostgreSQL)", () => {
+  let prisma: PrismaClient;
+  let repos: MediaRepos;
+  let closeDb: () => Promise<void>;
+  let mediaRoot: string;
+  let scratch: string;
+  const userIds: string[] = [];
+
+  beforeAll(async () => {
+    const db = createDb(databaseUrl as string, { poolMax: 8 });
+    prisma = db.prisma;
+    closeDb = async () => {
+      await db.prisma.$disconnect();
+      await db.pool.end();
+    };
+    repos = createMediaRepos(prisma);
+    scratch = await mkdtemp(path.join(tmpdir(), "media-ingest-"));
+    mediaRoot = path.join(scratch, "media");
+    await mkdir(mediaRoot, { recursive: true });
+  });
+
+  afterAll(async () => {
+    if (!prisma) return;
+    const memberships = await prisma.member.findMany({
+      where: { userId: { in: userIds } },
+      select: { organizationId: true },
+    });
+    await prisma.organization.deleteMany({
+      where: { id: { in: memberships.map((member) => member.organizationId) } },
+    });
+    await prisma.user.deleteMany({ where: { id: { in: userIds } } });
+    await closeDb();
+    await rm(scratch, { recursive: true, force: true });
+  });
+
+  async function newSpace() {
+    const id = `mi${randomUUID().replaceAll("-", "")}`;
+    userIds.push(id);
+    await prisma.user.create({ data: { id, name: "Test", email: `${id}@rakazo.test` } });
+    const { spaceId } = await bootstrapUserSpace(prisma, { id }, signupEnv, {
+      claimDeploymentOwner: false,
+    });
+    return { userId: id, spaceId };
+  }
+
+  function setup(
+    overrides: {
+      inference?: MediaInferenceProvider | null;
+      model?: MediaModelClient;
+      options?: MediaIngestHandlerOptions;
+    } = {},
+  ) {
+    const jobs = new RecordingJobs();
+    const events: Array<{ spaceId: string; event: MediaRealtimeEvent }> = [];
+    const inference =
+      overrides.inference === undefined
+        ? new EmulatorMediaInferenceProvider({ root: mediaRoot })
+        : overrides.inference;
+    const deps: MediaServiceDeps = {
+      prisma,
+      repos,
+      jobs,
+      inference,
+      renderer: {} as MediaServiceDeps["renderer"],
+      motion: {} as MediaServiceDeps["motion"],
+      publisher: {} as MediaServiceDeps["publisher"],
+      shortener: null,
+      sources: {
+        folder: new FolderMediaSourceProvider(),
+        emulator: new EmulatorMediaSourceProvider({
+          fixturesDir: path.join(mediaRoot, "emulator-library"),
+        }),
+      },
+      mediaRoot,
+      profiles: async () => {
+        throw new Error("not used");
+      },
+      artifacts: {} as MediaServiceDeps["artifacts"],
+      model: overrides.model ?? { generate: async () => null },
+      publish: async (spaceId, event) => {
+        events.push({ spaceId, event });
+      },
+      now: () => new Date(),
+    };
+    const handlers = createMediaIngestHandlers(deps, {
+      limits: LIMITS,
+      minFreeBytes: 0,
+      ...overrides.options,
+    });
+    return { deps, jobs, events, handlers, service: createMediaIngestService(deps) };
+  }
+
+  async function emulatorLibrary(count: number, seed: number) {
+    const space = await newSpace();
+    const source = await repos.createSource(space.spaceId, {
+      kind: "emulator",
+      label: "Emulator",
+      ownerUserId: space.userId,
+      config: { count, seed },
+    });
+    return { ...space, sourceId: source.id };
+  }
+
+  async function states(spaceId: string) {
+    const rows = await prisma.mediaAsset.findMany({
+      where: { spaceId },
+      select: { ingestState: true },
+    });
+    return rows.reduce<Record<string, number>>((acc, row) => {
+      acc[row.ingestState] = (acc[row.ingestState] ?? 0) + 1;
+      return acc;
+    }, {});
+  }
+
+  async function sessionMembers(spaceId: string) {
+    const rows = await prisma.mediaCollection.findMany({
+      where: { spaceId, kind: "session" },
+      select: {
+        id: true,
+        clusterKey: true,
+        labelSource: true,
+        items: { select: { assetId: true }, orderBy: { position: "asc" } },
+      },
+      orderBy: [{ startAt: "asc" }, { id: "asc" }],
+    });
+    return rows.map((row) => ({ ...row, assetIds: row.items.map((item) => item.assetId) }));
+  }
+
+  it("indexes a 1,000-file folder incrementally; an unchanged second sync enqueues nothing", async () => {
+    const space = await newSpace();
+    const folder = path.join(scratch, "folder-1000");
+    await mkdir(path.join(folder, "trip"), { recursive: true });
+    for (let index = 0; index < 1000; index++) {
+      const dir = index < 500 ? folder : path.join(folder, "trip");
+      await writeFile(path.join(dir, `clip-${String(index).padStart(4, "0")}.mp4`), "");
+    }
+    const source = await repos.createSource(space.spaceId, {
+      kind: "folder",
+      label: "Folder",
+      ownerUserId: space.userId,
+      config: { path: folder },
+    });
+    const { jobs, handlers, events } = setup();
+
+    await handlers["media.sync"]({ sourceId: source.id });
+    expect(await prisma.mediaAsset.count({ where: { spaceId: space.spaceId } })).toBe(1000);
+    // Only a window of proxy work is queued; finished jobs refill it.
+    const proxyKeys = new Set(jobs.named("media.proxy").map((job) => job.replaceKey));
+    expect(proxyKeys.size).toBe(LIMITS.cpu * 2);
+    expect(jobs.named("media.cluster")).toHaveLength(1);
+    expect(events.map(({ event }) => event)).toContainEqual({
+      type: "media.source",
+      sourceId: source.id,
+      status: "idle",
+      error: null,
+    });
+
+    jobs.reset();
+    await handlers["media.sync"]({ sourceId: source.id });
+    expect(jobs.log).toEqual([]);
+    expect(await prisma.mediaAsset.count({ where: { spaceId: space.spaceId } })).toBe(1000);
+
+    // One file changes and one disappears.
+    await writeFile(path.join(folder, "clip-0001.mp4"), "changed");
+    await rm(path.join(folder, "trip", "clip-0999.mp4"));
+    await handlers["media.sync"]({ sourceId: source.id });
+    expect(await prisma.mediaAsset.count({ where: { spaceId: space.spaceId } })).toBe(999);
+    const changed = await prisma.mediaAsset.findFirstOrThrow({
+      where: { spaceId: space.spaceId, externalId: "clip-0001.mp4" },
+    });
+    expect(changed.sizeBytes).toBe(7n);
+    expect(jobs.named("media.cluster")).toHaveLength(1);
+    const source2 = await repos.getSource(space.spaceId, source.id);
+    expect(source2.status).toBe("idle");
+  });
+
+  it("does not delete a library when its source suddenly lists nothing", async () => {
+    const space = await newSpace();
+    const folder = path.join(scratch, "unmounted");
+    await mkdir(folder, { recursive: true });
+    await writeFile(path.join(folder, "a.mp4"), "");
+    const source = await repos.createSource(space.spaceId, {
+      kind: "folder",
+      label: "Folder",
+      ownerUserId: space.userId,
+      config: { path: folder },
+    });
+    const { handlers } = setup();
+    await handlers["media.sync"]({ sourceId: source.id });
+    await rm(path.join(folder, "a.mp4"));
+    await handlers["media.sync"]({ sourceId: source.id });
+    expect(await prisma.mediaAsset.count({ where: { spaceId: space.spaceId } })).toBe(1);
+    const record = await repos.getSource(space.spaceId, source.id);
+    expect(record.status).toBe("error");
+    expect(record.error).toMatch(/listed no files/);
+  });
+
+  describe("emulator library end to end", () => {
+    let library: Awaited<ReturnType<typeof emulatorLibrary>>;
+    let env: ReturnType<typeof setup>;
+
+    beforeAll(async () => {
+      library = await emulatorLibrary(15, 3);
+      env = setup();
+      await env.handlers["media.sync"]({ sourceId: library.sourceId });
+      await drain(env.handlers, env.jobs);
+    }, 240_000);
+
+    it("proxies and analyses every asset and writes scored segments", async () => {
+      expect(await states(library.spaceId)).toEqual({ analyzed: 15 });
+      const assets = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } });
+      for (const asset of assets) {
+        expect(asset.proxyPath).toMatch(/^.+\/assets\/.+\/proxy\.(mp4|jpg)$/);
+        expect(asset.contentHash).toMatch(/^[0-9a-f]{64}$/);
+        expect(asset.analysisVersion).toBe(1);
+        expect((await stat(path.join(mediaRoot, asset.proxyPath as string))).size).toBeGreaterThan(
+          0,
+        );
+      }
+      const segments = await prisma.mediaSegment.count({ where: { spaceId: library.spaceId } });
+      expect(segments).toBeGreaterThanOrEqual(15);
+      const kinds = await prisma.mediaAnalysis.groupBy({
+        by: ["kind"],
+        where: { spaceId: library.spaceId },
+        _count: true,
+      });
+      expect(kinds.map((row) => row.kind).sort()).toEqual(
+        ["audio", "classifier", "embeddings", "faces", "saliency", "scenes", "transcript"].sort(),
+      );
+      const progress = env.events
+        .filter(({ event }) => event.type === "media.asset")
+        .map(({ event }) => (event as { ingestState: string }).ingestState);
+      expect(new Set(progress)).toEqual(new Set(["proxying", "proxied", "analyzing", "analyzed"]));
+    });
+
+    it("keeps face embeddings in face tracks only and suggests one person per identity", async () => {
+      const tracks = await prisma.mediaFaceTrack.findMany({ where: { spaceId: library.spaceId } });
+      expect(tracks).toHaveLength(4);
+      expect(tracks.every((track) => track.embedding && track.personId)).toBe(true);
+      const people = await prisma.mediaPerson.findMany({ where: { spaceId: library.spaceId } });
+      expect(people).toHaveLength(2);
+      expect(new Set(tracks.map((track) => track.personId)).size).toBe(2);
+      const faces = await prisma.mediaAnalysis.findMany({
+        where: { spaceId: library.spaceId, kind: "faces" },
+      });
+      for (const row of faces) {
+        expect(row.vectors).toBeNull();
+        expect(JSON.stringify(row.data)).not.toMatch(/embedding/i);
+      }
+    });
+
+    it("builds sessions by time and place and themes with rule labels without consent", async () => {
+      const sessions = await sessionMembers(library.spaceId);
+      expect(sessions).toHaveLength(3);
+      expect(sessions.flatMap((session) => session.assetIds)).toHaveLength(15);
+      const themes = await prisma.mediaCollection.findMany({
+        where: { spaceId: library.spaceId, kind: "theme" },
+        include: { items: true },
+      });
+      expect(themes).toHaveLength(2);
+      expect(themes.every((theme) => theme.labelSource === "rule")).toBe(true);
+      expect(env.events.map(({ event }) => event)).toContainEqual({
+        type: "media.collections",
+        kind: "theme",
+      });
+    });
+
+    it("labels themes through the model only from cover thumbnails, never calling out with face data", async () => {
+      const calls: Array<Parameters<MediaModelClient["generate"]>[0]> = [];
+      const model: MediaModelClient = {
+        generate: async <T>(input: Parameters<MediaModelClient["generate"]>[0]) => {
+          calls.push(input);
+          return { title: "Test pattern" } as T;
+        },
+      };
+      const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => {
+        throw new Error("no outbound calls during clustering");
+      });
+      try {
+        const withModel = setup({ model });
+        await withModel.handlers["media.cluster"]({ spaceId: library.spaceId });
+      } finally {
+        expect(fetchSpy).not.toHaveBeenCalled();
+        fetchSpy.mockRestore();
+      }
+      expect(calls).toHaveLength(2);
+      for (const call of calls) {
+        expect(call.userId).toBe(library.userId);
+        expect(call.images?.length).toBeGreaterThan(0);
+        expect(call.images?.length).toBeLessThanOrEqual(9);
+        expect(call.images?.every((image) => image.mimeType === "image/jpeg")).toBe(true);
+        expect(`${call.prompt}${call.instructions}`).not.toMatch(/person|face|embedding/i);
+      }
+      const themes = await prisma.mediaCollection.findMany({
+        where: { spaceId: library.spaceId, kind: "theme" },
+      });
+      expect(themes.map((theme) => [theme.title, theme.labelSource])).toEqual([
+        ["Test pattern", "model"],
+        ["Test pattern", "model"],
+      ]);
+    });
+
+    it("re-clustering keeps the user's merges and splits", async () => {
+      const [first, second, third] = await sessionMembers(library.spaceId);
+      if (!first || !second || !third) throw new Error("expected three sessions");
+      await repos.mergeCollections(library.spaceId, [first.id, second.id]);
+      const splitAt = third.assetIds[2] as string;
+      await repos.splitCollection(library.spaceId, third.id, splitAt);
+      const before = await sessionMembers(library.spaceId);
+
+      await env.handlers["media.cluster"]({ spaceId: library.spaceId });
+      const after = await sessionMembers(library.spaceId);
+      expect(after.map((session) => session.assetIds)).toEqual(
+        before.map((session) => session.assetIds),
+      );
+      const all = after.flatMap((session) => session.assetIds);
+      expect(new Set(all).size).toBe(all.length);
+      expect(all).toHaveLength(15);
+    });
+
+    it("re-runs nothing for unchanged content and adds rows for a new model version", async () => {
+      const asset = await prisma.mediaAsset.findFirstOrThrow({
+        where: { spaceId: library.spaceId, kind: "video" },
+        orderBy: { externalId: "asc" },
+      });
+      const analyses = await prisma.mediaAnalysis.count({ where: { assetId: asset.id } });
+      const proxyFile = path.join(mediaRoot, asset.proxyPath as string);
+      const proxyMtime = (await stat(proxyFile)).mtimeMs;
+      const inference = new EmulatorMediaInferenceProvider({ root: mediaRoot });
+      const scenes = vi.spyOn(inference, "detectScenes");
+      const run = setup({ inference });
+
+      // A repeated job for an analysed asset is a no-op.
+      await run.handlers["media.analyze"]({ assetId: asset.id });
+      expect(scenes).not.toHaveBeenCalled();
+
+      // Re-proxy with unchanged bytes keeps the proxy and reuses every analysis.
+      await prisma.mediaAsset.update({ where: { id: asset.id }, data: { ingestState: "indexed" } });
+      await run.handlers["media.proxy"]({ assetId: asset.id });
+      expect((await stat(proxyFile)).mtimeMs).toBe(proxyMtime);
+      await run.handlers["media.analyze"]({ assetId: asset.id });
+      expect(scenes).not.toHaveBeenCalled();
+      expect(await prisma.mediaAnalysis.count({ where: { assetId: asset.id } })).toBe(analyses);
+      expect(
+        (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: asset.id } })).ingestState,
+      ).toBe("analyzed");
+
+      // A new embedding model writes new rows and keeps the old ones.
+      const upgraded = new EmulatorMediaInferenceProvider({ root: mediaRoot });
+      const base = upgraded.capabilities.bind(upgraded);
+      upgraded.capabilities = async (signal?: AbortSignal): Promise<MediaCapabilities> => {
+        const caps = await base(signal);
+        return {
+          ...caps,
+          operations: {
+            ...caps.operations,
+            embedFrames: { ...caps.operations.embedFrames, model: "emulator-v2" },
+          },
+        };
+      };
+      await prisma.mediaAsset.update({ where: { id: asset.id }, data: { ingestState: "proxied" } });
+      await setup({ inference: upgraded }).handlers["media.analyze"]({ assetId: asset.id });
+      const versions = await prisma.mediaAnalysis.findMany({
+        where: { assetId: asset.id, kind: "embeddings" },
+        select: { modelVersion: true },
+        orderBy: { modelVersion: "asc" },
+      });
+      expect(versions.map((row) => row.modelVersion)).toEqual([
+        "emulator:emulator",
+        "emulator:emulator-v2",
+      ]);
+      expect(await prisma.mediaAnalysis.count({ where: { assetId: asset.id } })).toBe(analyses + 1);
+    });
+
+    it("removes a source with its assets and derived files", async () => {
+      const assets = await prisma.mediaAsset.findMany({
+        where: { spaceId: library.spaceId },
+        select: { id: true },
+      });
+      await env.service.removeSource(
+        { spaceId: library.spaceId, userId: library.userId },
+        library.sourceId,
+      );
+      expect(await prisma.mediaAsset.count({ where: { spaceId: library.spaceId } })).toBe(0);
+      const dir = path.join(mediaRoot, library.spaceId, "assets");
+      const left = await readdir(dir).catch(() => []);
+      expect(left.filter((name) => assets.some((asset) => asset.id === name))).toEqual([]);
+    });
+  });
+
+  it("pauses on a full disk with a clear message and resumes afterwards", async () => {
+    const library = await emulatorLibrary(2, 11);
+    const full = setup({ options: { minFreeBytes: Number.MAX_SAFE_INTEGER } });
+    await full.handlers["media.sync"]({ sourceId: library.sourceId });
+    const [asset] = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } });
+    if (!asset) throw new Error("expected an asset");
+    full.jobs.reset();
+    const before = Date.now();
+
+    await full.handlers["media.proxy"]({ assetId: asset.id });
+    expect(
+      (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: asset.id } })).ingestState,
+    ).toBe("indexed");
+    const source = await repos.getSource(library.spaceId, library.sourceId);
+    expect(source.error).toBe("Media processing is paused because the disk is full");
+    const [retry] = full.jobs.named("media.proxy");
+    expect(retry?.replaceKey).toBe(`media.proxy:${asset.id}`);
+    expect(retry?.availableAt?.getTime()).toBeGreaterThanOrEqual(before + MEDIA_PAUSE_RETRY_MS);
+
+    const roomy = setup();
+    await roomy.handlers["media.proxy"]({ assetId: asset.id });
+    expect(
+      (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: asset.id } })).ingestState,
+    ).toBe("proxied");
+    expect((await repos.getSource(library.spaceId, library.sourceId)).error).toBeNull();
+  });
+
+  it("defers jobs beyond the concurrency limit instead of holding a worker slot", async () => {
+    const library = await emulatorLibrary(2, 12);
+    const run = setup({ options: { limits: { gpu: 1, cpu: 1 } } });
+    await run.handlers["media.sync"]({ sourceId: library.sourceId });
+    const assets = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } });
+    run.jobs.reset();
+    const before = Date.now();
+    await Promise.all(assets.map((asset) => run.handlers["media.proxy"]({ assetId: asset.id })));
+    const deferred = run.jobs
+      .named("media.proxy")
+      .filter((job) => (job.availableAt?.getTime() ?? 0) >= before + MEDIA_BUSY_RETRY_MS);
+    expect(deferred).toHaveLength(1);
+    expect(await states(library.spaceId)).toEqual({ indexed: 1, proxied: 1 });
+  });
+
+  it("defers analysis while the sidecar is unreachable", async () => {
+    const library = await emulatorLibrary(1, 13);
+    const inference = new EmulatorMediaInferenceProvider({ root: mediaRoot });
+    vi.spyOn(inference, "detectScenes").mockRejectedValue(
+      new MediaSidecarError("unreachable", "media sidecar is unreachable", null, "detectScenes"),
+    );
+    const run = setup({ inference });
+    await run.handlers["media.sync"]({ sourceId: library.sourceId });
+    const [asset] = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } });
+    if (!asset) throw new Error("expected an asset");
+    await run.handlers["media.proxy"]({ assetId: asset.id });
+    run.jobs.reset();
+    await run.handlers["media.analyze"]({ assetId: asset.id });
+    expect(await states(library.spaceId)).toEqual({ proxied: 1 });
+    expect(run.jobs.named("media.analyze")[0]?.availableAt).toBeInstanceOf(Date);
+  });
+
+  it("returns an aborted proxy to the queue state and takes over stale claims", async () => {
+    const library = await emulatorLibrary(2, 14);
+    const controller = new AbortController();
+    controller.abort(new Error("worker stopping"));
+    const stopping = setup({ options: { signal: controller.signal } });
+    await setup().handlers["media.sync"]({ sourceId: library.sourceId });
+    const [first, second] = await prisma.mediaAsset.findMany({
+      where: { spaceId: library.spaceId },
+      orderBy: { externalId: "asc" },
+    });
+    if (!first || !second) throw new Error("expected two assets");
+    await expect(stopping.handlers["media.proxy"]({ assetId: first.id })).rejects.toThrow(
+      /abort|stopping/,
+    );
+    expect(
+      (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: first.id } })).ingestState,
+    ).toBe("indexed");
+
+    // A fresh claim by another worker is left alone; a stale one is taken over.
+    const run = setup();
+    await prisma.mediaAsset.update({ where: { id: second.id }, data: { ingestState: "proxying" } });
+    await run.handlers["media.proxy"]({ assetId: second.id });
+    expect(
+      (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: second.id } })).ingestState,
+    ).toBe("proxying");
+    const stale = new Date(Date.now() - MEDIA_STALE_CLAIM_MS - 60_000);
+    await prisma.$executeRaw`UPDATE media_assets SET "updatedAt" = ${stale} WHERE id = ${second.id}`;
+    await run.handlers["media.proxy"]({ assetId: second.id });
+    expect(
+      (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: second.id } })).ingestState,
+    ).toBe("proxied");
+  });
+
+  it("resumes stranded work after a restart", async () => {
+    const library = await emulatorLibrary(3, 15);
+    const run = setup();
+    await run.handlers["media.sync"]({ sourceId: library.sourceId });
+    const [stranded] = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } });
+    if (!stranded) throw new Error("expected an asset");
+    // The queue was lost and one asset was claimed by a worker that died.
+    run.jobs.reset();
+    const stale = new Date(Date.now() - MEDIA_STALE_CLAIM_MS - 60_000);
+    await prisma.$executeRaw`UPDATE media_assets SET "ingestState" = 'proxying', "updatedAt" = ${stale} WHERE id = ${stranded.id}`;
+
+    expect(await resumeMediaIngest(run.deps, LIMITS)).toBeGreaterThanOrEqual(1);
+    expect(run.jobs.named("media.proxy").map((job) => job.replaceKey)).toContain(
+      `media.proxy:${stranded.id}`,
+    );
+    // Other tests' spaces share this database; only this library's work runs here.
+    const ours = new Set(
+      (await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } })).map(
+        (asset) => `media.proxy:${asset.id}`,
+      ),
+    );
+    for (const key of [...run.jobs.pending.keys()]) {
+      if (!ours.has(key)) run.jobs.pending.delete(key);
+    }
+    await drain(run.handlers, run.jobs);
+    expect(await states(library.spaceId)).toEqual({ analyzed: 3 });
+    expect(run.jobs.named("media.cluster").length).toBeGreaterThan(0);
+  });
+
+  it("ignores jobs for assets and sources that no longer exist", async () => {
+    const { handlers } = setup();
+    await expect(handlers["media.proxy"]({ assetId: "missing" })).resolves.toBeUndefined();
+    await expect(handlers["media.analyze"]({ assetId: "missing" })).resolves.toBeUndefined();
+    await expect(handlers["media.sync"]({ sourceId: "missing" })).resolves.toBeUndefined();
+  });
+
+  it("moves a completed upload into place and starts its proxy", async () => {
+    const space = await newSpace();
+    const actor = { spaceId: space.spaceId, userId: space.userId };
+    const device = await repos.createSource(space.spaceId, {
+      kind: "ios_library",
+      label: "Phone",
+      ownerUserId: space.userId,
+      config: { deviceId: "device-1" },
+    });
+    const { service, jobs } = setup();
+    const pushed = await service.pushItems(actor, {
+      sourceId: device.id,
+      items: [
+        {
+          externalId: "IMG_0001",
+          kind: "video",
+          capturedAt: new Date(Date.UTC(2026, 0, 2)).toISOString(),
+          durationMs: 4000,
+          width: 320,
+          height: 180,
+          fps: 25,
+          gpsLat: null,
+          gpsLon: null,
+          deviceModel: null,
+          albumNames: [],
+          isFavorite: false,
+          fileName: "IMG_0001.mp4",
+        },
+      ],
+      deletedExternalIds: [],
+      syncToken: "token-1",
+      complete: true,
+    });
+    expect(pushed).toEqual({ created: 1, updated: 0, unchanged: 0, deleted: 0 });
+    const asset = await prisma.mediaAsset.findFirstOrThrow({ where: { spaceId: space.spaceId } });
+    expect(asset.ingestState).toBe("awaiting_upload");
+
+    const uploads = path.join(mediaRoot, space.spaceId, "uploads");
+    await mkdir(uploads, { recursive: true });
+    const temp = path.join(uploads, "part-1");
+    const bytes = Buffer.from("synthetic upload bytes");
+    await writeFile(temp, bytes);
+    const sha256 = createHash("sha256").update(bytes).digest("hex");
+
+    await expect(
+      service.uploadCompleted(actor, {
+        assetId: asset.id,
+        tempPath: temp,
+        sha256: "0".repeat(64),
+        fileName: "IMG_0001.mp4",
+      }),
+    ).rejects.toThrow(/hash mismatch/);
+    const outside = path.join(mediaRoot, "elsewhere");
+    await writeFile(outside, bytes);
+    await expect(
+      service.uploadCompleted(actor, {
+        assetId: asset.id,
+        tempPath: outside,
+        sha256,
+        fileName: "IMG_0001.mp4",
+      }),
+    ).rejects.toThrow(/upload area/);
+    await expect(
+      service.uploadCompleted(actor, {
+        assetId: asset.id,
+        tempPath: temp,
+        sha256,
+        fileName: "notes.txt",
+      }),
+    ).rejects.toThrow(/Unsupported/);
+
+    jobs.reset();
+    await service.uploadCompleted(actor, {
+      assetId: asset.id,
+      tempPath: temp,
+      sha256,
+      fileName: "IMG_0001.mp4",
+    });
+    const stored = await prisma.mediaAsset.findUniqueOrThrow({ where: { id: asset.id } });
+    expect(stored.storage).toBe("uploaded");
+    expect(stored.ingestState).toBe("indexed");
+    expect(stored.contentHash).toBe(sha256);
+    expect(stored.storagePath).toBe(`${space.spaceId}/assets/${asset.id}/original.mp4`);
+    expect((await stat(path.join(mediaRoot, stored.storagePath as string))).size).toBe(
+      bytes.length,
+    );
+    expect(jobs.named("media.proxy").map((job) => job.replaceKey)).toEqual([
+      `media.proxy:${asset.id}`,
+    ]);
+    await expect(service.requestSync(actor, device.id)).rejects.toThrow(/device/);
+  });
+});

packages/adapters/src/media-ingest.ts

@@ -0,0 +1,810 @@
+import { lstat, mkdir, rename } from "node:fs/promises";
+import path from "node:path";
+import type { BackgroundJob, MediaJobHandlers, MediaSourceProvider } from "@rakazo/adapter-kit";
+import { mediaAnalyzeJob, mediaClusterJob, mediaProxyJob, mediaSyncJob } from "@rakazo/adapter-kit";
+import type { MediaIngestState, MediaSourceKind } from "@rakazo/contracts";
+import { CLIENT_PUSHED_MEDIA_SOURCES } from "@rakazo/contracts";
+import type { MediaSourceRecord } from "@rakazo/db";
+import { clientMediaItemToSourceItem, InvalidMediaInputError, IsolationError } from "@rakazo/db";
+import { getLogger } from "@rakazo/logging";
+import type { SessionOptions } from "@rakazo/media";
+import { assertInside, resolveInside } from "@rakazo/media/node";
+import type { MediaIngestService, MediaServiceDeps } from "./media-context.js";
+import type { MediaToolPaths } from "./media-folder-source-probe.js";
+import {
+  analysisInputPath,
+  analyzeAsset,
+  isTransientInferenceError,
+  MEDIA_ANALYSIS_VERSION,
+} from "./media-ingest-analyze.js";
+import { clusterSpace } from "./media-ingest-cluster.js";
+import {
+  assetDir,
+  deleteMediaPaths,
+  errorMessage,
+  fileExists,
+  moveFile,
+  originalExtension,
+  publishSafe,
+  sha256File,
+} from "./media-ingest-files.js";
+import { buildProxy } from "./media-ingest-proxy.js";
+import type { MediaJobLimits } from "./media-jobs.js";
+import {
+  assertDiskSpace,
+  defaultMediaJobLimits,
+  isDiskFull,
+  MEDIA_MIN_FREE_BYTES,
+  MediaDiskFullError,
+  MediaJobLimiter,
+} from "./media-jobs.js";
+
+/*
+ * Media ingest: client pushes, uploads, and the sync → proxy → analyze →
+ * cluster job chain. Jobs carry only record ids; every step re-reads its
+ * record, scopes by the owning space, and is safe to repeat.
+ *
+ * Sync walks every page of a server-side source from the start: positional
+ * cursors (folder paths) cannot express "changed since", and per-item
+ * revisions make unchanged pages cheap. Unchanged items enqueue nothing.
+ *
+ * Proxy and analysis work is pumped, not fanned out: the queue holds at most a
+ * small window of proxy and analyze jobs per space (the oldest assets waiting
+ * in `indexed` or `proxied`), and every finished job refills the window. A
+ * 50k-asset library therefore never floods the shared job queue that chat runs
+ * also use. Job keys are per asset, so refills replace instead of duplicating.
+ *
+ * Idempotency: a proxy is rebuilt only when the original's content hash
+ * differs from the one it was built from; analysis runs only the kinds that
+ * have no result at the provider's current model version (explicit kinds
+ * always re-run). A crashed claim (proxying/analyzing) is taken over after
+ * MEDIA_STALE_CLAIM_MS.
+ */
+
+export const MEDIA_SYNC_PAGE = 500;
+/** Clustering waits for a quiet period so a burst of ingest work clusters once. */
+export const MEDIA_CLUSTER_DEBOUNCE_MS = 30_000;
+/** A job that found no free permit runs again after this delay. */
+export const MEDIA_BUSY_RETRY_MS = 15_000;
+/** A job paused by a full disk or an unreachable sidecar retries after this delay. */
+export const MEDIA_PAUSE_RETRY_MS = 5 * 60_000;
+/** An in-flight claim older than this belongs to a crashed worker. */
+export const MEDIA_STALE_CLAIM_MS = 30 * 60_000;
+
+const DISK_PAUSED = "Media processing is paused because the disk is full";
+const SERVER_SOURCES = ["folder", "emulator"] as const;
+const REMOVE_BATCH = 500;
+
+type ServerSourceKind = (typeof SERVER_SOURCES)[number];
+type AssetRecord = Awaited<ReturnType<MediaServiceDeps["repos"]["getAssetRecord"]>>;
+
+function isServerSource(kind: MediaSourceKind): kind is ServerSourceKind {
+  return (SERVER_SOURCES as readonly string[]).includes(kind);
+}
+
+function isClientPushed(kind: MediaSourceKind): boolean {
+  return (CLIENT_PUSHED_MEDIA_SOURCES as readonly string[]).includes(kind);
+}
+
+async function scheduleCluster(deps: MediaServiceDeps, spaceId: string): Promise<void> {
+  await deps.jobs.enqueue(
+    mediaClusterJob(spaceId, new Date(deps.now().getTime() + MEDIA_CLUSTER_DEBOUNCE_MS)),
+  );
+}
+
+/** Every derived file of a source's assets lives in their asset directories. */
+async function sourceAssetDirs(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  sourceId: string,
+  externalIds?: string[],
+): Promise<string[]> {
+  const rows = await deps.prisma.mediaAsset.findMany({
+    where: { spaceId, sourceId, ...(externalIds ? { externalId: { in: externalIds } } : {}) },
+    select: { id: true },
+  });
+  return rows.map((row) => assetDir(spaceId, row.id));
+}
+
+/** Delete assets by external id along with every file derived from them. */
+async function removeItems(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  sourceId: string,
+  externalIds: string[],
+): Promise<number> {
+  let deleted = 0;
+  for (let offset = 0; offset < externalIds.length; offset += REMOVE_BATCH) {
+    const batch = externalIds.slice(offset, offset + REMOVE_BATCH);
+    const dirs = await sourceAssetDirs(deps, spaceId, sourceId, batch);
+    const removed = await deps.repos.removeSourceItems(spaceId, sourceId, batch);
+    await deleteMediaPaths(deps.mediaRoot, [...removed.paths, ...dirs]);
+    deleted += removed.deleted;
+  }
+  return deleted;
+}
+
+/**
+ * Content changed under an asset: analyses, segments, and face tracks describe
+ * the old bytes and are dropped so the next analysis starts clean.
+ */
+async function dropStaleAnalysis(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  assetId: string,
+): Promise<void> {
+  await deps.repos.replaceFaceTracks(spaceId, assetId, "stale", []);
+  await deps.repos.replaceSegments(spaceId, assetId, 0, []);
+  await deps.prisma.mediaAnalysis.deleteMany({ where: { spaceId, assetId } });
+  await deleteMediaPaths(deps.mediaRoot, [`${assetDir(spaceId, assetId)}/faces`]);
+}
+
+/** Assets waiting in `waiting`, or stuck in `inFlight` by a worker that died. */
+function waitingWhere(
+  deps: MediaServiceDeps,
+  spaceId: string,
+  waiting: MediaIngestState,
+  inFlight: MediaIngestState,
+) {
+  const staleBefore = new Date(deps.now().getTime() - MEDIA_STALE_CLAIM_MS);
+  const storage = { in: ["referenced", "uploaded"] };
+  return {
+    spaceId,
+    storage,
+    OR: [{ ingestState: waiting }, { ingestState: inFlight, updatedAt: { lt: staleBefore } }],
+  };
+}
+
+/** Enqueue proxy jobs for the oldest assets waiting for one. */
+async function refillProxy(deps: MediaServiceDeps, spaceId: string, window: number) {
+  const rows = await deps.prisma.mediaAsset.findMany({
+    where: waitingWhere(deps, spaceId, "indexed", "proxying"),
+    orderBy: [{ createdAt: "asc" }, { id: "asc" }],
+    take: window,
+    select: { id: true },
+  });
+  for (const row of rows) await deps.jobs.enqueue(mediaProxyJob(row.id));
+}
+
+/** Enqueue analyze jobs for the oldest assets waiting for one. */
+async function refillAnalyze(deps: MediaServiceDeps, spaceId: string, window: number) {
+  if (!deps.inference) return;
+  const rows = await deps.prisma.mediaAsset.findMany({
+    where: waitingWhere(deps, spaceId, "proxied", "analyzing"),
+    orderBy: [{ updatedAt: "asc" }, { id: "asc" }],
+    take: window,
+    select: { id: true },
+  });
+  for (const row of rows) await deps.jobs.enqueue(mediaAnalyzeJob(row.id));
+}
+
+/**
+ * Restart the proxy and analysis pumps of every space with waiting or stranded
+ * work. The worker calls this on start (after a crash or a sidecar was added)
+ * and periodically; it only enqueues jobs, so repeating it is harmless.
+ */
+export async function resumeMediaIngest(
+  deps: MediaServiceDeps,
+  limits: MediaJobLimits = defaultMediaJobLimits(),
+): Promise<number> {
+  const states: MediaIngestState[] = deps.inference
+    ? ["indexed", "proxying", "proxied", "analyzing"]
+    : ["indexed", "proxying"];
+  const spaces = await deps.prisma.mediaAsset.findMany({
+    where: { ingestState: { in: states }, storage: { in: ["referenced", "uploaded"] } },
+    distinct: ["spaceId"],
+    select: { spaceId: true },
+  });
+  for (const { spaceId } of spaces) {
+    await refillProxy(deps, spaceId, limits.cpu * 2);
+    await refillAnalyze(deps, spaceId, limits.gpu * 2);
+  }
+  return spaces.length;
+}
+
+/** Schedule clustering once the space has no proxy or analysis work left. */
+async function clusterWhenDrained(deps: MediaServiceDeps, spaceId: string): Promise<void> {
+  const pending: MediaIngestState[] = deps.inference
+    ? ["proxying", "proxied", "analyzing"]
+    : ["proxying"];
+  const busy = await deps.prisma.mediaAsset.count({
+    where: {
+      spaceId,
+      OR: [
+        { ingestState: { in: pending } },
+        { ingestState: "indexed", storage: { in: ["referenced", "uploaded"] } },
+      ],
+    },
+  });
+  if (busy === 0) await scheduleCluster(deps, spaceId);
+}
+
+export function createMediaIngestService(deps: MediaServiceDeps): MediaIngestService {
+  return {
+    async pushItems(actor, input) {
+      const source = await deps.repos.getSource(actor.spaceId, input.sourceId);
+      if (!isClientPushed(source.kind)) {
+        throw new InvalidMediaInputError("Only device sources accept pushed items");
+      }
+      const upserted = await deps.repos.upsertSourceItems(
+        actor.spaceId,
+        source.id,
+        input.items.map(clientMediaItemToSourceItem),
+      );
+      const deleted = await removeItems(deps, actor.spaceId, source.id, [
+        ...new Set(input.deletedExternalIds),
+      ]);
+      const updated = await deps.repos.updateSourceSync(
+        actor.spaceId,
+        source.id,
+        input.complete
+          ? {
+              status: "idle",
+              lastSyncCursor: input.syncToken,
+              lastSyncedAt: deps.now(),
+              error: null,
+            }
+          : { status: "syncing", error: null },
+      );
+      await publishSafe(deps, actor.spaceId, {
+        type: "media.source",
+        sourceId: source.id,
+        status: updated.status,
+        error: null,
+      });
+      // Pushed assets wait for an upload; proxy work starts in uploadCompleted.
+      if (input.complete) await scheduleCluster(deps, actor.spaceId);
+      return {
+        created: upserted.created,
+        updated: upserted.updated,
+        unchanged: upserted.unchanged,
+        deleted,
+      };
+    },
+
+    async requestSync(actor, sourceId) {
+      const source = await deps.repos.getSource(actor.spaceId, sourceId);
+      if (!isServerSource(source.kind)) {
+        throw new InvalidMediaInputError("This source syncs from its device");
+      }
+      await deps.jobs.enqueue(mediaSyncJob(source.id));
+    },
+
+    async uploadCompleted(actor, input) {
+      const ext = originalExtension(input.fileName);
+      if (!ext) throw new InvalidMediaInputError("Unsupported media file type");
+      if (!/^[0-9a-f]{64}$/.test(input.sha256)) {
+        throw new InvalidMediaInputError("Invalid content hash");
+      }
+      const asset = await deps.repos.getAssetRecord(actor.spaceId, input.assetId);
+      if (asset.storage === "referenced") {
+        throw new InvalidMediaInputError("This asset is read from its source");
+      }
+      const temp = await assertInside(deps.mediaRoot, input.tempPath);
+      const uploads = await resolveInside(deps.mediaRoot, `${actor.spaceId}/uploads`);
+      const relative = path.relative(uploads, temp);
+      if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) {
+        throw new InvalidMediaInputError("Upload is outside this space's upload area");
+      }
+      if (!(await lstat(temp)).isFile()) throw new InvalidMediaInputError("Upload is not a file");
+      const actual = await sha256File(temp);
+      if (actual !== input.sha256) throw new InvalidMediaInputError("Content hash mismatch");
+
+      const storagePath = `${assetDir(actor.spaceId, asset.id)}/original.${ext}`;
+      const target = await resolveInside(deps.mediaRoot, storagePath);
+      await moveFile(temp, target);
+      try {
+        const upload = await deps.prisma.mediaUpload.findFirst({
+          where: { spaceId: actor.spaceId, assetId: asset.id, status: "open", sha256: actual },
+          orderBy: [{ createdAt: "desc" }, { id: "desc" }],
+          select: { id: true },
+        });
+        if (upload) {
+          // Closes the upload and points the asset at the stored original in one transaction.
+          await deps.repos.completeUpload(actor.spaceId, upload.id, { storagePath });
+        } else {
+          const result = await deps.prisma.mediaAsset.updateMany({
+            where: { id: asset.id, spaceId: actor.spaceId },
+            data: {
+              storage: "uploaded",
+              storagePath,
+              contentHash: actual,
+              ingestState: "indexed",
+              ingestError: null,
+            },
+          });
+          if (result.count !== 1) throw new InvalidMediaInputError("Asset no longer exists");
+        }
+      } catch (error) {
+        await rename(target, temp).catch(() => undefined);
+        throw error;
+      }
+      if (asset.storagePath && asset.storagePath !== storagePath) {
+        await deleteMediaPaths(deps.mediaRoot, [asset.storagePath]);
+      }
+      if (asset.contentHash && asset.contentHash !== actual) {
+        await dropStaleAnalysis(deps, actor.spaceId, asset.id);
+      }
+      await publishSafe(deps, actor.spaceId, {
+        type: "media.asset",
+        assetId: asset.id,
+        ingestState: "indexed",
+      });
+      await deps.jobs.enqueue(mediaProxyJob(asset.id));
+    },
+
+    async recluster(actor) {
+      await deps.jobs.enqueue(mediaClusterJob(actor.spaceId));
+    },
+
+    async removeSource(actor, sourceId) {
+      const source = await deps.repos.getSource(actor.spaceId, sourceId);
+      await deps.jobs.cancel(mediaSyncJob(source.id).replaceKey as string);
+      await deleteMediaPaths(deps.mediaRoot, await sourceAssetDirs(deps, actor.spaceId, source.id));
+      await deps.repos.removeSource(actor.spaceId, source.id);
+      await scheduleCluster(deps, actor.spaceId);
+    },
+  };
+}
+
+// ---------------------------------------------------------------------------
+// Job handlers
+// ---------------------------------------------------------------------------
+
+export interface MediaIngestHandlerOptions extends MediaToolPaths {
+  /** Aborts long-running ingest work when the worker shuts down. */
+  signal?: AbortSignal;
+  /** Per-process job limits; defaults to defaultMediaJobLimits(). */
+  limits?: MediaJobLimits;
+  /** Free bytes the media root keeps in reserve; jobs pause below it. 0 disables the check. */
+  minFreeBytes?: number;
+  /** Session rules (gap minutes and so on); the time zone defaults to the library owner's. */
+  sessionOptions?: Partial<SessionOptions>;
+}
+
+async function syncSource(
+  deps: MediaServiceDeps,
+  source: MediaSourceRecord,
+  provider: MediaSourceProvider,
+  onChanged: () => Promise<void>,
+  signal?: AbortSignal,
+): Promise<{ changed: number; deleted: number }> {
+  const seen = new Set<string>();
+  let cursor: string | null = null;
+  let changed = 0;
+  do {
+    signal?.throwIfAborted();
+    const page = await provider.list({
+      config: source.config,
+      cursor,
+      limit: MEDIA_SYNC_PAGE,
+      ...(signal ? { signal } : {}),
+    });
+    for (const item of page.items) seen.add(item.externalId);
+    const result = await deps.repos.upsertSourceItems(source.spaceId, source.id, page.items);
+    if (result.changedAssetIds.length > 0) {
+      // A changed revision means new bytes: the asset goes back through the pipeline.
+      await deps.prisma.mediaAsset.updateMany({
+        where: {
+          spaceId: source.spaceId,
+          id: { in: result.changedAssetIds },
+          storage: "referenced",
+          ingestState: { not: "indexed" },
+        },
+        data: { ingestState: "indexed", ingestError: null },
+      });
+      changed += result.changedAssetIds.length;
+      await onChanged();
+    }
+    cursor = page.nextCursor;
+  } while (cursor);
+
+  // Items gone from the source. An empty listing of a non-empty library is
+  // treated as an unavailable source (unmounted disk), never as mass deletion.
+  const known = await deps.prisma.mediaAsset.findMany({
+    where: { spaceId: source.spaceId, sourceId: source.id },
+    select: { externalId: true },
+  });
+  if (seen.size === 0 && known.length > 0) {
+    throw new Error("The source listed no files; check that it is reachable");
+  }
+  const gone = known.map((row) => row.externalId).filter((id) => !seen.has(id));
+  const deleted = await removeItems(deps, source.spaceId, source.id, gone);
+  return { changed, deleted };
+}
+
+export function createMediaIngestHandlers(
+  deps: MediaServiceDeps,
+  options: MediaIngestHandlerOptions = {},
+): Pick<MediaJobHandlers, "media.sync" | "media.proxy" | "media.analyze" | "media.cluster"> {
+  const { signal } = options;
+  const limiter = new MediaJobLimiter(options.limits);
+  const proxyWindow = limiter.limits.cpu * 2;
+  const analyzeWindow = limiter.limits.gpu * 2;
+  const minFreeBytes = options.minFreeBytes ?? MEDIA_MIN_FREE_BYTES;
+  const tools: MediaToolPaths = {
+    ...(options.ffmpegPath ? { ffmpegPath: options.ffmpegPath } : {}),
+    ...(options.ffprobePath ? { ffprobePath: options.ffprobePath } : {}),
+  };
+
+  const later = async (job: BackgroundJob, delayMs: number) => {
+    await deps.jobs.enqueue({ ...job, availableAt: new Date(deps.now().getTime() + delayMs) });
+  };
+
+  const checkDisk = async () => {
+    await mkdir(deps.mediaRoot, { recursive: true });
+    await assertDiskSpace(deps.mediaRoot, minFreeBytes);
+  };
+
+  /** Pause on a full disk: tell the source's owner why and try again later. */
+  const pauseForDisk = async (spaceId: string, sourceId: string, job: BackgroundJob) => {
+    getLogger().warn("media job paused: disk is full", { job: job.name });
+    const source = await deps.repos
+      .updateSourceSync(spaceId, sourceId, { error: DISK_PAUSED })
+      .catch(() => null);
+    if (source) {
+      await publishSafe(deps, spaceId, {
+        type: "media.source",
+        sourceId,
+        status: source.status,
+        error: DISK_PAUSED,
+      });
+    }
+    await later(job, MEDIA_PAUSE_RETRY_MS);
+  };
+
+  const resumeAfterDisk = async (spaceId: string, sourceId: string) => {
+    const result = await deps.prisma.mediaSource.updateMany({
+      where: { id: sourceId, spaceId, error: DISK_PAUSED },
+      data: { error: null },
+    });
+    if (result.count === 0) return;
+    const source = await deps.repos.getSource(spaceId, sourceId).catch(() => null);
+    if (source) {
+      await publishSafe(deps, spaceId, {
+        type: "media.source",
+        sourceId,
+        status: source.status,
+        error: null,
+      });
+    }
+  };
+
+  /**
+   * Claim an asset for a step: from one of `from`, or from `inFlight` when that
+   * claim is stale (its worker died). Only one worker wins.
+   */
+  const claim = async (
+    spaceId: string,
+    assetId: string,
+    from: MediaIngestState[],
+    inFlight: MediaIngestState,
+  ): Promise<boolean> => {
+    const staleBefore = new Date(deps.now().getTime() - MEDIA_STALE_CLAIM_MS);
+    const result = await deps.prisma.mediaAsset.updateMany({
+      where: {
+        id: assetId,
+        spaceId,
+        OR: [
+          { ingestState: { in: from } },
+          { ingestState: inFlight, updatedAt: { lt: staleBefore } },
+        ],
+      },
+      data: { ingestState: inFlight, ingestError: null },
+    });
+    if (result.count !== 1) return false;
+    await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: inFlight });
+    return true;
+  };
+
+  const setState = async (
+    spaceId: string,
+    assetId: string,
+    from: MediaIngestState,
+    to: MediaIngestState,
+    error?: string,
+  ) => {
+    const applied = await deps.repos
+      .transitionIngestState(spaceId, assetId, {
+        from: [from],
+        to,
+        ...(error !== undefined ? { error } : {}),
+      })
+      .catch((cause) => {
+        if (cause instanceof IsolationError) return false;
+        throw cause;
+      });
+    if (applied)
+      await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: to });
+    return applied;
+  };
+
+  /** Absolute path of the original, or null for a device asset that has not been uploaded. */
+  const originalOf = async (spaceId: string, asset: AssetRecord): Promise<string | null> => {
+    if (asset.storage === "uploaded" && asset.storagePath) {
+      return resolveInside(deps.mediaRoot, asset.storagePath);
+    }
+    if (asset.storage !== "referenced") return null;
+    const source = await deps.repos.getSource(spaceId, asset.sourceId);
+    const provider = isServerSource(source.kind) ? deps.sources[source.kind] : undefined;
+    if (!provider) throw new Error(`No ${source.kind} source provider is configured`);
+    return provider.resolve(source.config, asset.externalId);
+  };
+
+  /** Load the asset a job names; null when it (or its space) is gone. */
+  const loadAsset = async (assetId: string) => {
+    const spaceId = await deps.repos.mediaSpaceOf("asset", assetId);
+    if (!spaceId) return null;
+    try {
+      return { spaceId, asset: await deps.repos.getAssetRecord(spaceId, assetId) };
+    } catch (error) {
+      if (error instanceof IsolationError) return null;
+      throw error;
+    }
+  };
+
+  const proxyAsset = async (spaceId: string, asset: AssetRecord): Promise<void> => {
+    const original = await originalOf(spaceId, asset);
+    if (!original) throw new Error("The original has not been uploaded");
+    const contentHash = await sha256File(original, signal);
+    const unchanged =
+      contentHash === asset.contentHash &&
+      asset.proxyPath !== null &&
+      asset.thumbPath !== null &&
+      (await fileExists(deps.mediaRoot, asset.proxyPath)) &&
+      (await fileExists(deps.mediaRoot, asset.thumbPath));
+    if (unchanged) {
+      await deps.repos.recordProxy(spaceId, asset.id, {
+        contentHash,
+        proxyPath: asset.proxyPath,
+        thumbPath: asset.thumbPath,
+      });
+      return;
+    }
+    if (asset.contentHash && asset.contentHash !== contentHash) {
+      await dropStaleAnalysis(deps, spaceId, asset.id);
+    }
+    const outputs = await buildProxy({
+      mediaRoot: deps.mediaRoot,
+      spaceId,
+      assetId: asset.id,
+      original,
+      still: asset.kind === "photo",
+      tools,
+      ...(signal ? { signal } : {}),
+    });
+    const still = outputs.spritePath === null;
+    await deps.repos.recordProxy(spaceId, asset.id, {
+      contentHash,
+      proxyPath: outputs.proxyPath,
+      thumbPath: outputs.thumbPath,
+      spritePath: outputs.spritePath,
+      durationMs: still ? asset.durationMs : outputs.probe.durationMs,
+      width: outputs.probe.width,
+      height: outputs.probe.height,
+      fps: still ? asset.fps : outputs.probe.fps,
+      codec: outputs.probe.codec,
+      orientation: outputs.probe.rotation,
+    });
+  };
+
+  return {
+    async "media.sync"({ sourceId }) {
+      const spaceId = await deps.repos.mediaSpaceOf("source", sourceId);
+      if (!spaceId) return;
+      const source = await deps.repos.getSource(spaceId, sourceId);
+      const provider = isServerSource(source.kind) ? deps.sources[source.kind] : undefined;
+      if (!provider) {
+        const error = `No ${source.kind} source provider is configured`;
+        await deps.repos.updateSourceSync(spaceId, sourceId, { status: "unavailable", error });
+        await publishSafe(deps, spaceId, {
+          type: "media.source",
+          sourceId,
+          status: "unavailable",
+          error,
+        });
+        return;
+      }
+      await deps.repos.updateSourceSync(spaceId, sourceId, { status: "syncing", error: null });
+      await publishSafe(deps, spaceId, {
+        type: "media.source",
+        sourceId,
+        status: "syncing",
+        error: null,
+      });
+      try {
+        const { changed, deleted } = await syncSource(
+          deps,
+          source,
+          provider,
+          () => refillProxy(deps, spaceId, proxyWindow),
+          signal,
+        );
+        await deps.repos.updateSourceSync(spaceId, sourceId, {
+          status: "idle",
+          lastSyncedAt: deps.now(),
+          error: null,
+        });
+        await publishSafe(deps, spaceId, {
+          type: "media.source",
+          sourceId,
+          status: "idle",
+          error: null,
+        });
+        if (changed + deleted > 0) await scheduleCluster(deps, spaceId);
+      } catch (error) {
+        if (signal?.aborted) {
+          await deps.repos
+            .updateSourceSync(spaceId, sourceId, { status: "idle" })
+            .catch(() => undefined);
+          throw error;
+        }
+        // The source was removed while it synced.
+        if (error instanceof IsolationError) return;
+        const message = errorMessage(error);
+        getLogger().error("media sync failed", error, { sourceId });
+        await deps.repos
+          .updateSourceSync(spaceId, sourceId, { status: "error", error: message })
+          .catch(() => undefined);
+        await publishSafe(deps, spaceId, {
+          type: "media.source",
+          sourceId,
+          status: "error",
+          error: message,
+        });
+      }
+    },
+
+    async "media.proxy"({ assetId }) {
+      const loaded = await loadAsset(assetId);
+      if (!loaded || loaded.asset.storage === "remote") return;
+      const { spaceId, asset } = loaded;
+      const job = mediaProxyJob(assetId);
+      const release = limiter.tryAcquire("cpu");
+      if (!release) {
+        await later(job, MEDIA_BUSY_RETRY_MS);
+        return;
+      }
+      try {
+        try {
+          await checkDisk();
+        } catch (error) {
+          if (!(error instanceof MediaDiskFullError)) throw error;
+          await pauseForDisk(spaceId, asset.sourceId, job);
+          return;
+        }
+        if (!(await claim(spaceId, assetId, ["indexed", "failed"], "proxying"))) return;
+        try {
+          await proxyAsset(spaceId, asset);
+          await publishSafe(deps, spaceId, {
+            type: "media.asset",
+            assetId,
+            ingestState: "proxied",
+          });
+          await resumeAfterDisk(spaceId, asset.sourceId);
+        } catch (error) {
+          if (signal?.aborted) {
+            await setState(spaceId, assetId, "proxying", "indexed");
+            throw error;
+          }
+          if (error instanceof IsolationError) {
+            // Deleted while it was processed: drop whatever was written for it.
+            await deleteMediaPaths(deps.mediaRoot, [assetDir(spaceId, assetId)]);
+            return;
+          }
+          if (isDiskFull(error)) {
+            await setState(spaceId, assetId, "proxying", "indexed");
+            await pauseForDisk(spaceId, asset.sourceId, job);
+            return;
+          }
+          getLogger().error("media proxy failed", error, { assetId });
+          await setState(spaceId, assetId, "proxying", "failed", errorMessage(error));
+        }
+      } finally {
+        release();
+      }
+      await refillProxy(deps, spaceId, proxyWindow);
+      await refillAnalyze(deps, spaceId, analyzeWindow);
+      await clusterWhenDrained(deps, spaceId);
+    },
+
+    async "media.analyze"({ assetId, kinds }) {
+      const inference = deps.inference;
+      if (!inference) return;
+      const loaded = await loadAsset(assetId);
+      if (!loaded?.asset.proxyPath) return;
+      const { spaceId, asset } = loaded;
+      // A repeated default request for an asset already at the current pipeline version.
+      if (!kinds && asset.ingestState === "analyzed") {
+        if (asset.analysisVersion >= MEDIA_ANALYSIS_VERSION) return;
+      }
+      const job = mediaAnalyzeJob(assetId, kinds);
+      const release = limiter.tryAcquire("gpu");
+      if (!release) {
+        await later(job, MEDIA_BUSY_RETRY_MS);
+        return;
+      }
+      try {
+        try {
+          await checkDisk();
+        } catch (error) {
+          if (!(error instanceof MediaDiskFullError)) throw error;
+          await pauseForDisk(spaceId, asset.sourceId, job);
+          return;
+        }
+        if (!(await claim(spaceId, assetId, ["proxied", "analyzed", "failed"], "analyzing"))) {
+          return;
+        }
+        try {
+          const original = asset.storage === "referenced" ? await originalOf(spaceId, asset) : null;
+          const inputPath = await analysisInputPath({
+            mediaRoot: deps.mediaRoot,
+            proxyPath: asset.proxyPath as string,
+            referencedOriginal: original,
+          });
+          const outcome = await analyzeAsset({
+            deps,
+            inference,
+            spaceId,
+            asset,
+            inputPath,
+            ...(kinds ? { kinds } : {}),
+            ...(signal ? { signal } : {}),
+          });
+          await deps.repos.recordAnalysisVersion(
+            spaceId,
+            assetId,
+            MEDIA_ANALYSIS_VERSION,
+            outcome.qualityScore,
+          );
+          await publishSafe(deps, spaceId, {
+            type: "media.asset",
+            assetId,
+            ingestState: "analyzed",
+          });
+          await resumeAfterDisk(spaceId, asset.sourceId);
+        } catch (error) {
+          if (signal?.aborted) {
+            await setState(spaceId, assetId, "analyzing", "proxied");
+            throw error;
+          }
+          if (error instanceof IsolationError) return;
+          if (isDiskFull(error)) {
+            await setState(spaceId, assetId, "analyzing", "proxied");
+            await pauseForDisk(spaceId, asset.sourceId, job);
+            return;
+          }
+          if (isTransientInferenceError(error)) {
+            // Sidecar down or saturated: not the asset's fault, try again later.
+            getLogger().warn("media analysis deferred", { assetId, reason: errorMessage(error) });
+            await setState(spaceId, assetId, "analyzing", "proxied");
+            await later(job, MEDIA_PAUSE_RETRY_MS);
+            return;
+          }
+          getLogger().error("media analysis failed", error, { assetId });
+          await setState(spaceId, assetId, "analyzing", "failed", errorMessage(error));
+        }
+      } finally {
+        release();
+      }
+      await refillAnalyze(deps, spaceId, analyzeWindow);
+      await clusterWhenDrained(deps, spaceId);
+    },
+
+    async "media.cluster"({ spaceId }) {
+      const release = limiter.tryAcquire("cpu");
+      if (!release) {
+        await later(mediaClusterJob(spaceId), MEDIA_BUSY_RETRY_MS);
+        return;
+      }
+      try {
+        await clusterSpace(deps, spaceId, {
+          ...(signal ? { signal } : {}),
+          ...(options.sessionOptions ? { sessionOptions: options.sessionOptions } : {}),
+        });
+      } finally {
+        release();
+      }
+    },
+  };
+}

packages/adapters/src/media-jobs.test.ts

@@ -0,0 +1,58 @@
+import { tmpdir } from "node:os";
+import { describe, expect, it } from "vitest";
+import {
+  assertDiskSpace,
+  defaultMediaJobLimits,
+  freeBytes,
+  isDiskFull,
+  MediaDiskFullError,
+  MediaJobLimiter,
+} from "./media-jobs.js";
+
+describe("media job limits", () => {
+  it("defaults to one GPU job and a bounded number of CPU jobs", () => {
+    expect(defaultMediaJobLimits(1)).toEqual({ gpu: 1, cpu: 1 });
+    expect(defaultMediaJobLimits(4)).toEqual({ gpu: 1, cpu: 2 });
+    expect(defaultMediaJobLimits(64)).toEqual({ gpu: 1, cpu: 2 });
+  });
+
+  it("hands out permits up to the limit per class and never waits", () => {
+    const limiter = new MediaJobLimiter({ gpu: 1, cpu: 2 });
+    const gpu = limiter.tryAcquire("gpu");
+    expect(gpu).not.toBeNull();
+    expect(limiter.tryAcquire("gpu")).toBeNull();
+    const cpu = [limiter.tryAcquire("cpu"), limiter.tryAcquire("cpu")];
+    expect(cpu.every(Boolean)).toBe(true);
+    expect(limiter.tryAcquire("cpu")).toBeNull();
+
+    gpu?.();
+    gpu?.();
+    expect(limiter.running("gpu")).toBe(0);
+    expect(limiter.tryAcquire("gpu")).not.toBeNull();
+    cpu[0]?.();
+    expect(limiter.running("cpu")).toBe(1);
+  });
+
+  it("rejects limits below one", () => {
+    expect(() => new MediaJobLimiter({ gpu: 0, cpu: 1 })).toThrow();
+    expect(() => new MediaJobLimiter({ gpu: 1, cpu: 1.5 })).toThrow();
+  });
+});
+
+describe("disk backpressure", () => {
+  it("reports free space and pauses below the reserve", async () => {
+    const dir = tmpdir();
+    expect(await freeBytes(dir)).toBeGreaterThan(0);
+    await expect(assertDiskSpace(dir, 0)).resolves.toBeUndefined();
+    await expect(assertDiskSpace(dir, Number.MAX_SAFE_INTEGER)).rejects.toBeInstanceOf(
+      MediaDiskFullError,
+    );
+  });
+
+  it("recognizes a full disk from errno or ffmpeg output", () => {
+    expect(isDiskFull(Object.assign(new Error("write"), { code: "ENOSPC" }))).toBe(true);
+    expect(isDiskFull(new Error("ffmpeg failed: No space left on device"))).toBe(true);
+    expect(isDiskFull(new Error("Invalid data found"))).toBe(false);
+    expect(isDiskFull(null)).toBe(false);
+  });
+});

packages/adapters/src/media-jobs.ts

@@ -0,0 +1,85 @@
+import { statfs } from "node:fs/promises";
+import { availableParallelism } from "node:os";
+
+/*
+ * Resource limits for media jobs.
+ *
+ * Every job slot of the shared job worker also serves chat runs, so a media job
+ * never waits for capacity inside a slot: it takes a permit or steps aside
+ * (the caller re-enqueues it with a delay). Analysis is the GPU class
+ * (sidecar models), proxies and clustering are the CPU class. The limits are
+ * per worker process.
+ *
+ * Disk backpressure: before writing derived files a job checks the free space
+ * of the media root and pauses (re-enqueues itself later) when it falls under
+ * the reserve, instead of failing assets or crashing the worker.
+ */
+
+export type MediaJobClass = "gpu" | "cpu";
+
+export interface MediaJobLimits {
+  /** Concurrent analysis jobs. Default 1. */
+  gpu: number;
+  /** Concurrent proxy and cluster jobs. Default: half the cores, at most 2. */
+  cpu: number;
+}
+
+export function defaultMediaJobLimits(cores = availableParallelism()): MediaJobLimits {
+  return { gpu: 1, cpu: Math.max(1, Math.min(2, Math.floor(cores / 2))) };
+}
+
+/** Non-blocking counting semaphore. */
+export class MediaJobLimiter {
+  private readonly active: Record<MediaJobClass, number> = { gpu: 0, cpu: 0 };
+
+  constructor(readonly limits: MediaJobLimits = defaultMediaJobLimits()) {
+    for (const value of Object.values(limits)) {
+      if (!Number.isInteger(value) || value < 1) throw new Error("media job limits must be ≥ 1");
+    }
+  }
+
+  /** A release function when a permit was free, otherwise null. Release is idempotent. */
+  tryAcquire(kind: MediaJobClass): (() => void) | null {
+    if (this.active[kind] >= this.limits[kind]) return null;
+    this.active[kind] += 1;
+    let released = false;
+    return () => {
+      if (released) return;
+      released = true;
+      this.active[kind] -= 1;
+    };
+  }
+
+  running(kind: MediaJobClass): number {
+    return this.active[kind];
+  }
+}
+
+/** Free space under the media root reserve: 2 GiB. */
+export const MEDIA_MIN_FREE_BYTES = 2 * 1024 ** 3;
+
+/** Bytes available to this process on the file system holding `dir`. */
+export async function freeBytes(dir: string): Promise<number> {
+  const stats = await statfs(dir);
+  return Number(stats.bavail) * Number(stats.bsize);
+}
+
+/** A full disk is retried later instead of failing the asset for good. */
+export function isDiskFull(error: unknown): boolean {
+  if ((error as NodeJS.ErrnoException | null)?.code === "ENOSPC") return true;
+  return error instanceof Error && /No space left on device/i.test(error.message);
+}
+
+export class MediaDiskFullError extends Error {
+  constructor(readonly availableBytes: number | null) {
+    super("Media processing is paused because the disk is full");
+    this.name = "MediaDiskFullError";
+  }
+}
+
+/** Throws MediaDiskFullError when less than `reserve` bytes are free under `dir`. */
+export async function assertDiskSpace(dir: string, reserve: number): Promise<void> {
+  if (reserve <= 0) return;
+  const available = await freeBytes(dir);
+  if (available < reserve) throw new MediaDiskFullError(available);
+}