feat(video): normalize async job resource state

.github/workflows/red-video-job-registry-atomicity.yml

@@ -0,0 +1,29 @@
+name: Video job registry atomicity RED
+
+on:
+  push:
+    branches:
+      - feat/normalized-video-job-resource-20260829
+
+permissions:
+  contents: read
+
+jobs:
+  red:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+      - name: Set up Python
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1
+        with:
+          python-version: "3.12"
+      - name: Set up uv
+        uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e
+        with:
+          version: "0.12.5"
+      - name: Run atomicity regressions
+        run: >-
+          uv run --locked --extra api --extra db --extra queue --group dev
+          python -m pytest -q tests/test_video_job_registry_atomicity.py

CHANGELOG.md

@@ -17,7 +17,8 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
   readiness report while keeping local product evidence inspectable.
 - Provider-affine asynchronous video jobs now return an opaque gateway id and
   keep status polling and content download bound to the exact provider agent
-  that accepted the submission (ADR 0037).
+  that accepted the submission (ADR 0037); new ownership and first-complete
+  usage observations are stored as separate registry records.
 - A fail-closed, transactional evidence boundary for the optional NVIDIA NIM
   benchmark with immutable task/scorer identities and complete provenance.
 - Bounded first-valid-completion racing for operator-declared equivalent model

contextual_orchestrator/batch_job_registry.py

@@ -105,6 +105,13 @@ def __setitem__(self, job_id: str, value: Any) -> None:
         # expiry forward, so an active registry never expires mid-flight.
         self._client.expire(self._key, self._retention_seconds)
 
+    def set_if_absent(self, job_id: str, value: Any) -> bool:
+        """Store one value only when no concurrent writer stored it first."""
+        created = bool(self._client.hsetnx(self._key, job_id, _encode(value)))
+        if created:
+            self._client.expire(self._key, self._retention_seconds)
+        return created
+
     def __delitem__(self, job_id: str) -> None:
         if not self._client.hdel(self._key, job_id):
             raise KeyError(job_id)

contextual_orchestrator/video_jobs.py

@@ -5,6 +5,7 @@
 from dataclasses import dataclass
 import hashlib
 import json
+import threading
 import time
 from typing import Any
 import uuid
@@ -14,9 +15,30 @@ class VideoJobContractError(RuntimeError):
     """Raised when a provider submission cannot become a trackable video job."""
 
 
+@dataclass(frozen=True)
+class VideoJobRecord:
+    """Persist the immutable ownership tuple for one gateway video job."""
+
+    gateway_job_id: str
+    provider_job_id: str
+    agent_id: str
+    submitted_at: int
+    owner_id: str
+    agent_affinity_key: str = ""
+
+
+@dataclass(frozen=True)
+class VideoJobUsage:
+    """Persist the first complete provider usage report separately."""
+
+    prompt_tokens: int
+    completion_tokens: int
+    observed_at: int
+
+
 @dataclass(frozen=True)
 class VideoJobOwner:
-    """Bind one gateway video job to the exact provider agent that accepted it."""
+    """Combine an ownership record with its optional measured usage."""
 
     gateway_job_id: str
     provider_job_id: str
@@ -44,9 +66,19 @@ class VideoJobRegistry:
     """Store video-job ownership in the configured shared job registry."""
 
     def __init__(self, job_registry: Any) -> None:
