diff --git a/admin/app/globals.css b/admin/app/globals.css index 1fe4793..caf2bff 100644 --- a/admin/app/globals.css +++ b/admin/app/globals.css @@ -543,6 +543,29 @@ a:hover { font-weight: 600; color: var(--ink-2); } +.saved-inputs { + display: grid; + gap: 12px; + margin: 0 0 16px; + padding: 16px; + border: 1px solid var(--line-soft); + border-radius: var(--r-stage); +} +.saved-inputs__form { + display: flex; + flex-wrap: wrap; + align-items: end; + gap: 12px; +} +.saved-inputs__form .field { + flex: 1; + min-width: 0; +} +.saved-inputs__result { + display: grid; + justify-items: start; + gap: 8px; +} .field { display: grid; gap: 6px; diff --git a/admin/app/runs/[runId]/console.tsx b/admin/app/runs/[runId]/console.tsx index be537cb..75e4b94 100644 --- a/admin/app/runs/[runId]/console.tsx +++ b/admin/app/runs/[runId]/console.tsx @@ -8,6 +8,7 @@ import { leadArtifact } from '@/lib/blocks'; import { GraphCanvas } from '@/components/graph-canvas'; import { Progress } from '@/components/progress'; import { RunOverview } from '@/components/run-overview'; +import { SavedInputs } from '@/components/saved-inputs'; import { StageStrip } from '@/components/stage-strip'; import { Status } from '@/components/status'; import { useNow } from '@/components/use-now'; @@ -436,6 +437,7 @@ export default function RunConsole({ runId }: { runId: string }) { ? ['failed', 'blocked', 'succeeded', 'canceled'].includes(selected.status) : false; const paused = run.live?.paused === true || run.status === 'paused'; + const savedDelivery = run.configuration.retainedDelivery?.schema === 'wander.saved-inputs/1'; return (
@@ -526,6 +528,14 @@ export default function RunConsole({ runId }: { runId: string }) { + {savedDelivery ? ( +

+ Prepared from saved inputs. Resume to allow one local packaging attempt. Stage retries are + disabled; reconstruction quality remains unverified. +

+ ) : ( + + )} {run.liveError ? (

@@ -864,37 +874,39 @@ export default function RunConsole({ runId }: { runId: string }) { ) : null} -

- {isGate || selected.definition.kind === 'human' ? ( - - ) : null} - - - {isGate || selected.definition.kind === 'human' ? ( + {!savedDelivery ? ( +
+ {isGate || selected.definition.kind === 'human' ? ( + + ) : null} - ) : null} -
+ + {isGate || selected.definition.kind === 'human' ? ( + + ) : null} +
+ ) : null} )} diff --git a/admin/components/saved-inputs.tsx b/admin/components/saved-inputs.tsx new file mode 100644 index 0000000..382c1a1 --- /dev/null +++ b/admin/components/saved-inputs.tsx @@ -0,0 +1,147 @@ +'use client'; + +import Link from 'next/link'; +import { useRef, useState } from 'react'; +import { checkSavedInputs, prepareSavedInputRun } from '@/lib/client'; +import { formatBytes } from '@/lib/format'; +import type { SavedInputsAvailability } from '@/lib/types'; + +export function SavedInputs({ runId }: { runId: string }) { + const [availability, setAvailability] = useState(null); + const [checking, setChecking] = useState(false); + const [submitting, setSubmitting] = useState(false); + const [name, setName] = useState(''); + const [preparedId, setPreparedId] = useState(null); + const [error, setError] = useState(null); + const busyRef = useRef(false); + + async function check() { + if (busyRef.current) return; + busyRef.current = true; + setChecking(true); + setError(null); + setPreparedId(null); + try { + setAvailability(await checkSavedInputs(runId)); + } catch (cause) { + setError(`Saved files could not be checked: ${(cause as Error).message}`); + } finally { + busyRef.current = false; + setChecking(false); + } + } + + async function submitPreparation() { + if (!availability?.available || busyRef.current) return; + busyRef.current = true; + setSubmitting(true); + setError(null); + try { + const result = await prepareSavedInputRun(runId, name.trim() || undefined); + setPreparedId(result.id); + } catch (cause) { + setError(`Paused run could not be prepared: ${(cause as Error).message}`); + } finally { + busyRef.current = false; + setSubmitting(false); + } + } + + function prepare(event: React.FormEvent) { + event.preventDefault(); + void submitPreparation(); + } + + return ( +
+
+

+ Prepare a delivery run +

+

+ Reuses saved reconstruction. Local packaging only; no new inference or quality approval. +

+
+ {!availability ? ( + + ) : availability.available ? ( + preparedId ? null : ( + <> +

+ {availability.fileCount ?? 0} saved files · {formatBytes(availability.byteCount ?? 0)} +

+
+
+ + setName(event.target.value)} + /> +
+ +
+ + ) + ) : ( +
+

+ {availability.reason || 'Saved inputs are unavailable.'} +

+ +
+ )} + {error ? ( +
+

{error}

+ {availability?.available && !preparedId ? ( + + ) : ( + + )} +
+ ) : null} + {preparedId ? ( +

+ Paused run prepared.{' '} + Open prepared run +

+ ) : null} +
+ ); +} diff --git a/admin/lib/client.ts b/admin/lib/client.ts index cd71f10..7a2b4e7 100644 --- a/admin/lib/client.ts +++ b/admin/lib/client.ts @@ -10,6 +10,8 @@ import type { ReviewTranscript, RunArtifact, RunCommand, + PreparedSavedInputRun, + SavedInputsAvailability, TranscriptTail, } from './types'; @@ -49,6 +51,21 @@ export async function loadRun(runId: string): Promise { )) as PipelineRun; } +export async function checkSavedInputs(runId: string): Promise { + return (await unwrap( + await fetch(`${ROOT}/runs/${encodeURIComponent(runId)}/saved-inputs`, { cache: 'no-store' }), + )) as SavedInputsAvailability; +} + +export async function prepareSavedInputRun( + runId: string, + id?: string, +): Promise { + return (await send(`/runs/${encodeURIComponent(runId)}/saved-inputs`, { + id, + })) as PreparedSavedInputRun; +} + export async function uploadRun(input: { file: File; options: GraphOptions; diff --git a/admin/lib/types.ts b/admin/lib/types.ts index 96b2978..32ae007 100644 --- a/admin/lib/types.ts +++ b/admin/lib/types.ts @@ -162,7 +162,10 @@ export interface PipelineRun { branchKey?: string | null; graphVersion: string; sourceSha256: string; - configuration: { options: GraphOptions }; + configuration: { + options: GraphOptions; + retainedDelivery?: { schema: string; sourceRunId: string }; + }; budget?: { currency: string; maximum_cost: number | null }; createdAt?: string; updatedAt?: string; @@ -176,6 +179,19 @@ export interface PipelineRun { liveError?: string | null; } +export interface SavedInputsAvailability { + available: boolean; + reason?: string; + sourceRunId: string; + fileCount?: number; + byteCount?: number; +} + +export interface PreparedSavedInputRun { + id: string; + status: 'paused'; +} + export type RunCommand = 'reject' | 'retry' | 'pause' | 'resume' | 'cancel'; /** The outcomes an operator may record for a provider claim that ended unobserved. */ diff --git a/admin/tests/saved-inputs-client.test.ts b/admin/tests/saved-inputs-client.test.ts new file mode 100644 index 0000000..254c977 --- /dev/null +++ b/admin/tests/saved-inputs-client.test.ts @@ -0,0 +1,49 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { checkSavedInputs, prepareSavedInputRun } from '../lib/client'; + +test('saved input requests use the authenticated same-origin proxy and encode run ids', async () => { + const originalFetch = globalThis.fetch; + const requests: Array<{ input: RequestInfo | URL; init?: RequestInit }> = []; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + requests.push({ input, init }); + return Response.json({ + available: true, + sourceRunId: 'source/one', + fileCount: 3, + byteCount: 12, + }); + }) as typeof fetch; + try { + const result = await checkSavedInputs('run/one'); + assert.deepEqual(result, { + available: true, + sourceRunId: 'source/one', + fileCount: 3, + byteCount: 12, + }); + assert.equal(String(requests[0]?.input), '/api/pipeline/runs/run%2Fone/saved-inputs'); + assert.equal(requests[0]?.init?.cache, 'no-store'); + + await prepareSavedInputRun('run/one', 'delivery-1'); + assert.equal(String(requests[1]?.input), '/api/pipeline/runs/run%2Fone/saved-inputs'); + assert.equal(requests[1]?.init?.method, 'POST'); + assert.deepEqual(JSON.parse(String(requests[1]?.init?.body)), { id: 'delivery-1' }); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test('saved input requests surface proxy detail for retry messaging', async () => { + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => + Response.json( + { detail: 'source run is still active' }, + { status: 409 }, + )) as unknown as typeof fetch; + try { + await assert.rejects(prepareSavedInputRun('source'), /source run is still active/); + } finally { + globalThis.fetch = originalFetch; + } +}); diff --git a/docs/pipeline-setup.md b/docs/pipeline-setup.md index de4e05d..011a7ee 100644 --- a/docs/pipeline-setup.md +++ b/docs/pipeline-setup.md @@ -127,6 +127,31 @@ clips are never uploaded by the doctor, and no paid operation is launched. ## Submission and recovery +### Prepare a delivery from saved inputs + +On an existing run's admin page, choose **Use saved inputs**, then **Prepare paused run**. +The source run must be paused or finished, with no unresolved stage attempts. This supports +world-enabled tracked-person runs without fine-tuning, with complete local camera, tracking, +animation and exported scene files. Object models must be self-contained PLY or GLB files; +external glTF buffers and textures are not copied implicitly. + +The server verifies the source and exported frame hashes, copies the supported inputs, and +pins every copied file before publishing a separate paused run. It retains the source run link, +room, presentation, original audio and objects. It does not copy agent sessions, attempts, +quality decisions, credentials or provider archives. Missing, changed or symlinked inputs stop +preparation without publishing a partial run. Checking and preparing use no provider calls. + +After an operator resumes the prepared run, the local worker may launch **one** `package_people` +attempt using its pinned scene configuration. Other stages, extra parameters and retries are +rejected; the packager also disables its internal retry loop. The agent then asks for viewer +review. Reused stages are explicitly labeled as saved inputs, and packaging grants no fresh +inference or reconstruction quality acceptance. Preparation alone does not start a worker. + +The API exposes `GET /api/pipeline/runs/{id}/saved-inputs` for availability and +`POST /api/pipeline/runs/{id}/saved-inputs` with an optional `id` for preparation. + +### Submit new work + The admin reads its option defaults and supported world modes from `GET /api/pipeline/options`. `POST /api/pipeline/doctor` inspects the API server locally; it does not establish that a separate worker is ready. Uploads can be queued while worker setup is repaired. The API and supervisor diff --git a/docs/pipeline-validation.md b/docs/pipeline-validation.md index 7a9ca93..00a73d9 100644 --- a/docs/pipeline-validation.md +++ b/docs/pipeline-validation.md @@ -118,6 +118,17 @@ checks, with no automatic retry, provider commands, paid-ledger change, or quali It joins actual agent control to retained packaging and viewer delivery; fresh reconstruction and headset behavior remain unverified. Evidence: `.context/evidence/bottle-agent-delivery/`. +The admin's saved-input preparation flow was then exercised with that paused bottle run. +The browser checked 502 source files (about 1.35 GB), prepared a separate paused child run, +and opened its detail page. All 503 resulting pins, including the generated retention config, +verified. The page showed five reused prerequisites and only `package_people` pending, with +no stage retry controls. The child was excluded from supervisor discovery while paused; its +database contained zero attempts and zero quality reviews. The paid-stage ledger was unchanged. +This check covered preparation and delivery admission, without another packaging execution or +inference call. Tests cover detached admission, the one-attempt limit, internal retry suppression, +corrupt/missing/symlinked inputs, incomplete receipts, publication rollback, and external object +dependencies. Evidence: `.context/evidence/admin-saved-inputs/`. + Regression tests cover questions arriving before stage/agent launch, blocked watchdog resumes, retained transcript/session evidence, and answers arriving between a turn's final journal read and the supervisor's next pass. Quiet turns do not repeatedly wake on their own status notes. diff --git a/orchestrator/api/app.py b/orchestrator/api/app.py index 56e2151..0f06846 100644 --- a/orchestrator/api/app.py +++ b/orchestrator/api/app.py @@ -168,6 +168,11 @@ class CreateRunRequest(BaseModel): budget: BudgetPolicy = Field(default_factory=BudgetPolicy) +class SavedInputsRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") + + class MessageRequest(BaseModel): model_config = ConfigDict(extra="forbid") @@ -453,6 +458,39 @@ async def get_run(run_id: str, user: str = Depends(actor)): raise HTTPException(status_code=404, detail=str(error)) from error return with_live_state(summary) + @app.get("/api/pipeline/runs/{run_id}/saved-inputs") + def saved_inputs(run_id: str, user: str = Depends(actor)): + try: + repository.run_summary(run_id) + inspect = getattr(runs, "saved_inputs", None) + return ( + inspect(run_id) + if inspect + else { + "available": False, + "sourceRunId": run_id, + "reason": "Saved inputs are unavailable", + } + ) + except KeyError as error: + raise HTTPException(404, "Unknown run") from error + + @app.post("/api/pipeline/runs/{run_id}/saved-inputs", status_code=201) + def prepare_saved_inputs(run_id: str, request: SavedInputsRequest, user: str = Depends(actor)): + new_id = request.id or f"run-{uuid.uuid4().hex}" + require_new_run(new_id) + prepare = getattr(runs, "prepare_saved_inputs", None) + if prepare is None: + raise HTTPException(409, "Saved inputs are unavailable") + try: + repository.run_summary(run_id) + prepare(run_id, new_id, settings=settings, user=user) + except KeyError as error: + raise HTTPException(404, "Unknown run or missing saved inputs") from error + except (ValueError, OSError, IntegrityError) as error: + raise HTTPException(409, str(error)) from error + return {"id": new_id, "status": "paused"} + @app.post("/api/pipeline/runs/{run_id}/messages", status_code=202) async def add_message(run_id: str, request: MessageRequest, user: str = Depends(actor)): identifier = repository.add_message( @@ -524,7 +562,9 @@ def command( # Validate before recording a command. A stale operator tab must receive a # useful 404 without appending an event or creating a stray run directory. try: - repository.run_summary(run_id) + summary = repository.run_summary(run_id) + if command == "retry" and summary["configuration"].get("retainedDelivery") is not None: + raise HTTPException(409, "Stage retries are disabled for saved-input delivery runs") journal = runs.journal(run_id) except KeyError as error: raise HTTPException(status_code=404, detail=str(error)) from error diff --git a/orchestrator/api/progress.py b/orchestrator/api/progress.py index f7f92e6..5806df8 100644 --- a/orchestrator/api/progress.py +++ b/orchestrator/api/progress.py @@ -102,6 +102,7 @@ def _stage_snapshot(root: Path) -> StageSnapshot | None: Path(described["source"]), Path(described["repository"]), described.get("options") or {}, + retained_delivery=described.get("retainedDelivery"), ) journal = Journal(root) records = read_json(root / "state.json").get("stages", {}) @@ -134,6 +135,8 @@ def _stage_snapshot(root: Path) -> StageSnapshot | None: ) status = {"ok": "succeeded", "gate": "blocked"}.get(status, status) required = needs_of(name, people, many, context.options) + if context.retained_delivery is not None: + required = [dependency for dependency in required if dependency in plan_for(context)] required_dependencies[name] = required missing = [dep for dep in required if dep not in done] latest = latest_results.get(name, {}) diff --git a/orchestrator/api/service.py b/orchestrator/api/service.py index 16fe0e6..9c2d3e3 100644 --- a/orchestrator/api/service.py +++ b/orchestrator/api/service.py @@ -11,7 +11,7 @@ from orchestrator.api.app import ApiSettings, SourceStore, create_app from orchestrator.api.artifacts import ArtifactReader, safe_filename from orchestrator.api.storage import FilesystemArtifactReader, FilesystemSourceStore -from orchestrator.config import resolve_runs_root +from orchestrator.config import REPOSITORY_ROOT, resolve_runs_root from orchestrator.journal import Journal, RunProjection from orchestrator.repository import PipelineRepository from orchestrator.steps import STEPS @@ -85,6 +85,117 @@ def open( run_dir.rename(destination) return destination + def saved_inputs(self, run_id: str) -> dict: + from orchestrator.retained_inputs import inspect_inputs + + summary = self.repository.run_summary(run_id) + try: + snapshot = inspect_inputs(self._dir(run_id), summary) + copies = snapshot["copies"] + return { + "available": True, + "sourceRunId": run_id, + "fileCount": len(copies), + "byteCount": sum(path.stat().st_size for path in copies.values()), + } + except (ValueError, OSError, KeyError, TypeError) as error: + return {"available": False, "sourceRunId": run_id, "reason": str(error)} + + def prepare_saved_inputs(self, source_id: str, run_id: str, *, settings, user: str) -> Path: + from orchestrator.contracts import Run + from orchestrator.retained_inputs import ( + GOAL, + copy_inputs, + inspect_inputs, + verify_delivery_inputs, + ) + + source_dir = self._dir(source_id) + with Journal(source_dir).admission_lock(): + summary = self.repository.run_summary(source_id) + snapshot = inspect_inputs(source_dir, summary) + original = snapshot["manifest"] + options = original["options"] + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", run_id): + raise ValueError("invalid run id") + destination = self.root / name_for(self.root, run_id, run_id) + if destination.exists() or destination.is_symlink(): + raise FileExistsError("run directory already exists") + staging_root = self.root / ".submissions" + staging_root.mkdir(exist_ok=True) + created = False + try: + with tempfile.TemporaryDirectory(prefix="retained-", dir=staging_root) as staging: + run_dir = Path(staging) / "run" + run_dir.mkdir() + delivery = copy_inputs(snapshot, run_dir) + source_name = Path(original["source"]).name + described = { + "runId": run_id, + "repository": str(REPOSITORY_ROOT), + "sourceArtifactId": summary.get("sourceArtifactId") + or original.get("sourceArtifactId"), + "name": destination.name, + "source": str(destination / source_name), + "sourceSha256": original["sourceSha256"], + "options": options, + "goal": GOAL, + "retainedDelivery": delivery, + } + (run_dir / "run.json").write_text(json.dumps(described, indent=2)) + (run_dir / "paused").write_text( + "Prepared saved-input delivery; resume explicitly.\n" + ) + stages = [ + "pi3x", + "tracks", + *[f"lhm_motion_{i:02d}" for i in snapshot["tracks"]], + "frame_align", + ] + state = { + "stages": { + stage: { + "status": "skipped" if stage == "frame_align" else "ok", + "execution": "retained-input-replay", + "freshInference": False, + "sourceRun": source_id, + } + for stage in stages + } + } + (run_dir / "state.json").write_text(json.dumps(state, indent=2)) + verify_delivery_inputs(run_dir, delivery) + self.repository.create_run_row( + Run( + id=run_id, + graph_version="wander.agent-run/1", + code_revision=settings.code_revision, + container_digest=settings.container_digest, + source_sha256=original["sourceSha256"], + source_artifact_id=summary.get("sourceArtifactId") + or original.get("sourceArtifactId"), + configuration={"options": options, "retainedDelivery": delivery}, + status="paused", + created_by=user, + parent_run_id=source_id, + ) + ) + created = True + for stage in [*stages, "package_people"]: + self.repository.ensure_node( + run_id=run_id, + node_id=stage, + stage_type=stage, + definition={"id": stage}, + dependencies=[], + ) + run_dir.rename(destination) + return destination + except Exception: + if created: + self.repository.delete_unstarted_run(run_id) + raise + def journal(self, run_id: str) -> Journal: run_dir = self._dir(run_id) return Journal(run_dir, projection=RunProjection(self.repository, run_id, catalogue=STEPS)) diff --git a/orchestrator/cli.py b/orchestrator/cli.py index 45feda2..e693c7a 100644 --- a/orchestrator/cli.py +++ b/orchestrator/cli.py @@ -51,6 +51,7 @@ class RunContext: repository: Path options: dict[str, Any] source_sha256: str | None = None + retained_delivery: dict | None = None @classmethod def load(cls) -> RunContext: @@ -71,6 +72,7 @@ def load(cls) -> RunContext: ), options=described.get("options") or {}, source_sha256=described.get("sourceSha256"), + retained_delivery=described.get("retainedDelivery"), ) def journal(self) -> Journal: @@ -128,6 +130,8 @@ def run_clip_command(context: RunContext, step: str, extra: list[str]) -> list[s command.append("--no-objects") if not options.finetune: command.append("--skip-finetune") + if getattr(context, "retained_delivery", None) is not None: + command += ["--retained-scene", str(context.run_dir / "retained/scene.json")] return command + extra @@ -259,6 +263,20 @@ def admit_step( return 2 if refuse_controlled_step(context, journal, args.step, args.attempt): return 2 + if context.retained_delivery is not None: + problem = retained_delivery_problem(context, args, journal) + if problem: + if args.attempt: + journal.append( + "step.finished", + step=args.step, + attempt=args.attempt, + status="blocked", + error=problem, + preflight=True, + ) + print(problem, file=sys.stderr) + return 2 problem = source_problem(context) if refuse_controlled_step(context, journal, args.step, args.attempt): return 2 @@ -339,6 +357,21 @@ def admit_step( return AdmittedStep(attempt, command, log) +def retained_delivery_problem(context: RunContext, args: argparse.Namespace, journal: Journal): + """A prepared delivery can package its pinned inputs once, without inference or retries.""" + if args.step != "package_people" or args.rest: + return "Saved-input runs allow only package_people with the prepared inputs and no extra parameters." + if not args.attempt and any(entry.kind == "step.started" for entry in journal.entries()): + return "This saved-input run already used its packaging attempt. Inspect its result; retries are disabled." + from orchestrator.retained_inputs import verify_delivery_inputs + + try: + verify_delivery_inputs(context.run_dir, context.retained_delivery) + except (OSError, ValueError, TypeError, KeyError) as error: + return f"Saved inputs failed verification: {error}" + return None + + def command_step(context: RunContext, args: argparse.Namespace) -> int: """Admit one attempt atomically, then run it here or in a detached child.""" take_our_flags(args) @@ -575,9 +608,21 @@ def plan_for(context: RunContext) -> list[str]: will never exist. """ people, many = graph_shape(context) + if context.retained_delivery is not None: + return retained_delivery_steps(people) return people_steps(planned_steps(context.options), people, multiperson=many) +def retained_delivery_steps(people: int) -> list[str]: + return [ + "pi3x", + "tracks", + *[f"lhm_motion_{index:02d}" for index in range(people)], + "frame_align", + "package_people", + ] + + def waiting_on( step: str, done: set[str], people: int = 1, multiperson: bool = False, options=None ) -> list[str]: diff --git a/orchestrator/repository.py b/orchestrator/repository.py index ad9b6be..fb7f148 100644 --- a/orchestrator/repository.py +++ b/orchestrator/repository.py @@ -229,6 +229,7 @@ def run_summary(self, run_id: str) -> dict[str, Any]: ], "graphVersion": run.graph_version, "sourceSha256": run.source_sha256, + "sourceArtifactId": run.source_artifact_id, "configuration": run.configuration, "budget": run.budget, "createdAt": run.created_at.isoformat(), diff --git a/orchestrator/retained_inputs.py b/orchestrator/retained_inputs.py new file mode 100644 index 0000000..305b7cd --- /dev/null +++ b/orchestrator/retained_inputs.py @@ -0,0 +1,311 @@ +"""Confined, immutable snapshots for a single local retained-scene delivery.""" + +from __future__ import annotations + +import hashlib +import json +import re +import shutil +import stat +import struct +import tempfile +from pathlib import Path + +from orchestrator.api.viewer import WORLD_SUFFIXES, run_name +from orchestrator.journal import Journal +from scripts.retained_scene import RetainedScene, prepare_config + +SCHEMA = "wander.saved-inputs/1" +GOAL = """Perform exactly one detached local package_people stage using the pinned retained scene. No provider calls, inference, retries, other stages, edits to inputs, or quality acceptance. Start package_people once and end the turn. On the next turn inspect its result and ask for human viewer review. If blocked or failed, ask for help; never retry or finish the run automatically.""" + + +def digest(path: Path) -> str: + with path.open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def safe_file(root: Path, relative: str) -> Path: + if not isinstance(relative, str) or not relative or "\\" in relative or ":" in relative: + raise ValueError("invalid saved input path") + parts = relative.split("/") + if any(not part or part.startswith(".") for part in parts): + raise ValueError("unconfined saved input path") + path = root + if root.is_symlink(): + raise ValueError("symlink input root") + for part in parts: + path = path / part + if path.is_symlink(): + raise ValueError("symlink saved input") + if not stat.S_ISREG(path.stat().st_mode): + raise ValueError("nonregular saved input") + return path + + +def read(root: Path, relative: str): + return json.loads(safe_file(root, relative).read_text()) + + +def require_self_contained_model(path: Path) -> None: + """Only copy object formats whose viewer dependencies fit in the pinned file.""" + if path.suffix.lower() == ".ply": + return + if path.suffix.lower() != ".glb": + raise ValueError("saved object models must be self-contained PLY or GLB files") + with path.open("rb") as stream: + header = stream.read(20) + if len(header) != 20: + raise ValueError("invalid saved GLB header") + magic, version, total, length, chunk_type = struct.unpack("<4sIIII", header) + if ( + magic != b"glTF" + or version != 2 + or total != path.stat().st_size + or chunk_type != 0x4E4F534A + or not 0 < length <= min(total - 20, 16_000_000) + ): + raise ValueError("invalid saved GLB manifest") + manifest = json.loads(stream.read(length)) + for item in [*manifest.get("buffers", []), *manifest.get("images", [])]: + uri = item.get("uri") + if uri is not None and (not isinstance(uri, str) or not uri.startswith("data:")): + raise ValueError("saved GLB references external buffers or textures") + + +def verify_delivery_inputs(run_dir: Path, delivery: dict) -> None: + if ( + not isinstance(delivery, dict) + or delivery + != { + "schema": SCHEMA, + "sourceRunId": delivery.get("sourceRunId"), + "config": "retained/scene.json", + "inputs": "retained/inputs.json", + "stage": "package_people", + } + or not isinstance(delivery.get("sourceRunId"), str) + ): + raise ValueError("invalid retained delivery manifest") + entries = read(run_dir, delivery["inputs"]) + if not isinstance(entries, list) or not entries: + raise ValueError("missing saved input pins") + seen = set() + for entry in entries: + if not isinstance(entry, dict) or set(entry) != {"path", "sha256", "size"}: + raise ValueError("invalid saved input pin") + path = safe_file(run_dir, entry["path"]) + if ( + entry["path"] in seen + or path.stat().st_size != entry["size"] + or digest(path) != entry["sha256"] + ): + raise ValueError("saved input digest mismatch") + seen.add(entry["path"]) + source = Path(read(run_dir, "run.json")["source"]) + if "retained/scene.json" not in seen or source.name not in seen: + raise ValueError("missing source or config pin") + config = read(run_dir, "retained/scene.json") + if config.get("package") != "package" or config.get("world", {}).get("path") != "world.spz": + raise ValueError("retained config must reference copied assets") + scene = RetainedScene(run_dir / "retained/scene.json", safe_file(run_dir, source.name)) + needed = { + source.name, + "retained/scene.json", + "retained/world.spz", + "pi3x/cameras.json", + "tracks/tracks.json", + } + needed.update(f"retained/package/{name}" for name in scene.files) + for name, sequence in scene.sequences.items(): + needed.update( + f"retained/package/{Path(name).parent}/{frame}" for frame in sequence["frames"] + ) + tracks = read(run_dir, "tracks/tracks.json")["tracks"] + for track in tracks: + folder = f"lhm-motion-{track['track']:02d}" + needed.add(f"{folder}/sequence.json") + sequence = read(run_dir, f"{folder}/sequence.json") + needed.update(f"{folder}/{frame}" for frame in sequence["frames"]) + # Optional supported inputs must also be pinned if present; removal of a receipt entry + # cannot silently remove a packager dependency from verification. + for path in run_dir.rglob("*"): + relative = path.relative_to(run_dir).as_posix() + if relative.startswith(("retained/package/", "lhm-motion-", "pi3x/", "tracks/")) and ( + path.is_file() or path.is_symlink() + ): + needed.add(relative) + if not needed <= seen: + raise ValueError("incomplete saved input pins") + + +def inspect_inputs(root: Path, summary: dict) -> dict: + manifest = read(root, "run.json") + options = manifest.get("options", {}) + if ( + not (options.get("all_people") or options.get("people", 1) > 1) + or options.get("finetune") + or options.get("marble", "video") not in WORLD_SUFFIXES + ): + raise ValueError("requires tracked people, an enabled world, and skip-finetune") + if ( + summary.get("recordedStatus", summary["status"]) + not in {"paused", "succeeded", "failed", "canceled"} + and not (root / "paused").is_file() + ): + raise ValueError("source run must be paused or finished") + if Journal(root).active_attempts() or any( + a["status"] in {"pending", "running", "unknown"} for a in summary.get("attempts", []) + ): + raise ValueError("source has unresolved stage attempts") + source = Path(manifest["source"]) + if source.parent.resolve() != root.resolve(): + raise ValueError("source must be inside the run") + source = safe_file(root, source.name) + if digest(source) != manifest.get("sourceSha256") or digest(source) != summary["sourceSha256"]: + raise ValueError("source clip digest mismatch") + name = run_name(root) + package = root / "public" / "worlds" / f"{name}-4d" + world_rel = f"public/marble-{name}-{WORLD_SUFFIXES[options.get('marble', 'video')]}.spz" + world = safe_file(root, world_rel) + files = {"people.json", "cameras.json", "audio.json", "scene.json"} + people = read(package, "people.json") + for person in people["people"]: + sequence = person["sequence"] + files.add(sequence) + seq = read(package, sequence) + for frame, expected in zip(seq["frames"], seq["frame_sha256"], strict=True): + rel = str(Path(sequence).parent / frame) + if digest(safe_file(package, rel)) != expected: + raise ValueError("exported frame digest mismatch") + files.add(rel) + + def asset(value): + prefix = f"/worlds/{package.name}/" + if value.startswith(prefix): + value = value[len(prefix) :] + if Path(value).suffix.lower() not in { + ".ply", + ".glb", + ".gltf", + ".wav", + ".mp3", + ".m4a", + ".ogg", + }: + raise ValueError("unsupported retained asset") + safe_file(package, value) + return value + + audio = read(package, "audio.json") + if audio.get("original"): + files.add(asset(audio["original"]["url"])) + if (package / "objects.json").exists() or (package / "objects.json").is_symlink(): + files.add("objects.json") + for obj in read(package, "objects.json")["objects"]: + model = obj.get("appearance", {}).get("model") + if model: + relative = asset(model) + require_self_contained_model(safe_file(package, relative)) + files.add(relative) + copies = {source.name: source, "retained/world.spz": world} + for rel in files: + copies[f"retained/package/{rel}"] = safe_file(root, str((package / rel).relative_to(root))) + for rel in ("pi3x/cameras.json", "tracks/tracks.json"): + copies[rel] = safe_file(root, rel) + if not read(root, "pi3x/cameras.json").get("cameras"): + raise ValueError("missing saved cameras") + tracks = read(root, "tracks/tracks.json") + ids = [p["track"] for p in people["people"]] + if set(ids) != {t["track"] for t in tracks["tracks"]}: + raise ValueError("retained cast differs from tracked people") + if sorted(ids) != list(range(len(ids))) or len(ids) > options.get("people", 1): + raise ValueError("unsupported retained track slots") + for track in ids: + if type(track) is not int or track < 0: + raise ValueError("invalid track") + folder = f"lhm-motion-{track:02d}" + seq = read(root, f"{folder}/sequence.json") + if any( + not re.fullmatch(r"frame_[A-Za-z0-9_-]+\.ply", frame) for frame in seq.get("frames", []) + ): + raise ValueError("unsupported motion frame path") + if not seq.get("frames"): + raise ValueError("empty saved motion") + for rel in [f"{folder}/sequence.json", *[f"{folder}/{frame}" for frame in seq["frames"]]]: + copies[rel] = safe_file(root, rel) + if "frame_sha256" in seq: + for frame, expected in zip(seq["frames"], seq["frame_sha256"], strict=True): + if digest(copies[f"{folder}/{frame}"]) != expected: + raise ValueError("saved motion frame digest mismatch") + for extra in ("registration.json", "motion.json", "missing-poses.json"): + rel = f"{folder}/{extra}" + if (root / rel).exists() or (root / rel).is_symlink(): + copies[rel] = safe_file(root, rel) + if (root / "pi3x/framealign.json").exists() or (root / "pi3x/framealign.json").is_symlink(): + copies["pi3x/framealign.json"] = safe_file(root, "pi3x/framealign.json") + with tempfile.TemporaryDirectory() as tmp: + prepare_config(package, world, source, Path(tmp) / "scene.json") + return { + "manifest": manifest, + "copies": copies, + "tracks": ids, + "root": root, + "hashes": {name: digest(path) for name, path in copies.items()}, + } + + +def copy_inputs(snapshot: dict, destination: Path) -> dict: + pins = [] + for relative, source in sorted(snapshot["copies"].items()): + source = safe_file(snapshot["root"], str(source.relative_to(snapshot["root"]))) + expected = snapshot["hashes"][relative] + if digest(source) != expected: + raise ValueError("saved input changed before copy") + target = destination / relative + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(source, target) + if digest(target) != expected: + raise ValueError("saved input changed during copy") + pins.append({"path": relative, "sha256": expected, "size": target.stat().st_size}) + retained = destination / "retained" + # Exported viewer URLs name the old package. Normalize only supported asset + # references in our copies so the snapshot remains valid after relocation. + prefix = f"/worlds/{snapshot['manifest']['name']}-4d/" + audio_path = retained / "package/audio.json" + audio = json.loads(audio_path.read_text()) + if audio.get("original") and audio["original"]["url"].startswith(prefix): + audio["original"]["url"] = audio["original"]["url"][len(prefix) :] + audio_path.write_text(json.dumps(audio, indent=2) + "\n") + objects_path = retained / "package/objects.json" + if objects_path.exists(): + objects = json.loads(objects_path.read_text()) + changed = False + for obj in objects["objects"]: + appearance = obj.get("appearance", {}) + model = appearance.get("model") + if model and model.startswith(prefix): + appearance["model"] = model[len(prefix) :] + changed = True + if changed: + objects_path.write_text(json.dumps(objects, indent=2) + "\n") + for pin in pins: + copied = destination / pin["path"] + pin.update(sha256=digest(copied), size=copied.stat().st_size) + config = retained / "scene.json" + prepare_config( + retained / "package", + retained / "world.spz", + destination / Path(snapshot["manifest"]["source"]).name, + config, + ) + pins.append( + {"path": "retained/scene.json", "sha256": digest(config), "size": config.stat().st_size} + ) + (retained / "inputs.json").write_text(json.dumps(pins, indent=2) + "\n") + return { + "schema": SCHEMA, + "sourceRunId": snapshot["manifest"]["runId"], + "config": "retained/scene.json", + "inputs": "retained/inputs.json", + "stage": "package_people", + } diff --git a/orchestrator/session.py b/orchestrator/session.py index 9185122..d2e473f 100644 --- a/orchestrator/session.py +++ b/orchestrator/session.py @@ -241,6 +241,21 @@ def plan(self, options: dict) -> list[str]: real number. """ people = max(1, int(options.get("people") or 1)) + if self.described.get("retainedDelivery") is not None: + from orchestrator.cli import RunContext, plan_for + + described = self.described + return plan_for( + RunContext( + self.run_id, + self.run_dir, + described["name"], + Path(described["source"]), + Path(described["repository"]), + options, + retained_delivery=described["retainedDelivery"], + ) + ) many = bool(options.get("all_people")) or people > 1 return people_steps(planned_steps(options), people, multiperson=many) diff --git a/scripts/run_clip.py b/scripts/run_clip.py index dfdb21f..1fdb0b6 100644 --- a/scripts/run_clip.py +++ b/scripts/run_clip.py @@ -1554,7 +1554,10 @@ def package(destination): "--out", str(destination), ] - run(command, self.ctx / "package_people.log") + if retained: + run(command, self.ctx / "package_people.log", attempts=1) + else: + run(command, self.ctx / "package_people.log") if retained is None: source_audio_manifest(self.clip, tracks["duration"], out) diff --git a/scripts/tests/test_run_clip_retained_scene.py b/scripts/tests/test_run_clip_retained_scene.py index 7b1ed00..8787576 100644 --- a/scripts/tests/test_run_clip_retained_scene.py +++ b/scripts/tests/test_run_clip_retained_scene.py @@ -15,6 +15,7 @@ from scripts.tests import test_retained_scene as retained_fixtures scene = retained_fixtures.scene +original_run = run_clip.run @pytest.fixture @@ -37,7 +38,8 @@ def pipeline(scene, tmp_path): public = tmp_path / "public" public.mkdir() - def package(command, log): + def package(command, log, *, attempts): + assert attempts == 1 assert command[command.index("--shared-scale") + 1] == "1" shutil.copytree(fresh, Path(command[command.index("--out") + 1])) @@ -62,6 +64,24 @@ def test_delivers_package_and_selected_world_with_no_quality_acceptance(pipeline assert not list((public / "worlds").glob(".retained-*")) +def test_retained_packager_does_not_retry_rate_limit_errors(pipeline, monkeypatch): + p, _, _ = pipeline + calls = [] + + def fail(command, **kwargs): + calls.append(command) + kwargs["stdout"].write("rate limit: fixture failure\n") + kwargs["stdout"].flush() + return SimpleNamespace(returncode=1) + + monkeypatch.setattr(run_clip.subprocess, "run", fail) + # Use the real command runner under the package fixture's mock. + with patch.object(run_clip, "run", original_run): + with pytest.raises(RuntimeError): + p.package_people() + assert len(calls) == 1 + + @pytest.mark.parametrize("conflict", ["package", "world", "scale", "missing_reference", "finetune"]) def test_conflicts_preserve_existing_files_before_packaging(pipeline, conflict): p, public, execute = pipeline diff --git a/scripts/tests/test_saved_input_admission.py b/scripts/tests/test_saved_input_admission.py new file mode 100644 index 0000000..ae808ab --- /dev/null +++ b/scripts/tests/test_saved_input_admission.py @@ -0,0 +1,122 @@ +"""Prepared delivery admits one pinned local packaging attempt, never inference.""" + +import json +from dataclasses import replace +from unittest.mock import Mock + +import pytest + +from orchestrator import cli +from orchestrator.api.progress import stage_snapshot +from scripts.tests import test_run_controls as fixtures + +run = fixtures.run + + +@pytest.fixture +def saved_run(run, monkeypatch): + options = {"marble": "video", "people": 2, "all_people": True, "finetune": False} + delivery = { + "schema": "wander.saved-inputs/1", + "sourceRunId": "prior", + "config": "retained/scene.json", + "inputs": "retained/inputs.json", + "stage": "package_people", + } + run.context = replace(run.context, options=options, retained_delivery=delivery) + root = run.context.run_dir + (root / "run.json").write_text( + json.dumps( + { + "runId": run.context.run_id, + "name": run.context.name, + "source": str(run.context.source), + "sourceSha256": run.context.source_sha256, + "repository": str(run.context.repository), + "options": run.context.options, + "retainedDelivery": run.context.retained_delivery, + } + ) + ) + (root / "tracks").mkdir() + (root / "tracks/tracks.json").write_text('{"trackCount":2}') + verifier = Mock() + monkeypatch.setattr("orchestrator.retained_inputs.verify_delivery_inputs", verifier) + run.verifier = verifier + return run + + +def args(*tokens): + return cli.build_parser().parse_args(["step", *tokens]) + + +@pytest.mark.parametrize( + "tokens", + [ + ("pi3x",), + ("tracks",), + ("marble_video",), + ("package_people", "--force", "package_people"), + ("package_people", "--retained-scene", "other.json"), + ("package_people", "--no-gate"), + ], +) +def test_prepared_run_rejects_inference_and_overrides(saved_run, monkeypatch, tokens): + spawn = Mock() + monkeypatch.setattr(cli.subprocess, "Popen", spawn) + assert cli.command_step(saved_run.context, args(*tokens)) == 2 + assert not saved_run.journal.entries() + spawn.assert_not_called() + saved_run.verifier.assert_not_called() + + +def test_one_local_attempt_uses_prepared_config_then_refuses_retry(saved_run, monkeypatch): + spawn = Mock(return_value=fixtures.stage_process()) + monkeypatch.setattr(cli.subprocess, "Popen", spawn) + assert cli.command_step(saved_run.context, args("--wait", "package_people")) == 0 + launched = spawn.call_args.args[0] + assert launched[-2:] == [ + "--retained-scene", + str(saved_run.context.run_dir / "retained/scene.json"), + ] + assert "--no-publish" in launched + assert cli.command_step(saved_run.context, args("--wait", "package_people")) == 2 + spawn.assert_called_once() + assert len([e for e in saved_run.journal.entries() if e.kind == "step.started"]) == 1 + + +def test_detached_delivery_claims_original_attempt_once(saved_run, monkeypatch): + spawn = Mock(return_value=fixtures.stage_process()) + monkeypatch.setattr(cli.subprocess, "Popen", spawn) + assert cli.command_step(saved_run.context, args("package_people")) == 0 + child = cli.build_parser().parse_args(spawn.call_args.args[0][3:]) + assert cli.command_step(saved_run.context, child) == 0 + assert cli.command_step(saved_run.context, child) == 2 + assert cli.command_step(saved_run.context, args("package_people")) == 2 + assert spawn.call_count == 2 # detached waiter plus its one stage process + assert not saved_run.journal.active_attempts() + + +def test_changed_inputs_block_execution_and_close_pending_child(saved_run, monkeypatch): + spawn = Mock(return_value=fixtures.stage_process()) + monkeypatch.setattr(cli.subprocess, "Popen", spawn) + assert cli.command_step(saved_run.context, args("package_people")) == 0 + child = cli.build_parser().parse_args(spawn.call_args.args[0][3:]) + saved_run.verifier.side_effect = ValueError("input digest mismatch") + assert cli.command_step(saved_run.context, child) == 2 + spawn.assert_called_once() + assert not saved_run.journal.active_attempts() + result = saved_run.journal.entries()[-1] + assert result.kind == "step.finished" + assert result.data["status"] == "blocked" + assert "digest mismatch" in result.data["error"] + + +def test_api_cli_and_agent_plan_only_saved_prerequisites_and_packaging(saved_run): + expected = ["pi3x", "tracks", "lhm_motion_00", "lhm_motion_01", "frame_align", "package_people"] + assert cli.plan_for(saved_run.context) == expected + session = fixtures.session(saved_run) + assert session.plan(saved_run.context.options) == expected + snapshot = stage_snapshot(saved_run.context.run_dir) + assert [stage["name"] for stage in snapshot.stages] == expected + assert all(set(deps) <= set(expected) for deps in snapshot.required_dependencies.values()) diff --git a/scripts/tests/test_saved_inputs.py b/scripts/tests/test_saved_inputs.py new file mode 100644 index 0000000..db6351e --- /dev/null +++ b/scripts/tests/test_saved_inputs.py @@ -0,0 +1,252 @@ +"""Saved-input API preparation never executes a pipeline stage.""" + +import json +import shutil +from unittest.mock import patch + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from orchestrator.api.app import ApiSettings +from orchestrator.api.service import create_service +from orchestrator.contracts import Run +from orchestrator.database import Base +from orchestrator.retained_inputs import digest, verify_delivery_inputs +from scripts.tests.test_retained_scene import scene, write_json # noqa: F401 + +TOKEN = "saved-inputs-token-with-24-characters" +AUTH = {"Authorization": f"Bearer {TOKEN}"} + + +@pytest.fixture +def prepared(scene, tmp_path): # noqa: F811 + _, clip, old, fresh, world = scene + root = tmp_path / "runs" + source = root / "original" + package = source / "public/worlds/original-4d" + shutil.copytree(old, package) + for path in fresh.rglob("*.ply"): + target = package / path.relative_to(fresh) + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(path, target) + shutil.copyfile(world, source / "public/marble-original-clean.spz") + shutil.copyfile(clip, source / "source.mp4") + options = {"all_people": True, "people": 2, "marble": "video", "finetune": False} + write_json( + source / "run.json", + { + "runId": "original", + "sourceArtifactId": "source:original", + "name": "original", + "source": str(source / "source.mp4"), + "sourceSha256": digest(clip), + "options": options, + }, + ) + write_json(source / "pi3x/cameras.json", json.loads((package / "cameras.json").read_text())) + write_json(source / "tracks/tracks.json", {"tracks": [{"track": 0}, {"track": 1}]}) + for track, name in enumerate(("person", "person_01")): + shutil.copytree(package / name, source / f"lhm-motion-{track:02d}") + engine = create_engine( + f"sqlite:///{tmp_path / 'test.sqlite'}", connect_args={"check_same_thread": False} + ) + Base.metadata.create_all(engine) + from orchestrator.repository import PipelineRepository + + repository = PipelineRepository(sessionmaker(engine, expire_on_commit=False)) + repository.create_run_row( + Run( + id="original", + graph_version="test", + code_revision="1234567", + source_sha256=digest(clip), + source_artifact_id="source:original", + status="paused", + created_by="test", + configuration={"options": options}, + ) + ) + app = create_service( + repository=repository, + runs_root=root, + settings=ApiSettings(bearer_token=TOKEN, code_revision="1234567"), + artifact_root=tmp_path / "artifacts", + ) + with TestClient(app) as client: + yield client, app, source + engine.dispose() + + +def test_prepare_paused_and_pinned(prepared): + client, app, source = prepared + url = "/api/pipeline/runs/original/saved-inputs" + assert client.get(url).status_code == 401 + assert client.post(url, json={}).status_code == 401 + assert client.get(url, headers=AUTH).json()["available"] + response = client.post(url, headers=AUTH, json={"id": "delivery"}) + assert response.status_code == 201, response.text + assert response.json() == {"id": "delivery", "status": "paused"} + summary = app.state.repository.run_summary("delivery") + assert summary["status"] == "paused" + assert summary["parentRunId"] == "original" + assert not summary["attempts"] and not summary["reviews"] + destination = app.state.runs._dir("delivery") + manifest = json.loads((destination / "run.json").read_text()) + verify_delivery_inputs(destination, manifest["retainedDelivery"]) + assert (destination / "paused").is_file() + state = json.loads((destination / "state.json").read_text()) + assert all(record["freshInference"] is False for record in state["stages"].values()) + assert client.post(url, headers=AUTH, json={"id": "delivery"}).status_code == 409 + (destination / "pi3x/cameras.json").write_text("corrupt") + with pytest.raises(ValueError): + verify_delivery_inputs(destination, manifest["retainedDelivery"]) + + +@pytest.mark.parametrize("damage", ["missing", "corrupt", "symlink"]) +def test_unavailable(prepared, damage): + client, _, source = prepared + path = source / "public/worlds/original-4d/person/frame_000.ply" + if damage == "missing": + path.unlink() + elif damage == "corrupt": + path.write_text("bad") + else: + target = source / "other.ply" + path.rename(target) + path.symlink_to(target) + url = "/api/pipeline/runs/original/saved-inputs" + assert client.get(url, headers=AUTH).json()["available"] is False + assert client.post(url, headers=AUTH, json={"id": "delivery"}).status_code == 409 + + +def test_failure_rolls_back(prepared): + client, app, source = prepared + with patch.object(app.state.repository, "ensure_node", side_effect=ValueError("failed seed")): + response = client.post( + "/api/pipeline/runs/original/saved-inputs", headers=AUTH, json={"id": "delivery"} + ) + assert response.status_code == 409 + with pytest.raises(KeyError): + app.state.repository.run_summary("delivery") + assert not list((source.parent / ".submissions").iterdir()) + + +def test_absolute_export_urls_are_normalized_in_copies(prepared): + client, app, source = prepared + package = source / "public/worlds/original-4d" + audio_path = package / "audio.json" + audio = json.loads(audio_path.read_text()) + audio["original"]["url"] = "/worlds/original-4d/" + audio["original"]["url"] + write_json(audio_path, audio) + objects_path = package / "objects.json" + objects = json.loads(objects_path.read_text()) + for obj in objects["objects"]: + obj["appearance"]["model"] = "/worlds/original-4d/" + obj["appearance"]["model"] + write_json(objects_path, objects) + original_hashes = [digest(path) for path in (audio_path, objects_path)] + response = client.post( + "/api/pipeline/runs/original/saved-inputs", headers=AUTH, json={"id": "delivery"} + ) + assert response.status_code == 201, response.text + destination = app.state.runs._dir("delivery") + verify_delivery_inputs( + destination, json.loads((destination / "run.json").read_text())["retainedDelivery"] + ) + copied_audio = json.loads((destination / "retained/package/audio.json").read_text()) + assert not copied_audio["original"]["url"].startswith("/") + copied_objects = json.loads((destination / "retained/package/objects.json").read_text()) + assert all(not obj["appearance"]["model"].startswith("/") for obj in copied_objects["objects"]) + assert original_hashes == [digest(path) for path in (audio_path, objects_path)] + + +def test_unresolved_source_is_unavailable(prepared): + client, app, source = prepared + from orchestrator.journal import Journal + + Journal(source).append("step.started", step="package_people", attempt="pending") + url = "/api/pipeline/runs/original/saved-inputs" + assert not client.get(url, headers=AUTH).json()["available"] + assert client.post(url, headers=AUTH, json={"id": "delivery"}).status_code == 409 + + +def test_source_changed_after_inspection_is_rejected(prepared, tmp_path): + _, app, source = prepared + from orchestrator.retained_inputs import copy_inputs, inspect_inputs + + snapshot = inspect_inputs(source, app.state.repository.run_summary("original")) + (source / "source.mp4").write_bytes(b"changed") + with pytest.raises(ValueError, match="changed before copy"): + copy_inputs(snapshot, tmp_path / "copy") + + +def test_removed_pin_is_rejected(prepared): + client, app, _ = prepared + response = client.post( + "/api/pipeline/runs/original/saved-inputs", headers=AUTH, json={"id": "delivery"} + ) + assert response.status_code == 201 + destination = app.state.runs._dir("delivery") + pins_path = destination / "retained/inputs.json" + pins = json.loads(pins_path.read_text()) + pins = [pin for pin in pins if pin["path"] != "lhm-motion-00/frame_000.ply"] + pins_path.write_text(json.dumps(pins)) + with pytest.raises(ValueError, match="incomplete saved input pins"): + verify_delivery_inputs( + destination, json.loads((destination / "run.json").read_text())["retainedDelivery"] + ) + + +def test_symlink_ancestor_is_unavailable(prepared): + client, _, source = prepared + path = source / "pi3x" + target = source / "other-cameras" + path.rename(target) + path.symlink_to(target, target_is_directory=True) + response = client.get("/api/pipeline/runs/original/saved-inputs", headers=AUTH) + assert response.json()["available"] is False + + +@pytest.mark.parametrize("extension", ["gltf", "glb"]) +def test_external_object_dependencies_are_not_silently_dropped(prepared, extension): + import struct + + client, _, source = prepared + package = source / "public/worlds/original-4d" + objects = json.loads((package / "objects.json").read_text()) + model = f"objects/bottle/model.{extension}" + content = json.dumps({"asset": {"version": "2.0"}, "buffers": [{"uri": "mesh.bin"}]}).encode() + if extension == "glb": + content += b" " * (-len(content) % 4) + content = ( + struct.pack("<4sIIII", b"glTF", 2, 20 + len(content), len(content), 0x4E4F534A) + + content + ) + (package / model).write_bytes(content) + appearance = objects["objects"][0]["appearance"] + appearance["model"] = model + appearance["modelSha256"] = digest(package / model) + write_json(package / "objects.json", objects) + result = client.get("/api/pipeline/runs/original/saved-inputs", headers=AUTH).json() + assert not result["available"] + assert "self-contained" in result["reason"] or "external" in result["reason"] + + +def test_retry_cannot_change_prepared_run_or_unpause_it(prepared): + client, app, _ = prepared + assert ( + client.post( + "/api/pipeline/runs/original/saved-inputs", headers=AUTH, json={"id": "delivery"} + ).status_code + == 201 + ) + before = app.state.repository.events_after("delivery", 0) + response = client.post( + "/api/pipeline/runs/delivery/retry", + headers=AUTH, + json={"node_id": "pi3x", "rationale": "Try inference"}, + ) + assert response.status_code == 409 + assert app.state.repository.events_after("delivery", 0) == before + assert (app.state.runs._dir("delivery") / "paused").is_file()