generation quality

apps/api/app/ai/prompts/edit_plan.txt

@@ -13,8 +13,11 @@ Return one semantic decision and exactly one grounded variant:
   "decision": {
     "scope": "local" | "all_occurrences",
     "change_type": "appearance" | "removal" | "replacement" | "motion" | "scene",
-    "duration_policy": "bounded_action" | "continuous_occurrence" | "all_occurrences",
+    "duration_policy": "bounded_action" | "continuous_occurrence" | "trajectory_continuation" | "all_occurrences",
+    "temporal_behavior": "temporary" | "persistent_state" | "future_changing_motion",
     "target_description": "selected object or feature",
+    "selection_match": "match" | "mismatch" | "uncertain",
+    "selection_match_reason": "why the selected pixels do or do not match the request",
     "action_phases": ["ordered phase names for motion; empty for a state change"],
     "estimated_action_seconds": 3.5,
     "requires_recovery_motion": true,
@@ -36,14 +39,28 @@ Return one semantic decision and exactly one grounded variant:
 
 RULES
 - Inspect the reference image. Ground every noun and visual attribute in it.
+- The reference image is the selected target isolated from the source frame when
+  segmentation is available. Decide whether an explicit target noun in the user
+  request matches it. Use "mismatch" only for a clear contradiction, such as a
+  request for a man when the selection clearly contains only a bus. Pronouns and
+  deictic words such as "this", "it", "him", or "her" are not contradictions;
+  use "match". Use "uncertain" when the pixels are ambiguous. Never invent a
+  mismatch from a requested action or property.
 - Decide temporal meaning before any source clip is cut.
 - A state change such as "make this ball pink", "remove this logo", or "give
   this cat human-like eyes" normally means the change persists for this selected
   target's complete continuous appearance. Use scope "local" and duration_policy
-  "continuous_occurrence". Do not limit such changes to an arbitrary few seconds.
-- An action such as bounce, jump, wave, or turn is local and time-bounded. Use
-  duration_policy "bounded_action" and estimate enough time for preparation,
-  action, landing/recovery, and smooth return to source motion.
+  "continuous_occurrence" with temporal_behavior "persistent_state". Do not
+  limit such changes to an arbitrary few seconds.
+- An action such as blink, bounce, jump, wave, or turn can end near its original
+  state. Use temporal_behavior "temporary" and duration_policy "bounded_action".
+  Estimate enough time for preparation, action, landing/recovery, and a smooth
+  return to source motion.
+- Motion such as run, walk, fly, drive, leave, approach, follow, or chase changes
+  the subject's later position or trajectory. Use temporal_behavior
+  "future_changing_motion" and duration_policy "trajectory_continuation". Continue
+  until the generated motion can naturally rejoin the source or the target leaves
+  the shot. Never force it back into an old source position.
 - Use scope "all_occurrences" and duration_policy "all_occurrences" only when
   the user clearly asks for every appearance, everywhere, throughout, or the
   whole video. Never interpret an ordinary state change as permission to search

apps/api/app/ai/services/_stubs.py

@@ -36,7 +36,10 @@ async def _gemini_interpret_edit(
             "scope": intent.scope,
             "change_type": intent.change_type,
             "duration_policy": intent.duration_policy,
+            "temporal_behavior": intent.temporal_behavior,
             "target_description": intent.target_description or region,
+            "selection_match": "match",
+            "selection_match_reason": "selected region is the requested target",
             "action_phases": intent.action_phases,
             "estimated_action_seconds": intent.estimated_action_seconds,
             "requires_recovery_motion": intent.requires_recovery_motion,

apps/api/app/api/routes/accept.py

@@ -25,7 +25,12 @@
     record_accepted_range,
     update_project_edl_for_acceptance,
 )
-from app.services.generation_quality import acceptance_allowed, acceptance_block_reason
+from app.services.generation_quality import (
+    acceptance_allowed,
+    acceptance_block_reason,
+    cancel_waiting_retry,
+    quality_payload_for_candidate,
+)
 from app.services.timeline_response import build_timeline_response
 from app.workers import entity_job
 
@@ -58,16 +63,18 @@ async def accept(
     )
     if variant is None or variant.status != "done" or not variant.url:
         raise HTTPException(status_code=422, detail="variant not ready")
+    job.payload = cancel_waiting_retry(job.payload, reason="candidate applied")
+    quality_payload = quality_payload_for_candidate(job.payload, variant.id)
     if not acceptance_allowed(
-        job.payload,
+        quality_payload,
         override_requested=body.continuity_override,
         override_enabled=get_settings().allow_hard_failed_acceptance,
     ):
-        raise HTTPException(status_code=409, detail=acceptance_block_reason(job.payload))
+        raise HTTPException(status_code=409, detail=acceptance_block_reason(quality_payload))
 
     if job.start_ts is None or job.end_ts is None:
         raise HTTPException(status_code=422, detail="job has no segment range")
-    accepted_range = accepted_generation_range(job)
+    accepted_range = accepted_generation_range(job, variant=variant)
 
     # deactivate any existing generated segments that overlap this range.
     # the newest accept wins on overlap. we don't delete rows so we keep

apps/api/app/api/routes/edit_plans.py

@@ -131,13 +131,27 @@ async def create_edit_plan(
     structured_intent = body.structured_intent
     if structured_intent is None:
         planning_frame = ""
-        try:
-            frame_path, _ = storage.new_path("keyframes", "jpg")
-            source_path = await materialize_edit_source(project, edit_source)
-            await ffmpeg.extract_frame(source_path, selection.frame_ts, frame_path)
-            planning_frame = str(frame_path)
-        except Exception:
-            log.warning("semantic planning frame unavailable", exc_info=True)
+        if selection.subject_reference_url:
+            try:
+                planning_frame = str(
+                    await storage.path_from_url(selection.subject_reference_url)
+                )
+            except Exception:
+                log.warning("segmented subject reference unavailable", exc_info=True)
+        if not planning_frame:
+            try:
+                frame_path, _ = storage.new_path("keyframes", "jpg")
+                source_path = await materialize_edit_source(project, edit_source)
+                await ffmpeg.extract_frame(source_path, selection.frame_ts, frame_path)
+                crop_path, _ = storage.new_path("keyframes", "png")
+                await ffmpeg.crop_bbox_from_frame(
+                    frame_path,
+                    selection.bbox_json,
+                    crop_path,
+                )
+                planning_frame = str(crop_path)
+            except Exception:
+                log.warning("semantic planning frame unavailable", exc_info=True)
         try:
             interpretation = SemanticEditPlan.model_validate(
                 await ai.gemini.interpret_edit(
@@ -146,7 +160,21 @@ async def create_edit_plan(
                     planning_frame,
                 )
             )
+            if interpretation.decision.selection_match == "mismatch":
+                raise HTTPException(
+                    status_code=422,
+                    detail={
+                        "code": "selection_target_mismatch",
+                        "message": (
+                            interpretation.decision.selection_match_reason
+                            or "the selected object does not match the requested target"
+                        ),
+                        "selected_target": interpretation.decision.target_description,
+                    },
+                )
             structured_intent = interpretation.as_intent(body.prompt)
+        except HTTPException:
+            raise
         except Exception as exc:
             log.warning("semantic edit interpretation failed; using safe rules", exc_info=True)
             semantic_warning = (

apps/api/app/api/routes/jobs.py

@@ -1,7 +1,10 @@
 import asyncio
 import json
+import time
+from typing import Literal
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from pydantic import BaseModel
 from fastapi.responses import StreamingResponse
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -18,8 +21,13 @@
 router = APIRouter(tags=["jobs"])
 
 
+class RetryDecisionRequest(BaseModel):
+    action: Literal["cancel", "retry_now"]
+
+
 def _job_out(job: Job) -> JobOut:
     payload = job.payload or {}
+    candidate_reviews = payload.get("candidate_reviews") or {}
     return JobOut(
         job_id=job.id,
         kind=job.kind,
@@ -35,13 +43,15 @@ def _job_out(job: Job) -> JobOut:
         warnings=payload.get("warnings") or [],
         execution_window=payload.get("execution_window"),
         continuity_validation=payload.get("continuity_validation"),
+        selected_seams=payload.get("selected_seams"),
         generation_quality_state=payload.get("generation_quality_state"),
         generation_quality_evidence=payload.get("generation_quality_evidence") or [],
         generation_attempts=payload.get("generation_attempts"),
         generated_seconds=payload.get("generated_seconds"),
         provider_attempts=payload.get("provider_attempts") or [],
         localized_compositing=payload.get("localized_compositing") or [],
         local_flow_telemetry=payload.get("local_flow_telemetry"),
+        retry_state=payload.get("retry_state"),
         variants=[
             VariantOut(
                 id=v.id,
@@ -52,6 +62,13 @@ def _job_out(job: Job) -> JobOut:
                 visual_coherence=v.visual_coherence,
                 prompt_adherence=v.prompt_adherence,
                 error=v.error,
+                attempt_label=(candidate_reviews.get(v.id) or {}).get("label"),
+                quality_state=(candidate_reviews.get(v.id) or {}).get("quality_state"),
+                quality_evidence=(candidate_reviews.get(v.id) or {}).get("evidence") or [],
+                continuity_validation=(candidate_reviews.get(v.id) or {}).get(
+                    "continuity_validation"
+                ),
+                selected_seams=(candidate_reviews.get(v.id) or {}).get("selected_seams"),
             )
             for v in sorted(job.variants, key=lambda v: v.index)
         ],
@@ -103,6 +120,31 @@ async def get_job(
     return _job_out(job)
 
 
+@router.post("/jobs/{job_id}/retry-decision")
+async def decide_retry(
+    job_id: str,
+    body: RetryDecisionRequest,
+    session: SessionModel = Depends(get_session),
+    db: AsyncSession = Depends(get_db),
+):
+    job = await db.get(Job, job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="job not found")
+    project = await db.get(Project, job.project_id)
+    if project is None or project.session_id != session.id:
+        raise HTTPException(status_code=404, detail="job not found")
+    payload = dict(job.payload or {})
+    retry_state = dict(payload.get("retry_state") or {})
+    if retry_state.get("status") != "waiting":
+        raise HTTPException(status_code=409, detail="no retry is waiting for a decision")
+    retry_state["status"] = "cancelled" if body.action == "cancel" else "retry_now"
+    retry_state["decision_at"] = time.time()
+    payload["retry_state"] = retry_state
+    job.payload = payload
+    await db.commit()
+    return {"status": retry_state["status"]}
+
+
 @router.get("/jobs/{job_id}/stream")
 async def stream_job_events(
     job_id: str,

apps/api/app/api/routes/remix.py

@@ -21,7 +21,12 @@
     record_accepted_range,
     update_project_edl_for_acceptance,
 )
-from app.services.generation_quality import acceptance_allowed, acceptance_block_reason
+from app.services.generation_quality import (
+    acceptance_allowed,
+    acceptance_block_reason,
+    cancel_waiting_retry,
+    quality_payload_for_candidate,
+)
 from app.workers import entity_job
 from app.workers import generate_job as generate_worker
 
@@ -122,7 +127,7 @@ async def _accept_variant_for_job(
 ) -> tuple[str, str | None]:
     if job.start_ts is None or job.end_ts is None:
         raise HTTPException(status_code=422, detail="job has no segment range")
-    accepted_range = accepted_generation_range(job)
+    accepted_range = accepted_generation_range(job, variant=variant)
 
     overlapping = (
         await db.execute(
@@ -311,12 +316,17 @@ async def batch_accept(
         variant = next((v for v in job.variants if v.index == item.variant_index), None)
         if variant is None or variant.status != "done" or not variant.url:
             raise HTTPException(status_code=422, detail="variant not ready")
+        job.payload = cancel_waiting_retry(job.payload, reason="candidate applied")
+        quality_payload = quality_payload_for_candidate(job.payload, variant.id)
         if not acceptance_allowed(
-            job.payload,
+            quality_payload,
             override_requested=item.continuity_override,
             override_enabled=get_settings().allow_hard_failed_acceptance,
         ):
-            raise HTTPException(status_code=409, detail=acceptance_block_reason(job.payload))
+            raise HTTPException(
+                status_code=409,
+                detail=acceptance_block_reason(quality_payload),
+            )
 
         segment_id, entity_job_id = await _accept_variant_for_job(
             db=db,

apps/api/app/schemas/accept.py

@@ -5,7 +5,7 @@
 
 class AcceptRequest(BaseModel):
     job_id: str
-    # single-edit mode only ever produces index 0. kept for backward-compat.
+    # Index 0 is the first pass; index 1 is the optional corrected pass.
     variant_index: int = 0
     # Full-reel entity search is opt-in. Local acceptance stays local and cheap.
     discover_occurrences: bool = False

apps/api/app/schemas/edit_plan.py

@@ -17,9 +17,15 @@
 DurationPolicy = Literal[
     "bounded_action",
     "continuous_occurrence",
+    "trajectory_continuation",
     "explicit_range",
     "all_occurrences",
 ]
+TemporalBehavior = Literal[
+    "temporary",
+    "persistent_state",
+    "future_changing_motion",
+]
 
 
 class GroundedEditInstruction(BaseModel):
@@ -36,13 +42,31 @@ class SemanticEditDecision(BaseModel):
     scope: EditScope
     change_type: ChangeType
     duration_policy: DurationPolicy
+    temporal_behavior: TemporalBehavior
     target_description: str | None = None
+    selection_match: Literal["match", "mismatch", "uncertain"] = "match"
+    selection_match_reason: str | None = None
     action_phases: list[str] = Field(default_factory=list)
     estimated_action_seconds: float = Field(default=3.0, ge=0.5, le=15.0)
     requires_recovery_motion: bool = False
     preservation_requirements: list[str] = Field(default_factory=list)
     reasoning: str = Field(min_length=1, max_length=1000)
 
+    @model_validator(mode="before")
+    @classmethod
+    def infer_legacy_temporal_behavior(cls, value):
+        if isinstance(value, dict) and not value.get("temporal_behavior"):
+            value = dict(value)
+            policy = value.get("duration_policy")
+            value["temporal_behavior"] = (
+                "future_changing_motion"
+                if policy == "trajectory_continuation"
+                else "persistent_state"
+                if policy == "continuous_occurrence"
+                else "temporary"
+            )
+        return value
+
 
 class SemanticEditPlan(BaseModel):
     decision: SemanticEditDecision
@@ -55,6 +79,7 @@ def as_intent(self, raw_prompt: str) -> "EditIntent":
             scope=decision.scope,
             change_type=decision.change_type,
             duration_policy=decision.duration_policy,
+            temporal_behavior=decision.temporal_behavior,
             target_description=decision.target_description,
             action_phases=decision.action_phases,
             estimated_action_seconds=decision.estimated_action_seconds,
@@ -70,6 +95,7 @@ class EditIntent(BaseModel):
     scope: EditScope = "local"
     change_type: ChangeType
     duration_policy: DurationPolicy = "bounded_action"
+    temporal_behavior: TemporalBehavior = "temporary"
     target_description: str | None = None
     action_phases: list[str] = Field(default_factory=list)
     estimated_action_seconds: float = Field(default=3.0, gt=0.0)
@@ -78,6 +104,21 @@ class EditIntent(BaseModel):
     reasoning: str | None = None
     grounded_edit: GroundedEditInstruction | None = None
 
+    @model_validator(mode="before")
+    @classmethod
+    def infer_legacy_temporal_behavior(cls, value):
+        if isinstance(value, dict) and not value.get("temporal_behavior"):
+            value = dict(value)
+            policy = value.get("duration_policy")
+            value["temporal_behavior"] = (
+                "future_changing_motion"
+                if policy == "trajectory_continuation"
+                else "persistent_state"
+                if policy == "continuous_occurrence"
+                else "temporary"
+            )
+        return value
+
 
 class EditCore(BaseModel):
     start_ts: float = Field(ge=0.0)

apps/api/app/schemas/job.py

@@ -15,6 +15,11 @@ class VariantOut(BaseModel):
     visual_coherence: int | None = None
     prompt_adherence: int | None = None
     error: str | None = None
+    attempt_label: str | None = None
+    quality_state: str | None = None
+    quality_evidence: list[str] = Field(default_factory=list)
+    continuity_validation: dict[str, Any] | None = None
+    selected_seams: dict[str, Any] | None = None
 
 
 class JobOut(BaseModel):
@@ -34,10 +39,12 @@ class JobOut(BaseModel):
     warnings: list[str] = Field(default_factory=list)
     execution_window: dict[str, Any] | None = None
     continuity_validation: dict[str, Any] | None = None
+    selected_seams: dict[str, Any] | None = None
     generation_quality_state: str | None = None
     generation_quality_evidence: list[str] = Field(default_factory=list)
     generation_attempts: int | None = None
     generated_seconds: float | None = None
     provider_attempts: list[str] = Field(default_factory=list)
     localized_compositing: list[dict[str, Any]] = Field(default_factory=list)
     local_flow_telemetry: dict[str, Any] | None = None
+    retry_state: dict[str, Any] | None = None