-        self._owners = job_registry.mapping(
+        # Keep immutable ownership and mutable provider observations in separate
+        # records. The old mapping remains a compatibility path for jobs
+        # created before this normalized layout existed.
+        self._records = job_registry.mapping(
+            "video_job_records", decode=lambda raw: VideoJobRecord(**raw)
+        )
+        self._usages = job_registry.mapping(
+            "video_job_usages", decode=lambda raw: VideoJobUsage(**raw)
+        )
+        self._legacy_owners = job_registry.mapping(
             "video_job_owners", decode=lambda raw: VideoJobOwner(**raw)
         )
+        self._usage_lock = threading.Lock()
 
     def register(
         self,
@@ -65,20 +97,29 @@ def register(
         if not isinstance(owner_id, str) or not owner_id:
             raise VideoJobContractError("video job owner is unavailable")
         gateway_job_id = f"videojob_{uuid.uuid4().hex}"
-        reported_usage = self._provider_usage(provider_result)
-        self._owners[gateway_job_id] = VideoJobOwner(
+        submitted_at = int(time.time())
+        record = VideoJobRecord(
             gateway_job_id=gateway_job_id,
             provider_job_id=provider_job_id,
             agent_id=agent_id,
-            submitted_at=int(time.time()),
+            submitted_at=submitted_at,
             owner_id=owner_id,
-            usage_measurement_status=(
-                "measured" if reported_usage is not None else "unavailable"
-            ),
-            provider_usage=reported_usage,
             agent_affinity_key=agent_affinity_key,
         )
-        return self.public_response(provider_result, self._owners[gateway_job_id])
+        reported_usage = self._provider_usage(provider_result)
+        # Core ownership is written first: a failure after provider acceptance
+        # must never leave the job without a gateway-addressable owner.
+        self._records[gateway_job_id] = record
+        if reported_usage is not None:
+            self._store_usage_if_absent(
+                gateway_job_id,
+                VideoJobUsage(
+                    prompt_tokens=reported_usage["prompt_tokens"],
+                    completion_tokens=reported_usage["completion_tokens"],
+                    observed_at=submitted_at,
+                ),
+            )
+        return self.public_response(provider_result, self._owner_from_record(record))
 
     @staticmethod
     def public_response(
@@ -111,30 +152,97 @@ def replace(value: Any) -> Any:
 
     def owner(self, gateway_job_id: str, owner_id: str) -> VideoJobOwner:
         """Return a principal-owned record, hiding unknown and foreign identities alike."""
-        owner = self._owners[gateway_job_id]
-        if not owner_id or owner.owner_id != owner_id:
+        record, legacy = self._record(gateway_job_id)
+        stored_owner_id = record.owner_id
+        if not owner_id or stored_owner_id != owner_id:
             raise KeyError(gateway_job_id)
-        return owner
+        if legacy:
+            assert isinstance(record, VideoJobOwner)
+            return record
+        assert isinstance(record, VideoJobRecord)
+        return self._owner_from_record(record)
+
+    def _record(self, gateway_job_id: str) -> tuple[VideoJobRecord | VideoJobOwner, bool]:
+        """Load the normalized record, falling back to the legacy map."""
+        try:
+            return self._records[gateway_job_id], False
+        except KeyError:
+            return self._legacy_owners[gateway_job_id], True
+
+    def _owner_from_record(self, record: VideoJobRecord) -> VideoJobOwner:
+        """Join one normalized ownership record to its optional usage row."""
+        usage = self._usages.get(record.gateway_job_id)
+        provider_usage = (
+            None
+            if usage is None
+            else {
+                "prompt_tokens": usage.prompt_tokens,
+                "completion_tokens": usage.completion_tokens,
+            }
+        )
+        return VideoJobOwner(
+            gateway_job_id=record.gateway_job_id,
+            provider_job_id=record.provider_job_id,
+            agent_id=record.agent_id,
+            submitted_at=record.submitted_at,
+            owner_id=record.owner_id,
+            usage_measurement_status=(
+                "measured" if provider_usage is not None else "unavailable"
+            ),
+            provider_usage=provider_usage,
+            agent_affinity_key=record.agent_affinity_key,
+        )
 
     def observe_provider_result(
         self, owner: VideoJobOwner, provider_result: dict[str, Any]
     ) -> VideoJobOwner:
         """Persist provider-reported counts while retaining honest unknown state."""
         usage = self._provider_usage(provider_result)
-        if usage is None or owner.provider_usage is not None:
-            return owner
-        updated = VideoJobOwner(
-            gateway_job_id=owner.gateway_job_id,
-            provider_job_id=owner.provider_job_id,
-            agent_id=owner.agent_id,
-            submitted_at=owner.submitted_at,
-            owner_id=owner.owner_id,
-            usage_measurement_status="measured",
-            provider_usage=usage,
-            agent_affinity_key=owner.agent_affinity_key,
-        )
-        self._owners[owner.gateway_job_id] = updated
-        return updated
+        record, legacy = self._record(owner.gateway_job_id)
+        if legacy:
+            if usage is None or owner.provider_usage is not None:
+                return owner
+            # Preserve the old payload shape until its configured retention
+            # expires; new submissions never write this denormalized mapping.
+            with self._usage_lock:
+                current = self._legacy_owners[owner.gateway_job_id]
+                if current.provider_usage is not None:
+                    return current
+                updated = VideoJobOwner(
+                    gateway_job_id=current.gateway_job_id,
+                    provider_job_id=current.provider_job_id,
+                    agent_id=current.agent_id,
+                    submitted_at=current.submitted_at,
+                    owner_id=current.owner_id,
+                    usage_measurement_status="measured",
+                    provider_usage=usage,
+                    agent_affinity_key=current.agent_affinity_key,
+                )
+                self._legacy_owners[owner.gateway_job_id] = updated
+                return updated
+        assert isinstance(record, VideoJobRecord)
+        if usage is not None and owner.provider_usage is None:
+            self._store_usage_if_absent(
+                owner.gateway_job_id,
+                VideoJobUsage(
+                    prompt_tokens=usage["prompt_tokens"],
+                    completion_tokens=usage["completion_tokens"],
+                    observed_at=int(time.time()),
+                ),
+            )
+        return self._owner_from_record(record)
+
+    def _store_usage_if_absent(self, gateway_job_id: str, usage: VideoJobUsage) -> None:
+        """Keep the first complete report when replicas observe stale owners."""
+        setter = getattr(self._usages, "set_if_absent", None)
+        if callable(setter):
+            setter(gateway_job_id, usage)
+            return
+        # ponytail: plain dict fallback has no compare-and-set; this process-local
+        # lock serializes its threaded request path. Valkey uses hsetnx above.
+        with self._usage_lock:
+            if self._usages.get(gateway_job_id) is None:
+                self._usages[gateway_job_id] = usage
 
     @staticmethod
     def _provider_usage(document: dict[str, Any]) -> dict[str, int] | None:

docs/architecture.md

@@ -68,6 +68,11 @@ bounded, authenticated recursion protocol; it is not administratively disabled.
 - `WorkflowStep.access`: Conductor-style visibility control.
 - `ModelClient`: OpenAI-compatible HTTP client, with `mock://` for local checks.
 - `contextual_orchestrator.server`: small `/v1/chat/completions` HTTP server.
+- `contextual_orchestrator.video_jobs.VideoJobRegistry`: provider-affine async
+  video resources. New submissions join immutable `video_job_records` with an
+  optional first-complete `video_job_usages` row; legacy owner payloads are only
+  a compatibility read path. Provider status is observed in provider responses,
+  never persisted or inferred.
 - Streamed `/v1/responses` workflow runs preserve optional provider usage on
   each trace step and record one `stream` cost-ledger row per completed step.
   Missing provider counts remain `unavailable`; the gateway never derives

docs/doctoring/video-job-resource.md

@@ -0,0 +1,28 @@
+# Video job resource contract
+
+The gateway treats video generation as an asynchronous resource. Submission
+returns an opaque gateway identifier; later status and content requests resolve
+the stored provider owner rather than selecting a model again. The provider job
+identifier remains internal, and a provider lifecycle value is copied only as
+observed data.
+
+New registry entries keep immutable ownership in `video_job_records` and the
+first complete token report in `video_job_usages`. Missing usage remains
+unavailable, repeated polling cannot revise the first measured report, and
+provider status is returned only as observed by the current provider response.
+Standalone process-local storage makes no durability claim. Valkey is required
+when restart or replica continuity matters.
+
+The contract follows the HTTP asynchronous-resource model: the upstream video
+API accepts work and exposes a polling resource, while HTTP semantics require
+the gateway to describe acceptance without claiming completion. The gateway's
+opaque id also prevents a provider-specific resource identifier from becoming
+the public ownership key.
+
+## References
+
+OpenRouter. (2026). *Video generation*. OpenRouter documentation.
+https://openrouter.ai/docs/guides/overview/multimodal/video-generation
+
+Fielding, R. T., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC
+9110). Internet Engineering Task Force. https://www.rfc-editor.org/rfc/rfc9110

docs/model-group-product-technical-spec.md

@@ -61,8 +61,10 @@ failover; see `docs/doctoring/equivalent-endpoint-racing.md`.
   inference.
 - The current success-per-second member order is not a distinct-model quality
   router. Quality/cost routing requires calibrated evaluation and ablation.
-- A synchronous video submission does not prove provider-affine polling or durable
-  result ownership; that remains a production Gap.
+- Asynchronous video submission returns an opaque gateway resource whose status
+  and content reads stay bound to the accepting provider account. New ownership
+  and measured-usage records are normalized inside the existing job-registry
+  boundary; Valkey is still required for restart and replica durability.
 
 ## TRD
 
@@ -108,8 +110,22 @@ classDiagram
     +agent_id: text
     +group_name: text
   }
+  class VideoJobRecord {
+    +gateway_job_id: text
+    +provider_job_id: text
+    +agent_id: text
+    +owner_id: text
+    +submitted_at: integer
+  }
+  class VideoJobUsage {
+    +gateway_job_id: text
+    +prompt_tokens: integer
+    +completion_tokens: integer
+    +observed_at: integer
+  }
   ModelGroup "1" --> "0..*" ModelGroupMember
   AgentPool "1" --> "0..1" ModelGroupMember
+  VideoJobRecord "1" --> "0..1" VideoJobUsage
 ```
 
 `model_group_member.agent_id` is both its primary key and a foreign key, so one
@@ -143,9 +159,11 @@ transactionally into these relations.
 - Touch targets, responsive layout, typography/color tokens, navigation, form
   validation, performance, and accessibility remain governed by the existing
   Admin design system and regression contracts.
-- Multi-replica observation durability, video job ownership, and opt-in spend-
-  capped live canaries remain explicitly tracked in
-  `docs/product-technical-gap-baseline.md`.
+- Multi-replica routing observation calibration, cancellable conducted-answer
+  streaming, and opt-in spend-capped live canaries remain explicitly tracked in
+  `docs/product-technical-gap-baseline.md`. Video job durability still depends
+  on the configured shared job registry; standalone mode makes no durability
+  claim.
 
 ## Evidence and limits
 

docs/planning/adrs/0037-provider-affine-video-job-ownership.md

@@ -71,6 +71,13 @@ configured, ownership survives process restart and is shared across replicas;
 without it, the established standalone registry is process-local and makes no
 durability claim. This decision adds no inferred TTL or new retention policy.
 
+The registry representation is normalized for new submissions: immutable
+ownership lives in `video_job_records`, while the first complete provider usage
+report is joined from `video_job_usages`. The pre-0037 `video_job_owners` map is
+read and updated only for legacy records during its configured retention. A
+provider's lifecycle status remains an observed response field; the gateway
+does not persist or infer a status enum from model names or timing.
+
 ## Consequences
 
 - Polling and content retrieval cannot cross provider boundaries.

docs/product-technical-gap-baseline.md

@@ -1,5 +1,22 @@
 # Product and Technical Gap Baseline
 
+## 2026-08-29 KST normalized video job resource slice
+
+Protected `main` at `b21645116b352967e50fc497b87eb745b9cc8c61` already contains
+PR #883's provider-affine video follow-up contract: the client receives an
+opaque gateway id, and polling/content reads use the exact accepting provider
+agent. This candidate slice closes the remaining representation gap for new
+submissions by separating immutable ownership in `video_job_records` from the
+first complete usage observation in `video_job_usages`, while retaining the
+existing `video_job_owners` shape only for legacy compatibility. The registry's
+existing Valkey configuration remains the only restart/multi-replica durability claim;
+standalone mode remains process-local. No provider status is inferred and no
+new retention policy is introduced.
+
+The next unclosed customer-visible gap is cancellable answer-token streaming
+for conducted workflows. It requires an asynchronous dependency graph that can
+cancel or safely drain dependent provider work; this slice does not fabricate
+partial answers before verification and synthesis.
 ## 2026-08-29 streamed Responses usage boundary
 
 Protected `main` remains