From e5aae298d0a1fd7ee94f10bedb8f96aeb878a811 Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 11:47:30 +0530 Subject: [PATCH 01/14] docs: define generation quality plan --- docs/docs.json | 1 + docs/product/generation-quality-plan.mdx | 230 +++++++++++++++++++++++ 2 files changed, 231 insertions(+) create mode 100644 docs/product/generation-quality-plan.mdx diff --git a/docs/docs.json b/docs/docs.json index 655ccf3..4f8908e 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -32,6 +32,7 @@ "pages": [ "product/editor-workflow", "product/continuity", + "product/generation-quality-plan", "product/library-and-projects" ] }, diff --git a/docs/product/generation-quality-plan.mdx b/docs/product/generation-quality-plan.mdx new file mode 100644 index 0000000..773ae4d --- /dev/null +++ b/docs/product/generation-quality-plan.mdx @@ -0,0 +1,230 @@ +--- +title: "Generation quality work plan" +description: "The implementation plan for scoped edits, frame-matched seams, live review, and one evidence-driven retry." +--- + +# Generation quality work plan + +This document defines the target generation flow. It is a work plan, not a description of behavior already available in production. + +## Product decisions + +The completed flow must follow these rules: + +- show every generated candidate as soon as it is available +- keep the first candidate while it is reviewed +- allow at most one automatic retry, using specific evidence from the first result +- never switch video providers automatically during a quality retry +- preserve and use the complete generated context, including its handles +- choose splice points by matching source and generated frames +- do not hide a candidate merely because semantic review dislikes it +- block Apply only when the media or seam is technically unsafe, and show the reason before the user clicks Apply +- do not use opacity fades as the default seam repair +- do not generate a separate transition clip + +## Target flow + +```text +request + -> validate selected target + -> classify edit behavior + -> resolve natural edit range + -> add locked source handles + -> generate first candidate + -> show candidate immediately + -> review meaning and measure seams + -> choose matching entry and exit frames + -> pass, or announce one evidence-driven retry + -> allow the user to stop the retry + -> show the final candidate and Apply state +``` + +Each stage must be persisted on the job and streamed to chat. Leaving the editor, refreshing the page, or opening the dashboard must not stop the worker or lose its result. + +## 1. Validate the target before spending generation credits + +The playhead, bounding box, still-image segmentation, object description, and user request must refer to the same target. + +If the selected object and requested object conflict, generation stops before calling a video provider. The chat asks the user to move the playhead or select the correct object. It must not warn about a mismatch and continue generating anyway. + +SAM is useful here for a precise still-image target mask. A bounding-box crop remains the fallback. + +## 2. Classify how the edit affects time + +Prompt planning must classify every edit into one of three behaviors: + +| Behavior | Example | Range rule | +| --- | --- | --- | +| Temporary | "Make him blink" | Use a local core with handles and rejoin the original motion. | +| Persistent state | "Make this car green" | Continue while the target remains visible. Do not return to its old state while it is still present. | +| Future-changing motion | "Make the man run" | Continue until the generated motion can naturally rejoin the source, or until the target leaves the shot. | + +The planner must return the behavior, confidence, reason, requested target, and a provider-ready edit instruction in one structured response. + +This classification prevents impossible timelines such as a green car suddenly returning to red or a running person teleporting back to an old walking position. + +## 3. Resolve the range from the playhead outward + +Range search starts at the selected playhead and expands only as far as the request requires. + +- temporary edits stop at a natural local action boundary +- persistent edits follow the target until it disappears, the shot changes, or the video ends +- future-changing motion continues until source and generated motion can plausibly rejoin +- an explicit user range overrides automatic scope when it is valid + +Object detection and tracking provide occurrence evidence. They must not scan the entire video when a local action is sufficient. + +If the resolved range exceeds the supported limit, finish range analysis first, then stop before video generation and show the product limit message. Long supported ranges are divided into bounded chunks only after the complete occurrence is known. + +## 4. Send strict, locked handles + +The provider receives the edit range plus source context on both sides whenever both sides exist. + +The generation prompt must include exact relative timestamps and a strict contract similar to: + +> The first `{pre_handle}` seconds and final `{post_handle}` seconds are locked source handles. Keep their visual content identical to the source: objects, colour, lighting, pose, camera, background, and motion. Do not introduce cuts, fades, freezes, or appearing or disappearing objects. Inside the editable interval, begin from the same pose and motion, perform the requested change continuously frame by frame, and leave in a state that can continue naturally. + +The task-specific prompt must also name what may change and what must remain untouched. For example, a colour edit should identify the exact surface, required colour, and protected windows, wheels, reflections, road, and nearby objects. + +Prompting is a constraint, not proof. Handle preservation is verified after generation. + +## 5. Keep the full generated context + +The generated handles must not be discarded before editing the timeline. They provide the overlap needed to find safe splice points. + +For an entry seam, compare source and generated frames throughout the pre-handle. For an exit seam, do the same throughout the post-handle. The accepted media range begins and ends at the selected matching frames rather than always at the nominal edit-core boundaries. + +Persistent edits need an exit handle only after the target leaves view or reaches a state where returning to source is semantically valid. + +## 6. Choose seams by frame and motion matching + +Seam selection is deterministic. It does not rely only on a visual-language model. + +For each possible cut frame in a handle, measure: + +- full-frame visual difference +- background difference outside the target +- target position and shape difference +- camera and optical-motion difference across neighboring frames +- temporal direction, so the cut does not reverse or freeze motion + +Choose the lowest-cost valid entry and exit pair. Require a minimum safety threshold. Save the chosen timestamps, scores, and rejected alternatives in job telemetry. + +If no safe pair exists, keep the candidate visible but disable Apply with the seam evidence. That evidence may drive the single corrective retry. + +An opacity dissolve can create ghosts and false motion, so it is not the default solution. A separately generated transition adds another model inconsistency and two more seams, so it is not part of this flow. + +## 7. Review the candidate while the user can see it + +As soon as the first provider result is stored, stream it to the chat as `First pass`. Do not wait for semantic scoring or seam analysis before showing it. + +Two reviews then run: + +### Semantic review + +A visual model checks chronological full frames and target crops against explicit questions: + +- was the correct target changed? +- was the requested property or action produced? +- did the change spill outside the target? +- did unrelated content change? + +Semantic review writes a short, concrete explanation. It helps decide whether to retry, but it does not hide the candidate or override the user's visual judgment. + +### Structural review + +Deterministic media checks verify duration, decoding, frame order, audio, handle preservation, and entry and exit seam quality. A structural failure disables Apply immediately and displays its reason beside the candidate. + +## 8. Allow one visible, evidence-driven retry + +A retry is allowed only when the first candidate has a clear, high-confidence failure that can be expressed as a concrete correction. + +Example: + +```text +First pass missed the requested colour: the car became white instead of green. +Retrying with: change only the painted body panels to saturated green; preserve all other surfaces. +[Stop retry] +``` + +The first candidate remains visible. Before dispatching the retry, the client receives the review and a short cancellation window with `Retry now` and `Stop retry` actions. + +If the provider call has already started, cancellation is best effort: stop polling and downstream processing and discard a late result. The UI must not imply that already-spent provider credits can always be recovered. + +The retry must: + +- use the same provider +- use the same resolved range and locked handles +- append the precise review evidence to the prompt +- create a separate candidate record instead of overwriting the first result +- never trigger another automatic retry + +## 9. Use SAM only where it provides real evidence + +The lightweight SAM service provides still-image segmentation. That is useful for: + +- validating the selected target +- making a clean target crop for prompt grounding +- weighting seam measurements toward background preservation +- giving semantic review a precise target crop + +The normal Wan source-video path does not gain full video-mask control from that still mask. Per-frame compositing is enabled only when a real SAM2 video tracker is available and tracking confidence passes its safety threshold. Otherwise the system must describe the fallback honestly and continue with bbox and frame evidence. + +## 10. Stream useful progress + +Chat should show persisted stage messages such as: + +- checking the selected object +- deciding whether the change is temporary or persistent +- following the target from the playhead +- preparing locked context +- generating first pass with Wan +- first pass ready +- checking requested change +- matching entry and exit frames +- first pass missed the requested colour +- retry begins in 10 seconds — Stop retry +- corrected pass ready +- ready to Apply, or Apply blocked because no safe seam was found + +Repeated generic messages such as `Creating edit` are not sufficient. + +## 11. Preserve audio and timeline state + +Generated video is conformed to the source frame rate, dimensions, duration, and audio layout. Original audio remains authoritative unless the request explicitly changes sound. + +Apply commits the frame-matched generated range into the EDL. A follow-up edit uses the current applied timeline as its source, so it cannot silently revert earlier edits. + +Candidate records, review evidence, selected seams, retry state, and final Apply state survive refresh and navigation. + +## Implementation sequence + +1. Remove the experimental default opacity seam blend before shipping this work. +2. Add the three-way temporal behavior to prompt planning and range resolution. +3. Store the complete generated context and candidate attempts separately. +4. Implement frame and motion seam search across both handles. +5. Change acceptance to use the selected media timestamps. +6. Stream the first candidate before review finishes. +7. Add semantic and structural review cards with correct Apply state. +8. Add the retry grace period, cancel endpoint, and one-retry state machine. +9. Connect SAM crops to target validation, semantic review, and seam weighting. +10. Verify audio, refresh recovery, follow-up edits, Compare, export, and Apply end to end. + +## Verification gates + +The work is ready to merge only when all of these pass: + +- unit tests for all three temporal behaviors and range rules +- synthetic seam tests with known best entry and exit frames +- motion tests that reject freeze, reversal, and teleport seams +- integration test proving the first candidate appears before review completes +- integration test proving Stop retry prevents a second dispatch +- integration test proving only one targeted retry can run +- integration test proving both candidates remain stored +- integration test proving Apply uses selected seam timestamps +- integration test proving a follow-up edit uses the applied timeline +- browser test covering navigation away, refresh, return, Compare, Apply, and export +- audio test proving original sound remains synchronized +- production build, backend suite, frontend suite, and CLI suite + +Before enabling the new acceptance path in production, run a small fixed evaluation set containing colour changes, appearance changes, temporary actions, persistent state edits, trajectory-changing motion, target mismatches, and sequential edits. Record generation time, provider calls, semantic success, seam score, Apply success, and human review. From 65c4b6c018f4d27a8c3e26e05a354560d17c972f Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 11:48:27 +0530 Subject: [PATCH 02/14] fix: stabilize edited clip playback --- apps/web/src/components/editor-port/CompareOverlay.tsx | 5 ++++- apps/web/src/components/editor-port/Preview.tsx | 7 +++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/apps/web/src/components/editor-port/CompareOverlay.tsx b/apps/web/src/components/editor-port/CompareOverlay.tsx index a62e5c9..fab5a8f 100644 --- a/apps/web/src/components/editor-port/CompareOverlay.tsx +++ b/apps/web/src/components/editor-port/CompareOverlay.tsx @@ -88,9 +88,12 @@ export function CompareOverlay({ onClose }: { onClose: () => void }) { const wantedTime = sourceTimeFor(clip, hit.offsetInClip); if (editedClipIdRef.current !== clip.id) { editedClipIdRef.current = clip.id; + const seekWhenReady = () => { + editedVideo.currentTime = wantedTime; + }; + editedVideo.addEventListener("loadedmetadata", seekWhenReady, { once: true }); editedVideo.src = clip.url; editedVideo.load(); - editedVideo.currentTime = wantedTime; } else if (force || Math.abs(editedVideo.currentTime - wantedTime) > 0.12) { editedVideo.currentTime = wantedTime; } diff --git a/apps/web/src/components/editor-port/Preview.tsx b/apps/web/src/components/editor-port/Preview.tsx index dba04c6..43357b3 100644 --- a/apps/web/src/components/editor-port/Preview.tsx +++ b/apps/web/src/components/editor-port/Preview.tsx @@ -195,11 +195,14 @@ export function Preview() { }; v.addEventListener("loadedmetadata", onLoaded, { once: true }); v.addEventListener("seeked", onSeeked, { once: true }); + // Do not drop the freeze canvas merely because playhead state updates + // while the replacement clip is decoding. That was exposing a black, + // glitchy frame at every edited/source boundary. + const fallbackRelease = window.setTimeout(releaseFreeze, 1500); return () => { v.removeEventListener("loadedmetadata", onLoaded); v.removeEventListener("seeked", onSeeked); - // safety: if effect re-runs before seeked fires, release the freeze - releaseFreeze(); + window.clearTimeout(fallbackRelease); }; } From 4665451e4fc97b5bf4bf8b913286a364ef09d67f Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 11:48:53 +0530 Subject: [PATCH 03/14] fix: keep quality retry on one provider --- apps/api/app/services/generation_quality.py | 35 +++------------------ apps/api/tests/test_generation_quality.py | 10 +++--- 2 files changed, 10 insertions(+), 35 deletions(-) diff --git a/apps/api/app/services/generation_quality.py b/apps/api/app/services/generation_quality.py index 5400a9d..8a0cd58 100644 --- a/apps/api/app/services/generation_quality.py +++ b/apps/api/app/services/generation_quality.py @@ -11,7 +11,7 @@ from app.services.continuity_validator import ContinuityReport -MAX_GENERATION_ATTEMPTS = 3 +MAX_GENERATION_ATTEMPTS = 2 MAX_GENERATED_SECONDS = 30.0 CONTINUITY_UNAVAILABLE = "continuity validation unavailable" @@ -134,35 +134,10 @@ def decide_generation_quality( if not can_generate_again: return GenerationQualityDecision(GenerationQualityAction.HARD_FAIL, evidence) - current_capabilities = VIDEO_PROVIDER_CAPABILITIES.get(current_provider) - fallback = ( - select_fallback_provider(current_provider, duration) - if source_video_available - else None - ) - provider_limited = ( - generation_error is not None - or current_capabilities is None - or not current_capabilities.source_video_edit - ) - if provider_limited and not fallback_used and fallback: - return GenerationQualityDecision( - GenerationQualityAction.PROVIDER_FALLBACK, - evidence, - fallback, - ) - if attempts == 1: - return GenerationQualityDecision( - GenerationQualityAction.CORRECTIVE_RETRY, - evidence, - ) - if not fallback_used and fallback: - return GenerationQualityDecision( - GenerationQualityAction.PROVIDER_FALLBACK, - evidence, - fallback, - ) - return GenerationQualityDecision(GenerationQualityAction.HARD_FAIL, evidence) + # A provider switch is not a seam repair. It historically spent another + # paid render and replaced Wan with a weaker result. Retry this provider + # once with concrete evidence, then fail closed. + return GenerationQualityDecision(GenerationQualityAction.CORRECTIVE_RETRY, evidence) def corrective_prompt(evidence: tuple[str, ...]) -> str: diff --git a/apps/api/tests/test_generation_quality.py b/apps/api/tests/test_generation_quality.py index 9fef6c9..7422b72 100644 --- a/apps/api/tests/test_generation_quality.py +++ b/apps/api/tests/test_generation_quality.py @@ -83,7 +83,7 @@ def test_first_source_edit_failure_gets_evidence_driven_retry(): assert "exit_subject_motion_jump" in corrective_prompt(decision.evidence) -def test_image_provider_failure_routes_to_source_video_fallback(): +def test_failed_source_edit_never_switches_provider_automatically(): decision = decide_generation_quality( score={"visual_coherence": 8, "prompt_adherence": 8}, continuity=_failed_continuity(), @@ -94,11 +94,11 @@ def test_image_provider_failure_routes_to_source_video_fallback(): fallback_used=False, source_video_available=True, ) - assert decision.action == GenerationQualityAction.PROVIDER_FALLBACK - assert decision.next_provider == "wan" + assert decision.action == GenerationQualityAction.CORRECTIVE_RETRY + assert decision.next_provider is None -def test_retry_then_fallback_is_capped_by_generated_seconds(): +def test_corrective_retry_is_capped_by_generated_seconds(): decision = decide_generation_quality( score={"visual_coherence": 4, "prompt_adherence": 8}, continuity=_failed_continuity(), @@ -112,7 +112,7 @@ def test_retry_then_fallback_is_capped_by_generated_seconds(): assert decision.action == GenerationQualityAction.HARD_FAIL -def test_provider_fallback_respects_full_context_limit(): +def test_fallback_helper_remains_duration_aware_for_explicit_use(): assert select_fallback_provider("veo", 8.0) == "wan" assert select_fallback_provider("wan", 12.0) == "happyhorse" assert select_fallback_provider("happyhorse", 12.0) is None From 9ba673d0714bedb67d95f7fe268469cc9cadae04 Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 11:54:10 +0530 Subject: [PATCH 04/14] feat: classify edit behavior over time --- apps/api/app/ai/prompts/edit_plan.txt | 18 +++++++++---- apps/api/app/ai/services/_stubs.py | 1 + apps/api/app/schemas/edit_plan.py | 39 +++++++++++++++++++++++++++ apps/api/app/services/edit_scope.py | 23 +++++++++++++--- apps/api/app/services/local_range.py | 19 +++++++------ apps/api/tests/test_edit_scope.py | 11 ++++++++ apps/api/tests/test_local_range.py | 26 ++++++++++++++++++ apps/web/src/lib/api.ts | 3 ++- 8 files changed, 123 insertions(+), 17 deletions(-) diff --git a/apps/api/app/ai/prompts/edit_plan.txt b/apps/api/app/ai/prompts/edit_plan.txt index 91b41f2..a88836e 100644 --- a/apps/api/app/ai/prompts/edit_plan.txt +++ b/apps/api/app/ai/prompts/edit_plan.txt @@ -13,7 +13,8 @@ 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", "action_phases": ["ordered phase names for motion; empty for a state change"], "estimated_action_seconds": 3.5, @@ -40,10 +41,17 @@ RULES - 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 diff --git a/apps/api/app/ai/services/_stubs.py b/apps/api/app/ai/services/_stubs.py index 1c10148..48151d4 100644 --- a/apps/api/app/ai/services/_stubs.py +++ b/apps/api/app/ai/services/_stubs.py @@ -36,6 +36,7 @@ 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, "action_phases": intent.action_phases, "estimated_action_seconds": intent.estimated_action_seconds, diff --git a/apps/api/app/schemas/edit_plan.py b/apps/api/app/schemas/edit_plan.py index f71bdcd..264038e 100644 --- a/apps/api/app/schemas/edit_plan.py +++ b/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,6 +42,7 @@ class SemanticEditDecision(BaseModel): scope: EditScope change_type: ChangeType duration_policy: DurationPolicy + temporal_behavior: TemporalBehavior target_description: 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) @@ -43,6 +50,21 @@ class SemanticEditDecision(BaseModel): 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 +77,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 +93,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 +102,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) diff --git a/apps/api/app/services/edit_scope.py b/apps/api/app/services/edit_scope.py index c9d14b5..9e291e1 100644 --- a/apps/api/app/services/edit_scope.py +++ b/apps/api/app/services/edit_scope.py @@ -21,6 +21,11 @@ r"clap|nod|bow|fall|leap)(s|ed|ing)?\b", re.IGNORECASE, ) +_TRAJECTORY_RE = re.compile( + r"\b(run|walk|fly|drive|move|leave|approach|follow|chase|crawl|swim)" + r"(s|ed|ing)?\b", + re.IGNORECASE, +) _REMOVAL_RE = re.compile(r"\b(remove|erase|delete|make .+ disappear)\b", re.IGNORECASE) _REPLACEMENT_RE = re.compile(r"\b(replace|swap|turn .+ into)\b", re.IGNORECASE) _APPEARANCE_RE = re.compile( @@ -104,6 +109,12 @@ def plan_prompt_intent( change_type = _change_type(prompt) phases, action_seconds, recovery = _motion_contract(prompt) persistent = bool(_PERSISTENT_RE.search(prompt)) + if change_type == "motion" and _TRAJECTORY_RE.search(prompt): + temporal_behavior = "future_changing_motion" + elif change_type == "motion": + temporal_behavior = "temporary" + else: + temporal_behavior = "persistent_state" preservation = ["preserve unedited subjects", "preserve camera and background"] if persistent: if "apply change for complete visible occurrence" not in preservation: @@ -112,8 +123,10 @@ def plan_prompt_intent( duration_policy = "all_occurrences" elif scope == "explicit_range": duration_policy = "explicit_range" - elif change_type == "motion": + elif temporal_behavior == "temporary": duration_policy = "bounded_action" + elif temporal_behavior == "future_changing_motion": + duration_policy = "trajectory_continuation" else: # State changes should not visibly revert while the selected target # remains in the same continuous appearance. @@ -125,6 +138,7 @@ def plan_prompt_intent( scope=scope, change_type=change_type, duration_policy=duration_policy, + temporal_behavior=temporal_behavior, action_phases=phases, estimated_action_seconds=action_seconds, requires_recovery_motion=recovery, @@ -132,14 +146,17 @@ def plan_prompt_intent( ) global_search = intent.scope == "all_occurrences" - persistent_local = intent.scope == "local" and bool(_PERSISTENT_RE.search(intent.raw_prompt)) + continuous_local = intent.scope == "local" and intent.temporal_behavior in { + "persistent_state", + "future_changing_motion", + } if global_search: mode = "coarse_global_then_dense_candidates" tracking_seconds = None elif intent.scope == "selected_occurrences": mode = "selected_occurrences_only" tracking_seconds = None - elif persistent_local: + elif continuous_local: mode = "complete_local_occurrence" tracking_seconds = None elif intent.scope == "explicit_range": diff --git a/apps/api/app/services/local_range.py b/apps/api/app/services/local_range.py index 3eb6a0d..cb61f8b 100644 --- a/apps/api/app/services/local_range.py +++ b/apps/api/app/services/local_range.py @@ -44,8 +44,11 @@ def __init__(self, duration: float, limit: float = MAX_CONTINUOUS_EDIT_SECONDS): ) -def _persistent_change(intent: EditIntent) -> bool: - return intent.duration_policy == "continuous_occurrence" +def _continuous_change(intent: EditIntent) -> bool: + return intent.duration_policy in { + "continuous_occurrence", + "trajectory_continuation", + } def analysis_window( @@ -61,7 +64,7 @@ def analysis_window( raise ValueError("active source clip must have positive duration") if seed_ts < source_start - 1e-3 or seed_ts > upper + 1e-3: raise ValueError("selection is outside the active source clip") - if _persistent_change(intent): + if _continuous_change(intent): radius = INITIAL_CONTINUOUS_RADIUS_SECONDS else: radius = intent.estimated_action_seconds / 2 + HANDLE_SECONDS + LOCAL_MARGIN_SECONDS @@ -118,7 +121,7 @@ def resolve_window_from_evidence( core_end = min(occurrence_end, explicit_core.end_ts) if core_end <= core_start: raise ValueError("explicit range does not overlap selected occurrence") - elif _persistent_change(intent): + elif _continuous_change(intent): core_start, core_end = occurrence_start, occurrence_end if tracking_reached_budget: warnings.append("target remains visible at local tracking budget; adjust range to continue") @@ -256,7 +259,7 @@ async def resolve_local_range( source_start=source_start, source_end=bounded_end, ) - analysis_fps = CONTINUOUS_ANALYSIS_FPS if _persistent_change(intent) else ANALYSIS_FPS + analysis_fps = CONTINUOUS_ANALYSIS_FPS if _continuous_change(intent) else ANALYSIS_FPS total_frames_inspected = 0 track: sam.TrackResult | None = None shot_start = start @@ -311,7 +314,7 @@ async def resolve_local_range( include_masks=False, ) except Exception as exc: - if _persistent_change(intent): + if _continuous_change(intent): raise ValueError( "could not reliably track the selected subject; please choose a tighter box" ) from exc @@ -359,7 +362,7 @@ async def resolve_local_range( "confidence": frame.get("confidence"), "bbox": frame.get("bbox") or selection.bbox_json, } - if not _persistent_change(intent): + if not _continuous_change(intent): break tolerance = 1.1 / analysis_fps @@ -426,7 +429,7 @@ async def resolve_local_range( source_end=bounded_end, ) if ( - _persistent_change(intent) + _continuous_change(intent) and resolution.edit_core.duration > MAX_CONTINUOUS_EDIT_SECONDS + 0.05 ): raise EditWindowLimitError(resolution.edit_core.duration) diff --git a/apps/api/tests/test_edit_scope.py b/apps/api/tests/test_edit_scope.py index c5cb8f6..eea20b5 100644 --- a/apps/api/tests/test_edit_scope.py +++ b/apps/api/tests/test_edit_scope.py @@ -48,6 +48,7 @@ def test_state_change_defaults_to_current_continuous_occurrence(): assert result.intent.scope == "local" assert result.intent.duration_policy == "continuous_occurrence" + assert result.intent.temporal_behavior == "persistent_state" def test_action_defaults_to_bounded_local_window(): @@ -55,3 +56,13 @@ def test_action_defaults_to_bounded_local_window(): assert result.intent.scope == "local" assert result.intent.duration_policy == "bounded_action" + assert result.intent.temporal_behavior == "temporary" + + +def test_trajectory_change_covers_the_current_occurrence(): + result = plan_prompt_intent("make this man run") + + assert result.intent.scope == "local" + assert result.intent.duration_policy == "trajectory_continuation" + assert result.intent.temporal_behavior == "future_changing_motion" + assert result.estimate.analysis_mode == "complete_local_occurrence" diff --git a/apps/api/tests/test_local_range.py b/apps/api/tests/test_local_range.py index 4a7a987..ea65709 100644 --- a/apps/api/tests/test_local_range.py +++ b/apps/api/tests/test_local_range.py @@ -49,6 +49,32 @@ def test_range_uses_two_sided_handles_without_covering_whole_occurrence(): assert result.generation_context.end_ts == pytest.approx(12.5) +def test_trajectory_change_covers_occurrence_instead_of_old_local_position(): + intent = EditIntent( + raw_prompt="make this person run", + change_type="motion", + duration_policy="trajectory_continuation", + temporal_behavior="future_changing_motion", + ) + result = resolve_window_from_evidence( + intent=intent, + seed_ts=10.0, + duration=30.0, + analysis_start=7.0, + analysis_end=13.0, + shot_start=4.0, + shot_end=18.0, + tracked_start=5.0, + tracked_end=17.0, + frames_inspected=25, + ) + + assert result.edit_core.start_ts == pytest.approx(5.0) + assert result.edit_core.end_ts == pytest.approx(17.0) + assert result.generation_context.start_ts == pytest.approx(4.25) + assert result.generation_context.end_ts == pytest.approx(17.75) + + def test_range_never_crosses_active_source_clip(): intent = EditIntent( raw_prompt="make this person jump", diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index 2b040d7..de73410 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -197,7 +197,8 @@ export type EditPlanResp = { intent: { raw_prompt: string; change_type: "appearance" | "removal" | "replacement" | "motion" | "scene"; - duration_policy: "bounded_action" | "continuous_occurrence" | "explicit_range" | "all_occurrences"; + duration_policy: "bounded_action" | "continuous_occurrence" | "trajectory_continuation" | "explicit_range" | "all_occurrences"; + temporal_behavior: "temporary" | "persistent_state" | "future_changing_motion"; action_phases: string[]; estimated_action_seconds: number; requires_recovery_motion: boolean; From 5f3df1885edb307241e250e9057a6c5ffaae72de Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 11:59:33 +0530 Subject: [PATCH 05/14] refactor: share frame seam scoring --- apps/api/app/services/global_seam.py | 81 +++---------------------- apps/api/app/services/seam_matching.py | 84 ++++++++++++++++++++++++++ apps/api/tests/test_seam_matching.py | 45 ++++++++++++++ 3 files changed, 137 insertions(+), 73 deletions(-) create mode 100644 apps/api/app/services/seam_matching.py create mode 100644 apps/api/tests/test_seam_matching.py diff --git a/apps/api/app/services/global_seam.py b/apps/api/app/services/global_seam.py index 3c1a69c..6e8dd65 100644 --- a/apps/api/app/services/global_seam.py +++ b/apps/api/app/services/global_seam.py @@ -19,8 +19,15 @@ from app.services.global_chunk_execution import target_bbox +from app.services.seam_matching import ( + SeamChoice, + SeamFrames, + seam_score, + select_best_seam, +) + + SEAM_SAMPLES = 9 -MAX_SEAM_SCORE = 0.22 class GlobalSeamError(ValueError): @@ -29,29 +36,6 @@ def __init__(self, message: str, *, retry_chunk_index: int): self.retry_chunk_index = retry_chunk_index -@dataclass(frozen=True, slots=True) -class SeamFrames: - timestamp: float - left_before: np.ndarray - left_at: np.ndarray - right_at: np.ndarray - right_after: np.ndarray - - -@dataclass(frozen=True, slots=True) -class SeamChoice: - timestamp: float - score: float - samples: int - - def metadata(self) -> dict[str, float | int]: - return { - "timestamp": self.timestamp, - "score": self.score, - "samples": self.samples, - } - - @dataclass(frozen=True, slots=True) class AssemblyResult: output_url: str @@ -59,55 +43,6 @@ class AssemblyResult: continuity: dict -def _bbox_mask(shape: tuple[int, ...], bbox: dict[str, float], *, invert: bool) -> np.ndarray: - height, width = shape[:2] - left = max(0, round(float(bbox["x"]) * width)) - top = max(0, round(float(bbox["y"]) * height)) - right = min(width, round((float(bbox["x"]) + float(bbox["w"])) * width)) - bottom = min(height, round((float(bbox["y"]) + float(bbox["h"])) * height)) - mask = np.zeros((height, width), dtype=bool) - mask[top:bottom, left:right] = True - return ~mask if invert else mask - - -def _mean_delta(left: np.ndarray, right: np.ndarray, mask: np.ndarray) -> float: - if left.shape != right.shape: - right = cv2.resize(right, (left.shape[1], left.shape[0])) - selected = cv2.absdiff(left, right).astype(np.float32)[mask] / 255.0 - return float(selected.mean()) if selected.size else 0.0 - - -def seam_score(sample: SeamFrames, bbox: dict[str, float]) -> float: - target = _bbox_mask(sample.left_at.shape, bbox, invert=False) - background = _bbox_mask(sample.left_at.shape, bbox, invert=True) - appearance_background = _mean_delta(sample.left_at, sample.right_at, background) - appearance_target = _mean_delta(sample.left_at, sample.right_at, target) - left_motion = cv2.absdiff(sample.left_before, sample.left_at) - right_motion = cv2.absdiff(sample.right_at, sample.right_after) - motion_delta = _mean_delta(left_motion, right_motion, np.ones_like(target)) - return ( - 0.50 * appearance_background - + 0.20 * appearance_target - + 0.30 * motion_delta - ) - - -def select_best_seam( - samples: list[SeamFrames], - *, - bbox: dict[str, float], -) -> SeamChoice: - if not samples: - raise ValueError("global chunk overlap has no seam samples") - scored = [(seam_score(sample, bbox), sample.timestamp) for sample in samples] - score, timestamp = min(scored) - if score > MAX_SEAM_SCORE: - raise ValueError( - f"global chunk overlap failed seam validation ({score:.3f} > {MAX_SEAM_SCORE:.3f})" - ) - return SeamChoice(timestamp=timestamp, score=score, samples=len(samples)) - - def _read_frame(path: Path, timestamp: float) -> np.ndarray: capture = cv2.VideoCapture(str(path)) if not capture.isOpened(): diff --git a/apps/api/app/services/seam_matching.py b/apps/api/app/services/seam_matching.py new file mode 100644 index 0000000..6bf437a --- /dev/null +++ b/apps/api/app/services/seam_matching.py @@ -0,0 +1,84 @@ +"""Shared visual and motion scoring for hard-cut seam selection.""" +from __future__ import annotations + +from dataclasses import dataclass + +import cv2 +import numpy as np + + +MAX_SEAM_SCORE = 0.22 + + +@dataclass(frozen=True, slots=True) +class SeamFrames: + timestamp: float + left_before: np.ndarray + left_at: np.ndarray + right_at: np.ndarray + right_after: np.ndarray + + +@dataclass(frozen=True, slots=True) +class SeamChoice: + timestamp: float + score: float + samples: int + + def metadata(self) -> dict[str, float | int]: + return { + "timestamp": self.timestamp, + "score": self.score, + "samples": self.samples, + } + + +def _bbox_mask(shape: tuple[int, ...], bbox: dict[str, float], *, invert: bool) -> np.ndarray: + height, width = shape[:2] + left = max(0, round(float(bbox["x"]) * width)) + top = max(0, round(float(bbox["y"]) * height)) + right = min(width, round((float(bbox["x"]) + float(bbox["w"])) * width)) + bottom = min(height, round((float(bbox["y"]) + float(bbox["h"])) * height)) + mask = np.zeros((height, width), dtype=bool) + mask[top:bottom, left:right] = True + return ~mask if invert else mask + + +def _mean_delta(left: np.ndarray, right: np.ndarray, mask: np.ndarray) -> float: + if left.shape != right.shape: + right = cv2.resize(right, (left.shape[1], left.shape[0])) + selected = cv2.absdiff(left, right).astype(np.float32)[mask] / 255.0 + return float(selected.mean()) if selected.size else 0.0 + + +def seam_score(sample: SeamFrames, bbox: dict[str, float]) -> float: + """Weight protected background and cross-cut motion above target appearance.""" + target = _bbox_mask(sample.left_at.shape, bbox, invert=False) + background = _bbox_mask(sample.left_at.shape, bbox, invert=True) + appearance_background = _mean_delta(sample.left_at, sample.right_at, background) + appearance_target = _mean_delta(sample.left_at, sample.right_at, target) + left_motion = cv2.absdiff(sample.left_before, sample.left_at) + right_motion = cv2.absdiff(sample.right_at, sample.right_after) + motion_delta = _mean_delta(left_motion, right_motion, np.ones_like(target)) + return 0.50 * appearance_background + 0.20 * appearance_target + 0.30 * motion_delta + + +def select_best_seam( + samples: list[SeamFrames], + *, + bbox: dict[str, float], + prefer_late: bool = False, + max_score: float = MAX_SEAM_SCORE, +) -> SeamChoice: + if not samples: + raise ValueError("overlap has no seam samples") + scored = [(seam_score(sample, bbox), sample.timestamp) for sample in samples] + score, timestamp = min( + scored, + key=lambda item: (item[0], -item[1] if prefer_late else item[1]), + ) + if score > max_score: + raise ValueError( + f"overlap failed seam validation ({score:.3f} > {max_score:.3f})" + ) + return SeamChoice(timestamp=timestamp, score=score, samples=len(samples)) diff --git a/apps/api/tests/test_seam_matching.py b/apps/api/tests/test_seam_matching.py new file mode 100644 index 0000000..97017fd --- /dev/null +++ b/apps/api/tests/test_seam_matching.py @@ -0,0 +1,45 @@ +import numpy as np +import pytest + +from app.services.seam_matching import SeamFrames, select_best_seam + + +BBOX = {"x": 0.25, "y": 0.2, "w": 0.3, "h": 0.6} + + +def _frame(value: int) -> np.ndarray: + return np.full((32, 48, 3), value, dtype=np.uint8) + + +def _sample(timestamp: float, left: int, right: int) -> SeamFrames: + return SeamFrames( + timestamp=timestamp, + left_before=_frame(left), + left_at=_frame(left), + right_at=_frame(right), + right_after=_frame(right), + ) + + +def test_entry_prefers_latest_equally_safe_frame(): + choice = select_best_seam( + [_sample(0.25, 40, 40), _sample(0.50, 40, 40)], + bbox=BBOX, + prefer_late=True, + ) + + assert choice.timestamp == pytest.approx(0.50) + + +def test_exit_prefers_earliest_equally_safe_frame(): + choice = select_best_seam( + [_sample(3.25, 40, 40), _sample(3.50, 40, 40)], + bbox=BBOX, + ) + + assert choice.timestamp == pytest.approx(3.25) + + +def test_unrelated_frames_are_rejected_without_a_fade(): + with pytest.raises(ValueError, match="failed seam validation"): + select_best_seam([_sample(0.5, 0, 255)], bbox=BBOX) From 61f1e72e6e1edd19b9b02245791cae6bd215096e Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 12:00:59 +0530 Subject: [PATCH 06/14] feat: select local seams from matching frames --- apps/api/app/services/local_seam.py | 193 +++++++++++++++++++++++++ apps/api/app/services/seam_matching.py | 19 ++- apps/api/tests/test_local_seam.py | 46 ++++++ 3 files changed, 254 insertions(+), 4 deletions(-) create mode 100644 apps/api/app/services/local_seam.py create mode 100644 apps/api/tests/test_local_seam.py diff --git a/apps/api/app/services/local_seam.py b/apps/api/app/services/local_seam.py new file mode 100644 index 0000000..5778611 --- /dev/null +++ b/apps/api/app/services/local_seam.py @@ -0,0 +1,193 @@ +"""Find source-to-generated hard cuts inside protected context handles.""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from pathlib import Path + +import cv2 +import numpy as np + +from app.services.generation_window import GenerationWindow +from app.services.seam_matching import ( + MAX_SEAM_SCORE, + SeamChoice, + SeamFrames, + rank_best_seam, +) + + +LOCAL_SEAM_SAMPLES = 9 + + +@dataclass(frozen=True, slots=True) +class LocalSeamSelection: + entry: SeamChoice | None + exit: SeamChoice | None + context_start: float + context_duration: float + max_score: float = MAX_SEAM_SCORE + + @property + def passed(self) -> bool: + return all( + choice is None or choice.score <= self.max_score + for choice in (self.entry, self.exit) + ) + + @property + def media_start(self) -> float: + return self.entry.timestamp if self.entry is not None else 0.0 + + @property + def media_end(self) -> float: + return self.exit.timestamp if self.exit is not None else self.context_duration + + @property + def timeline_start(self) -> float: + return self.context_start + self.media_start + + @property + def timeline_end(self) -> float: + return self.context_start + self.media_end + + @property + def issues(self) -> tuple[str, ...]: + output = [] + for boundary, choice in (("entry", self.entry), ("exit", self.exit)): + if choice is not None and choice.score > self.max_score: + output.append( + f"{boundary}_frame_match_score at {boundary}: " + f"measured {choice.score:.3f}, limit {self.max_score:.3f}" + ) + return tuple(output) + + def metadata(self) -> dict: + def boundary(choice: SeamChoice | None) -> dict | None: + if choice is None: + return None + return { + "media_timestamp": choice.timestamp, + "source_timestamp": self.context_start + choice.timestamp, + "score": choice.score, + "samples": choice.samples, + "safe": choice.score <= self.max_score, + } + + return { + "passed": self.passed, + "max_score": self.max_score, + "entry": boundary(self.entry), + "exit": boundary(self.exit), + "media_start": self.media_start, + "media_end": self.media_end, + "timeline_start": self.timeline_start, + "timeline_end": self.timeline_end, + "issues": list(self.issues), + } + + +def select_local_seams( + *, + entry_samples: list[SeamFrames], + exit_samples: list[SeamFrames], + bbox: dict[str, float], + window: GenerationWindow, +) -> LocalSeamSelection: + entry = ( + rank_best_seam(entry_samples, bbox=bbox, prefer_late=True) + if entry_samples + else None + ) + exit = rank_best_seam(exit_samples, bbox=bbox) if exit_samples else None + return LocalSeamSelection( + entry=entry, + exit=exit, + context_start=window.context_start, + context_duration=window.context_duration, + ) + + +def _candidate_times(start: float, end: float, fps: float) -> list[float]: + step = max(1.0 / max(fps, 1.0), 0.04) + lower = start + step + upper = end - step + if upper < lower: + return [] + return np.linspace(lower, upper, LOCAL_SEAM_SAMPLES).tolist() + + +def _read(capture: cv2.VideoCapture, timestamp: float) -> np.ndarray: + capture.set(cv2.CAP_PROP_POS_MSEC, max(0.0, timestamp) * 1000.0) + ok, frame = capture.read() + if not ok or frame is None: + raise ValueError(f"could not read seam frame at {timestamp:.3f}s") + return frame + + +def _match_local_context_sync( + source_path: Path, + generated_path: Path, + window: GenerationWindow, + bbox: dict[str, float], +) -> LocalSeamSelection: + source = cv2.VideoCapture(str(source_path)) + generated = cv2.VideoCapture(str(generated_path)) + if not source.isOpened() or not generated.isOpened(): + source.release() + generated.release() + raise ValueError("could not open source or generated video for seam matching") + try: + fps = float(source.get(cv2.CAP_PROP_FPS) or generated.get(cv2.CAP_PROP_FPS) or 0.0) + if fps <= 0: + raise ValueError("seam matching requires a valid frame rate") + step = max(1.0 / fps, 0.04) + entry_samples = [ + SeamFrames( + timestamp=timestamp, + left_before=_read(source, timestamp - step), + left_at=_read(source, timestamp), + right_at=_read(generated, timestamp), + right_after=_read(generated, timestamp + step), + ) + for timestamp in _candidate_times(0.0, window.edit_start_offset, fps) + ] + exit_samples = [ + SeamFrames( + timestamp=timestamp, + left_before=_read(generated, timestamp - step), + left_at=_read(generated, timestamp), + right_at=_read(source, timestamp), + right_after=_read(source, timestamp + step), + ) + for timestamp in _candidate_times( + window.edit_end_offset, + window.context_duration, + fps, + ) + ] + finally: + source.release() + generated.release() + return select_local_seams( + entry_samples=entry_samples, + exit_samples=exit_samples, + bbox=bbox, + window=window, + ) + + +async def match_local_context( + *, + source_path: Path, + generated_path: Path, + window: GenerationWindow, + bbox: dict[str, float], +) -> LocalSeamSelection: + return await asyncio.to_thread( + _match_local_context_sync, + source_path, + generated_path, + window, + bbox, + ) diff --git a/apps/api/app/services/seam_matching.py b/apps/api/app/services/seam_matching.py index 6bf437a..a552956 100644 --- a/apps/api/app/services/seam_matching.py +++ b/apps/api/app/services/seam_matching.py @@ -70,6 +70,21 @@ def select_best_seam( prefer_late: bool = False, max_score: float = MAX_SEAM_SCORE, ) -> SeamChoice: + choice = rank_best_seam(samples, bbox=bbox, prefer_late=prefer_late) + if choice.score > max_score: + raise ValueError( + f"overlap failed seam validation ({choice.score:.3f} > {max_score:.3f})" + ) + return choice + + +def rank_best_seam( + samples: list[SeamFrames], + *, + bbox: dict[str, float], + prefer_late: bool = False, +) -> SeamChoice: + """Return the strongest candidate even when it is too weak to accept.""" if not samples: raise ValueError("overlap has no seam samples") scored = [(seam_score(sample, bbox), sample.timestamp) for sample in samples] @@ -77,8 +92,4 @@ def select_best_seam( scored, key=lambda item: (item[0], -item[1] if prefer_late else item[1]), ) - if score > max_score: - raise ValueError( - f"overlap failed seam validation ({score:.3f} > {max_score:.3f})" - ) return SeamChoice(timestamp=timestamp, score=score, samples=len(samples)) diff --git a/apps/api/tests/test_local_seam.py b/apps/api/tests/test_local_seam.py new file mode 100644 index 0000000..96a734c --- /dev/null +++ b/apps/api/tests/test_local_seam.py @@ -0,0 +1,46 @@ +import numpy as np +import pytest + +from app.services.generation_window import GenerationWindow +from app.services.local_seam import select_local_seams +from app.services.seam_matching import SeamFrames + + +BBOX = {"x": 0.25, "y": 0.2, "w": 0.3, "h": 0.6} +WINDOW = GenerationWindow(3.0, 5.0, 2.0, 6.0, adaptive=True) + + +def _frame(value: int) -> np.ndarray: + return np.full((24, 32, 3), value, dtype=np.uint8) + + +def _sample(timestamp: float, left: int, right: int) -> SeamFrames: + return SeamFrames(timestamp, _frame(left), _frame(left), _frame(right), _frame(right)) + + +def test_local_seams_keep_full_context_until_matching_cut_frames(): + selection = select_local_seams( + entry_samples=[_sample(0.25, 30, 31), _sample(0.75, 30, 30)], + exit_samples=[_sample(3.25, 30, 30), _sample(3.75, 30, 31)], + bbox=BBOX, + window=WINDOW, + ) + + assert selection.passed is True + assert selection.media_start == pytest.approx(0.75) + assert selection.media_end == pytest.approx(3.25) + assert selection.timeline_start == pytest.approx(2.75) + assert selection.timeline_end == pytest.approx(5.25) + + +def test_local_seams_report_unsafe_boundaries_without_blending(): + selection = select_local_seams( + entry_samples=[_sample(0.5, 0, 255)], + exit_samples=[_sample(3.5, 0, 255)], + bbox=BBOX, + window=WINDOW, + ) + + assert selection.passed is False + assert len(selection.issues) == 2 + assert "entry_frame_match_score" in selection.issues[0] From 5c23d234f8da3f9da0e902d924c06159f8fa4419 Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 12:04:03 +0530 Subject: [PATCH 07/14] feat: apply edits at frame-matched seams --- apps/api/app/api/routes/jobs.py | 1 + apps/api/app/schemas/job.py | 1 + apps/api/app/services/accepted_generation.py | 15 +++++ apps/api/app/services/local_seam.py | 35 +++++++++++ apps/api/app/workers/generate_job.py | 61 ++++++++++++++++--- apps/api/tests/test_accepted_generation.py | 17 ++++++ apps/api/tests/test_local_seam.py | 27 +++++++- .../src/features/reveal/GenerationReveal.tsx | 3 + apps/web/src/hooks/useAgentStream.ts | 2 + apps/web/src/lib/api.ts | 9 +++ 10 files changed, 163 insertions(+), 8 deletions(-) diff --git a/apps/api/app/api/routes/jobs.py b/apps/api/app/api/routes/jobs.py index 6cb3e63..cb971d3 100644 --- a/apps/api/app/api/routes/jobs.py +++ b/apps/api/app/api/routes/jobs.py @@ -35,6 +35,7 @@ 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"), diff --git a/apps/api/app/schemas/job.py b/apps/api/app/schemas/job.py index 39e43e6..f6b3346 100644 --- a/apps/api/app/schemas/job.py +++ b/apps/api/app/schemas/job.py @@ -34,6 +34,7 @@ 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 diff --git a/apps/api/app/services/accepted_generation.py b/apps/api/app/services/accepted_generation.py index ae604fe..df7ebd2 100644 --- a/apps/api/app/services/accepted_generation.py +++ b/apps/api/app/services/accepted_generation.py @@ -43,6 +43,21 @@ def accepted_generation_range(job: Job) -> AcceptedGenerationRange: committed = raw_committed if isinstance(raw_committed, dict) else {} committed_start = float(committed.get("start", core_start)) committed_end = float(committed.get("end", core_end)) + raw_seams = payload.get("selected_seams") + seams = raw_seams if isinstance(raw_seams, dict) else {} + if seams.get("passed") is True: + candidate_media_start = float(seams.get("media_start", media_start)) + candidate_media_end = float(seams.get("media_end", media_end)) + candidate_timeline_start = float(seams.get("timeline_start", core_start)) + candidate_timeline_end = float(seams.get("timeline_end", core_end)) + if ( + 0.0 <= candidate_media_start < candidate_media_end <= context_end - context_start + 0.05 + and context_start - 0.05 <= candidate_timeline_start < candidate_timeline_end <= context_end + 0.05 + ): + media_start = candidate_media_start + media_end = candidate_media_end + committed_start += candidate_timeline_start - core_start + committed_end += candidate_timeline_end - core_end return AcceptedGenerationRange( requested_start=requested_start, requested_end=requested_end, diff --git a/apps/api/app/services/local_seam.py b/apps/api/app/services/local_seam.py index 5778611..0d9919e 100644 --- a/apps/api/app/services/local_seam.py +++ b/apps/api/app/services/local_seam.py @@ -9,6 +9,7 @@ import numpy as np from app.services.generation_window import GenerationWindow +from app.services.continuity_validator import ContinuityIssue, ContinuityReport from app.services.seam_matching import ( MAX_SEAM_SCORE, SeamChoice, @@ -191,3 +192,37 @@ async def match_local_context( window, bbox, ) + + +def continuity_at_selected_seams( + base: ContinuityReport, + selection: LocalSeamSelection, +) -> ContinuityReport: + """Keep media-integrity failures and replace nominal cuts with chosen cuts.""" + issues = [ + issue + for issue in base.issues + if issue.code in {"duration_delta_s", "fps_delta_ratio", "frozen_tail"} + ] + metrics = dict(base.metrics) + for boundary, choice in (("entry", selection.entry), ("exit", selection.exit)): + if choice is None: + continue + code = f"{boundary}_frame_match_score" + metrics[code] = choice.score + if choice.score > selection.max_score: + issues.append( + ContinuityIssue( + code, + choice.score, + selection.max_score, + boundary, + ) + ) + return ContinuityReport( + passed=not issues, + metrics=metrics, + issues=issues, + sampled_frames=base.sampled_frames + + sum(choice.samples for choice in (selection.entry, selection.exit) if choice), + ) diff --git a/apps/api/app/workers/generate_job.py b/apps/api/app/workers/generate_job.py index 663a85d..2100bfc 100644 --- a/apps/api/app/workers/generate_job.py +++ b/apps/api/app/workers/generate_job.py @@ -47,6 +47,7 @@ from app.schemas.edit_plan import EditIntent from app.services import ffmpeg, job_events, storage from app.services.continuity_validator import ( + ContinuityIssue, ContinuityReport, validate_generated_continuity, ) @@ -64,6 +65,7 @@ ) from app.services.generation_telemetry import build_local_flow_telemetry from app.services.local_compositor import composite_generated_target +from app.services.local_seam import continuity_at_selected_seams, match_local_context from app.services.edit_prompt import planned_edit_prompt log = logging.getLogger("fiebatt.jobs.generate") @@ -985,9 +987,12 @@ async def score_variant(variant: Variant) -> dict | None: reference_target_path=(str(crop_path) if crop_path is not None else None), ) + validation_by_url: dict[str, tuple[ContinuityReport, dict | None]] = {} + async def validate_continuity(variant: Variant) -> ContinuityReport | None: if not variant.url: return None + seam_metadata: dict | None = None try: generated_path = await storage.path_from_url(variant.url) report = await validate_generated_continuity( @@ -996,6 +1001,48 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: window=generation_window, bbox=bbox, ) + if generation_window.adaptive: + try: + seam_selection = await match_local_context( + source_path=clip_path, + generated_path=generated_path, + window=generation_window, + bbox=bbox, + ) + except Exception as seam_exc: + log.exception("frame-matched seam selection failed") + report = ContinuityReport( + passed=False, + metrics={**report.metrics, "frame_matching_unavailable": 1.0}, + issues=[ + *report.issues, + ContinuityIssue( + "frame_matching_unavailable", + 1.0, + 0.0, + "clip", + ), + ], + sampled_frames=report.sampled_frames, + ) + await _emit( + job_id, + "seam_match_unavailable", + f"couldn't match safe cut frames: {seam_exc}", + ) + else: + report = continuity_at_selected_seams(report, seam_selection) + seam_metadata = seam_selection.metadata() + await _emit( + job_id, + "seam_match_done", + ( + "found safe source-to-edit cut frames" + if seam_selection.passed + else "no safe source-to-edit cut frames found" + ), + **seam_metadata, + ) except Exception as exc: log.exception("continuity validation failed") await _emit( @@ -1004,13 +1051,7 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: f"couldn't validate generated seams: {exc}", ) return None - async with AsyncSessionLocal() as db: - current_job = await db.get(Job, job_id) - if current_job is not None: - current_payload = dict(current_job.payload or {}) - current_payload["continuity_validation"] = report.metadata() - current_job.payload = current_payload - await db.commit() + validation_by_url[variant.url] = (report, seam_metadata) await _emit( job_id, "continuity_validation_done", @@ -1256,6 +1297,12 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: "selected_edit_mode": final_edit_mode, } ) + final_validation = validation_by_url.get(done[0].url or "") + if final_validation is not None: + final_report, final_seams = final_validation + current_payload["continuity_validation"] = final_report.metadata() + if final_seams is not None: + current_payload["selected_seams"] = final_seams local_flow_telemetry = build_local_flow_telemetry( payload=current_payload, window=generation_window, diff --git a/apps/api/tests/test_accepted_generation.py b/apps/api/tests/test_accepted_generation.py index 81fcc81..80c4624 100644 --- a/apps/api/tests/test_accepted_generation.py +++ b/apps/api/tests/test_accepted_generation.py @@ -44,6 +44,23 @@ def test_range_keeps_core_context_and_media_offsets_distinct(): assert job.payload["accepted_ranges"]["segment-1"]["media_start"] == 1.0 +def test_range_uses_safe_frame_matched_context_instead_of_nominal_core(): + job = _job() + job.payload["committed_timeline_range"] = {"start": 10.0, "end": 12.0} + job.payload["selected_seams"] = { + "passed": True, + "media_start": 0.75, + "media_end": 3.5, + "timeline_start": 2.75, + "timeline_end": 5.5, + } + + accepted = accepted_generation_range(job) + + assert (accepted.media_start, accepted.media_end) == pytest.approx((0.75, 3.5)) + assert (accepted.committed_start, accepted.committed_end) == pytest.approx((9.75, 12.5)) + + def test_edl_splice_preserves_manual_order_and_uses_core_inside_padded_media(): edl = PersistedEDL( clips=[ diff --git a/apps/api/tests/test_local_seam.py b/apps/api/tests/test_local_seam.py index 96a734c..ab9220f 100644 --- a/apps/api/tests/test_local_seam.py +++ b/apps/api/tests/test_local_seam.py @@ -2,7 +2,8 @@ import pytest from app.services.generation_window import GenerationWindow -from app.services.local_seam import select_local_seams +from app.services.continuity_validator import ContinuityIssue, ContinuityReport +from app.services.local_seam import continuity_at_selected_seams, select_local_seams from app.services.seam_matching import SeamFrames @@ -44,3 +45,27 @@ def test_local_seams_report_unsafe_boundaries_without_blending(): assert selection.passed is False assert len(selection.issues) == 2 assert "entry_frame_match_score" in selection.issues[0] + + +def test_selected_seams_replace_nominal_handle_failures_but_keep_media_failures(): + selection = select_local_seams( + entry_samples=[_sample(0.5, 30, 30)], + exit_samples=[_sample(3.5, 30, 30)], + bbox=BBOX, + window=WINDOW, + ) + base = ContinuityReport( + passed=False, + metrics={"pre_handle_background_delta": 0.113, "duration_delta_s": 0.2}, + issues=[ + ContinuityIssue("pre_handle_background_delta", 0.113, 0.09, "pre"), + ContinuityIssue("duration_delta_s", 0.2, 0.16), + ], + sampled_frames=12, + ) + + report = continuity_at_selected_seams(base, selection) + + assert report.passed is False + assert [issue.code for issue in report.issues] == ["duration_delta_s"] + assert report.metrics["entry_frame_match_score"] == pytest.approx(0.0) diff --git a/apps/web/src/features/reveal/GenerationReveal.tsx b/apps/web/src/features/reveal/GenerationReveal.tsx index 90e5651..f1f9795 100644 --- a/apps/web/src/features/reveal/GenerationReveal.tsx +++ b/apps/web/src/features/reveal/GenerationReveal.tsx @@ -483,6 +483,7 @@ function ContextPill({ k, v }: { k: string; v: string }) { function GenerationOutcome({ result }: { result: JobResp }) { const validation = result.continuity_validation; + const seams = result.selected_seams; const composite = result.localized_compositing?.at(-1); const quality = result.generation_quality_state ?? "not reported"; const providers = result.provider_attempts?.length @@ -497,6 +498,8 @@ function GenerationOutcome({ result }: { result: JobResp }) {
quality {quality.replaceAll("_", " ")} seams {validation ? (validation.passed ? "passed" : "failed") : "unavailable"} + {seams?.entry ? entry match {seams.entry.score.toFixed(3)} : null} + {seams?.exit ? exit match {seams.exit.score.toFixed(3)} : null} {result.generation_attempts ?? 1} attempt(s) {result.generated_seconds != null ? ( {result.generated_seconds.toFixed(1)} generated seconds diff --git a/apps/web/src/hooks/useAgentStream.ts b/apps/web/src/hooks/useAgentStream.ts index ee70818..654dabd 100644 --- a/apps/web/src/hooks/useAgentStream.ts +++ b/apps/web/src/hooks/useAgentStream.ts @@ -107,6 +107,8 @@ function generationActivity(event: JobStreamEvent): string { score_start: "checking visual quality and transitions…", continuity_validation_done: "transition check complete…", continuity_validation_unavailable: "finishing quality review…", + seam_match_done: "matching source and edit frames…", + seam_match_unavailable: "could not find safe cut frames…", gen_retry: "improving first render…", gen_provider_fallback: "trying backup video model…", gen_retry_rejected: "keeping stronger render…", diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index de73410..afd02df 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -140,6 +140,15 @@ export type JobResp = { warnings?: string[]; execution_window?: GenerationExecutionWindow | null; continuity_validation?: ContinuityValidation | null; + selected_seams?: { + passed: boolean; + media_start: number; + media_end: number; + timeline_start: number; + timeline_end: number; + entry?: { score: number; source_timestamp: number; media_timestamp: number } | null; + exit?: { score: number; source_timestamp: number; media_timestamp: number } | null; + } | null; generation_quality_state?: string | null; generation_quality_evidence?: string[]; generation_attempts?: number | null; From 770b966c1b39ef1cc03b77a5596986a7a10f3964 Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 12:04:38 +0530 Subject: [PATCH 08/14] feat: lock generation context handles --- apps/api/app/services/generation_window.py | 17 ++++++++++------- apps/api/tests/test_generation_window.py | 3 +++ 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/apps/api/app/services/generation_window.py b/apps/api/app/services/generation_window.py index 22422c5..a59b845 100644 --- a/apps/api/app/services/generation_window.py +++ b/apps/api/app/services/generation_window.py @@ -104,13 +104,16 @@ def protected_context_prompt(prompt: str, window: GenerationWindow) -> str: if not window.adaptive: return prompt return ( - "PADDED SOURCE EDIT CONTRACT: The supplied video includes protected " - "motion context before and after the requested edit. Modify only seconds " + "LOCKED SOURCE-CONTINUITY CONTRACT: Modify only seconds " f"{window.edit_start_offset:.3f} through {window.edit_end_offset:.3f} " - "relative to the supplied clip. Preserve frames, subject motion, camera " - f"motion, background, and audio in the first {window.pre_handle:.3f} " - f"seconds and final {window.post_handle:.3f} seconds. Use those handles " - "only to enter and leave the edit continuously; do not introduce a cut, " - "fade, freeze, or transition.\n\n" + f"relative to the supplied clip. The first {window.pre_handle:.3f} seconds " + f"and final {window.post_handle:.3f} seconds are locked source handles. " + "Keep their visual and temporal content identical to the source: the same " + "objects, colours, lighting, pose, camera, background, and motion. Do not " + "change, restyle, regenerate, freeze, or replace any handle frame. Inside " + "the editable interval, begin from the source pose and velocity, perform " + "the requested change continuously frame by frame, and leave in a state " + "that can continue naturally into the final locked handle. Do not use a " + "cut, fade, dissolve, freeze, or sudden appearance or disappearance.\n\n" + prompt ) diff --git a/apps/api/tests/test_generation_window.py b/apps/api/tests/test_generation_window.py index f4a1b37..95f5ad8 100644 --- a/apps/api/tests/test_generation_window.py +++ b/apps/api/tests/test_generation_window.py @@ -97,4 +97,7 @@ def test_protected_context_prompt_names_edit_offsets_and_handles(): assert "0.500 through 4.000" in rendered assert "first 0.500 seconds" in rendered assert "final 2.000 seconds" in rendered + assert "locked source handles" in rendered + assert "identical to the source" in rendered + assert "Do not use a cut, fade, dissolve" in rendered assert rendered.endswith("Make the person jump.") From 4edccb38a8cb6b0d92440e429c75b35446b46a77 Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 12:13:37 +0530 Subject: [PATCH 09/14] feat: show and preserve generation attempts --- apps/api/app/api/routes/accept.py | 13 +- apps/api/app/api/routes/jobs.py | 9 + apps/api/app/api/routes/remix.py | 16 +- apps/api/app/schemas/accept.py | 2 +- apps/api/app/schemas/job.py | 6 + apps/api/app/services/accepted_generation.py | 11 +- apps/api/app/services/agent_tools.py | 13 +- apps/api/app/services/generation_quality.py | 42 ++++- apps/api/app/workers/generate_job.py | 157 +++++++++++------- apps/api/tests/test_accepted_generation.py | 32 ++++ apps/api/tests/test_generation_quality.py | 41 ++++- .../src/components/editor-port/AgentChat.tsx | 42 ++++- apps/web/src/hooks/useAgentStream.ts | 32 +++- apps/web/src/lib/api.ts | 11 ++ apps/web/src/stores/agent.tsx | 3 + 15 files changed, 347 insertions(+), 83 deletions(-) diff --git a/apps/api/app/api/routes/accept.py b/apps/api/app/api/routes/accept.py index 3767db4..38783dc 100644 --- a/apps/api/app/api/routes/accept.py +++ b/apps/api/app/api/routes/accept.py @@ -25,7 +25,11 @@ 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, + quality_payload_for_candidate, +) from app.services.timeline_response import build_timeline_response from app.workers import entity_job @@ -58,16 +62,17 @@ async def accept( ) if variant is None or variant.status != "done" or not variant.url: raise HTTPException(status_code=422, detail="variant not ready") + 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 diff --git a/apps/api/app/api/routes/jobs.py b/apps/api/app/api/routes/jobs.py index cb971d3..f327516 100644 --- a/apps/api/app/api/routes/jobs.py +++ b/apps/api/app/api/routes/jobs.py @@ -20,6 +20,7 @@ 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, @@ -43,6 +44,7 @@ def _job_out(job: Job) -> JobOut: 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, @@ -53,6 +55,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) ], diff --git a/apps/api/app/api/routes/remix.py b/apps/api/app/api/routes/remix.py index 3437450..4ca7a8c 100644 --- a/apps/api/app/api/routes/remix.py +++ b/apps/api/app/api/routes/remix.py @@ -21,7 +21,11 @@ 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, + quality_payload_for_candidate, +) from app.workers import entity_job from app.workers import generate_job as generate_worker @@ -122,7 +126,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 +315,16 @@ 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") + 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, diff --git a/apps/api/app/schemas/accept.py b/apps/api/app/schemas/accept.py index 2a47104..11ff390 100644 --- a/apps/api/app/schemas/accept.py +++ b/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 diff --git a/apps/api/app/schemas/job.py b/apps/api/app/schemas/job.py index f6b3346..7e842c8 100644 --- a/apps/api/app/schemas/job.py +++ b/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): @@ -42,3 +47,4 @@ class JobOut(BaseModel): 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 diff --git a/apps/api/app/services/accepted_generation.py b/apps/api/app/services/accepted_generation.py index df7ebd2..111e636 100644 --- a/apps/api/app/services/accepted_generation.py +++ b/apps/api/app/services/accepted_generation.py @@ -25,12 +25,21 @@ def metadata(self) -> dict[str, float]: return asdict(self) -def accepted_generation_range(job: Job) -> AcceptedGenerationRange: +def accepted_generation_range( + job: Job, + *, + variant: Variant | None = None, +) -> AcceptedGenerationRange: if job.start_ts is None or job.end_ts is None: raise ValueError("generation job has no requested range") requested_start = float(job.start_ts) requested_end = float(job.end_ts) payload = dict(job.payload or {}) + if variant is not None: + reviews = payload.get("candidate_reviews") + review = reviews.get(variant.id) if isinstance(reviews, dict) else None + if isinstance(review, dict): + payload["selected_seams"] = review.get("selected_seams") raw_window = payload.get("execution_window") window = raw_window if isinstance(raw_window, dict) else {} core_start = float(window.get("core_start", requested_start)) diff --git a/apps/api/app/services/agent_tools.py b/apps/api/app/services/agent_tools.py index 43f35b0..2181f01 100644 --- a/apps/api/app/services/agent_tools.py +++ b/apps/api/app/services/agent_tools.py @@ -820,7 +820,11 @@ async def _accept_variant( 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, + quality_payload_for_candidate, + ) from app.services.timeline_response import build_timeline_response from app.workers import entity_job @@ -846,16 +850,17 @@ async def _accept_variant( ) if variant is None or variant.status != "done" or not variant.url: raise ValueError("variant not ready") + quality_payload = quality_payload_for_candidate(job.payload, variant.id) if not acceptance_allowed( - job.payload, + quality_payload, override_requested=continuity_override, override_enabled=_get_backend_settings().allow_hard_failed_acceptance, ): - raise ValueError(acceptance_block_reason(job.payload)) + raise ValueError(acceptance_block_reason(quality_payload)) if job.start_ts is None or job.end_ts is None: raise ValueError("job has no segment range") - accepted_range = accepted_generation_range(job) + accepted_range = accepted_generation_range(job, variant=variant) # deactivate overlapping generated segments overlapping = ( diff --git a/apps/api/app/services/generation_quality.py b/apps/api/app/services/generation_quality.py index 8a0cd58..867ae42 100644 --- a/apps/api/app/services/generation_quality.py +++ b/apps/api/app/services/generation_quality.py @@ -18,6 +18,7 @@ class GenerationQualityAction(StrEnum): PASS = "pass" + REVIEW_WARNING = "review_warning" CORRECTIVE_RETRY = "corrective_retry" PROVIDER_FALLBACK = "provider_fallback" HARD_FAIL = "hard_fail" @@ -90,11 +91,27 @@ def attempt_quality_rank( def final_semantic_quality(score: dict | None) -> GenerationQualityDecision: evidence = semantic_quality_evidence(score) return GenerationQualityDecision( - GenerationQualityAction.HARD_FAIL if evidence else GenerationQualityAction.PASS, + GenerationQualityAction.REVIEW_WARNING if evidence else GenerationQualityAction.PASS, evidence, ) +def final_candidate_quality( + score: dict | None, + continuity: ContinuityReport | None, +) -> GenerationQualityDecision: + """Semantic uncertainty warns; unsafe or unavailable seams block Apply.""" + if continuity is None: + return GenerationQualityDecision( + GenerationQualityAction.HARD_FAIL, + ("frame matching unavailable",), + ) + if not continuity.passed: + evidence = quality_evidence(score, continuity) + return GenerationQualityDecision(GenerationQualityAction.HARD_FAIL, evidence) + return final_semantic_quality(score) + + def select_fallback_provider(current_provider: str, duration: float) -> str | None: """Prefer a different source-video editor that fits the full context.""" order = ("wan", "happyhorse") if current_provider != "wan" else ("happyhorse",) @@ -132,7 +149,12 @@ def decide_generation_quality( and generated_seconds + duration <= MAX_GENERATED_SECONDS + 0.05 ) if not can_generate_again: - return GenerationQualityDecision(GenerationQualityAction.HARD_FAIL, evidence) + return final_candidate_quality(score, continuity) + + # Missing semantic scoring cannot describe a useful correction. Missing + # structural validation cannot be repaired by asking the model to guess. + if score is None or continuity is None: + return final_candidate_quality(score, continuity) # A provider switch is not a seam repair. It historically spent another # paid render and replaced Wan with a weaker result. Retry this provider @@ -172,6 +194,22 @@ def acceptance_block_reason(payload: dict | None) -> str | None: return "generation quality hard-failed" +def quality_payload_for_candidate( + payload: dict | None, + variant_id: str, +) -> dict: + """Overlay one candidate's review onto the legacy job-level fields.""" + output = dict(payload or {}) + reviews = output.get("candidate_reviews") + review = reviews.get(variant_id) if isinstance(reviews, dict) else None + if isinstance(review, dict): + output["generation_quality_state"] = review.get("quality_state") + output["generation_quality_evidence"] = review.get("evidence") or [] + output["continuity_validation"] = review.get("continuity_validation") + output["selected_seams"] = review.get("selected_seams") + return output + + def acceptance_allowed( payload: dict | None, *, diff --git a/apps/api/app/workers/generate_job.py b/apps/api/app/workers/generate_job.py index 2100bfc..aab68ed 100644 --- a/apps/api/app/workers/generate_job.py +++ b/apps/api/app/workers/generate_job.py @@ -60,6 +60,7 @@ attempt_quality_rank, corrective_prompt, decide_generation_quality, + final_candidate_quality, final_semantic_quality, semantic_quality_evidence, ) @@ -679,13 +680,17 @@ async def _do_extract_end_frame() -> tuple[Path, bool]: variant_count=len(list(plans)), ) - # create the single Variant row so the polling endpoint sees it from t0 - async with AsyncSessionLocal() as db: - v = Variant(job_id=job_id, index=0, status="pending") - db.add(v) - await db.flush() - variant_id = v.id - await db.commit() + async def create_attempt_variant(index: int) -> str: + async with AsyncSessionLocal() as db: + variant = Variant(job_id=job_id, index=index, status="pending") + db.add(variant) + await db.flush() + variant_id = variant.id + await db.commit() + return variant_id + + # The first row exists before dispatch so polling can show its live state. + variant_id = await create_attempt_variant(0) source_video_url = _public_url_or_none(clip_url) mask_public_url = _public_url_or_none(mask_image_url or "") @@ -749,7 +754,12 @@ def prompt_for_provider(provider: str, correction: str = "") -> str: ) return base + correction - async def dispatch(provider: str, correction: str = "") -> bool: + async def dispatch( + provider: str, + correction: str = "", + *, + target_variant_id: str, + ) -> bool: attempt_mode = select_source_edit_mode( provider, duration=generation_duration, @@ -764,7 +774,7 @@ async def dispatch(provider: str, correction: str = "") -> bool: ) return await _run_variant( job_id=job_id, - variant_id=variant_id, + variant_id=target_variant_id, clip_path=clip_path, clip_url=clip_url, plan={ @@ -795,7 +805,7 @@ async def dispatch(provider: str, correction: str = "") -> bool: generated_seconds = generation_duration provider_attempts = [video_provider] fallback_used = False - initial_ok = await dispatch(video_provider) + initial_ok = await dispatch(video_provider, target_variant_id=variant_id) if generation_window.adaptive and not initial_ok: async with AsyncSessionLocal() as db: @@ -823,6 +833,7 @@ async def dispatch(provider: str, correction: str = "") -> bool: attempts += 1 generated_seconds += generation_duration provider_attempts.append(video_provider) + retry_variant_id = await create_attempt_variant(attempts - 1) await _emit( job_id, ( @@ -838,6 +849,7 @@ async def dispatch(provider: str, correction: str = "") -> bool: initial_ok = await dispatch( video_provider, corrective_prompt(failure_decision.evidence), + target_variant_id=retry_variant_id, ) # collect outcome @@ -1060,6 +1072,57 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: ) return report + async def persist_candidate_review( + variant: Variant, + candidate_score: dict | None, + candidate_continuity: ContinuityReport | None, + ) -> None: + decision = ( + final_candidate_quality(candidate_score, candidate_continuity) + if generation_window.adaptive + else final_semantic_quality(candidate_score) + ) + final_validation = validation_by_url.get(variant.url or "") + seams = final_validation[1] if final_validation is not None else None + review = { + "attempt": variant.index + 1, + "label": "First pass" if variant.index == 0 else "Corrected pass", + "quality_state": decision.action.value, + "evidence": list(decision.evidence), + "continuity_validation": ( + candidate_continuity.metadata() if candidate_continuity is not None else None + ), + "selected_seams": seams, + "semantic_score": candidate_score, + } + async with AsyncSessionLocal() as db: + current_job = await db.get(Job, job_id) + if current_job is not None: + current_payload = dict(current_job.payload or {}) + reviews = dict(current_payload.get("candidate_reviews") or {}) + reviews[variant.id] = review + current_payload["candidate_reviews"] = reviews + current_job.payload = current_payload + await _update_variant( + db, + variant.id, + visual_coherence=(candidate_score or {}).get("visual_coherence"), + prompt_adherence=(candidate_score or {}).get("prompt_adherence"), + ) + await db.commit() + await _emit( + job_id, + "candidate_review_done", + ( + "first pass review complete" + if variant.index == 0 + else "corrected pass review complete" + ), + variant_id=variant.id, + variant_index=variant.index, + **review, + ) + await _emit( job_id, "score_start", @@ -1069,6 +1132,7 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: score_variant(done[0]), validate_continuity(done[0]), ) + await persist_candidate_review(done[0], score, continuity_report) quality_state = GenerationQualityAction.PASS quality_evidence: list[str] = [] @@ -1087,8 +1151,7 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: GenerationQualityAction.CORRECTIVE_RETRY, GenerationQualityAction.PROVIDER_FALLBACK, }: - previous_url = done[0].url - previous_description = done[0].description + previous_variant = done[0] previous_score = score previous_continuity = continuity_report previous_provider = video_provider @@ -1100,6 +1163,7 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: attempts += 1 generated_seconds += generation_duration provider_attempts.append(next_provider) + retry_variant_id = await create_attempt_variant(attempts - 1) await _emit( job_id, ( @@ -1118,10 +1182,11 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: succeeded = await dispatch( next_provider, corrective_prompt(decision.evidence), + target_variant_id=retry_variant_id, ) video_provider = next_provider async with AsyncSessionLocal() as db: - retried = await db.get(Variant, variant_id) + retried = await db.get(Variant, retry_variant_id) generation_error = ( retried.error if retried is not None and not succeeded @@ -1129,24 +1194,17 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: ) if retried is not None and succeeded and retried.status == "done": done = [retried] - elif retried is not None: - await _update_variant( - db, - variant_id, - status="done", - url=previous_url, - description=previous_description, - error=None, - ) - restored = await db.get(Variant, variant_id) - if restored is not None: - done = [restored] + else: + done = [previous_variant] if succeeded: done = [await maybe_composite(done[0], video_provider)] candidate_score, candidate_continuity = await asyncio.gather( score_variant(done[0]), validate_continuity(done[0]), ) + await persist_candidate_review( + done[0], candidate_score, candidate_continuity + ) if attempt_quality_rank( candidate_score, candidate_continuity, @@ -1154,18 +1212,7 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: score = candidate_score continuity_report = candidate_continuity else: - async with AsyncSessionLocal() as db: - await _update_variant( - db, - variant_id, - status="done", - url=previous_url, - description=previous_description, - error=None, - ) - restored = await db.get(Variant, variant_id) - if restored is not None: - done = [restored] + done = [previous_variant] score = previous_score continuity_report = previous_continuity video_provider = previous_provider @@ -1202,13 +1249,13 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: or int(score.get("prompt_adherence") or 0) < 6 ) if needs_retry: - previous_url = done[0].url - previous_description = done[0].description + previous_variant = done[0] previous_score = score previous_continuity = continuity_report attempts += 1 generated_seconds += generation_duration provider_attempts.append(video_provider) + retry_variant_id = await create_attempt_variant(attempts - 1) await _emit( job_id, "gen_retry", @@ -1217,15 +1264,22 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: previous_score=score, ) correction = corrective_prompt(semantic_quality_evidence(score)) - succeeded = await dispatch(video_provider, correction) + succeeded = await dispatch( + video_provider, + correction, + target_variant_id=retry_variant_id, + ) async with AsyncSessionLocal() as db: - retried = await db.get(Variant, variant_id) + retried = await db.get(Variant, retry_variant_id) if retried is not None and succeeded and retried.status == "done": done = [retried] candidate_score, candidate_continuity = await asyncio.gather( score_variant(retried), validate_continuity(retried), ) + await persist_candidate_review( + retried, candidate_score, candidate_continuity + ) if attempt_quality_rank( candidate_score, candidate_continuity, @@ -1233,17 +1287,7 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: score = candidate_score continuity_report = candidate_continuity else: - await _update_variant( - db, - variant_id, - status="done", - url=previous_url, - description=previous_description, - error=None, - ) - restored = await db.get(Variant, variant_id) - if restored is not None: - done = [restored] + done = [previous_variant] score = previous_score continuity_report = previous_continuity await _emit( @@ -1253,14 +1297,7 @@ async def validate_continuity(variant: Variant) -> ContinuityReport | None: attempt=attempts, ) elif retried is not None: - await _update_variant( - db, - variant_id, - status="done", - url=previous_url, - description=previous_description, - error=None, - ) + done = [previous_variant] await _emit( job_id, "gen_retry_failed", diff --git a/apps/api/tests/test_accepted_generation.py b/apps/api/tests/test_accepted_generation.py index 80c4624..538b0c7 100644 --- a/apps/api/tests/test_accepted_generation.py +++ b/apps/api/tests/test_accepted_generation.py @@ -61,6 +61,38 @@ def test_range_uses_safe_frame_matched_context_instead_of_nominal_core(): assert (accepted.committed_start, accepted.committed_end) == pytest.approx((9.75, 12.5)) +def test_range_uses_the_selected_candidate_seams(): + job = _job() + first = Variant(id="first", job_id=job.id, index=0, url="first.mp4") + corrected = Variant(id="corrected", job_id=job.id, index=1, url="corrected.mp4") + job.payload["candidate_reviews"] = { + first.id: { + "selected_seams": { + "passed": True, + "media_start": 0.25, + "media_end": 3.25, + "timeline_start": 2.25, + "timeline_end": 5.25, + } + }, + corrected.id: { + "selected_seams": { + "passed": True, + "media_start": 0.75, + "media_end": 3.5, + "timeline_start": 2.75, + "timeline_end": 5.5, + } + }, + } + + first_range = accepted_generation_range(job, variant=first) + corrected_range = accepted_generation_range(job, variant=corrected) + + assert (first_range.media_start, first_range.media_end) == pytest.approx((0.25, 3.25)) + assert (corrected_range.media_start, corrected_range.media_end) == pytest.approx((0.75, 3.5)) + + def test_edl_splice_preserves_manual_order_and_uses_core_inside_padded_media(): edl = PersistedEDL( clips=[ diff --git a/apps/api/tests/test_generation_quality.py b/apps/api/tests/test_generation_quality.py index 7422b72..2c72f39 100644 --- a/apps/api/tests/test_generation_quality.py +++ b/apps/api/tests/test_generation_quality.py @@ -7,6 +7,8 @@ corrective_prompt, decide_generation_quality, final_semantic_quality, + final_candidate_quality, + quality_payload_for_candidate, semantic_quality_evidence, select_fallback_provider, ) @@ -147,15 +149,50 @@ def test_production_bad_outputs_cannot_be_marked_clean(): assert semantic_quality_evidence(None) == ("semantic quality scoring unavailable",) assert final_semantic_quality( {"visual_coherence": 8, "prompt_adherence": 1} - ).action == GenerationQualityAction.HARD_FAIL + ).action == GenerationQualityAction.REVIEW_WARNING assert final_semantic_quality( {"visual_coherence": 4, "prompt_adherence": 2} - ).action == GenerationQualityAction.HARD_FAIL + ).action == GenerationQualityAction.REVIEW_WARNING assert final_semantic_quality( {"visual_coherence": 8, "prompt_adherence": 8} ).action == GenerationQualityAction.PASS +def test_semantic_miss_warns_but_unsafe_or_missing_seam_blocks_apply(): + clean = ContinuityReport(True, {}) + assert final_candidate_quality( + {"visual_coherence": 8, "prompt_adherence": 2}, clean + ).action == GenerationQualityAction.REVIEW_WARNING + assert final_candidate_quality( + {"visual_coherence": 8, "prompt_adherence": 8}, _failed_continuity() + ).action == GenerationQualityAction.HARD_FAIL + assert final_candidate_quality( + {"visual_coherence": 8, "prompt_adherence": 8}, None + ).action == GenerationQualityAction.HARD_FAIL + + +def test_candidate_review_overrides_job_level_acceptance_state(): + payload = { + "generation_quality_state": "pass", + "candidate_reviews": { + "variant-unsafe": { + "quality_state": "hard_fail", + "evidence": ["exit_frame_match_score at exit"], + "selected_seams": {"passed": False}, + } + }, + } + + candidate = quality_payload_for_candidate(payload, "variant-unsafe") + + assert candidate["generation_quality_state"] == "hard_fail" + assert not acceptance_allowed( + candidate, + override_requested=False, + override_enabled=False, + ) + + def test_retry_replaces_previous_result_only_when_quality_improves(): continuity = ContinuityReport(True, {}) car_failure = {"visual_coherence": 8, "prompt_adherence": 1} diff --git a/apps/web/src/components/editor-port/AgentChat.tsx b/apps/web/src/components/editor-port/AgentChat.tsx index ad6f3a6..e20580c 100644 --- a/apps/web/src/components/editor-port/AgentChat.tsx +++ b/apps/web/src/components/editor-port/AgentChat.tsx @@ -347,6 +347,22 @@ export function AgentChat({ projectId }: AgentChatProps) { text-align: center; margin-top: 6px; } + .variant-preview__attempt { + display: inline-block; + margin: 0 0 6px; + color: var(--foreground); + font-size: 12px; + font-weight: 600; + } + .variant-preview__review { + margin: 6px 0 0; + color: var(--ink-fade); + font-size: 11px; + line-height: 1.45; + } + .variant-preview__review--unsafe { + color: rgba(255, 120, 120, 0.9); + } /* ── prompt plan brief ─ */ @@ -708,8 +724,13 @@ function VariantPreviewCard({ const key = `${jobId}:${v.index}`; const isApplying = applyingVariant === key; const isApplied = appliedVariant === key; + const reviewPending = !v.quality_state; + const unsafe = v.quality_state === "hard_fail"; return (
+ + {v.attempt_label ?? (v.index === 0 ? "First pass" : "Corrected pass")} +
{v.url ? (
diff --git a/apps/web/src/hooks/useAgentStream.ts b/apps/web/src/hooks/useAgentStream.ts index 654dabd..1b03712 100644 --- a/apps/web/src/hooks/useAgentStream.ts +++ b/apps/web/src/hooks/useAgentStream.ts @@ -147,6 +147,7 @@ export function useAgentStream(projectId?: string | null) { let detailedProgressAt = 0; let detailedStage = ""; let lastHeartbeatAt = 0; + let previewSignature = ""; const eventStream = streamJobEvents(jobId, { onEvent: (event) => { detailedProgressAt = Date.now(); @@ -173,6 +174,30 @@ export function useAgentStream(projectId?: string | null) { if (watch.signal.aborted) return; const ready = job.variants.filter((variant) => variant.url); const failed = job.variants.filter((variant) => variant.status === "error"); + const nextPreviewSignature = ready + .map((variant) => `${variant.id}:${variant.url}`) + .join("|"); + if (ready.length > 0 && nextPreviewSignature !== previewSignature) { + previewSignature = nextPreviewSignature; + dispatch({ + type: "add_variant_preview", + jobId, + variants: ready, + timelineStart: job.selected_seams?.timeline_start ?? job.execution_window?.core_start ?? job.start_ts, + timelineEnd: job.selected_seams?.timeline_end ?? job.execution_window?.core_end ?? job.end_ts, + mediaStart: job.selected_seams?.media_start ?? job.execution_window?.edit_start_offset ?? 0, + mediaEnd: + job.selected_seams?.media_end ?? + job.execution_window?.edit_end_offset ?? + (job.start_ts != null && job.end_ts != null + ? job.end_ts - job.start_ts + : null), + }); + dispatch({ + type: "set_activity", + activity: ready.length === 1 ? "first pass ready — reviewing it now…" : "corrected pass ready — finishing review…", + }); + } if (job.status === "done" || job.status === "error") { if (projectId) clearPendingTurn(projectId); if (ready.length > 0) { @@ -180,10 +205,11 @@ export function useAgentStream(projectId?: string | null) { type: "add_variant_preview", jobId, variants: ready, - timelineStart: job.execution_window?.core_start ?? job.start_ts, - timelineEnd: job.execution_window?.core_end ?? job.end_ts, - mediaStart: job.execution_window?.edit_start_offset ?? 0, + timelineStart: job.selected_seams?.timeline_start ?? job.execution_window?.core_start ?? job.start_ts, + timelineEnd: job.selected_seams?.timeline_end ?? job.execution_window?.core_end ?? job.end_ts, + mediaStart: job.selected_seams?.media_start ?? job.execution_window?.edit_start_offset ?? 0, mediaEnd: + job.selected_seams?.media_end ?? job.execution_window?.edit_end_offset ?? (job.start_ts != null && job.end_ts != null ? job.end_ts - job.start_ts diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index afd02df..8edcb75 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -121,6 +121,11 @@ export type Variant = { visual_coherence: number | null; prompt_adherence: number | null; error: string | null; + attempt_label?: string | null; + quality_state?: string | null; + quality_evidence?: string[]; + continuity_validation?: ContinuityValidation | null; + selected_seams?: JobResp["selected_seams"]; }; export type JobResp = { @@ -156,6 +161,12 @@ export type JobResp = { provider_attempts?: string[]; localized_compositing?: Array<{ applied?: boolean; reason?: string }>; local_flow_telemetry?: Record | null; + retry_state?: { + status: string; + retry_at?: number; + evidence?: string[]; + correction?: string; + } | null; }; export type GenerationExecutionWindow = { diff --git a/apps/web/src/stores/agent.tsx b/apps/web/src/stores/agent.tsx index 831da19..30c7c3c 100644 --- a/apps/web/src/stores/agent.tsx +++ b/apps/web/src/stores/agent.tsx @@ -52,6 +52,9 @@ export interface VariantPreview { description: string | null; visual_coherence: number | null; prompt_adherence: number | null; + attempt_label?: string | null; + quality_state?: string | null; + quality_evidence?: string[]; } export interface GenerationProgressEntry { From 08b943875bd6bfd570d33405d2da7e9db1aece67 Mon Sep 17 00:00:00 2001 From: snow Date: Sat, 18 Jul 2026 12:23:13 +0530 Subject: [PATCH 10/14] feat: add cancellable corrective retry --- apps/api/app/api/routes/accept.py | 2 + apps/api/app/api/routes/jobs.py | 32 +++++ apps/api/app/api/routes/remix.py | 2 + apps/api/app/services/agent_tools.py | 2 + apps/api/app/services/generation_quality.py | 10 ++ apps/api/app/workers/generate_job.py | 112 +++++++++++++++--- apps/api/tests/test_generation_quality.py | 11 ++ apps/api/tests/test_pipeline.py | 44 +++++++ .../src/components/editor-port/AgentChat.tsx | 79 +++++++++++- apps/web/src/hooks/useAgentStream.ts | 15 +++ apps/web/src/lib/api.ts | 10 ++ apps/web/src/stores/agent.tsx | 44 +++++++ apps/web/tests/e2e/frontend-readiness.spec.ts | 9 ++ 13 files changed, 353 insertions(+), 19 deletions(-) diff --git a/apps/api/app/api/routes/accept.py b/apps/api/app/api/routes/accept.py index 38783dc..d16b500 100644 --- a/apps/api/app/api/routes/accept.py +++ b/apps/api/app/api/routes/accept.py @@ -28,6 +28,7 @@ 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 @@ -62,6 +63,7 @@ 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( quality_payload, diff --git a/apps/api/app/api/routes/jobs.py b/apps/api/app/api/routes/jobs.py index f327516..c45a128 100644 --- a/apps/api/app/api/routes/jobs.py +++ b/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,6 +21,10 @@ 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 {} @@ -113,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, diff --git a/apps/api/app/api/routes/remix.py b/apps/api/app/api/routes/remix.py index 4ca7a8c..0400318 100644 --- a/apps/api/app/api/routes/remix.py +++ b/apps/api/app/api/routes/remix.py @@ -24,6 +24,7 @@ from app.services.generation_quality import ( acceptance_allowed, acceptance_block_reason, + cancel_waiting_retry, quality_payload_for_candidate, ) from app.workers import entity_job @@ -315,6 +316,7 @@ 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( quality_payload, diff --git a/apps/api/app/services/agent_tools.py b/apps/api/app/services/agent_tools.py index 2181f01..5e5b6e1 100644 --- a/apps/api/app/services/agent_tools.py +++ b/apps/api/app/services/agent_tools.py @@ -823,6 +823,7 @@ async def _accept_variant( 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 @@ -850,6 +851,7 @@ async def _accept_variant( ) if variant is None or variant.status != "done" or not variant.url: raise ValueError("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( quality_payload, diff --git a/apps/api/app/services/generation_quality.py b/apps/api/app/services/generation_quality.py index 867ae42..8c64b65 100644 --- a/apps/api/app/services/generation_quality.py +++ b/apps/api/app/services/generation_quality.py @@ -210,6 +210,16 @@ def quality_payload_for_candidate( return output +def cancel_waiting_retry(payload: dict | None, *, reason: str) -> dict: + output = dict(payload or {}) + retry_state = dict(output.get("retry_state") or {}) + if retry_state.get("status") == "waiting": + retry_state["status"] = "cancelled" + retry_state["cancel_reason"] = reason + output["retry_state"] = retry_state + return output + + def acceptance_allowed( payload: dict | None, *, diff --git a/apps/api/app/workers/generate_job.py b/apps/api/app/workers/generate_job.py index aab68ed..57dac40 100644 --- a/apps/api/app/workers/generate_job.py +++ b/apps/api/app/workers/generate_job.py @@ -91,6 +91,7 @@ def _provider_label(provider: str) -> str: }.get(provider, _PROVIDER_LABEL) VARIANT_COUNT = 1 +RETRY_GRACE_SECONDS = 10.0 def _provider_model(provider: str, edit_mode: str | None = None) -> str: @@ -1134,6 +1135,62 @@ async def persist_candidate_review( ) await persist_candidate_review(done[0], score, continuity_report) + async def await_retry_permission(evidence: tuple[str, ...]) -> bool: + retry_at = time.time() + RETRY_GRACE_SECONDS + correction = corrective_prompt(evidence) + retry_state = { + "status": "waiting", + "retry_at": retry_at, + "evidence": list(evidence), + "correction": correction, + } + async with AsyncSessionLocal() as db: + current_job = await db.get(Job, job_id) + if current_job is not None: + current_payload = dict(current_job.payload or {}) + current_payload["retry_state"] = retry_state + current_job.payload = current_payload + await db.commit() + await _emit( + job_id, + "retry_pending", + "first pass review found a specific issue; corrective retry is waiting", + **retry_state, + ) + while time.time() < retry_at: + await asyncio.sleep(0.5) + async with AsyncSessionLocal() as db: + current_job = await db.get(Job, job_id) + current_state = dict((current_job.payload or {}).get("retry_state") or {}) + if current_state.get("status") == "cancelled": + await _emit( + job_id, + "retry_cancelled", + "corrective retry stopped; keeping the first pass", + ) + return False + if current_state.get("status") == "retry_now": + break + async with AsyncSessionLocal() as db: + current_job = await db.get(Job, job_id) + if current_job is not None: + current_payload = dict(current_job.payload or {}) + current_state = dict(current_payload.get("retry_state") or retry_state) + if current_state.get("status") == "cancelled": + return False + current_state["status"] = "dispatched" + current_state["dispatched_at"] = time.time() + current_payload["retry_state"] = current_state + current_job.payload = current_payload + await db.commit() + await _emit( + job_id, + "retry_dispatched", + "starting one evidence-driven corrective retry", + evidence=list(evidence), + ) + return True + quality_state = GenerationQualityAction.PASS quality_evidence: list[str] = [] if generation_window.adaptive: @@ -1151,6 +1208,9 @@ async def persist_candidate_review( GenerationQualityAction.CORRECTIVE_RETRY, GenerationQualityAction.PROVIDER_FALLBACK, }: + if not await await_retry_permission(decision.evidence): + decision = final_candidate_quality(score, continuity_report) + break previous_variant = done[0] previous_score = score previous_continuity = continuity_report @@ -1249,28 +1309,38 @@ async def persist_candidate_review( or int(score.get("prompt_adherence") or 0) < 6 ) if needs_retry: + retry_evidence = semantic_quality_evidence(score) + if not await await_retry_permission(retry_evidence): + needs_retry = False previous_variant = done[0] previous_score = score previous_continuity = continuity_report - attempts += 1 - generated_seconds += generation_duration - provider_attempts.append(video_provider) - retry_variant_id = await create_attempt_variant(attempts - 1) - await _emit( - job_id, - "gen_retry", - "quality validation failed; making one corrective generation attempt", - attempt=attempts, - previous_score=score, - ) - correction = corrective_prompt(semantic_quality_evidence(score)) - succeeded = await dispatch( + attempts += int(needs_retry) + if not needs_retry: + final_quality = final_semantic_quality(score) + quality_state = final_quality.action + quality_evidence = list(final_quality.evidence) + else: + generated_seconds += generation_duration + provider_attempts.append(video_provider) + retry_variant_id = await create_attempt_variant(attempts - 1) + if needs_retry: + await _emit( + job_id, + "gen_retry", + "quality validation failed; making one corrective generation attempt", + attempt=attempts, + previous_score=score, + ) + correction = corrective_prompt(retry_evidence) + succeeded = needs_retry and await dispatch( video_provider, correction, target_variant_id=retry_variant_id, ) - async with AsyncSessionLocal() as db: - retried = await db.get(Variant, retry_variant_id) + if needs_retry: + async with AsyncSessionLocal() as db: + retried = await db.get(Variant, retry_variant_id) if retried is not None and succeeded and retried.status == "done": done = [retried] candidate_score, candidate_continuity = await asyncio.gather( @@ -1303,9 +1373,10 @@ async def persist_candidate_review( "gen_retry_failed", "corrective attempt failed; retaining the first generated result", ) - final_quality = final_semantic_quality(score) - quality_state = final_quality.action - quality_evidence = list(final_quality.evidence) + if needs_retry: + final_quality = final_semantic_quality(score) + quality_state = final_quality.action + quality_evidence = list(final_quality.evidence) else: final_quality = final_semantic_quality(score) quality_state = final_quality.action @@ -1334,6 +1405,11 @@ async def persist_candidate_review( "selected_edit_mode": final_edit_mode, } ) + retry_state = dict(current_payload.get("retry_state") or {}) + if retry_state.get("status") == "dispatched": + retry_state["status"] = "completed" + retry_state["completed_at"] = time.time() + current_payload["retry_state"] = retry_state final_validation = validation_by_url.get(done[0].url or "") if final_validation is not None: final_report, final_seams = final_validation diff --git a/apps/api/tests/test_generation_quality.py b/apps/api/tests/test_generation_quality.py index 2c72f39..06630fa 100644 --- a/apps/api/tests/test_generation_quality.py +++ b/apps/api/tests/test_generation_quality.py @@ -9,6 +9,7 @@ final_semantic_quality, final_candidate_quality, quality_payload_for_candidate, + cancel_waiting_retry, semantic_quality_evidence, select_fallback_provider, ) @@ -193,6 +194,16 @@ def test_candidate_review_overrides_job_level_acceptance_state(): ) +def test_applying_first_pass_cancels_a_waiting_retry(): + payload = cancel_waiting_retry( + {"retry_state": {"status": "waiting", "evidence": ["wrong colour"]}}, + reason="candidate applied", + ) + + assert payload["retry_state"]["status"] == "cancelled" + assert payload["retry_state"]["cancel_reason"] == "candidate applied" + + def test_retry_replaces_previous_result_only_when_quality_improves(): continuity = ContinuityReport(True, {}) car_failure = {"visual_coherence": 8, "prompt_adherence": 1} diff --git a/apps/api/tests/test_pipeline.py b/apps/api/tests/test_pipeline.py index 0f60513..9582d6a 100644 --- a/apps/api/tests/test_pipeline.py +++ b/apps/api/tests/test_pipeline.py @@ -16,6 +16,8 @@ from app.main import app # noqa: E402 from app.db.init import create_all # noqa: E402 +from app.db.session import AsyncSessionLocal # noqa: E402 +from app.models.job import Job # noqa: E402 from app.workers.runner import JobRunner # noqa: E402 @@ -108,6 +110,48 @@ async def test_upload_and_project(client: AsyncClient): assert renamed.json()["name"] == "Launch cut" +@pytest.mark.asyncio +async def test_pending_corrective_retry_can_be_cancelled(client: AsyncClient): + if not os.path.exists(FIXTURE_VIDEO): + pytest.skip("test fixture video not found") + with open(FIXTURE_VIDEO, "rb") as file: + uploaded = await client.post( + "/api/upload", + files={"file": ("test.mp4", file, "video/mp4")}, + headers=headers(), + ) + project_id = uploaded.json()["project_id"] + async with AsyncSessionLocal() as db: + job = Job( + project_id=project_id, + kind="generate", + status="processing", + payload={ + "retry_state": { + "status": "waiting", + "retry_at": 9999999999.0, + "evidence": ["requested green was missing"], + } + }, + ) + db.add(job) + await db.commit() + await db.refresh(job) + job_id = job.id + + response = await client.post( + f"/api/jobs/{job_id}/retry-decision", + json={"action": "cancel"}, + headers=headers(), + ) + + assert response.status_code == 200 + assert response.json()["status"] == "cancelled" + async with AsyncSessionLocal() as db: + saved = await db.get(Job, job_id) + assert saved.payload["retry_state"]["status"] == "cancelled" + + @pytest.mark.asyncio async def test_identify_endpoint(client: AsyncClient): """identify region returns entity description with dict attributes""" diff --git a/apps/web/src/components/editor-port/AgentChat.tsx b/apps/web/src/components/editor-port/AgentChat.tsx index e20580c..af02250 100644 --- a/apps/web/src/components/editor-port/AgentChat.tsx +++ b/apps/web/src/components/editor-port/AgentChat.tsx @@ -7,7 +7,7 @@ import { MessageContent, } from "@/components/ui/message"; import { useAgentStream } from "@/hooks/useAgentStream"; -import { accept } from "@/lib/api"; +import { accept, decideGenerationRetry } from "@/lib/api"; import { useEDL, totalDuration } from "@/stores/edl"; import { useAgent, @@ -124,6 +124,25 @@ export function AgentChat({ projectId }: AgentChatProps) { [projectId, applyingVariant, agentDispatch], ); + const handleRetryDecision = useCallback( + async (jobId: string, action: "cancel" | "retry_now") => { + try { + const result = await decideGenerationRetry(jobId, action); + agentDispatch({ + type: "update_retry_control", + jobId, + status: result.status, + }); + } catch (error) { + agentDispatch({ + type: "add_error", + message: error instanceof Error ? error.message : "Could not update retry.", + }); + } + }, + [agentDispatch], + ); + return ( <>