diff --git a/app/services/generation_record.py b/app/services/generation_record.py new file mode 100644 index 00000000..f3c890aa --- /dev/null +++ b/app/services/generation_record.py @@ -0,0 +1,1077 @@ +"""Typed generation-record v1: one attempt to produce an asset. + +This module does not import FastAPI, WanGP or launch. It is a portable +read/write projection over asset-manifest v1, generation provenance and the +job lifecycle — not a second media store. + +Identity policy (b): a retry mints a new ``generation_id`` and links the +parent attempt in ``lineage.parents``. ``asset_id`` is reused only when the +bytes are the same artifact. Resume of a queued/running record keeps both IDs +and the last durable status; it never invents success. + +Public status is ``planned | queued | running | completed | failed | +cancelled``. Asset-manifest ``prepared`` projects to ``planned``. Manifest +``partial`` is not a seventh public status: it becomes ``completed`` with +``result.kind = "partial"`` when a filename exists, otherwise ``failed``. +A running record with ``cancellation.requested`` is the public form of the +in-process ``cancelling`` job state. +""" + +from __future__ import annotations + +import json +import os +import threading +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping, Sequence, TypedDict + +from .asset_manifest import _iso, _milliseconds, _redact, sidecar_path +from .generation_provenance import provenance_from_manifest, resolve_generation_location + + +SCHEMA_NAME = "hocuspocus.generation-record" +SCHEMA_VERSION = 1 +PROMPT_DISPLAY_MAX = 180 +ATTEMPT_IDENTITY_POLICY = "new_generation_id" + +PRODUCTS = frozenset({ + "studio", "story_lab", "series_lab", "director", "comic", "tools", + "wizard", "video_editor", "video_3d", "character_kit", "system", "unknown", +}) +STATUSES = frozenset({ + "planned", "queued", "running", "completed", "failed", "cancelled", +}) +TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled"}) +LEGAL_TRANSITIONS: dict[str, frozenset[str]] = { + "planned": frozenset({"queued", "cancelled"}), + "queued": frozenset({"running", "cancelled"}), + "running": frozenset({"queued", "completed", "failed", "cancelled"}), + "completed": frozenset(), + "failed": frozenset(), + "cancelled": frozenset(), +} +_PRODUCT_ALIASES = { + "studio-image": "studio", + "studio-video": "studio", + "studio-audio": "studio", + "story-lab": "story_lab", + "story-music-video": "story_lab", + "series-lab": "series_lab", + "comics": "comic", + "video-editor": "video_editor", + "scene-animator-3d": "video_3d", + "hunyuan3d": "video_3d", + "3d": "video_3d", + "character-kit": "character_kit", + "filesystem-import": "system", + "legacy": "unknown", + "upscale": "tools", + "revoice": "tools", + "remove_background": "tools", +} +_PRODUCT_FROM_CAPABILITY = { + "generate_story_song": "story_lab", + "start_director_production": "director", + "upscale": "tools", + "revoice": "tools", + "remove_background": "tools", +} +_LANGUAGE_KEYS = ( + "conversation_language", "content_language", + "spoken_language", "technical_prompt_language", +) +_CORRELATION_KEYS = ( + "command_id", "workflow_id", "run_id", "task_id", + "job_id", "pipeline_id", +) +_CANONICAL_FIELDS = ( + "schema", "schema_version", "generation_id", "asset_id", "product", + "workspace_id", "output_folder", "project_id", "production_id", "cue_id", + "candidate_id", "song_version", "prompt_full", "prompt_display", "model", + "languages", "timestamps", "status", "lineage", "error", "retry_count", + "cancellation", "location", "links", "result", "provenance", "correlations", +) + + +class GenerationRecordError(ValueError): + """The generation record cannot be normalized without losing its contract.""" + + +class GenerationModel(TypedDict, total=False): + provider: str | None + id: str | None + version: str | None + configuration: dict[str, Any] + + +class GenerationLanguages(TypedDict, total=False): + conversation_language: str | None + content_language: str | None + spoken_language: str | None + technical_prompt_language: str | None + + +class GenerationTimestamps(TypedDict, total=False): + created_at: str | None + queued_at: str | None + started_at: str | None + completed_at: str | None + duration_ms: int | None + + +class GenerationLineageRef(TypedDict, total=False): + generation_id: str | None + asset_id: str | None + kind: str | None + uri: str | None + + +class GenerationCancellation(TypedDict, total=False): + requested: bool + at: str | None + reason: str | None + + +class GenerationLocation(TypedDict, total=False): + filename: str | None + uri: str | None + sidecar: str | None + + +class GenerationLinks(TypedDict, total=False): + activity_id: str | None + catalog_id: str | None + ui_href: str | None + + +class GenerationRecord(TypedDict, total=False): + schema: str + schema_version: int + generation_id: str + asset_id: str + product: str + workspace_id: str + output_folder: str + project_id: str | None + production_id: str | None + cue_id: str | None + candidate_id: str | None + song_version: str | None + prompt_full: str + prompt_display: str + model: GenerationModel + languages: GenerationLanguages + timestamps: GenerationTimestamps + status: str + lineage: dict[str, list[GenerationLineageRef]] + error: dict[str, Any] | None + retry_count: int + cancellation: GenerationCancellation + location: GenerationLocation + links: GenerationLinks + result: dict[str, Any] + provenance: dict[str, Any] + correlations: dict[str, Any] + + +def _clean(value: Any) -> str | None: + text = str(value or "").strip() + return text or None + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _json_copy(value: Mapping[str, Any]) -> dict[str, Any]: + return json.loads(json.dumps(value, ensure_ascii=False)) + + +def _is_host_path(value: str | None) -> bool: + text = str(value or "").strip() + if not text: + return False + if os.path.isabs(text) or text.startswith(("/", "\\")): + return True + return len(text) >= 3 and text[1] == ":" and text[2] in "/\\" + + +def _is_token(value: str | None) -> bool: + text = str(value or "").strip() + if not text or text in {".", ".."} or _is_host_path(text): + return False + return os.path.basename(text) == text and "/" not in text and "\\" not in text + + +def _identity_token(value: Any, field: str, *, required: bool) -> str | None: + text = _clean(value) + if not text: + if required: + raise GenerationRecordError(f"{field} is required") + return None + if not _is_token(text) or len(text) > 240: + raise GenerationRecordError(f"{field} must be a stable ID, never a path") + return text + + +def _portable_filename(value: Any) -> str | None: + text = _clean(value) + if not text: + return None + name = os.path.basename(text.replace("\\", "/")) + if not name or name in {".", ".."} or _is_host_path(name): + raise GenerationRecordError("location values must be relative filenames") + return name + + +def _count(value: Any, default: int = 0) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return default + return value + + +def prompt_display_text(value: Any, limit: int = PROMPT_DISPLAY_MAX) -> str: + """Truncated, secret-free prompt for lists and inspectors.""" + text = str(value or "").strip() + if len(text) <= limit: + return text + if limit <= 1: + return "…" + return text[: limit - 1].rstrip() + "…" + + +def map_product(value: Any, *, capability: str | None = None) -> str: + token = (_clean(value) or "").casefold().replace(" ", "_") + if token in PRODUCTS: + return token + if token in _PRODUCT_ALIASES: + return _PRODUCT_ALIASES[token] + mapped = _PRODUCT_FROM_CAPABILITY.get(_clean(capability) or "") + return mapped if mapped in PRODUCTS else "unknown" + + +def map_manifest_status(status: Any, *, has_filename: bool = False) -> tuple[str, dict[str, Any], dict[str, Any] | None]: + """Project asset-manifest execution.status onto the six public values.""" + raw = (_clean(status) or "").casefold() + result: dict[str, Any] = {"kind": None} + if raw == "prepared": + return "planned", result, None + if raw == "partial": + if has_filename: + return "completed", {"kind": "partial"}, None + return "failed", result, { + "code": "partial", + "message": "Generation finished without a complete artifact", + } + if raw in STATUSES: + result["kind"] = raw if raw in {"completed", "failed", "cancelled"} else None + return raw, result, None + if not raw: + return "planned", result, None + return "failed", result, {"code": "invalid_status", "message": f"Unsupported status {raw!r}"} + + +def map_record_status_to_manifest(status: str, *, result_kind: str | None = None) -> str: + if status == "planned": + return "prepared" + if status == "completed" and result_kind == "partial": + # Public enum stays completed; callers may inspect result.kind. + return "completed" + if status in STATUSES: + return status + raise GenerationRecordError(f"Unsupported generation status: {status}") + + +def is_legal_transition(current: str, target: str) -> bool: + return target in LEGAL_TRANSITIONS.get(current, frozenset()) + + +def belongs_to_workspace(record: Mapping[str, Any], workspace_id: str) -> bool: + wanted = _clean(workspace_id) + return bool(wanted) and _clean(record.get("workspace_id")) == wanted + + +def _mapping(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, Mapping) else {} + + +def _coalesce(*values: Any) -> Any: + for value in values: + if value not in (None, ""): + return value + return None + + +def _sidecar_for(filename: str | None) -> str | None: + return sidecar_path(filename).name if filename else None + + +def _model_block(value: Any) -> dict[str, Any]: + raw = _mapping(value) + configuration = raw.get("configuration") + if not isinstance(configuration, Mapping): + configuration = raw.get("parameters") if isinstance(raw.get("parameters"), Mapping) else {} + return { + "provider": _clean(raw.get("provider")), + "id": _clean(raw.get("id")), + "version": _clean(raw.get("version") or raw.get("revision")), + "configuration": dict(configuration), + } + + +def _languages_block(value: Any) -> dict[str, Any]: + raw = _mapping(value) + return {key: _clean(raw.get(key)) for key in _LANGUAGE_KEYS} + + +def _timestamps_block(value: Any, *, created_fallback: str) -> dict[str, Any]: + raw = _mapping(value) + created_at = _iso(raw.get("created_at")) or created_fallback + queued_at = _iso(raw.get("queued_at")) + started_at = _iso(raw.get("started_at")) + completed_at = _iso(raw.get("completed_at")) + duration = raw.get("duration_ms") + if not isinstance(duration, int) or isinstance(duration, bool) or duration < 0: + duration = _milliseconds(started_at or created_at, completed_at) + return { + "created_at": created_at, + "queued_at": queued_at, + "started_at": started_at, + "completed_at": completed_at, + "duration_ms": duration, + } + + +def _require_lineage_token(value: str | None, field: str) -> str | None: + if value is None: + return None + if not _is_token(value): + raise GenerationRecordError(f"lineage {field} must be a stable ID") + return value + + +def _lineage_ref(value: Any) -> dict[str, Any] | None: + raw = _mapping(value) + generation_id = _require_lineage_token(_clean(raw.get("generation_id")), "generation_id") + asset_id = _require_lineage_token(_clean(_coalesce(raw.get("asset_id"), raw.get("id"))), "asset_id") + if generation_id is None and asset_id is None: + return None + item: dict[str, Any] = {} + for key, token in (("generation_id", generation_id), ("asset_id", asset_id), + ("kind", _clean(_coalesce(raw.get("kind"), raw.get("role")))), + ("uri", _portable_filename(raw.get("uri")))): + if token: + item[key] = token + return item + + +def _lineage_list(values: Any) -> list[dict[str, Any]]: + if not isinstance(values, Sequence) or isinstance(values, (str, bytes, bytearray)): + return [] + result: list[dict[str, Any]] = [] + seen: set[tuple[str | None, str | None]] = set() + for value in values: + item = _lineage_ref(value) + if item is None: + continue + key = (item.get("generation_id"), item.get("asset_id")) + if key in seen: + continue + seen.add(key) + result.append(item) + return result + + +def _error_block(value: Any) -> dict[str, Any] | None: + raw = _mapping(value) + if not raw: + return None + code = _clean(raw.get("code")) + message = _clean(raw.get("message")) + details = raw.get("details") if isinstance(raw.get("details"), Mapping) else None + if not code and not message and not details: + return None + error: dict[str, Any] = {} + if code: + error["code"] = code[:120] + if message: + error["message"] = message[:2000] + if details: + error["details"] = dict(details) + return error + + +def _cancellation_block(value: Any) -> dict[str, Any]: + raw = _mapping(value) + return { + "requested": bool(raw.get("requested")), + "at": _iso(raw.get("at")), + "reason": _clean(raw.get("reason")), + } + + +def _location_block(value: Any) -> dict[str, Any]: + raw = _mapping(value) + filename = _portable_filename(raw.get("filename") or raw.get("uri")) + uri = _portable_filename(raw.get("uri")) or filename + sidecar = _portable_filename(raw.get("sidecar")) + if filename and sidecar is None: + sidecar = sidecar_path(filename).name + return {"filename": filename, "uri": uri, "sidecar": sidecar} + + +def _links_block(value: Any) -> dict[str, Any]: + raw = _mapping(value) + href = _clean(raw.get("ui_href")) + if href and _is_host_path(href) and not href.startswith(("/", "#")): + # Host filesystem paths are not portable. App routes such as /activity/x stay. + if "://" not in href: + raise GenerationRecordError("links.ui_href must not be a host filesystem path") + return { + "activity_id": _clean(raw.get("activity_id")), + "catalog_id": _clean(raw.get("catalog_id")), + "ui_href": href, + } + + +def _result_block(value: Any) -> dict[str, Any]: + raw = _mapping(value) + return {"kind": _clean(raw.get("kind"))} + + +def _provenance_block(value: Any, *, actor: str | None, capability: str | None) -> dict[str, Any]: + raw = _mapping(value) + resolved_actor = _clean(raw.get("actor") or actor) or "unknown" + if resolved_actor not in {"user", "wizard", "system", "unknown"}: + resolved_actor = "unknown" + return { + "actor": resolved_actor, + "capability": _clean(raw.get("capability") or capability), + } + + +def _correlations_block(value: Any) -> dict[str, Any]: + raw = _mapping(value) + return {key: _clean(raw.get(key)) for key in _CORRELATION_KEYS} + + +def build_generation_record( + *, + generation_id: str | None = None, + asset_id: str | None = None, + product: str | None = None, + workspace_id: str | None = None, + output_folder: str | None = None, + project_id: str | None = None, + production_id: str | None = None, + cue_id: str | None = None, + candidate_id: str | None = None, + song_version: str | None = None, + prompt_full: str | None = None, + model: Mapping[str, Any] | None = None, + languages: Mapping[str, Any] | None = None, + timestamps: Mapping[str, Any] | None = None, + status: str = "planned", + parents: Sequence[Mapping[str, Any]] | None = None, + derivatives: Sequence[Mapping[str, Any]] | None = None, + error: Mapping[str, Any] | None = None, + retry_count: int = 0, + cancellation: Mapping[str, Any] | None = None, + location: Mapping[str, Any] | None = None, + links: Mapping[str, Any] | None = None, + result: Mapping[str, Any] | None = None, + provenance: Mapping[str, Any] | None = None, + correlations: Mapping[str, Any] | None = None, + actor: str | None = None, + capability: str | None = None, + mint_ids: bool = True, +) -> dict[str, Any]: + """Build one JSON-safe generation attempt without host filesystem paths.""" + resolved_status = (_clean(status) or "planned").casefold() + if resolved_status not in STATUSES: + raise GenerationRecordError(f"Unsupported generation status: {resolved_status}") + resolved_generation_id = _identity_token( + generation_id, "generation_id", required=not mint_ids, + ) or f"gen_{uuid.uuid4().hex}" + resolved_asset_id = _identity_token( + asset_id, "asset_id", required=not mint_ids, + ) or f"asset_{uuid.uuid4().hex}" + location_ids = resolve_generation_location( + workspace_id=workspace_id, output_folder=output_folder, + ) + collection = _identity_token( + location_ids.get("workspace_id"), "workspace_id", required=True, + ) + folder = _portable_filename(location_ids.get("output_folder")) or collection + prompt = str(prompt_full or "") + record = { + "schema": SCHEMA_NAME, + "schema_version": SCHEMA_VERSION, + "generation_id": resolved_generation_id, + "asset_id": resolved_asset_id, + "product": map_product(product, capability=capability), + "workspace_id": collection, + "output_folder": folder, + "project_id": _clean(project_id), + "production_id": _clean(production_id), + "cue_id": _clean(cue_id), + "candidate_id": _clean(candidate_id), + "song_version": _clean(song_version), + "prompt_full": prompt, + "prompt_display": prompt_display_text(prompt), + "model": _model_block(model), + "languages": _languages_block(languages), + "timestamps": _timestamps_block(timestamps, created_fallback=_now_iso()), + "status": resolved_status, + "lineage": { + "parents": _lineage_list(parents), + "derivatives": _lineage_list(derivatives), + }, + "error": _error_block(error), + "retry_count": _count(retry_count), + "cancellation": _cancellation_block(cancellation), + "location": _location_block(location), + "links": _links_block(links), + "result": _result_block(result), + "provenance": _provenance_block(provenance, actor=actor, capability=capability), + "correlations": _correlations_block(correlations), + } + if not record["links"].get("catalog_id"): + record["links"]["catalog_id"] = resolved_asset_id + return _redact(record) + + +def _require_record_object(value: Mapping[str, Any]) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise GenerationRecordError("Generation record must be an object") + if value.get("schema") != SCHEMA_NAME: + raise GenerationRecordError("Unsupported generation record schema") + if value.get("schema_version") != SCHEMA_VERSION: + raise GenerationRecordError("Unsupported generation record schema") + return value + + +def validate_generation_record(value: Mapping[str, Any]) -> dict[str, Any]: + payload = _require_record_object(value) + lineage = _mapping(payload.get("lineage")) + normalized = build_generation_record( + generation_id=payload.get("generation_id"), + asset_id=payload.get("asset_id"), + product=payload.get("product"), + workspace_id=payload.get("workspace_id"), + output_folder=payload.get("output_folder"), + project_id=payload.get("project_id"), + production_id=payload.get("production_id"), + cue_id=payload.get("cue_id"), + candidate_id=payload.get("candidate_id"), + song_version=payload.get("song_version"), + prompt_full=payload.get("prompt_full"), + model=_mapping(payload.get("model")), + languages=_mapping(payload.get("languages")), + timestamps=_mapping(payload.get("timestamps")), + status=str(payload.get("status") or "planned"), + parents=lineage.get("parents"), + derivatives=lineage.get("derivatives"), + error=_mapping(payload.get("error")) or None, + retry_count=_count(payload.get("retry_count")), + cancellation=_mapping(payload.get("cancellation")), + location=_mapping(payload.get("location")), + links=_mapping(payload.get("links")), + result=_mapping(payload.get("result")), + provenance=_mapping(payload.get("provenance")), + correlations=_mapping(payload.get("correlations")), + mint_ids=False, + ) + if _is_host_path(normalized["workspace_id"]): + raise GenerationRecordError("workspace_id and output_folder must never be host paths") + if _is_host_path(normalized["output_folder"]): + raise GenerationRecordError("workspace_id and output_folder must never be host paths") + return {key: normalized[key] for key in _CANONICAL_FIELDS} + + +def _languages_from_manifest(generation: Mapping[str, Any]) -> dict[str, Any]: + prompts = _mapping(generation.get("prompts")) + languages = _mapping(generation.get("languages")) + intent = _mapping(prompts.get("language_intent")) + merged = {**intent, **languages} + content = _clean(_coalesce(merged.get("content_language"), prompts.get("language"))) + if content: + merged["content_language"] = content + return merged + + +def _prompt_from_manifest(generation: Mapping[str, Any]) -> str: + prompts = _mapping(generation.get("prompts")) + for key in ("effective", "original", "audio", "instruction"): + text = _clean(prompts.get(key)) + if text: + return text + return "" + + +def _generation_id_from_manifest( + technical: Mapping[str, Any], + execution: Mapping[str, Any], + asset_id: str, +) -> str: + return _coalesce( + _clean(technical.get("generation_id")), + _clean(execution.get("job_id")), + f"gen_{asset_id}", + ) + + +def _manifest_parts(manifest: Mapping[str, Any]) -> dict[str, Any]: + value = _mapping(manifest) + generation = _mapping(value.get("generation")) + origin = _mapping(value.get("origin")) + return { + "asset": _mapping(value.get("asset")), + "origin": origin, + "execution": _mapping(value.get("execution")), + "generation": generation, + "timing": _mapping(value.get("timing")), + "lineage": _mapping(value.get("lineage")), + "technical": _mapping(value.get("technical")), + "model": _mapping(generation.get("model")), + "parameters": _mapping(generation.get("parameters")), + "project": _mapping(origin.get("project")), + "production": _mapping(origin.get("production")), + "provenance": provenance_from_manifest(value), + } + + +def _manifest_error(execution: Mapping[str, Any], mapped_error: dict[str, Any] | None) -> dict[str, Any] | None: + error = execution.get("error") + if isinstance(error, Mapping): + return dict(error) + return mapped_error + + +def _project_kwargs(parts: Mapping[str, Any]) -> dict[str, Any]: + origin = parts["origin"] + execution = parts["execution"] + provenance = parts["provenance"] + technical = parts["technical"] + asset = parts["asset"] + filename = _portable_filename(_coalesce(asset.get("filename"), asset.get("uri"))) + status, result, mapped_error = map_manifest_status( + execution.get("status"), has_filename=bool(filename), + ) + asset_id = _identity_token(asset.get("id"), "asset_id", required=False) or f"asset_{uuid.uuid4().hex}" + model = dict(parts["model"]) + model["configuration"] = parts["parameters"] + model["version"] = _coalesce(model.get("revision"), model.get("version")) + timing = parts["timing"] + return { + "generation_id": _generation_id_from_manifest(technical, execution, asset_id), + "asset_id": asset_id, + "product": map_product( + _coalesce(origin.get("tool"), provenance.get("tool")), + capability=origin.get("capability"), + ), + "workspace_id": _coalesce(origin.get("workspace_id"), provenance.get("workspace_id")), + "output_folder": _coalesce(origin.get("output_folder"), provenance.get("output_folder")), + "project_id": _coalesce(parts["project"].get("id"), provenance.get("project_id")), + "production_id": _coalesce(parts["production"].get("id"), provenance.get("production_id")), + "cue_id": _coalesce(execution.get("cue_id"), provenance.get("cue_id")), + "candidate_id": _coalesce(execution.get("candidate_id"), provenance.get("candidate_id")), + "song_version": _coalesce(execution.get("song_version"), provenance.get("song_version")), + "prompt_full": _prompt_from_manifest(parts["generation"]), + "model": model, + "languages": _languages_from_manifest(parts["generation"]), + "timestamps": { + "created_at": timing.get("created_at"), + "queued_at": timing.get("queued_at"), + "started_at": timing.get("started_at"), + "completed_at": timing.get("completed_at"), + "duration_ms": _coalesce(timing.get("total_ms"), timing.get("inference_ms")), + }, + "status": status, + "parents": _lineage_list(parts["lineage"].get("parents")), + "error": _manifest_error(execution, mapped_error), + "location": {"filename": filename, "uri": filename, "sidecar": _sidecar_for(filename)}, + "links": { + "catalog_id": asset_id, + "activity_id": _coalesce(technical.get("activity_id"), execution.get("task_id")), + }, + "result": result, + "provenance": {"actor": provenance.get("actor"), "capability": provenance.get("capability")}, + "correlations": {key: execution.get(key) for key in _CORRELATION_KEYS}, + "actor": provenance.get("actor"), + "capability": provenance.get("capability"), + "mint_ids": False, + } + + +def project_from_asset_manifest(manifest: Mapping[str, Any]) -> dict[str, Any]: + """Project a canonical asset-manifest onto GenerationRecord v1.""" + return build_generation_record(**_project_kwargs(_manifest_parts(manifest))) + + +def _artifact_parent(item: Mapping[str, Any]) -> dict[str, Any] | None: + asset_id = item.get("asset_id") + if not asset_id: + return None + parent = {"id": asset_id, "kind": item.get("kind") or "other"} + if item.get("uri"): + parent["uri"] = item["uri"] + return parent + + +def _entity_patch(kind: str, identifier: str | None) -> dict[str, str] | None: + if not identifier: + return None + return {"kind": kind, "id": identifier} + + +def to_asset_manifest_patch(record: Mapping[str, Any]) -> dict[str, Any]: + """Return the asset-manifest fields implied by a generation record.""" + value = validate_generation_record(record) + location = value["location"] + model = value["model"] + timestamps = value["timestamps"] + provenance = value["provenance"] + filename = location.get("filename") + parents = [item for item in (_artifact_parent(parent) for parent in value["lineage"]["parents"]) if item] + result_kind = _mapping(value.get("result")).get("kind") + return { + "asset": { + "id": value["asset_id"], + "filename": filename, + "uri": _coalesce(location.get("uri"), filename), + }, + "origin": { + "tool": value["product"], + "capability": provenance.get("capability"), + "actor": _coalesce(provenance.get("actor"), "unknown"), + "workspace_id": value["workspace_id"], + "output_folder": value["output_folder"], + "project": _entity_patch("project", value.get("project_id")), + "production": _entity_patch("production", value.get("production_id")), + }, + "execution": { + "status": map_record_status_to_manifest(value["status"], result_kind=result_kind), + "error": value.get("error"), + "cue_id": value.get("cue_id"), + "candidate_id": value.get("candidate_id"), + "song_version": value.get("song_version"), + **value["correlations"], + }, + "generation": { + "prompts": { + "original": value["prompt_full"], + "effective": value["prompt_full"], + "language": value["languages"].get("content_language"), + }, + "model": { + "provider": model.get("provider"), + "id": model.get("id"), + "revision": model.get("version"), + }, + "parameters": _mapping(model.get("configuration")), + "inputs": parents, + }, + "timing": { + "created_at": timestamps.get("created_at"), + "queued_at": timestamps.get("queued_at"), + "started_at": timestamps.get("started_at"), + "completed_at": timestamps.get("completed_at"), + "total_ms": timestamps.get("duration_ms"), + }, + "lineage": {"parents": parents, "transformations": []}, + "technical": { + "generation_id": value["generation_id"], + "result": value.get("result"), + }, + } + + +def _stamp_transition(record: dict[str, Any], target: str, at: str) -> None: + times = dict(record["timestamps"]) + if target == "queued": + times["queued_at"] = times.get("queued_at") or at + elif target == "running": + times["started_at"] = times.get("started_at") or at + elif target in TERMINAL_STATUSES: + times["completed_at"] = times.get("completed_at") or at + times["duration_ms"] = _milliseconds( + times.get("started_at") or times.get("created_at"), + times["completed_at"], + ) + record["timestamps"] = times + + +def transition_status( + record: Mapping[str, Any], + target: str, + *, + at: Any = None, + error: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Apply one legal lifecycle transition. Cancellation already requested wins.""" + current = validate_generation_record(record) + resolved = (_clean(target) or "").casefold() + if current["cancellation"]["requested"] and resolved != "cancelled": + return apply_cancel(current, reason=current["cancellation"].get("reason"), at=at) + if not is_legal_transition(current["status"], resolved): + raise GenerationRecordError( + f"Illegal generation transition {current['status']!r} -> {resolved!r}", + ) + current["status"] = resolved + if resolved == "failed": + current["error"] = _error_block(error) or current.get("error") + elif resolved == "completed": + current["error"] = None + _stamp_transition(current, resolved, _iso(at) or _now_iso()) + return validate_generation_record(current) + + +def request_cancel( + record: Mapping[str, Any], + *, + reason: str | None = None, + at: Any = None, +) -> dict[str, Any]: + """Mark cancellation. Planned/queued settle immediately; running waits for apply.""" + current = validate_generation_record(record) + if current["status"] in TERMINAL_STATUSES: + return current + stamp = _iso(at) or _now_iso() + current["cancellation"] = { + "requested": True, + "at": stamp, + "reason": _clean(reason) or current["cancellation"].get("reason"), + } + if current["status"] in {"planned", "queued"}: + return apply_cancel(current, reason=reason, at=stamp) + return validate_generation_record(current) + + +def apply_cancel( + record: Mapping[str, Any], + *, + reason: str | None = None, + at: Any = None, +) -> dict[str, Any]: + """Settle a cancellation request, matching job-lifecycle acknowledgement.""" + current = validate_generation_record(record) + if current["status"] == "cancelled": + return current + if current["status"] in TERMINAL_STATUSES: + raise GenerationRecordError("Cannot cancel a finished generation") + stamp = _iso(at) or current["cancellation"].get("at") or _now_iso() + current["cancellation"] = { + "requested": True, + "at": stamp, + "reason": _clean(reason) or current["cancellation"].get("reason"), + } + current["status"] = "cancelled" + current["error"] = None + _stamp_transition(current, "cancelled", stamp) + return validate_generation_record(current) + + +def retry_generation( + record: Mapping[str, Any], + *, + same_artifact: bool = False, +) -> dict[str, Any]: + """Start a new attempt (policy b) linked to the parent generation_id.""" + parent = validate_generation_record(record) + child = build_generation_record( + asset_id=parent["asset_id"] if same_artifact else None, + product=parent["product"], + workspace_id=parent["workspace_id"], + output_folder=parent["output_folder"], + project_id=parent.get("project_id"), + production_id=parent.get("production_id"), + cue_id=parent.get("cue_id"), + candidate_id=parent.get("candidate_id"), + song_version=parent.get("song_version"), + prompt_full=parent.get("prompt_full"), + model=parent.get("model"), + languages=parent.get("languages"), + status="planned", + parents=[{ + "generation_id": parent["generation_id"], + "asset_id": parent["asset_id"], + "kind": "attempt", + }], + retry_count=parent["retry_count"] + 1, + location=parent.get("location") if same_artifact else None, + links={"catalog_id": parent["asset_id"] if same_artifact else None}, + provenance=parent.get("provenance"), + correlations=parent.get("correlations"), + actor=(parent.get("provenance") or {}).get("actor"), + capability=(parent.get("provenance") or {}).get("capability"), + ) + return child + + +def attach_derivative(parent: Mapping[str, Any], child: Mapping[str, Any]) -> dict[str, Any]: + """Return the parent with the child linked in derivatives[]. IDs stay put.""" + current = validate_generation_record(parent) + attempt = validate_generation_record(child) + current["lineage"] = { + "parents": current["lineage"]["parents"], + "derivatives": _lineage_list([ + *current["lineage"]["derivatives"], + { + "generation_id": attempt["generation_id"], + "asset_id": attempt["asset_id"], + "kind": "attempt", + }, + ]), + } + return validate_generation_record(current) + + +def resume_generation_record(record: Mapping[str, Any]) -> dict[str, Any]: + """Continue from the last durable status after a process restart. + + Queued and running records stay queued/running. Success is never inferred. + """ + current = validate_generation_record(record) + if current["status"] in {"queued", "running", "planned"}: + return current + return current + + +def _existing_identity(path: Path) -> tuple[str | None, str | None, str | None]: + if not path.is_file(): + return None, None, None + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None, None, None + if not isinstance(value, Mapping): + return None, None, None + return ( + _clean(value.get("generation_id")), + _clean(value.get("asset_id")), + _clean(value.get("workspace_id")), + ) + + +def persist_generation_record(path: str | os.PathLike[str], record: Mapping[str, Any]) -> Path: + """Atomically replace one generation-record JSON file.""" + normalized = validate_generation_record(record) + target = Path(path) + existing_generation_id, existing_asset_id, existing_workspace_id = _existing_identity(target) + if existing_generation_id and existing_generation_id != normalized["generation_id"]: + raise GenerationRecordError( + f"Refusing to replace generation identity {existing_generation_id!r}", + ) + if existing_asset_id and existing_asset_id != normalized["asset_id"]: + raise GenerationRecordError( + f"Refusing to replace asset identity {existing_asset_id!r}", + ) + if existing_workspace_id and existing_workspace_id != normalized["workspace_id"]: + raise GenerationRecordError("cross-workspace adoption is not allowed") + payload = _json_copy(normalized) + target.parent.mkdir(parents=True, exist_ok=True) + temporary = target.parent / f".{target.name}.{uuid.uuid4().hex}.tmp" + try: + with open(temporary, "w", encoding="utf-8") as handle: + json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, target) + if hasattr(os, "O_DIRECTORY"): + directory_fd = os.open(target.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + except Exception: + try: + temporary.unlink(missing_ok=True) + except OSError: + pass + raise + return target + + +def load_generation_record( + path: str | os.PathLike[str], + *, + workspace_id: str, +) -> dict[str, Any]: + """Load one record and refuse to adopt it into a different workspace.""" + target = Path(path) + try: + value = json.loads(target.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise GenerationRecordError("Generation record is unreadable") from exc + record = validate_generation_record(value) + if not belongs_to_workspace(record, workspace_id): + raise GenerationRecordError("cross-workspace adoption is not allowed") + return record + + +class GenerationRecordStore: + """Workspace-scoped JSON files with atomic replacement.""" + + def __init__(self, root: str | os.PathLike[str]): + self.root = Path(root) + self._lock = threading.RLock() + + def _path(self, workspace_id: str, generation_id: str) -> Path: + collection = _identity_token(workspace_id, "workspace_id", required=True) + identifier = _identity_token(generation_id, "generation_id", required=True) + folder = (self.root / collection).resolve() + root = self.root.resolve() + if folder != root and root not in folder.parents: + raise GenerationRecordError("workspace path escapes the store") + return folder / f"{identifier}.json" + + def persist(self, record: Mapping[str, Any]) -> Path: + normalized = validate_generation_record(record) + with self._lock: + return persist_generation_record( + self._path(normalized["workspace_id"], normalized["generation_id"]), + normalized, + ) + + def load(self, generation_id: str, *, workspace_id: str) -> dict[str, Any]: + with self._lock: + return load_generation_record( + self._path(workspace_id, generation_id), workspace_id=workspace_id, + ) + + def list(self, *, workspace_id: str) -> list[dict[str, Any]]: + collection = _identity_token(workspace_id, "workspace_id", required=True) + folder = self.root / collection + records: list[dict[str, Any]] = [] + if not folder.is_dir(): + return records + with self._lock: + for path in sorted(folder.glob("*.json")): + try: + record = load_generation_record(path, workspace_id=collection) + except GenerationRecordError: + continue + records.append(record) + records.sort(key=lambda item: str((item.get("timestamps") or {}).get("created_at") or "")) + return records + + def resume(self, generation_id: str, *, workspace_id: str) -> dict[str, Any]: + return resume_generation_record(self.load(generation_id, workspace_id=workspace_id)) + + +__all__ = [ + "ATTEMPT_IDENTITY_POLICY", "LEGAL_TRANSITIONS", "PRODUCTS", "PROMPT_DISPLAY_MAX", + "SCHEMA_NAME", "SCHEMA_VERSION", "STATUSES", "TERMINAL_STATUSES", + "GenerationRecord", "GenerationRecordError", "GenerationRecordStore", + "apply_cancel", "attach_derivative", "belongs_to_workspace", + "build_generation_record", "is_legal_transition", "load_generation_record", + "map_manifest_status", "map_product", "map_record_status_to_manifest", + "persist_generation_record", "project_from_asset_manifest", + "prompt_display_text", "request_cancel", "resume_generation_record", + "retry_generation", "to_asset_manifest_patch", "transition_status", + "validate_generation_record", +] diff --git a/docs/development/DOMAIN_MODEL_AND_ASSET_PROVENANCE.md b/docs/development/DOMAIN_MODEL_AND_ASSET_PROVENANCE.md index 3d70689f..db693be5 100644 --- a/docs/development/DOMAIN_MODEL_AND_ASSET_PROVENANCE.md +++ b/docs/development/DOMAIN_MODEL_AND_ASSET_PROVENANCE.md @@ -21,6 +21,10 @@ It extends the existing command plane; it does not create a second Wizard API. - **Run** is one execution attempt of a production or command. Retrying creates a new run while retaining lineage to the prior attempt. - **Task** is a technical queued step belonging to a run. +- **Generation** is one attempt to produce an asset, identified by a stable + `generation_id`. A retry mints a new generation with lineage to the parent + attempt; `asset_id` is reused only when the bytes are the same artifact. + See `GENERATION_RECORD.md` and `generation-record-v1.schema.json`. The identity chain is therefore: @@ -151,6 +155,17 @@ Secrets, credentials, authorization headers and tokens are recursively redacted. Absolute local paths are not part of the portable contract. Legacy top-level keys such as `params` may coexist during migration. +## Generation record v1 + +The generation record is a portable read/write projection over asset-manifest +v1, generation provenance and the job lifecycle. It is not a second media +store: published bytes remain the sidecar, and this contract adds typed attempt +identity, status, cancellation and resume. Helpers live in +`app/services/generation_record.py`; UI types live in +`ui/src/lib/generationRecord.ts`. The schema is +`generation-record-v1.schema.json`. Details and the attempt-identity policy are +in `GENERATION_RECORD.md`. + ## UI requirement: Extra info The existing **Extra info** action becomes the human-readable inspector for diff --git a/docs/development/GENERATION_RECORD.md b/docs/development/GENERATION_RECORD.md new file mode 100644 index 00000000..bd1797c9 --- /dev/null +++ b/docs/development/GENERATION_RECORD.md @@ -0,0 +1,128 @@ +# Generation record v1 + +Status: accepted contract (2026-09-04) + +A **Generation** is one attempt to produce an asset. This document is the +portable read/write contract for that attempt. It projects existing sources; it +does not invent a second media store. + +Authoritative bytes and published provenance remain the adjacent +`.meta.json` asset-manifest v1 sidecar. The generation record adds a +typed attempt identity, lifecycle, cancellation and resume vocabulary on top of: + +- `app/services/asset_manifest.py` / `asset-manifest-v1.schema.json` +- `app/services/generation_provenance.py` +- `app/services/job_lifecycle.py` +- crash-safe JSON replacement as in `app/services/durable_generation_queue.py` + +Python helpers live in `app/services/generation_record.py`. UI types and pure +mappers live in `ui/src/lib/generationRecord.ts`. The JSON schema is +`generation-record-v1.schema.json`. + +This module does not import FastAPI, WanGP or launch. Wiring into +`_launch_runtime.py`, Activity and the Library catalog is a later sequential PR. + +## Identity + +| Field | Rule | +|---|---| +| `generation_id` | Stable attempt ID. Never recycled. | +| `asset_id` | Stable artifact ID. Never recycled. Matches `asset.id` when published. | + +Titles, prompts, filenames and display labels are never identity. Two attempts +with the same prompt are still two generations. + +**Attempt policy (b):** a retry mints a **new** `generation_id`, increments +`retry_count` on the new attempt, and links the parent `generation_id` in +`lineage.parents` (`kind: "attempt"`). The parent may list the child in +`lineage.derivatives`. `asset_id` is copied only when the bytes are the same +artifact (`same_artifact=True`). Distinct output files get a new `asset_id`. + +Resume after a process restart reloads the JSON and continues from the last +durable status. It does not mint IDs and does not invent `completed`. + +Policy (a) — mutate one `generation_id` and only bump `retry_count` — is +rejected because a retry is a new run in the domain model. + +## Product / origin + +`product` is the originating UI surface, projected from `origin.tool` (and, when +needed, a trusted capability) onto: + +`studio | story_lab | series_lab | director | comic | tools | wizard | +video_editor | video_3d | character_kit | system | unknown` + +Who started the work (`provenance.actor`) stays separate from what computed the +bytes (`model.provider` / `model.id` / `model.version`). + +## Location + +`workspace_id` is a Workspace **collection** ID. `output_folder` is the physical +folder **name**. `location.filename`, `location.uri` and `location.sidecar` are +relative filenames (`clip.mp4`, `clip.meta.json`). Host absolute paths are +rejected. + +A record persisted under workspace A cannot be loaded, listed or adopted as +workspace B. `load` / `list` always take `workspace_id` and compare it to the +document. + +## Status + +Public enum (six values): + +`planned | queued | running | completed | failed | cancelled` + +| Asset-manifest `execution.status` | Generation record | +|---|---| +| `prepared` | `planned` | +| `queued` | `queued` | +| `running` | `running` | +| `completed` | `completed` | +| `failed` | `failed` | +| `cancelled` | `cancelled` | +| `partial` | `completed` + `result.kind = "partial"` when a filename exists; otherwise `failed` with `error.code = "partial"` | + +`partial` is **not** a seventh public status. Legal transitions match the +in-process job lifecycle, including `running -> queued` for multi-phase work. +Terminal states do not transition. A requested cancellation beats a late +`completed` / `failed` write, as in `job_lifecycle.finish_job`. + +Cancellation: + +- **before running** (`planned` / `queued`): `request_cancel` settles immediately + to `cancelled`; +- **while running**: `request_cancel` sets `cancellation.requested` and keeps + `status=running` (the public form of job `cancelling`); `apply_cancel` is the + worker acknowledgement that moves the record to `cancelled`. + +## Persistence and resume + +`persist_generation_record` / `GenerationRecordStore` write one JSON document +per attempt with atomic temp-file replace and `fsync`, the same durability +pattern as `DurableGenerationQueue`. Identity fields on an existing file cannot +be replaced. + +After a simulated process restart, construct a new store, `load` / `resume` the +id, and continue from `queued` or `running`. Do not mark the attempt completed +just because the process came back. + +## Projection + +- `project_from_asset_manifest(manifest)` — read model over a canonical sidecar. + `generation_id` comes from `technical.generation_id`, else `execution.job_id`, + else `gen_{asset_id}`. +- `to_asset_manifest_patch(record)` — the asset-manifest fields implied by the + record (`planned` writes back as `prepared`). Patching does not rewrite media + bytes and does not create a parallel catalog. + +## Secrets and prompts + +`model.configuration` (and any nested parameters) recursively redact credentials, +tokens and API keys using the asset-manifest policy. `prompt_display` is at most +180 characters and is derived from `prompt_full` after that redaction. + +## Follow-up + +Launch, Activity and Library wiring is deferred: `_launch_runtime.py` is owned +by another in-flight PR. This contract is the portable layer those writers +should adopt next. diff --git a/docs/development/generation-record-v1.schema.json b/docs/development/generation-record-v1.schema.json new file mode 100644 index 00000000..371ee5ca --- /dev/null +++ b/docs/development/generation-record-v1.schema.json @@ -0,0 +1,200 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://hocuspocus.local/schemas/generation-record-v1.schema.json", + "title": "HocusPocus Generation Record", + "type": "object", + "required": [ + "schema", + "schema_version", + "generation_id", + "asset_id", + "product", + "workspace_id", + "output_folder", + "prompt_full", + "prompt_display", + "model", + "languages", + "timestamps", + "status", + "lineage", + "retry_count", + "cancellation", + "location", + "links" + ], + "properties": { + "schema": { "const": "hocuspocus.generation-record" }, + "schema_version": { "const": 1 }, + "generation_id": { + "type": "string", + "minLength": 1, + "maxLength": 240, + "description": "Stable attempt ID. Never recycled. A retry mints a new ID." + }, + "asset_id": { + "type": "string", + "minLength": 1, + "maxLength": 240, + "description": "Stable artifact ID. Matches asset-manifest asset.id when published. Reused only when the bytes are the same artifact." + }, + "product": { + "enum": [ + "studio", + "story_lab", + "series_lab", + "director", + "comic", + "tools", + "wizard", + "video_editor", + "video_3d", + "character_kit", + "system", + "unknown" + ], + "description": "Originating product surface (origin.tool projected onto a closed vocabulary)." + }, + "workspace_id": { + "type": "string", + "minLength": 1, + "maxLength": 240, + "description": "Workspace collection ID, never an absolute path." + }, + "output_folder": { + "type": "string", + "minLength": 1, + "maxLength": 240, + "description": "Physical output-folder name, never an absolute path." + }, + "project_id": { "type": ["string", "null"] }, + "production_id": { "type": ["string", "null"] }, + "cue_id": { "type": ["string", "null"] }, + "candidate_id": { "type": ["string", "null"] }, + "song_version": { "type": ["string", "null"] }, + "prompt_full": { "type": "string" }, + "prompt_display": { + "type": "string", + "maxLength": 180, + "description": "Truncated prompt for lists. Secrets are not stored here." + }, + "model": { "$ref": "#/$defs/model" }, + "languages": { "$ref": "#/$defs/languages" }, + "timestamps": { "$ref": "#/$defs/timestamps" }, + "status": { + "enum": ["planned", "queued", "running", "completed", "failed", "cancelled"], + "description": "Asset-manifest prepared maps to planned. Manifest partial is result.kind, not a seventh status." + }, + "lineage": { + "type": "object", + "required": ["parents", "derivatives"], + "properties": { + "parents": { "type": "array", "items": { "$ref": "#/$defs/lineageRef" } }, + "derivatives": { "type": "array", "items": { "$ref": "#/$defs/lineageRef" } } + }, + "additionalProperties": false + }, + "error": { "anyOf": [{ "$ref": "#/$defs/error" }, { "type": "null" }] }, + "retry_count": { "type": "integer", "minimum": 0 }, + "cancellation": { "$ref": "#/$defs/cancellation" }, + "location": { "$ref": "#/$defs/location" }, + "links": { "$ref": "#/$defs/links" }, + "result": { + "type": "object", + "properties": { + "kind": { "type": ["string", "null"] } + }, + "additionalProperties": true + }, + "provenance": { + "type": "object", + "properties": { + "actor": { "enum": ["user", "wizard", "system", "unknown"] }, + "capability": { "type": ["string", "null"] } + }, + "additionalProperties": true + }, + "correlations": { "type": "object" } + }, + "$defs": { + "model": { + "type": "object", + "properties": { + "provider": { "type": ["string", "null"] }, + "id": { "type": ["string", "null"] }, + "version": { "type": ["string", "null"] }, + "configuration": { "type": "object" } + }, + "additionalProperties": true + }, + "languages": { + "type": "object", + "properties": { + "conversation_language": { "type": ["string", "null"] }, + "content_language": { "type": ["string", "null"] }, + "spoken_language": { "type": ["string", "null"] }, + "technical_prompt_language": { "type": ["string", "null"] } + }, + "additionalProperties": false + }, + "timestamps": { + "type": "object", + "properties": { + "created_at": { "type": ["string", "null"], "format": "date-time" }, + "queued_at": { "type": ["string", "null"], "format": "date-time" }, + "started_at": { "type": ["string", "null"], "format": "date-time" }, + "completed_at": { "type": ["string", "null"], "format": "date-time" }, + "duration_ms": { "type": ["integer", "null"], "minimum": 0 } + }, + "additionalProperties": false + }, + "lineageRef": { + "type": "object", + "properties": { + "generation_id": { "type": "string", "minLength": 1 }, + "asset_id": { "type": "string", "minLength": 1 }, + "kind": { "type": "string" }, + "uri": { "type": "string" } + }, + "additionalProperties": true + }, + "error": { + "type": "object", + "properties": { + "code": { "type": "string" }, + "message": { "type": "string" }, + "details": { "type": "object" } + }, + "additionalProperties": true + }, + "cancellation": { + "type": "object", + "required": ["requested"], + "properties": { + "requested": { "type": "boolean" }, + "at": { "type": ["string", "null"], "format": "date-time" }, + "reason": { "type": ["string", "null"] } + }, + "additionalProperties": false + }, + "location": { + "type": "object", + "properties": { + "filename": { "type": ["string", "null"] }, + "uri": { "type": ["string", "null"], "description": "Canonical relative filename, never a host absolute path." }, + "sidecar": { "type": ["string", "null"], "description": "Relative metadata JSON filename such as clip.meta.json." } + }, + "additionalProperties": false + }, + "links": { + "type": "object", + "properties": { + "activity_id": { "type": ["string", "null"] }, + "catalog_id": { "type": ["string", "null"], "description": "Library / asset catalog id." }, + "ui_href": { "type": ["string", "null"] } + }, + "additionalProperties": true + } + }, + "additionalProperties": false +} diff --git a/tests/test_generation_record.py b/tests/test_generation_record.py new file mode 100644 index 00000000..f2ec0912 --- /dev/null +++ b/tests/test_generation_record.py @@ -0,0 +1,320 @@ +from __future__ import annotations + +import ast +import json +from pathlib import Path + +import pytest + +from app.services.asset_manifest import build_asset_manifest +from app.services.generation_record import ( + ATTEMPT_IDENTITY_POLICY, + PRODUCTS, + PROMPT_DISPLAY_MAX, + SCHEMA_NAME, + SCHEMA_VERSION, + STATUSES, + GenerationRecordError, + GenerationRecordStore, + apply_cancel, + attach_derivative, + belongs_to_workspace, + build_generation_record, + load_generation_record, + map_manifest_status, + persist_generation_record, + project_from_asset_manifest, + prompt_display_text, + request_cancel, + resume_generation_record, + retry_generation, + to_asset_manifest_patch, + transition_status, + validate_generation_record, +) + + +ROOT = Path(__file__).resolve().parents[1] +SCHEMA_PATH = ROOT / "docs" / "development" / "generation-record-v1.schema.json" +MODULE_PATH = ROOT / "app" / "services" / "generation_record.py" + + +def _record(**overrides): + payload = dict( + workspace_id="collection-a", + output_folder="night-shift", + product="studio", + prompt_full="A sysadmin choir in the server room", + model={"provider": "local", "id": "minimax-h3", "version": "1", "configuration": {"seed": 7}}, + location={"filename": "choir.mp4"}, + ) + payload.update(overrides) + return build_generation_record(**payload) + + +def test_contract_schema_and_required_fields(): + schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + record = _record( + generation_id="gen_fixed", + asset_id="asset_fixed", + project_id="story-1", + production_id="production-1", + cue_id="cue-1", + candidate_id="candidate-1", + song_version="2", + languages={ + "conversation_language": "es", + "content_language": "es", + "spoken_language": "es", + "technical_prompt_language": "en", + }, + ) + assert record["schema"] == SCHEMA_NAME == schema["properties"]["schema"]["const"] + assert record["schema_version"] == SCHEMA_VERSION == schema["properties"]["schema_version"]["const"] + assert set(schema["required"]) <= set(record) + assert record["product"] in PRODUCTS + assert record["status"] in STATUSES + assert "title" not in record + assert record["generation_id"] == "gen_fixed" + assert record["asset_id"] == "asset_fixed" + assert record["workspace_id"] == "collection-a" + assert record["output_folder"] == "night-shift" + assert record["location"]["filename"] == "choir.mp4" + assert record["location"]["sidecar"] == "choir.meta.json" + assert ATTEMPT_IDENTITY_POLICY == "new_generation_id" + + +def test_rejects_host_paths_and_missing_workspace(): + with pytest.raises(GenerationRecordError, match="workspace_id"): + build_generation_record(output_folder="night-shift", prompt_full="x") + with pytest.raises(GenerationRecordError, match="never a path"): + build_generation_record(workspace_id="/tmp/outputs", prompt_full="x") + record = build_generation_record( + workspace_id="collection-a", + output_folder="/tmp/outputs/night-shift", + location={"filename": "/tmp/outputs/choir.mp4", "uri": "/tmp/outputs/choir.mp4"}, + ) + encoded = json.dumps(record) + assert record["output_folder"] == "night-shift" + assert record["location"]["filename"] == "choir.mp4" + assert record["location"]["uri"] == "choir.mp4" + assert "/tmp" not in encoded + assert "title" not in record + + +def test_prompt_display_truncation_and_secret_redaction(): + long_prompt = "α" * (PROMPT_DISPLAY_MAX + 40) + record = _record( + prompt_full=long_prompt, + model={"provider": "local", "id": "h3", "configuration": { + "api_key": "do-not-save", + "nested": {"authorization": "Bearer secret"}, + "prompt": "safe", + }}, + ) + encoded = json.dumps(record) + assert record["prompt_full"] == long_prompt + assert len(record["prompt_display"]) <= PROMPT_DISPLAY_MAX + assert record["prompt_display"].endswith("…") + assert prompt_display_text(long_prompt) == record["prompt_display"] + assert "do-not-save" not in encoded + assert "Bearer secret" not in encoded + assert record["model"]["configuration"]["prompt"] == "safe" + assert record["model"]["configuration"]["api_key"] == "[REDACTED]" + + +def test_identity_is_not_title_or_prompt(): + first = _record(prompt_full="same prompt", location={"filename": "same.mp4"}) + second = _record(prompt_full="same prompt", location={"filename": "same.mp4"}) + assert first["generation_id"] != second["generation_id"] + assert first["asset_id"] != second["asset_id"] + resumed = validate_generation_record({**first, "prompt_full": "a different prompt"}) + assert resumed["generation_id"] == first["generation_id"] + assert resumed["asset_id"] == first["asset_id"] + + +def test_persistence_round_trip_is_atomic(tmp_path: Path): + record = _record(generation_id="gen_persist", asset_id="asset_persist", status="queued") + path = tmp_path / "collection-a" / "gen_persist.json" + written = persist_generation_record(path, record) + loaded = load_generation_record(written, workspace_id="collection-a") + assert loaded["generation_id"] == "gen_persist" + assert loaded["asset_id"] == "asset_persist" + assert loaded["status"] == "queued" + assert loaded["prompt_full"] == record["prompt_full"] + assert json.loads(written.read_text(encoding="utf-8"))["schema"] == SCHEMA_NAME + assert not list(path.parent.glob("*.tmp")) + + +def test_resume_after_simulated_restart_keeps_running(tmp_path: Path): + store = GenerationRecordStore(tmp_path / "records") + record = transition_status( + transition_status(_record(generation_id="gen_live", asset_id="asset_live"), "queued"), + "running", + ) + store.persist(record) + recovered = GenerationRecordStore(tmp_path / "records").resume( + "gen_live", workspace_id="collection-a", + ) + assert recovered["status"] == "running" + assert recovered["generation_id"] == "gen_live" + assert recovered["asset_id"] == "asset_live" + assert resume_generation_record(recovered)["status"] != "completed" + + +def test_cancellation_before_and_during_running(): + planned = request_cancel(_record(status="planned"), reason="user") + assert planned["status"] == "cancelled" + assert planned["cancellation"]["requested"] is True + assert planned["cancellation"]["reason"] == "user" + + queued = request_cancel(transition_status(_record(), "queued"), reason="queue") + assert queued["status"] == "cancelled" + + running = transition_status(transition_status(_record(), "queued"), "running") + requested = request_cancel(running, reason="stop") + assert requested["status"] == "running" + assert requested["cancellation"]["requested"] is True + settled = apply_cancel(requested, reason="stop") + assert settled["status"] == "cancelled" + assert settled["cancellation"]["requested"] is True + late = transition_status(requested, "completed") + assert late["status"] == "cancelled" + requeued = transition_status(requested, "queued") + assert requeued["status"] == "cancelled" + + +def test_cross_workspace_isolation(tmp_path: Path): + store = GenerationRecordStore(tmp_path / "records") + record = _record(generation_id="gen_a", asset_id="asset_a", workspace_id="workspace-a") + path = store.persist(record) + assert belongs_to_workspace(record, "workspace-a") + assert not belongs_to_workspace(record, "workspace-b") + with pytest.raises(GenerationRecordError, match="cross-workspace"): + load_generation_record(path, workspace_id="workspace-b") + assert store.list(workspace_id="workspace-b") == [] + assert [item["generation_id"] for item in store.list(workspace_id="workspace-a")] == ["gen_a"] + cloned = tmp_path / "records" / "workspace-b" / "gen_a.json" + cloned.parent.mkdir(parents=True) + cloned.write_text(path.read_text(encoding="utf-8"), encoding="utf-8") + with pytest.raises(GenerationRecordError, match="cross-workspace"): + load_generation_record(cloned, workspace_id="workspace-b") + assert store.list(workspace_id="workspace-b") == [] + + +def test_retry_mints_new_generation_id_and_lineage(): + parent = _record(generation_id="gen_parent", asset_id="asset_parent", status="failed") + child = retry_generation(parent) + linked = attach_derivative(parent, child) + assert ATTEMPT_IDENTITY_POLICY == "new_generation_id" + assert child["generation_id"] != parent["generation_id"] + assert child["asset_id"] != parent["asset_id"] + assert child["retry_count"] == 1 + assert child["status"] == "planned" + assert child["lineage"]["parents"][0]["generation_id"] == "gen_parent" + assert linked["generation_id"] == "gen_parent" + assert linked["lineage"]["derivatives"][0]["generation_id"] == child["generation_id"] + same_bytes = retry_generation(parent, same_artifact=True) + assert same_bytes["asset_id"] == "asset_parent" + assert same_bytes["generation_id"] != "gen_parent" + + +def test_resume_and_retry_do_not_recycle_parent_ids(tmp_path: Path): + store = GenerationRecordStore(tmp_path) + parent = transition_status(transition_status(_record(generation_id="gen_r", asset_id="asset_r"), "queued"), "running") + store.persist(parent) + resumed = store.resume("gen_r", workspace_id="collection-a") + assert resumed["generation_id"] == "gen_r" + assert resumed["asset_id"] == "asset_r" + child = retry_generation(resumed) + store.persist(attach_derivative(resumed, child)) + store.persist(child) + assert store.load("gen_r", workspace_id="collection-a")["generation_id"] == "gen_r" + + +def test_illegal_transitions_are_rejected(): + completed = transition_status( + transition_status(transition_status(_record(), "queued"), "running"), + "completed", + ) + assert completed["status"] == "completed" + with pytest.raises(GenerationRecordError, match="Illegal generation transition"): + transition_status(completed, "running") + + +def test_project_from_asset_manifest_and_patch_round_trip(tmp_path: Path): + output = tmp_path / "choir.mp4" + output.write_bytes(b"video") + manifest = build_asset_manifest( + output, + asset_id="asset_video_1", + workspace_id="collection-a", + output_folder="night-shift", + project={"kind": "story", "id": "story_1"}, + production={"kind": "music_video", "id": "production_1"}, + tool="story_lab", + capability="generate_story_song", + actor="wizard", + status="prepared", + correlations={"job_id": "job-1", "cue_id": "cue-1", "candidate_id": "candidate-1", "song_version": "2"}, + prompts={"effective": "Metal fantástico", "language": "es"}, + model={"provider": "local", "id": "minimax-h3", "revision": "r1"}, + parameters={"seed": 3, "api_key": "secret"}, + parents=[{"id": "asset_song_1", "kind": "audio", "uri": "song.wav", "role": "soundtrack"}], + timing={"created_at": 1_700_000_000, "queued_at": 1_700_000_001}, + ) + record = project_from_asset_manifest(manifest) + assert record["status"] == "planned" + assert record["generation_id"] == "job-1" + assert record["asset_id"] == "asset_video_1" + assert record["product"] == "story_lab" + assert record["project_id"] == "story_1" + assert record["cue_id"] == "cue-1" + assert record["prompt_full"] == "Metal fantástico" + assert record["languages"]["content_language"] == "es" + assert record["model"]["configuration"]["api_key"] == "[REDACTED]" + assert record["lineage"]["parents"][0]["asset_id"] == "asset_song_1" + patch = to_asset_manifest_patch(record) + assert patch["execution"]["status"] == "prepared" + assert patch["asset"]["id"] == "asset_video_1" + assert patch["origin"]["workspace_id"] == "collection-a" + assert patch["technical"]["generation_id"] == "job-1" + assert "secret" not in json.dumps(patch) + + +def test_manifest_partial_maps_to_result_kind(tmp_path: Path): + output = tmp_path / "clip.mp4" + output.write_bytes(b"video") + with_file = project_from_asset_manifest(build_asset_manifest( + output, asset_id="asset_partial", workspace_id="ws", status="partial", tool="studio", + )) + assert with_file["status"] == "completed" + assert with_file["result"]["kind"] == "partial" + failed = project_from_asset_manifest({ + "schema": "hocuspocus.asset-manifest", + "schema_version": 1, + "asset": {"id": "asset_empty", "kind": "video", "filename": None}, + "origin": {"tool": "studio", "workspace_id": "ws", "output_folder": "ws"}, + "execution": {"status": "partial", "mode": "real"}, + "generation": {"prompts": {}, "model": {}, "parameters": {}, "inputs": []}, + "timing": {}, + "lineage": {"parents": [], "transformations": []}, + }) + assert failed["status"] == "failed" + assert failed["error"]["code"] == "partial" + status, _result, error = map_manifest_status("bogus") + assert status == "failed" + assert error["code"] == "invalid_status" + + +def test_module_does_not_import_runtime_engines(): + tree = ast.parse(MODULE_PATH.read_text(encoding="utf-8"), filename=str(MODULE_PATH)) + imported: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imported.update(alias.name.split(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + imported.add(node.module.split(".")[0]) + assert all("fastapi" not in name for name in imported) + assert all("wgp" not in name for name in imported) + assert all("launch" not in name for name in imported) diff --git a/ui/src/lib/generationRecord.ts b/ui/src/lib/generationRecord.ts new file mode 100644 index 00000000..e164158c --- /dev/null +++ b/ui/src/lib/generationRecord.ts @@ -0,0 +1,436 @@ +/** + * Portable GenerationRecord v1 helpers. Keep this aligned with + * app/services/generation_record.py; it must not import stores or launch. + */ + +export const GENERATION_RECORD_SCHEMA = 'hocuspocus.generation-record' as const +export const GENERATION_RECORD_SCHEMA_VERSION = 1 as const +export const PROMPT_DISPLAY_MAX = 180 +export const ATTEMPT_IDENTITY_POLICY = 'new_generation_id' as const + +export const GENERATION_PRODUCTS = [ + 'studio', + 'story_lab', + 'series_lab', + 'director', + 'comic', + 'tools', + 'wizard', + 'video_editor', + 'video_3d', + 'character_kit', + 'system', + 'unknown', +] as const + +export const GENERATION_STATUSES = [ + 'planned', + 'queued', + 'running', + 'completed', + 'failed', + 'cancelled', +] as const + +export type GenerationProduct = typeof GENERATION_PRODUCTS[number] +export type GenerationStatus = typeof GENERATION_STATUSES[number] + +export const LEGAL_GENERATION_TRANSITIONS: Record = { + planned: ['queued', 'cancelled'], + queued: ['running', 'cancelled'], + running: ['queued', 'completed', 'failed', 'cancelled'], + completed: [], + failed: [], + cancelled: [], +} + +const PRODUCT_SET = new Set(GENERATION_PRODUCTS) +const STATUS_SET = new Set(GENERATION_STATUSES) +const PRODUCT_ALIASES: Record = { + 'studio-image': 'studio', + 'studio-video': 'studio', + 'studio-audio': 'studio', + 'story-lab': 'story_lab', + 'story-music-video': 'story_lab', + 'series-lab': 'series_lab', + comics: 'comic', + 'video-editor': 'video_editor', + 'scene-animator-3d': 'video_3d', + hunyuan3d: 'video_3d', + '3d': 'video_3d', + 'character-kit': 'character_kit', + 'filesystem-import': 'system', + legacy: 'unknown', + upscale: 'tools', + revoice: 'tools', + remove_background: 'tools', +} +const PRODUCT_FROM_CAPABILITY: Record = { + generate_story_song: 'story_lab', + start_director_production: 'director', + upscale: 'tools', + revoice: 'tools', + remove_background: 'tools', +} +const SENSITIVE_KEYS = new Set([ + 'api_key', 'apikey', 'authorization', 'credential', 'credentials', + 'password', 'secret', 'token', 'access_token', 'refresh_token', + 'bearer_token', 'auth_token', 'private_key', +]) + +export interface GenerationLineageRef { + generation_id?: string + asset_id?: string + kind?: string + uri?: string +} + +export interface GenerationRecord { + schema: typeof GENERATION_RECORD_SCHEMA + schema_version: typeof GENERATION_RECORD_SCHEMA_VERSION + generation_id: string + asset_id: string + product: GenerationProduct + workspace_id: string + output_folder: string + project_id: string | null + production_id: string | null + cue_id: string | null + candidate_id: string | null + song_version: string | null + prompt_full: string + prompt_display: string + model: { + provider: string | null + id: string | null + version: string | null + configuration: Record + } + languages: { + conversation_language: string | null + content_language: string | null + spoken_language: string | null + technical_prompt_language: string | null + } + timestamps: { + created_at: string | null + queued_at: string | null + started_at: string | null + completed_at: string | null + duration_ms: number | null + } + status: GenerationStatus + lineage: { + parents: GenerationLineageRef[] + derivatives: GenerationLineageRef[] + } + error: { code?: string; message?: string; details?: Record } | null + retry_count: number + cancellation: { requested: boolean; at: string | null; reason: string | null } + location: { filename: string | null; uri: string | null; sidecar: string | null } + links: { activity_id: string | null; catalog_id: string | null; ui_href: string | null } + result: { kind: string | null } +} + +type JsonMap = Record + +function text(value: unknown): string | null { + if (typeof value !== 'string') return value == null ? null : String(value).trim() || null + return value.trim() || null +} + +function asMap(value: unknown): JsonMap { + return value && typeof value === 'object' && !Array.isArray(value) ? value as JsonMap : {} +} + +export function isHostPath(value: string | null | undefined): boolean { + const candidate = (value || '').trim() + if (!candidate) return false + if (candidate.startsWith('/') || candidate.startsWith('\\')) return true + return candidate.length >= 3 && candidate[1] === ':' && (candidate[2] === '/' || candidate[2] === '\\') +} + +export function portableFilename(value: unknown): string | null { + const candidate = text(value) + if (!candidate) return null + const name = candidate.replace(/\\/g, '/').split('/').pop() || '' + if (!name || name === '.' || name === '..' || isHostPath(name)) return null + return name +} + +export function truncatePromptDisplay(value: unknown, limit = PROMPT_DISPLAY_MAX): string { + const candidate = String(value || '').trim() + if (candidate.length <= limit) return candidate + if (limit <= 1) return '…' + return `${candidate.slice(0, limit - 1).trimEnd()}…` +} + +export function redactSecrets(value: unknown, key = ''): unknown { + const lowered = key.toLowerCase().replace(/-/g, '_') + if ( + SENSITIVE_KEYS.has(lowered) + || lowered.endsWith('_api_key') + || lowered.endsWith('_password') + || lowered.endsWith('_secret') + || lowered.endsWith('_token') + ) { + return '[REDACTED]' + } + if (value == null || typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') { + return value + } + if (Array.isArray(value)) return value.map(item => redactSecrets(item)) + if (typeof value === 'object') { + return Object.fromEntries( + Object.entries(value as JsonMap).map(([child, item]) => [child, redactSecrets(item, child)]), + ) + } + return String(value) +} + +export function mapGenerationProduct(value: unknown, capability?: unknown): GenerationProduct { + const token = (text(value) || '').toLowerCase().replace(/ /g, '_') + if (PRODUCT_SET.has(token)) return token as GenerationProduct + if (token in PRODUCT_ALIASES) return PRODUCT_ALIASES[token] + const mapped = PRODUCT_FROM_CAPABILITY[text(capability) || ''] + return mapped || 'unknown' +} + +export function mapAssetManifestStatus( + status: unknown, + hasFilename = false, +): { status: GenerationStatus; resultKind: string | null; error: GenerationRecord['error'] } { + const raw = (text(status) || '').toLowerCase() + if (raw === 'prepared') return { status: 'planned', resultKind: null, error: null } + if (raw === 'partial') { + if (hasFilename) return { status: 'completed', resultKind: 'partial', error: null } + return { + status: 'failed', + resultKind: null, + error: { code: 'partial', message: 'Generation finished without a complete artifact' }, + } + } + if (STATUS_SET.has(raw)) { + return { status: raw as GenerationStatus, resultKind: null, error: null } + } + if (!raw) return { status: 'planned', resultKind: null, error: null } + return { + status: 'failed', + resultKind: null, + error: { code: 'invalid_status', message: `Unsupported status '${raw}'` }, + } +} + +export function mapGenerationStatusToManifest( + status: GenerationStatus, + resultKind?: string | null, +): string { + if (status === 'planned') return 'prepared' + if (status === 'completed' && resultKind === 'partial') return 'completed' + return status +} + +export function isLegalGenerationTransition(current: GenerationStatus, target: GenerationStatus): boolean { + return LEGAL_GENERATION_TRANSITIONS[current].includes(target) +} + +export function recordBelongsToWorkspace(record: Pick, workspaceId: string): boolean { + return Boolean(workspaceId) && record.workspace_id === workspaceId +} + +function sidecarName(filename: string | null): string | null { + if (!filename) return null + const dot = filename.lastIndexOf('.') + const stem = dot > 0 ? filename.slice(0, dot) : filename + return `${stem}.meta.json` +} + +function firstText(...values: unknown[]): string | null { + for (const value of values) { + const candidate = text(value) + if (candidate) return candidate + } + return null +} + +function lineageRef(value: unknown): GenerationLineageRef | null { + const raw = asMap(value) + const generationId = text(raw.generation_id) + const assetId = firstText(raw.asset_id, raw.id) + if (!generationId && !assetId) return null + const item: GenerationLineageRef = {} + if (generationId) item.generation_id = generationId + if (assetId) item.asset_id = assetId + const kind = firstText(raw.kind, raw.role) + if (kind) item.kind = kind + const uri = portableFilename(raw.uri) + if (uri) item.uri = uri + return item +} + +function lineageParents(value: unknown): GenerationLineageRef[] { + if (!Array.isArray(value)) return [] + return value.map(lineageRef).filter((item): item is GenerationLineageRef => item != null) +} + +function manifestPrompt(prompts: JsonMap): string { + return firstText(prompts.effective, prompts.original, prompts.audio) || '' +} + +function manifestError(execution: JsonMap, fallback: GenerationRecord['error']): GenerationRecord['error'] { + const error = asMap(execution.error) + if (error.code || error.message) return error as GenerationRecord['error'] + return fallback +} + +export function projectFromAssetManifest(manifest: unknown): GenerationRecord { + const value = asMap(manifest) + const asset = asMap(value.asset) + const origin = asMap(value.origin) + const execution = asMap(value.execution) + const generation = asMap(value.generation) + const timing = asMap(value.timing) + const technical = asMap(value.technical) + const model = asMap(generation.model) + const prompts = asMap(generation.prompts) + const languages = asMap(generation.languages) + const filename = portableFilename(firstText(asset.filename, asset.uri)) + const mapped = mapAssetManifestStatus(execution.status, Boolean(filename)) + const assetId = firstText(asset.id) || 'asset_unknown' + const workspaceId = firstText(origin.workspace_id) || 'unknown' + const prompt = manifestPrompt(prompts) + return { + schema: GENERATION_RECORD_SCHEMA, + schema_version: GENERATION_RECORD_SCHEMA_VERSION, + generation_id: firstText(technical.generation_id, execution.job_id) || `gen_${assetId}`, + asset_id: assetId, + product: mapGenerationProduct(origin.tool, origin.capability), + workspace_id: workspaceId, + output_folder: portableFilename(origin.output_folder) || workspaceId, + project_id: text(asMap(origin.project).id), + production_id: text(asMap(origin.production).id), + cue_id: text(execution.cue_id), + candidate_id: text(execution.candidate_id), + song_version: text(execution.song_version), + prompt_full: prompt, + prompt_display: truncatePromptDisplay(prompt), + model: { + provider: text(model.provider), + id: text(model.id), + version: firstText(model.version, model.revision), + configuration: redactSecrets(asMap(generation.parameters)) as Record, + }, + languages: { + conversation_language: text(languages.conversation_language), + content_language: firstText(languages.content_language, prompts.language), + spoken_language: text(languages.spoken_language), + technical_prompt_language: text(languages.technical_prompt_language), + }, + timestamps: { + created_at: text(timing.created_at), + queued_at: text(timing.queued_at), + started_at: text(timing.started_at), + completed_at: text(timing.completed_at), + duration_ms: typeof timing.total_ms === 'number' ? timing.total_ms : null, + }, + status: mapped.status, + lineage: { parents: lineageParents(asMap(value.lineage).parents), derivatives: [] }, + error: manifestError(execution, mapped.error), + retry_count: 0, + cancellation: { requested: false, at: null, reason: null }, + location: { filename, uri: filename, sidecar: sidecarName(filename) }, + links: { + activity_id: firstText(technical.activity_id, execution.task_id), + catalog_id: assetId, + ui_href: null, + }, + result: { kind: mapped.resultKind }, + } +} + +export function toAssetManifestPatch(record: GenerationRecord): JsonMap { + const filename = record.location.filename + const parents = record.lineage.parents.flatMap(item => ( + item.asset_id ? [{ id: item.asset_id, kind: item.kind || 'other', ...(item.uri ? { uri: item.uri } : {}) }] : [] + )) + return { + asset: { id: record.asset_id, filename, uri: record.location.uri || filename }, + origin: { + tool: record.product, + workspace_id: record.workspace_id, + output_folder: record.output_folder, + project: record.project_id ? { kind: 'project', id: record.project_id } : null, + production: record.production_id ? { kind: 'production', id: record.production_id } : null, + }, + execution: { + status: mapGenerationStatusToManifest(record.status, record.result.kind), + error: record.error, + cue_id: record.cue_id, + candidate_id: record.candidate_id, + song_version: record.song_version, + }, + generation: { + prompts: { original: record.prompt_full, effective: record.prompt_full }, + model: { provider: record.model.provider, id: record.model.id, revision: record.model.version }, + parameters: record.model.configuration, + inputs: parents, + }, + timing: { + created_at: record.timestamps.created_at, + queued_at: record.timestamps.queued_at, + started_at: record.timestamps.started_at, + completed_at: record.timestamps.completed_at, + total_ms: record.timestamps.duration_ms, + }, + lineage: { parents, transformations: [] }, + technical: { generation_id: record.generation_id, result: record.result }, + } +} + +function mintAttemptId(prefix: string): string { + const token = globalThis.crypto?.randomUUID?.().replace(/-/g, '') + || `${Date.now().toString(16)}${Math.random().toString(16).slice(2)}` + return `${prefix}_${token.slice(0, 24)}` +} + +export function retryGeneration(record: GenerationRecord, sameArtifact = false): Pick< + GenerationRecord, + 'asset_id' | 'generation_id' | 'retry_count' | 'lineage' | 'status' | 'workspace_id' +> { + return { + generation_id: mintAttemptId('gen'), + asset_id: sameArtifact ? record.asset_id : mintAttemptId('asset'), + retry_count: record.retry_count + 1, + status: 'planned', + workspace_id: record.workspace_id, + lineage: { + parents: [{ + generation_id: record.generation_id, + asset_id: record.asset_id, + kind: 'attempt', + }], + derivatives: [], + }, + } +} + +export function requestCancel(record: GenerationRecord): Pick { + if (record.status === 'completed' || record.status === 'failed' || record.status === 'cancelled') { + return { status: record.status, cancellation: record.cancellation } + } + const cancellation = { requested: true, at: record.cancellation.at, reason: record.cancellation.reason } + if (record.status === 'planned' || record.status === 'queued') { + return { status: 'cancelled', cancellation } + } + return { status: record.status, cancellation } +} + +export function applyCancel(record: GenerationRecord): Pick { + if (record.status === 'completed' || record.status === 'failed' || record.status === 'cancelled') { + return { status: record.status, cancellation: record.cancellation } + } + return { + status: 'cancelled', + cancellation: { requested: true, at: record.cancellation.at, reason: record.cancellation.reason }, + } +} diff --git a/ui/tests/generationRecord.test.ts b/ui/tests/generationRecord.test.ts new file mode 100644 index 00000000..c0ef0422 --- /dev/null +++ b/ui/tests/generationRecord.test.ts @@ -0,0 +1,162 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { + ATTEMPT_IDENTITY_POLICY, + GENERATION_RECORD_SCHEMA, + PROMPT_DISPLAY_MAX, + applyCancel, + isHostPath, + isLegalGenerationTransition, + mapAssetManifestStatus, + mapGenerationProduct, + mapGenerationStatusToManifest, + portableFilename, + projectFromAssetManifest, + recordBelongsToWorkspace, + redactSecrets, + requestCancel, + retryGeneration, + toAssetManifestPatch, + truncatePromptDisplay, + type GenerationRecord, +} from '../src/lib/generationRecord.ts' + +const sample: GenerationRecord = projectFromAssetManifest({ + asset: { id: 'asset_video_1', filename: 'choir.mp4' }, + origin: { + tool: 'story_lab', + capability: 'generate_story_song', + workspace_id: 'collection-a', + output_folder: 'night-shift', + project: { kind: 'story', id: 'story_1' }, + }, + execution: { + status: 'prepared', + job_id: 'job-1', + cue_id: 'cue-1', + candidate_id: 'candidate-1', + song_version: '2', + }, + generation: { + prompts: { effective: 'Metal fantástico de 1981', language: 'es' }, + model: { provider: 'local', id: 'minimax-h3', revision: 'r1' }, + parameters: { seed: 3, api_key: 'do-not-save', nested: { authorization: 'Bearer secret' } }, + inputs: [], + }, + timing: { created_at: '2026-09-01T00:00:00Z' }, + lineage: { + parents: [{ id: 'asset_song_1', kind: 'audio', uri: 'song.wav' }], + transformations: [], + }, + technical: { generation_id: 'gen_fixed' }, +}) + +test('projects the generation-record contract from an asset manifest', () => { + assert.equal(sample.schema, GENERATION_RECORD_SCHEMA) + assert.equal(sample.generation_id, 'gen_fixed') + assert.equal(sample.asset_id, 'asset_video_1') + assert.equal(sample.product, 'story_lab') + assert.equal(sample.status, 'planned') + assert.equal(sample.workspace_id, 'collection-a') + assert.equal(sample.output_folder, 'night-shift') + assert.equal(sample.cue_id, 'cue-1') + assert.equal(sample.location.filename, 'choir.mp4') + assert.equal(sample.location.sidecar, 'choir.meta.json') + assert.equal(sample.lineage.parents[0]?.asset_id, 'asset_song_1') + assert.equal(mapGenerationStatusToManifest(sample.status), 'prepared') + assert.equal(ATTEMPT_IDENTITY_POLICY, 'new_generation_id') + assert.equal('title' in sample, false) +}) + +test('truncates prompt_display and redacts secrets', () => { + const longPrompt = 'α'.repeat(PROMPT_DISPLAY_MAX + 40) + const display = truncatePromptDisplay(longPrompt) + assert.equal(display.length <= PROMPT_DISPLAY_MAX, true) + assert.equal(display.endsWith('…'), true) + const redacted = redactSecrets({ + api_key: 'do-not-save', + nested: { authorization: 'Bearer secret' }, + prompt: 'safe', + }) as { api_key: string; nested: { authorization: string }; prompt: string } + assert.equal(redacted.api_key, '[REDACTED]') + assert.equal(redacted.nested.authorization, '[REDACTED]') + assert.equal(redacted.prompt, 'safe') + assert.equal(sample.model.configuration.api_key, '[REDACTED]') + assert.notEqual(JSON.stringify(sample).includes('do-not-save'), true) +}) + +test('maps prepared and partial without a seventh public status', () => { + assert.deepEqual(mapAssetManifestStatus('prepared'), { + status: 'planned', resultKind: null, error: null, + }) + assert.deepEqual(mapAssetManifestStatus('partial', true), { + status: 'completed', resultKind: 'partial', error: null, + }) + assert.equal(mapAssetManifestStatus('partial', false).status, 'failed') + assert.equal(mapAssetManifestStatus('bogus').status, 'failed') + assert.equal(mapAssetManifestStatus('bogus').error?.code, 'invalid_status') + assert.equal(mapGenerationProduct('scene-animator-3d'), 'video_3d') + assert.equal(mapGenerationProduct('spoofed', 'generate_story_song'), 'story_lab') +}) + +test('identity is generation_id/asset_id, never title or prompt', () => { + const other = projectFromAssetManifest({ + asset: { id: 'asset_other', filename: 'choir.mp4' }, + origin: { tool: 'studio', workspace_id: 'collection-a', output_folder: 'night-shift' }, + execution: { status: 'queued', job_id: 'job-other' }, + generation: { + prompts: { effective: sample.prompt_full }, + model: {}, + parameters: {}, + inputs: [], + }, + lineage: { parents: [] }, + }) + assert.notEqual(other.generation_id, sample.generation_id) + assert.notEqual(other.asset_id, sample.asset_id) + assert.equal(other.prompt_full, sample.prompt_full) +}) + +test('retry mints a new generation_id and parent lineage', () => { + const child = retryGeneration(sample) + const sameBytes = retryGeneration(sample, true) + const second = retryGeneration(sample) + assert.notEqual(child.generation_id, sample.generation_id) + assert.notEqual(second.generation_id, child.generation_id) + assert.notEqual(child.asset_id, sample.asset_id) + assert.equal(child.retry_count, 1) + assert.equal(child.status, 'planned') + assert.equal(child.lineage.parents[0]?.generation_id, sample.generation_id) + assert.equal(sameBytes.asset_id, sample.asset_id) + assert.notEqual(sameBytes.generation_id, sample.generation_id) +}) + +test('cancellation before running settles; during running waits for apply', () => { + const queued = { ...sample, status: 'queued' as const } + assert.equal(requestCancel(queued).status, 'cancelled') + const running = { ...sample, status: 'running' as const } + const requested = requestCancel(running) + assert.equal(requested.status, 'running') + assert.equal(requested.cancellation.requested, true) + assert.equal(applyCancel(running).status, 'cancelled') + assert.equal(applyCancel({ ...sample, status: 'completed' }).status, 'completed') + assert.equal(applyCancel({ ...sample, status: 'failed' }).status, 'failed') + assert.equal(isLegalGenerationTransition('running', 'completed'), true) + assert.equal(isLegalGenerationTransition('completed', 'running'), false) +}) + +test('records cannot be adopted across workspaces and paths stay relative', () => { + assert.equal(recordBelongsToWorkspace(sample, 'collection-a'), true) + assert.equal(recordBelongsToWorkspace(sample, 'collection-b'), false) + assert.equal(isHostPath('/tmp/outputs'), true) + assert.equal(isHostPath('night-shift'), false) + assert.equal(portableFilename('/tmp/outputs/choir.mp4'), 'choir.mp4') + const patch = toAssetManifestPatch({ + ...sample, + timestamps: { ...sample.timestamps, duration_ms: 4120 }, + }) + assert.equal((patch.timing as { total_ms: number }).total_ms, 4120) + assert.equal((patch.origin as { workspace_id: string }).workspace_id, 'collection-a') + assert.equal((patch.asset as { id: string }).id, 'asset_video_1') + assert.equal(JSON.stringify(patch).includes('/tmp'), false) +})