diff --git a/.env.example b/.env.example index f56446e..fac1012 100644 --- a/.env.example +++ b/.env.example @@ -29,7 +29,7 @@ MESH_VIDEO_MODEL="google/veo-3" MESH_VIDEO_ENDPOINT="/video/generations" # Leave true for local development. Production startup requires false. USE_AI_STUBS="true" -ADAPTIVE_EDIT_PLANNING="false" +ADAPTIVE_EDIT_PLANNING="true" GLOBAL_EDIT_PLANNING="false" # Emergency-only: keep false so continuity hard-fails cannot be accepted. ALLOW_HARD_FAILED_ACCEPTANCE="false" diff --git a/README.md b/README.md index 3075595..08acfb1 100644 --- a/README.md +++ b/README.md @@ -249,7 +249,7 @@ Important environment groups: - auth: `AUTH_JWT_SECRET`, `AUTH_JWT_EXPIRES_MINUTES` - media storage: `S3_BUCKET` and standard AWS settings, or local media fallback - AI mode: `USE_AI_STUBS` -- local edit rollout: `ADAPTIVE_EDIT_PLANNING` (defaults off; fixed-window fallback stays available) +- local edit windows: `ADAPTIVE_EDIT_PLANNING` (defaults on; fixed-window fallback stays available) - emergency continuity override: `ALLOW_HARD_FAILED_ACCEPTANCE` (keep off during normal rollout) - Mesh API gateway: `MESH_API_KEY`, `MESH_API_BASE_URL`, `MESH_MODEL`, `MESH_VIDEO_MODEL`, `MESH_VIDEO_ENDPOINT` - generation provider: `VIDEO_GEN_PROVIDER`, `VEO_MODEL`, `VIDEO_GENERATION_TIMEOUT` diff --git a/apps/api/app/ai/prompts/edit_plan.txt b/apps/api/app/ai/prompts/edit_plan.txt index 27a12c0..91b41f2 100644 --- a/apps/api/app/ai/prompts/edit_plan.txt +++ b/apps/api/app/ai/prompts/edit_plan.txt @@ -7,9 +7,20 @@ INPUT - entity_description: optional identified subject - reference image: actual frame at the selected time -Return exactly one grounded variant: +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", + "target_description": "selected object or feature", + "action_phases": ["ordered phase names for motion; empty for a state change"], + "estimated_action_seconds": 3.5, + "requires_recovery_motion": true, + "preservation_requirements": ["specific things that must remain unchanged"], + "reasoning": "short explanation of temporal meaning" + }, "variants": [ { "intent": "remove" | "replace" | "transform" | "add" | "restyle", @@ -25,6 +36,20 @@ Return exactly one grounded variant: RULES - Inspect the reference image. Ground every noun and visual attribute in it. +- Decide temporal meaning before any source clip is cut. +- A state change such as "make this ball pink", "remove this logo", or "give + this cat human-like eyes" normally means the change persists for this selected + target's complete continuous appearance. Use scope "local" and duration_policy + "continuous_occurrence". Do not limit such changes to an arbitrary few seconds. +- An action such as bounce, jump, wave, or turn is local and time-bounded. Use + duration_policy "bounded_action" and estimate enough time for preparation, + action, landing/recovery, and smooth return to source motion. +- 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 + separated appearances across the full video. +- estimated_action_seconds describes action core only. For state changes it is + only a safe fallback; tracking determines the continuous occurrence boundary. - Preserve the source video's timing, motion, framing, camera, lighting, background, and every unselected subject unless the user explicitly changes them. - State the requested result positively and precisely. Never weaken or reinterpret it. diff --git a/apps/api/app/ai/services/__init__.py b/apps/api/app/ai/services/__init__.py index 6e6e074..2fd0967 100644 --- a/apps/api/app/ai/services/__init__.py +++ b/apps/api/app/ai/services/__init__.py @@ -61,7 +61,7 @@ def _resolve_use_ai_stubs() -> bool: _MOTION_EDIT_RE = re.compile( - r"\b(jump|jumping|run|running|walk|walking|dance|dancing|wave|waving|" + r"\b(jump|jumping|bounce|bouncing|run|running|walk|walking|dance|dancing|wave|waving|" r"turn|turning|spin|spinning|sit|sitting|stand|standing|kick|throw|" r"catch|clap|clapping|nod|bow|fall|falls|leap|leaping)\b" ) @@ -130,13 +130,14 @@ def _rewrite_motion_prompt(prompt_text: str) -> tuple[bool, bool, str]: # ------------------------ gemini adapter ------------------------ # simple in-memory plan cache: keyed on (prompt, bbox_string), max 32 entries - _plan_cache: dict[str, list[dict]] = {} + _plan_cache: dict[str, dict] = {} _plan_cache_order: list[str] = [] _PLAN_CACHE_MAX = 32 - async def _plan_variants(prompt: str, bbox: dict, frame_path: str) -> list[dict]: - # only cache text-only plans (frame_path makes every call unique anyway) - cache_key = f"{prompt}|{json.dumps(bbox, sort_keys=True)}" + async def _interpret_edit(prompt: str, bbox: dict, frame_path: str) -> dict: + # Include visual source identity. Same words and bbox coordinates can + # describe completely different targets in different frames. + cache_key = f"{prompt}|{json.dumps(bbox, sort_keys=True)}|{frame_path}" cached = _plan_cache.get(cache_key) if cached is not None: return cached @@ -144,6 +145,15 @@ async def _plan_variants(prompt: str, bbox: dict, frame_path: str) -> list[dict] plan = await _gemini_real.create_edit_plan( prompt, bbox, frame_path=frame_path ) + _plan_cache[cache_key] = plan + _plan_cache_order.append(cache_key) + if len(_plan_cache_order) > _PLAN_CACHE_MAX: + oldest = _plan_cache_order.pop(0) + _plan_cache.pop(oldest, None) + return plan + + async def _plan_variants(prompt: str, bbox: dict, frame_path: str) -> list[dict]: + plan = await _interpret_edit(prompt, bbox, frame_path) variants = plan.get("variants") if isinstance(plan, dict) else None if not variants: return [] @@ -163,13 +173,6 @@ async def _plan_variants(prompt: str, bbox: dict, frame_path: str) -> list[dict] # because veo has no visual context for the scene v["conditioning_strategy"] = "first_frame" - # cache the result for repeated prompts (propagation reuses same prompt) - _plan_cache[cache_key] = variants - _plan_cache_order.append(cache_key) - if len(_plan_cache_order) > _PLAN_CACHE_MAX: - oldest = _plan_cache_order.pop(0) - _plan_cache.pop(oldest, None) - return variants async def _score_variant( @@ -221,6 +224,7 @@ async def _find_entity_in_keyframes(entity: dict, keyframes: list[str]) -> list[ return out gemini = _types.SimpleNamespace( + interpret_edit=_interpret_edit, plan_variants=_plan_variants, score_variant=_score_variant, identify_entity=_identify_entity, diff --git a/apps/api/app/ai/services/_stubs.py b/apps/api/app/ai/services/_stubs.py index 7f5ca51..1c10148 100644 --- a/apps/api/app/ai/services/_stubs.py +++ b/apps/api/app/ai/services/_stubs.py @@ -22,27 +22,50 @@ # ------------------------------ gemini ------------------------------ +async def _gemini_interpret_edit( + prompt: str, + bbox: BBoxDict, + frame_path: str, +) -> dict: + from app.services.edit_scope import plan_prompt_intent + + intent = plan_prompt_intent(prompt).intent + region = f"region at ({bbox['x']:.2f},{bbox['y']:.2f}) {bbox['w']:.2f}x{bbox['h']:.2f}" + return { + "decision": { + "scope": intent.scope, + "change_type": intent.change_type, + "duration_policy": intent.duration_policy, + "target_description": intent.target_description or region, + "action_phases": intent.action_phases, + "estimated_action_seconds": intent.estimated_action_seconds, + "requires_recovery_motion": intent.requires_recovery_motion, + "preservation_requirements": intent.preservation_requirements, + "reasoning": "deterministic test interpretation", + }, + "variants": [ + { + "intent": intent.change_type, + "conditioning_strategy": "first_frame", + "description": f"Apply requested edit to {region}", + "tone": "original visual tone unless explicitly changed", + "color_grading": "original grade unless explicitly changed", + "region_emphasis": region, + "prompt_for_video_edit": ( + f"Edit only the selected target in {region}. {prompt}. " + "Preserve all unselected content. Do not regenerate the scene." + ), + } + ], + } + async def _gemini_plan_variants( prompt: str, bbox: BBoxDict, frame_path: str, ) -> list[EditPlan]: - tones = [ - ("cinematic warm", "warm shadows, amber highlights, subtle vignette"), - ("moody cool", "desaturated blues, crushed blacks, cyan highlights"), - ("high contrast", "punchy contrast, saturated mids, clean whites"), - ] - region = f"region at ({bbox['x']:.2f},{bbox['y']:.2f}) {bbox['w']:.2f}x{bbox['h']:.2f}" - return [ - EditPlan( - description=f"{tone} take: {prompt}", - tone=tone, - color_grading=grade, - region_emphasis=region, - prompt_for_runway=f"{prompt}. Focus changes on {region}. Apply {tone} look with {grade}.", - ) - for tone, grade in tones - ] + interpreted = await _gemini_interpret_edit(prompt, bbox, frame_path) + return interpreted["variants"] async def _gemini_score_variant( @@ -91,6 +114,7 @@ async def _gemini_find_entity_in_keyframes( gemini = types.SimpleNamespace( + interpret_edit=_gemini_interpret_edit, plan_variants=_gemini_plan_variants, score_variant=_gemini_score_variant, identify_entity=_gemini_identify_entity, diff --git a/apps/api/app/ai/services/sam.py b/apps/api/app/ai/services/sam.py index f126bf0..e5f24a1 100644 --- a/apps/api/app/ai/services/sam.py +++ b/apps/api/app/ai/services/sam.py @@ -157,6 +157,29 @@ async def is_available() -> bool: return False +async def video_tracking_available() -> bool: + """Check for the full worker used by SAM2 video tracking. + + The lightweight Hugging Face Space only exposes still-image segmentation. + Treating it as a video worker adds a slow, guaranteed-failing request to + every edit. + """ + try: + worker_url = get_settings().vision_worker_url.strip() + if not worker_url or _is_huggingface_space(worker_url): + return False + health_url = ( + worker_url.replace("segment", "health") + if "modal.run" in worker_url + else f"{worker_url.rstrip('/')}/health" + ) + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.get(health_url) + return response.status_code == 200 + except Exception: + return False + + async def track_frames( frame_paths: list[str], *, diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 03bff18..b515c57 100644 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -1,4 +1,5 @@ import time +import logging from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select @@ -10,6 +11,7 @@ select_video_provider, validate_provider_duration, ) +from app.ai import services as ai from app.config.settings import get_settings from app.db.session import get_db from app.deps import get_session @@ -17,7 +19,13 @@ from app.models.project import Project from app.models.selection import SelectionArtifact from app.models.session import Session as SessionModel -from app.schemas.edit_plan import EditCore, EditIntent, LocalRangeResolution +from app.schemas.edit_plan import ( + EditCore, + EditIntent, + GenerationContext, + LocalRangeResolution, + SemanticEditPlan, +) from app.schemas.edit_plan_api import ( EditPlanRequest, EditPlanResponse, @@ -25,10 +33,17 @@ PlanEstimateResponse, ) from app.services.edit_scope import plan_prompt_intent -from app.services.local_range import resolve_local_range +from app.services.local_range import EditWindowLimitError, resolve_local_range +from app.services.global_chunk_planner import ( + PlannedGlobalChunk, + plan_occurrence_chunks, + split_evidence_from_track_frames, +) +from app.services import ffmpeg, storage router = APIRouter(prefix="/edit-plans", tags=["edit-plans"]) +log = logging.getLogger("fiebatt.edit_plans") def _response(plan: EditPlanRecord) -> EditPlanResponse: @@ -39,7 +54,14 @@ def _response(plan: EditPlanRecord) -> EditPlanResponse: id=chunk.id, index=chunk.index, edit_core=EditCore(start_ts=chunk.edit_start, end_ts=chunk.edit_end), - generation_context=resolution.generation_context, + generation_context=GenerationContext( + start_ts=chunk.context_start, + end_ts=chunk.context_end, + edit_core=EditCore( + start_ts=chunk.edit_start, + end_ts=chunk.edit_end, + ), + ), provider=chunk.provider, status=chunk.status, ) @@ -82,19 +104,68 @@ async def create_edit_plan( if selection.source_revision != project.video_url: raise HTTPException(status_code=409, detail="selection is stale for current source") + source_start = body.source_start_ts if body.source_start_ts is not None else 0.0 + source_end = body.source_end_ts if body.source_end_ts is not None else project.duration + if source_end > project.duration + 1e-3: + raise HTTPException(status_code=422, detail="source range exceeds project") + if not source_start - 1e-3 <= selection.frame_ts <= source_end + 1e-3: + raise HTTPException(status_code=422, detail="selection is outside the active source clip") + explicit_core = None if body.explicit_start_ts is not None and body.explicit_end_ts is not None: if body.explicit_end_ts > project.duration + 1e-3: raise HTTPException(status_code=422, detail="explicit range exceeds project") + if ( + body.explicit_start_ts < source_start - 1e-3 + or body.explicit_end_ts > source_end + 1e-3 + ): + raise HTTPException(status_code=422, detail="explicit range exceeds active source clip") explicit_core = EditCore( start_ts=body.explicit_start_ts, end_ts=body.explicit_end_ts, ) + semantic_warning: str | None = None + structured_intent = body.structured_intent + if structured_intent is None: + planning_frame = "" + try: + frame_path, _ = storage.new_path("keyframes", "jpg") + await ffmpeg.extract_frame(project.video_path, selection.frame_ts, frame_path) + planning_frame = str(frame_path) + except Exception: + log.warning("semantic planning frame unavailable", exc_info=True) + try: + interpretation = SemanticEditPlan.model_validate( + await ai.gemini.interpret_edit( + body.prompt, + selection.bbox_json, + planning_frame, + ) + ) + structured_intent = interpretation.as_intent(body.prompt) + except Exception as exc: + log.warning("semantic edit interpretation failed; using safe rules", exc_info=True) + semantic_warning = ( + "semantic prompt interpretation unavailable; used conservative temporal rules " + f"({type(exc).__name__})" + ) + + if structured_intent is not None: + updates: dict[str, str] = {} + if explicit_core is not None: + updates = {"scope": "explicit_range", "duration_policy": "explicit_range"} + elif body.requested_scope is not None: + updates["scope"] = body.requested_scope + if body.requested_scope in {"all_occurrences", "selected_occurrences"}: + updates["duration_policy"] = "all_occurrences" + if updates: + structured_intent = structured_intent.model_copy(update=updates) + gate = plan_prompt_intent( body.prompt, explicit_range=explicit_core is not None, requested_scope=body.requested_scope, - structured_intent=body.structured_intent, + structured_intent=structured_intent, ) if gate.intent.scope in {"all_occurrences", "selected_occurrences"}: raise HTTPException( @@ -104,20 +175,64 @@ async def create_edit_plan( try: resolution = await resolve_local_range( - project, selection, gate.intent, explicit_core=explicit_core + project, + selection, + gate.intent, + explicit_core=explicit_core, + source_start=source_start, + source_end=source_end, ) + except EditWindowLimitError as exc: + raise HTTPException( + status_code=422, + detail={ + "code": "edit_window_too_long", + "message": str(exc), + "limit_seconds": exc.limit, + "detected_seconds": round(exc.duration, 3), + }, + ) from exc except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc - context_duration = resolution.generation_context.duration - provider = select_video_provider( - body.video_gen_provider, - source_video=True, - duration=context_duration, - ) - provider_error = validate_provider_duration(provider, context_duration) - if provider_error: - raise HTTPException(status_code=422, detail=provider_error) + if resolution.edit_core.duration > 30.05: + raise HTTPException( + status_code=422, + detail={ + "code": "edit_window_too_long", + "message": "this edit covers more than 30 seconds; choose a shorter range", + "limit_seconds": 30.0, + "detected_seconds": round(resolution.edit_core.duration, 3), + }, + ) + context = resolution.generation_context + source_edit_request = body.video_gen_provider in {"auto", "wan", "happyhorse"} + planned_chunks = [] + if source_edit_request: + try: + planned_chunks = plan_occurrence_chunks( + occurrence_start=resolution.edit_core.start_ts, + occurrence_end=resolution.edit_core.end_ts, + project_duration=project.duration, + requested_provider=body.video_gen_provider, + split_evidence=split_evidence_from_track_frames( + resolution.tracked_frames + ), + source_start=context.start_ts, + source_end=context.end_ts, + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + provider = planned_chunks[0].provider + else: + provider = select_video_provider( + body.video_gen_provider, + source_video=True, + duration=context.duration, + ) + provider_error = validate_provider_duration(provider, context.duration) + if provider_error: + raise HTTPException(status_code=422, detail=provider_error) capabilities = VIDEO_PROVIDER_CAPABILITIES[provider] provider_reason = ( f"selected {provider} for source-video context" @@ -125,21 +240,33 @@ async def create_edit_plan( else f"selected {provider} image conditioning; source-video context is unsupported" ) warnings = list(resolution.warnings) + if semantic_warning: + warnings.append(semantic_warning) if not get_settings().adaptive_edit_planning: warnings.append( "adaptive generation rollout is disabled; render will use the legacy fixed window" ) if not capabilities.source_video_edit: warnings.append(f"{provider} cannot consume source-video motion context") + if len(planned_chunks) > 1: + warnings.append( + f"edit will render as {len(planned_chunks)} overlapping clips and join at validated seams" + ) + expected_calls = len(planned_chunks) if planned_chunks else 1 + expected_seconds = ( + sum(chunk.context_duration for chunk in planned_chunks) + if planned_chunks + else context.duration + ) estimate = { "analysis_mode": gate.estimate.analysis_mode, "analysis_duration_ms": round( (time.perf_counter() - analysis_started) * 1000.0, 3 ), "frames_inspected": resolution.frames_inspected, - "expected_generation_calls": 1, - "expected_generated_seconds": context_duration, + "expected_generation_calls": expected_calls, + "expected_generated_seconds": expected_seconds, "requires_global_discovery": gate.estimate.requires_global_discovery, } plan = EditPlanRecord( @@ -157,23 +284,51 @@ async def create_edit_plan( ) db.add(plan) await db.flush() - context = resolution.generation_context - chunk = GenerationChunk( - plan_id=plan.id, - index=0, - edit_start=resolution.edit_core.start_ts, - edit_end=resolution.edit_core.end_ts, - context_start=context.start_ts, - context_end=context.end_ts, - provider=provider, - payload_json={ - "selection_id": selection.id, - "subject_reference_url": selection.subject_reference_url, - "mask_url": selection.mask_url, - "source_revision": selection.source_revision, - }, - ) - db.add(chunk) + chunks_to_store = planned_chunks or [ + PlannedGlobalChunk( + index=0, + edit_start=resolution.edit_core.start_ts, + edit_end=resolution.edit_core.end_ts, + context_start=context.start_ts, + context_end=context.end_ts, + provider=provider, + split_reason="occurrence_end", + ) + ] + for chunk in chunks_to_store: + first_chunk = chunk.index == 0 + last_chunk = chunk.index == len(chunks_to_store) - 1 + db.add( + GenerationChunk( + plan_id=plan.id, + index=chunk.index, + edit_start=chunk.edit_start, + edit_end=chunk.edit_end, + context_start=chunk.context_start, + context_end=chunk.context_end, + provider=chunk.provider, + payload_json={ + "selection_id": selection.id, + "subject_reference_url": selection.subject_reference_url, + "mask_url": selection.mask_url, + "source_revision": selection.source_revision, + "split_reason": chunk.split_reason, + "track_frames": [ + frame + for frame in resolution.tracked_frames + if chunk.context_start + <= float(frame.get("timestamp") or 0.0) + <= chunk.context_end + ], + "boundary_contract": { + "protect_source_before": first_chunk, + "protect_source_after": last_chunk, + "handoff_from_previous": not first_chunk, + "handoff_to_next": not last_chunk, + }, + }, + ) + ) await db.commit() loaded = ( await db.execute( diff --git a/apps/api/app/api/routes/generate.py b/apps/api/app/api/routes/generate.py index da4104f..7946919 100644 --- a/apps/api/app/api/routes/generate.py +++ b/apps/api/app/api/routes/generate.py @@ -1,5 +1,7 @@ from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload from app.ai.services.provider_capabilities import validate_provider_duration from app.config.settings import get_settings @@ -13,7 +15,7 @@ from app.schemas.edit_plan import LocalRangeResolution from app.schemas.generate import GenerateRequest, GenerateResponse from app.services.accepted_generation import resolve_committed_timeline_range -from app.workers import generate_job +from app.workers import generate_job, local_chunk_job router = APIRouter(tags=["generate"]) @@ -36,7 +38,13 @@ async def generate( plan: EditPlanRecord | None = None resolution: LocalRangeResolution | None = None if body.plan_id: - plan = await db.get(EditPlanRecord, body.plan_id) + plan = ( + await db.execute( + select(EditPlanRecord) + .where(EditPlanRecord.id == body.plan_id) + .options(selectinload(EditPlanRecord.chunks)) + ) + ).scalar_one_or_none() if plan is None or plan.project_id != proj.id: raise HTTPException(status_code=404, detail="edit plan not found") if plan.source_revision != proj.video_url: @@ -76,8 +84,19 @@ async def generate( reference_frame_ts = float(body.reference_frame_ts) video_gen_provider = body.video_gen_provider or "auto" + chunked_plan = bool(plan is not None and len(plan.chunks) > 1) length = end_ts - start_ts - if length < MIN_SEG_LEN or length > MAX_SEG_LEN: + if length > 30.05: + raise HTTPException( + status_code=422, + detail={ + "code": "edit_window_too_long", + "message": "this edit covers more than 30 seconds; choose a shorter range", + "limit_seconds": 30.0, + "detected_seconds": round(length, 3), + }, + ) + if length < MIN_SEG_LEN or (length > MAX_SEG_LEN and not chunked_plan): raise HTTPException( status_code=422, detail=( @@ -92,7 +111,7 @@ async def generate( if use_plan_range and resolution is not None else length ) - if generation_length > MAX_SEG_LEN: + if generation_length > MAX_SEG_LEN and not chunked_plan: raise HTTPException( status_code=422, detail=( @@ -100,7 +119,13 @@ async def generate( f"{MAX_SEG_LEN:g}s (got {generation_length:.2f}s)" ), ) - if video_gen_provider != "auto": + if chunked_plan: + for chunk in plan.chunks: + chunk_duration = chunk.context_end - chunk.context_start + provider_error = validate_provider_duration(chunk.provider, chunk_duration) + if provider_error: + raise HTTPException(status_code=422, detail=provider_error) + elif video_gen_provider != "auto": provider_error = validate_provider_duration(video_gen_provider, generation_length) if provider_error: raise HTTPException(status_code=422, detail=provider_error) @@ -130,6 +155,7 @@ async def generate( "video_gen_provider": video_gen_provider, "plan_id": plan.id if plan else None, "planned_context": plan.range_json if plan else None, + "planned_intent": plan.intent_json if plan else None, "adaptive_context_enabled": use_plan_range, "target_clip_id": body.target_clip_id, "plan_scope": plan.scope if plan else "legacy", @@ -143,6 +169,7 @@ async def generate( if plan else 0 ), + "chunk_count": len(plan.chunks) if plan else 1, "fixed_window_baseline_seconds": ( float(body.end_ts) - float(body.start_ts) if body.start_ts is not None and body.end_ts is not None @@ -158,6 +185,7 @@ async def generate( await db.commit() await db.refresh(job) - runner.submit(job.id, lambda: generate_job.run(job.id)) + worker = local_chunk_job.run if chunked_plan else generate_job.run + runner.submit(job.id, lambda: worker(job.id)) return GenerateResponse(job_id=job.id) diff --git a/apps/api/app/config/settings.py b/apps/api/app/config/settings.py index ee5fc56..b5534db 100644 --- a/apps/api/app/config/settings.py +++ b/apps/api/app/config/settings.py @@ -46,8 +46,9 @@ class Settings(BaseSettings): # when true, worker calls resolve to ai/services/_stubs.py use_ai_stubs: bool = True - # Plans can be previewed before PR2 changes generation semantics. - adaptive_edit_planning: bool = False + # Local requests use a prompt- and tracking-aware window by default. + # Operators can still disable this as an emergency fallback. + adaptive_edit_planning: bool = True # PR3 global discovery/chunk orchestration. Local and legacy paths stay # unchanged until this independent rollout switch is enabled. global_edit_planning: bool = False diff --git a/apps/api/app/schemas/edit_plan.py b/apps/api/app/schemas/edit_plan.py index bcf2fdd..f71bdcd 100644 --- a/apps/api/app/schemas/edit_plan.py +++ b/apps/api/app/schemas/edit_plan.py @@ -14,17 +14,69 @@ "all_occurrences", ] ChangeType = Literal["appearance", "removal", "replacement", "motion", "scene"] +DurationPolicy = Literal[ + "bounded_action", + "continuous_occurrence", + "explicit_range", + "all_occurrences", +] + + +class GroundedEditInstruction(BaseModel): + intent: str + conditioning_strategy: Literal["first_frame", "text_only"] = "first_frame" + description: str + tone: str = "original visual tone unless explicitly changed" + color_grading: str = "original grade unless explicitly changed" + region_emphasis: str + prompt_for_video_edit: str = Field(min_length=1, max_length=4000) + + +class SemanticEditDecision(BaseModel): + scope: EditScope + change_type: ChangeType + duration_policy: DurationPolicy + 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) + requires_recovery_motion: bool = False + preservation_requirements: list[str] = Field(default_factory=list) + reasoning: str = Field(min_length=1, max_length=1000) + + +class SemanticEditPlan(BaseModel): + decision: SemanticEditDecision + variants: list[GroundedEditInstruction] = Field(min_length=1, max_length=1) + + def as_intent(self, raw_prompt: str) -> "EditIntent": + decision = self.decision + return EditIntent( + raw_prompt=raw_prompt, + scope=decision.scope, + change_type=decision.change_type, + duration_policy=decision.duration_policy, + target_description=decision.target_description, + action_phases=decision.action_phases, + estimated_action_seconds=decision.estimated_action_seconds, + requires_recovery_motion=decision.requires_recovery_motion, + preservation_requirements=decision.preservation_requirements, + reasoning=decision.reasoning, + grounded_edit=self.variants[0], + ) class EditIntent(BaseModel): raw_prompt: str = Field(min_length=1, max_length=2000) scope: EditScope = "local" change_type: ChangeType + duration_policy: DurationPolicy = "bounded_action" target_description: str | None = None action_phases: list[str] = Field(default_factory=list) estimated_action_seconds: float = Field(default=3.0, gt=0.0) requires_recovery_motion: bool = False preservation_requirements: list[str] = Field(default_factory=list) + reasoning: str | None = None + grounded_edit: GroundedEditInstruction | None = None class EditCore(BaseModel): @@ -78,6 +130,7 @@ class LocalRangeResolution(BaseModel): analysis_start: float analysis_end: float frames_inspected: int = 0 + tracked_frames: list[dict] = Field(default_factory=list) confidence: float = Field(ge=0.0, le=1.0) warnings: list[str] = Field(default_factory=list) diff --git a/apps/api/app/schemas/edit_plan_api.py b/apps/api/app/schemas/edit_plan_api.py index ace854a..5f2a8f8 100644 --- a/apps/api/app/schemas/edit_plan_api.py +++ b/apps/api/app/schemas/edit_plan_api.py @@ -20,6 +20,8 @@ class EditPlanRequest(BaseModel): requested_scope: EditScope | None = None explicit_start_ts: float | None = Field(default=None, ge=0.0) explicit_end_ts: float | None = Field(default=None, gt=0.0) + source_start_ts: float | None = Field(default=None, ge=0.0) + source_end_ts: float | None = Field(default=None, gt=0.0) video_gen_provider: VideoProvider = "auto" structured_intent: EditIntent | None = None @@ -30,6 +32,11 @@ def validate_explicit_range(self) -> "EditPlanRequest": raise ValueError("explicit_start_ts and explicit_end_ts must be supplied together") if supplied and self.explicit_end_ts <= self.explicit_start_ts: raise ValueError("explicit_end_ts must be after explicit_start_ts") + source_supplied = self.source_start_ts is not None or self.source_end_ts is not None + if source_supplied and (self.source_start_ts is None or self.source_end_ts is None): + raise ValueError("source_start_ts and source_end_ts must be supplied together") + if source_supplied and self.source_end_ts <= self.source_start_ts: + raise ValueError("source_end_ts must be after source_start_ts") return self diff --git a/apps/api/app/services/agent_tools.py b/apps/api/app/services/agent_tools.py index 9918211..0268bb4 100644 --- a/apps/api/app/services/agent_tools.py +++ b/apps/api/app/services/agent_tools.py @@ -657,6 +657,7 @@ async def _generate_edit( "video_gen_provider": video_gen_provider or "auto", "plan_id": plan.id if plan else None, "planned_context": plan.range_json if plan else None, + "planned_intent": plan.intent_json if plan else None, "adaptive_context_enabled": use_plan_range, }, ) diff --git a/apps/api/app/services/cpu_tracking.py b/apps/api/app/services/cpu_tracking.py new file mode 100644 index 0000000..7aaa2de --- /dev/null +++ b/apps/api/app/services/cpu_tracking.py @@ -0,0 +1,242 @@ +"""Bounded local object tracking without a separate GPU service.""" +from __future__ import annotations + +import asyncio +from pathlib import Path + +import cv2 +import numpy as np + +from app.ai.services.sam import TrackResult + + +MIN_APPEARANCE_SCORE = 0.05 +MIN_BOX_PIXELS = 8 +MAX_CENTER_JUMP_BOXES = 1.5 + + +def _normalized_box( + pixel_box: tuple[float, float, float, float], + width: int, + height: int, +) -> dict[str, float]: + x, y, box_width, box_height = pixel_box + return { + "x": max(0.0, x / width), + "y": max(0.0, y / height), + "w": min(1.0, box_width / width), + "h": min(1.0, box_height / height), + } + + +def _pixel_box( + bbox: dict[str, float], width: int, height: int +) -> tuple[float, float, float, float]: + x = max(0, min(width - 1, round(float(bbox["x"]) * width))) + y = max(0, min(height - 1, round(float(bbox["y"]) * height))) + right = max(x + 1, min(width, round((float(bbox["x"]) + float(bbox["w"])) * width))) + bottom = max(y + 1, min(height, round((float(bbox["y"]) + float(bbox["h"])) * height))) + return float(x), float(y), float(right - x), float(bottom - y) + + +def _box_from_mask( + seed_mask_path: str | None, + *, + width: int, + height: int, +) -> tuple[float, float, float, float] | None: + if not seed_mask_path or not Path(seed_mask_path).is_file(): + return None + mask = cv2.imread(seed_mask_path, cv2.IMREAD_GRAYSCALE) + if mask is None: + return None + if mask.shape != (height, width): + mask = cv2.resize(mask, (width, height), interpolation=cv2.INTER_NEAREST) + points = cv2.findNonZero((mask > 127).astype(np.uint8)) + if points is None: + return None + x, y, box_width, box_height = cv2.boundingRect(points) + if box_width < MIN_BOX_PIXELS or box_height < MIN_BOX_PIXELS: + return None + return float(x), float(y), float(box_width), float(box_height) + + +def _histogram(frame: np.ndarray, box: tuple[float, float, float, float]) -> np.ndarray: + height, width = frame.shape[:2] + x, y, box_width, box_height = box + left = max(0, min(width - 1, round(x))) + top = max(0, min(height - 1, round(y))) + right = max(left + 1, min(width, round(x + box_width))) + bottom = max(top + 1, min(height, round(y + box_height))) + crop = frame[top:bottom, left:right] + hsv = cv2.cvtColor(crop, cv2.COLOR_BGR2HSV) + histogram = cv2.calcHist([hsv], [0, 1], None, [24, 24], [0, 180, 0, 256]) + return cv2.normalize(histogram, histogram).flatten() + + +def _valid_box( + box: tuple[float, float, float, float], width: int, height: int +) -> bool: + x, y, box_width, box_height = box + return ( + np.isfinite(box).all() + and box_width >= MIN_BOX_PIXELS + and box_height >= MIN_BOX_PIXELS + and x < width + and y < height + and x + box_width > 0 + and y + box_height > 0 + and box_width * box_height < width * height * 0.85 + ) + + +def _track_direction( + frames: list[np.ndarray], + *, + seed_index: int, + seed_box: tuple[float, float, float, float], + seed_histogram: np.ndarray, + step: int, +) -> dict[int, dict]: + tracker = cv2.TrackerMIL_create() + tracker.init(frames[seed_index], tuple(round(value) for value in seed_box)) + height, width = frames[seed_index].shape[:2] + output: dict[int, dict] = {} + previous_box = seed_box + index = seed_index + step + while 0 <= index < len(frames): + tracked, raw_box = tracker.update(frames[index]) + box = tuple(float(value) for value in raw_box) + if not tracked or not _valid_box(box, width, height): + output[index] = { + "frame_index": index, + "state": "lost", + "confidence": 0.0, + "bbox": None, + } + break + previous_center = ( + previous_box[0] + previous_box[2] / 2, + previous_box[1] + previous_box[3] / 2, + ) + current_center = (box[0] + box[2] / 2, box[1] + box[3] / 2) + center_jump = ( + (current_center[0] - previous_center[0]) ** 2 + + (current_center[1] - previous_center[1]) ** 2 + ) ** 0.5 + allowed_jump = MAX_CENTER_JUMP_BOXES * max( + MIN_BOX_PIXELS, + min(previous_box[2], previous_box[3]), + ) + if center_jump > allowed_jump: + output[index] = { + "frame_index": index, + "state": "lost", + "confidence": 0.0, + "bbox": None, + } + break + appearance = cv2.compareHist( + seed_histogram, + _histogram(frames[index], box), + cv2.HISTCMP_CORREL, + ) + confidence = max(0.0, min(1.0, (float(appearance) + 1.0) / 2.0)) + if appearance < MIN_APPEARANCE_SCORE: + output[index] = { + "frame_index": index, + "state": "lost", + "confidence": confidence, + "bbox": None, + } + break + output[index] = { + "frame_index": index, + "state": "tracked", + "confidence": confidence, + "bbox": _normalized_box(box, width, height), + } + previous_box = box + index += step + return output + + +def _track( + frame_paths: list[str], + *, + seed_frame_index: int, + bbox: dict[str, float], + seed_mask_path: str | None, +) -> TrackResult: + frames = [cv2.imread(path, cv2.IMREAD_COLOR) for path in frame_paths] + if any(frame is None for frame in frames): + raise ValueError("local tracker could not read analysis frames") + values = [frame for frame in frames if frame is not None] + if not 0 <= seed_frame_index < len(values): + raise ValueError("seed frame is outside analysis frames") + height, width = values[seed_frame_index].shape[:2] + seed_box = _box_from_mask( + seed_mask_path, + width=width, + height=height, + ) or _pixel_box(bbox, width, height) + seed_histogram = _histogram(values[seed_frame_index], seed_box) + output = { + seed_frame_index: { + "frame_index": seed_frame_index, + "state": "tracked", + "confidence": 1.0, + "bbox": _normalized_box(seed_box, width, height), + } + } + output.update( + _track_direction( + values, + seed_index=seed_frame_index, + seed_box=seed_box, + seed_histogram=seed_histogram, + step=-1, + ) + ) + output.update( + _track_direction( + values, + seed_index=seed_frame_index, + seed_box=seed_box, + seed_histogram=seed_histogram, + step=1, + ) + ) + ordered = [output[index] for index in sorted(output)] + return TrackResult( + tracker="opencv_mil", + frames=ordered, + processed_start_index=ordered[0]["frame_index"], + processed_end_index=ordered[-1]["frame_index"], + ) + + +async def track_frames( + frame_paths: list[str], + *, + seed_frame_index: int, + bbox: dict[str, float] | None = None, + seed_mask_path: str | None = None, + max_frames: int = 240, + max_seconds: float = 30.0, + include_masks: bool = False, +) -> TrackResult: + """Track selected target locally; exact per-frame masks are not fabricated.""" + del max_seconds + if include_masks: + raise ValueError("CPU occurrence tracker does not produce precise video masks") + if bbox is None: + raise ValueError("CPU occurrence tracker requires a bounding box") + bounded = frame_paths[:max_frames] + return await asyncio.to_thread( + _track, + bounded, + seed_frame_index=seed_frame_index, + bbox=bbox, + seed_mask_path=seed_mask_path, + ) diff --git a/apps/api/app/services/dense_occurrence.py b/apps/api/app/services/dense_occurrence.py index f0187cd..8df1c88 100644 --- a/apps/api/app/services/dense_occurrence.py +++ b/apps/api/app/services/dense_occurrence.py @@ -20,6 +20,7 @@ OccurrenceTrack, ) from app.services import ffmpeg +from app.services import cpu_tracking from app.services.local_range import detect_shot_span, tracked_span @@ -112,7 +113,7 @@ async def refine_occurrence_candidate( reference_crop_path: str, extract_frame: Callable[[str | Path, float, str | Path], Awaitable[Path]] = ffmpeg.extract_frame, locate_bbox: Callable[..., Awaitable[tuple[dict[str, float] | None, float]]] = locate_reference_bbox, - track_frames: Callable[..., Awaitable[sam.TrackResult]] = sam.track_frames, + track_frames: Callable[..., Awaitable[sam.TrackResult]] = cpu_tracking.track_frames, ) -> DenseTrackEvidence: start = max(0.0, candidate.keyframe_ts - DENSE_TRACK_RADIUS_SECONDS) end = min(project_duration, candidate.keyframe_ts + DENSE_TRACK_RADIUS_SECONDS) @@ -159,11 +160,11 @@ async def refine_occurrence_candidate( include_masks=False, ) - if track.tracker != "sam2_video" or track.cancelled: + if track.tracker not in {"sam2_video", "opencv_mil"} or track.cancelled: return DenseTrackEvidence( candidate.id, False, - "real SAM2 tracking is unavailable or incomplete", + "subject tracking is unavailable or incomplete", candidate.keyframe_ts, candidate.keyframe_ts, candidate.keyframe_ts, diff --git a/apps/api/app/services/edit_prompt.py b/apps/api/app/services/edit_prompt.py new file mode 100644 index 0000000..f538705 --- /dev/null +++ b/apps/api/app/services/edit_prompt.py @@ -0,0 +1,22 @@ +"""Build provider prompts without weakening the user's exact request.""" +from __future__ import annotations + +from typing import Any + + +def planned_edit_prompt(user_prompt: str, plan: dict[str, Any]) -> str: + planned = str( + plan.get("prompt_for_video_edit") + or plan.get("prompt_for_runway") + or plan.get("prompt_for_veo") + or "" + ).strip() + requirement = user_prompt.strip() + if not planned or planned.casefold() == requirement.casefold(): + return requirement + return ( + "NON-NEGOTIABLE USER REQUIREMENT:\n" + f"{requirement}\n\n" + "GROUNDED VIDEO-EDIT INSTRUCTION:\n" + f"{planned}" + ) diff --git a/apps/api/app/services/edit_scope.py b/apps/api/app/services/edit_scope.py index 17914c2..c9d14b5 100644 --- a/apps/api/app/services/edit_scope.py +++ b/apps/api/app/services/edit_scope.py @@ -17,7 +17,7 @@ re.IGNORECASE, ) _MOTION_RE = re.compile( - r"\b(jump|run|walk|dance|wave|turn|spin|sit|stand|kick|throw|catch|" + r"\b(jump|bounce|run|walk|dance|wave|turn|spin|sit|stand|kick|throw|catch|" r"clap|nod|bow|fall|leap)(s|ed|ing)?\b", re.IGNORECASE, ) @@ -106,11 +106,25 @@ def plan_prompt_intent( persistent = bool(_PERSISTENT_RE.search(prompt)) preservation = ["preserve unedited subjects", "preserve camera and background"] if persistent: - preservation.append("apply change for complete visible occurrence") + if "apply change for complete visible occurrence" not in preservation: + preservation.append("apply change for complete visible occurrence") + if scope == "all_occurrences": + duration_policy = "all_occurrences" + elif scope == "explicit_range": + duration_policy = "explicit_range" + elif change_type == "motion": + duration_policy = "bounded_action" + else: + # State changes should not visibly revert while the selected target + # remains in the same continuous appearance. + duration_policy = "continuous_occurrence" + if "apply change for complete visible occurrence" not in preservation: + preservation.append("apply change for complete visible occurrence") intent = EditIntent( raw_prompt=prompt, scope=scope, change_type=change_type, + duration_policy=duration_policy, action_phases=phases, estimated_action_seconds=action_seconds, requires_recovery_motion=recovery, diff --git a/apps/api/app/services/ffmpeg.py b/apps/api/app/services/ffmpeg.py index 5e4c4e1..a5fa098 100644 --- a/apps/api/app/services/ffmpeg.py +++ b/apps/api/app/services/ffmpeg.py @@ -604,6 +604,49 @@ async def extract_frame(src: str | Path, ts: float, out: str | Path) -> Path: return out +async def extract_sampled_frames( + src: str | Path, + *, + start_ts: float, + end_ts: float, + fps: float, + output_dir: str | Path, + max_width: int = 640, +) -> tuple[list[str], list[float]]: + """Decode a bounded analysis window once instead of one process per frame.""" + if start_ts < 0.0 or end_ts <= start_ts: + raise ValueError("sample window must have positive duration") + if fps <= 0.0: + raise ValueError("sample fps must be positive") + destination = Path(output_dir) + destination.mkdir(parents=True, exist_ok=True) + pattern = destination / "frame_%06d.jpg" + sample_duration = end_ts - start_ts + 0.5 / fps + cmd = [ + "ffmpeg", + "-y", + "-ss", + f"{start_ts:.6f}", + "-i", + str(src), + "-t", + f"{sample_duration:.6f}", + "-vf", + f"fps={fps:.6f},scale=w=min({max_width}\\,iw):h=-2", + "-q:v", + "3", + "-start_number", + "0", + str(pattern), + ] + await _run(cmd) + paths = sorted(destination.glob("frame_*.jpg")) + if not paths: + raise FfmpegError(cmd, "ffmpeg produced no analysis frames", 0) + timestamps = [min(end_ts, start_ts + index / fps) for index in range(len(paths))] + return [str(path) for path in paths], timestamps + + async def crop_bbox_from_frame( frame_path: str | Path, bbox: dict[str, float], diff --git a/apps/api/app/services/global_chunk_execution.py b/apps/api/app/services/global_chunk_execution.py index 33347f2..c37446f 100644 --- a/apps/api/app/services/global_chunk_execution.py +++ b/apps/api/app/services/global_chunk_execution.py @@ -7,7 +7,6 @@ from urllib.parse import urlparse from app.ai import services as ai -from app.ai.services import sam from app.ai.services.provider_capabilities import select_source_edit_mode from app.config.settings import get_settings from app.models.propagation import GlobalGenerationChunk @@ -204,19 +203,13 @@ async def execute_global_chunk( seed_offset = min(max(0.0, midpoint - chunk.context_start), duration) await ffmpeg.extract_frame(source_path, seed_offset, target_frame_path) - mask_url: str | None = None mask_frame_id = max(1, round(seed_offset * float(project.fps or 1.0)) + 1) - if chunk.provider == "wan" and duration <= 5.001 and await sam.is_available(): - mask_path = await sam.bbox_to_mask(str(target_frame_path), bbox) - mask_url = await storage.publish(Path(mask_path), content_type="image/png") - if not settings.use_ai_stubs and not _is_public_http_url(mask_url): - mask_url = None edit_mode = select_source_edit_mode( chunk.provider, duration=duration, source_video=True, - mask_available=mask_url is not None, + mask_available=False, ) window = GenerationWindow( core_start=chunk.edit_start, @@ -239,7 +232,7 @@ async def execute_global_chunk( }, subject_reference_path=str(reference_subject_path), source_video_url=source_url, - mask_image_url=mask_url if edit_mode == "tracked_mask" else None, + mask_image_url=None, mask_frame_id=mask_frame_id, duration=math.ceil(duration - 1e-6), resolution="720P", @@ -267,6 +260,6 @@ async def execute_global_chunk( "context_duration": round(duration, 3), "target_bbox": bbox, "handoff_used": previous is not None, - "mask_used": mask_url is not None and edit_mode == "tracked_mask", + "mask_used": False, }, ) diff --git a/apps/api/app/services/global_chunk_planner.py b/apps/api/app/services/global_chunk_planner.py index 03c4f85..b8a21e7 100644 --- a/apps/api/app/services/global_chunk_planner.py +++ b/apps/api/app/services/global_chunk_planner.py @@ -118,13 +118,20 @@ def plan_occurrence_chunks( project_duration: float, requested_provider: str = "auto", split_evidence: list[SplitEvidence] | None = None, + source_start: float = 0.0, + source_end: float | None = None, ) -> list[PlannedGlobalChunk]: + source_upper = ( + project_duration if source_end is None else min(project_duration, source_end) + ) if occurrence_end <= occurrence_start: raise ValueError("occurrence must have positive duration") - if occurrence_start < 0 or occurrence_end > project_duration + 0.05: + if source_start < 0 or source_upper <= source_start: + raise ValueError("source range must have positive duration") + if occurrence_start < source_start - 0.05 or occurrence_end > source_upper + 0.05: raise ValueError("occurrence is outside the source video") - full_context_start = max(0.0, occurrence_start - HANDLE_SECONDS) - full_context_end = min(project_duration, occurrence_end + HANDLE_SECONDS) + full_context_start = max(source_start, occurrence_start - HANDLE_SECONDS) + full_context_end = min(source_upper, occurrence_end + HANDLE_SECONDS) provider = _provider_for_occurrence( requested_provider, full_context_end - full_context_start ) @@ -136,11 +143,11 @@ def plan_occurrence_chunks( core_ranges: list[tuple[float, float, str]] = [] cursor = occurrence_start while ( - min(project_duration, occurrence_end + HANDLE_SECONDS) - - max(0.0, cursor - HANDLE_SECONDS) + min(source_upper, occurrence_end + HANDLE_SECONDS) + - max(source_start, cursor - HANDLE_SECONDS) > max_context + 1e-6 ): - context_start = max(0.0, cursor - HANDLE_SECONDS) + context_start = max(source_start, cursor - HANDLE_SECONDS) target = context_start + max_context - HANDLE_SECONDS lower = cursor + MIN_CORE_SECONDS upper = min( @@ -162,17 +169,17 @@ def plan_occurrence_chunks( chunks: list[PlannedGlobalChunk] = [] minimum_total = MIN_TOTAL_SECONDS.get(provider, 0.0) for index, (start, end, reason) in enumerate(core_ranges): - context_start = max(0.0, start - HANDLE_SECONDS) - context_end = min(project_duration, end + HANDLE_SECONDS) + context_start = max(source_start, start - HANDLE_SECONDS) + context_end = min(source_upper, end + HANDLE_SECONDS) missing = minimum_total - (context_end - context_start) if missing > 0: - grow_left = min(context_start, missing / 2) + grow_left = min(context_start - source_start, missing / 2) context_start -= grow_left missing -= grow_left - grow_right = min(project_duration - context_end, missing) + grow_right = min(source_upper - context_end, missing) context_end += grow_right missing -= grow_right - context_start = max(0.0, context_start - missing) + context_start = max(source_start, context_start - missing) chunks.append( PlannedGlobalChunk( index=index, diff --git a/apps/api/app/services/local_range.py b/apps/api/app/services/local_range.py index 0c975c4..3c80c79 100644 --- a/apps/api/app/services/local_range.py +++ b/apps/api/app/services/local_range.py @@ -20,27 +20,52 @@ LocalRangeResolution, ) from app.services import ffmpeg, storage +from app.services import cpu_tracking ANALYSIS_FPS = 4.0 +CONTINUOUS_ANALYSIS_FPS = 2.0 HANDLE_SECONDS = 0.75 LOCAL_MARGIN_SECONDS = 1.0 -PERSISTENT_TRACK_BUDGET_SECONDS = 30.0 +INITIAL_CONTINUOUS_RADIUS_SECONDS = 3.0 +MAX_CONTINUOUS_EDIT_SECONDS = 30.0 +CONTINUOUS_EDGE_PROBE_SECONDS = 1.0 SHOT_CHANGE_THRESHOLD = 0.32 _RANGE_CACHE: dict[str, LocalRangeResolution] = {} +class EditWindowLimitError(ValueError): + def __init__(self, duration: float, limit: float = MAX_CONTINUOUS_EDIT_SECONDS): + self.duration = duration + self.limit = limit + super().__init__( + f"this edit covers more than {limit:g} seconds; choose a shorter range" + ) + + def _persistent_change(intent: EditIntent) -> bool: - return any("complete visible occurrence" in item for item in intent.preservation_requirements) + return intent.duration_policy == "continuous_occurrence" -def analysis_window(intent: EditIntent, seed_ts: float, duration: float) -> tuple[float, float]: +def analysis_window( + intent: EditIntent, + seed_ts: float, + duration: float, + *, + source_start: float = 0.0, + source_end: float | None = None, +) -> tuple[float, float]: + upper = duration if source_end is None else min(duration, source_end) + if source_start < 0.0 or upper <= source_start: + 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): - radius = PERSISTENT_TRACK_BUDGET_SECONDS / 2 + radius = INITIAL_CONTINUOUS_RADIUS_SECONDS else: radius = intent.estimated_action_seconds / 2 + HANDLE_SECONDS + LOCAL_MARGIN_SECONDS - return max(0.0, seed_ts - radius), min(duration, seed_ts + radius) + return max(source_start, seed_ts - radius), min(upper, seed_ts + radius) def _fit_interval(center: float, length: float, lower: float, upper: float) -> tuple[float, float]: @@ -70,15 +95,25 @@ def resolve_window_from_evidence( frames_inspected: int, explicit_core: EditCore | None = None, tracking_reached_budget: bool = False, + source_start: float = 0.0, + source_end: float | None = None, ) -> LocalRangeResolution: - occurrence_start = max(0.0, shot_start, tracked_start or 0.0) - occurrence_end = min(duration, shot_end, tracked_end or duration) + upper = duration if source_end is None else min(duration, source_end) + tracked_lower = tracked_start if tracked_start is not None else source_start + tracked_upper = tracked_end if tracked_end is not None else upper + occurrence_start = max(source_start, shot_start, tracked_lower) + occurrence_end = min(upper, shot_end, tracked_upper) warnings: list[str] = [] if occurrence_end <= occurrence_start: occurrence_start, occurrence_end = analysis_start, analysis_end warnings.append("target track was inconclusive; using analyzed shot window") if explicit_core is not None: + if ( + explicit_core.start_ts < source_start - 1e-3 + or explicit_core.end_ts > upper + 1e-3 + ): + raise ValueError("explicit range exceeds active source clip") core_start = max(occurrence_start, explicit_core.start_ts) core_end = min(occurrence_end, explicit_core.end_ts) if core_end <= core_start: @@ -95,8 +130,10 @@ def resolve_window_from_evidence( occurrence_end, ) - context_start = max(occurrence_start, core_start - HANDLE_SECONDS) - context_end = min(occurrence_end, core_end + HANDLE_SECONDS) + # Context belongs to the surrounding shot, not only frames where target + # is visible. This preserves real entrance/exit motion for state changes. + context_start = max(source_start, shot_start, core_start - HANDLE_SECONDS) + context_end = min(upper, shot_end, core_end + HANDLE_SECONDS) pre_handle = core_start - context_start post_handle = context_end - core_end if pre_handle < HANDLE_SECONDS - 0.05: @@ -162,7 +199,11 @@ def tracked_span( def _cache_key( - selection: SelectionArtifact, intent: EditIntent, explicit_core: EditCore | None + selection: SelectionArtifact, + intent: EditIntent, + explicit_core: EditCore | None, + source_start: float, + source_end: float, ) -> str: value = json.dumps( { @@ -170,6 +211,8 @@ def _cache_key( "source_revision": selection.source_revision, "intent": intent.model_dump(mode="json"), "explicit_core": explicit_core.model_dump() if explicit_core else None, + "source_start": source_start, + "source_end": source_end, }, sort_keys=True, ) @@ -183,9 +226,12 @@ async def resolve_local_range( *, explicit_core: EditCore | None = None, extract_frame: Callable[[str | Path, float, str | Path], Awaitable[Path]] = ffmpeg.extract_frame, - track_frames: Callable[..., Awaitable[sam.TrackResult]] = sam.track_frames, + track_frames: Callable[..., Awaitable[sam.TrackResult]] = cpu_tracking.track_frames, + source_start: float = 0.0, + source_end: float | None = None, ) -> LocalRangeResolution: - key = _cache_key(selection, intent, explicit_core) + bounded_end = project.duration if source_end is None else min(project.duration, source_end) + key = _cache_key(selection, intent, explicit_core, source_start, bounded_end) cached = _RANGE_CACHE.get(key) if cached is not None: return cached.model_copy(deep=True) @@ -200,68 +246,193 @@ async def resolve_local_range( seed_mask_path = str(await storage.path_from_url(mask_url)) except Exception: seed_mask_path = None - start, end = analysis_window(intent, selection.frame_ts, project.duration) - frame_count = max(2, math.floor((end - start) * ANALYSIS_FPS) + 1) - timestamps = [ - min(end, start + index / ANALYSIS_FPS) for index in range(frame_count) - ] - seed_index = min(range(len(timestamps)), key=lambda index: abs(timestamps[index] - selection.frame_ts)) - + start, end = analysis_window( + intent, + selection.frame_ts, + project.duration, + source_start=source_start, + source_end=bounded_end, + ) + analysis_fps = CONTINUOUS_ANALYSIS_FPS if _persistent_change(intent) else ANALYSIS_FPS + total_frames_inspected = 0 + track: sam.TrackResult | None = None + shot_start = start + shot_end = end + track_start: float | None = None + track_end: float | None = None + observed_track_start: float | None = None + observed_track_end: float | None = None + observed_shot_start = start + observed_shot_end = end + observed_frames: dict[float, dict] = {} with tempfile.TemporaryDirectory(prefix="fiebatt-local-range-") as temp_dir: - frame_paths = [] - for index, timestamp in enumerate(timestamps): - path = Path(temp_dir) / f"{index:06d}.jpg" - await extract_frame(source, timestamp, path) - frame_paths.append(str(path)) - shot_start, shot_end = detect_shot_span(frame_paths, timestamps, seed_index) - try: - track = await track_frames( - frame_paths, - seed_frame_index=seed_index, - bbox=selection.bbox_json, - seed_mask_path=seed_mask_path, - max_frames=len(frame_paths), - max_seconds=PERSISTENT_TRACK_BUDGET_SECONDS, - include_masks=False, + iteration = 0 + while True: + frame_dir = Path(temp_dir) / f"window-{iteration}" + if extract_frame is ffmpeg.extract_frame: + frame_paths, timestamps = await ffmpeg.extract_sampled_frames( + source, + start_ts=start, + end_ts=end, + fps=analysis_fps, + output_dir=frame_dir, + ) + else: + frame_count = max(2, math.floor((end - start) * analysis_fps) + 1) + timestamps = [ + min(end, start + index / analysis_fps) + for index in range(frame_count) + ] + frame_paths = [] + frame_dir.mkdir(parents=True, exist_ok=True) + for index, timestamp in enumerate(timestamps): + path = frame_dir / f"{index:06d}.jpg" + await extract_frame(source, timestamp, path) + frame_paths.append(str(path)) + total_frames_inspected += len(timestamps) + seed_index = min( + range(len(timestamps)), + key=lambda index: abs(timestamps[index] - selection.frame_ts), + ) + shot_start, shot_end = detect_shot_span( + frame_paths, timestamps, seed_index + ) + try: + track = await track_frames( + frame_paths, + seed_frame_index=seed_index, + bbox=selection.bbox_json, + seed_mask_path=seed_mask_path, + max_frames=len(frame_paths), + max_seconds=30.0, + include_masks=False, + ) + except Exception as exc: + if _persistent_change(intent): + raise ValueError( + "could not reliably track the selected subject; please choose a tighter box" + ) from exc + track = sam.TrackResult( + tracker="bbox_fallback", + frames=[ + { + "frame_index": index, + "state": "tracked", + "confidence": 0.0, + } + for index in range(len(frame_paths)) + ], + processed_start_index=0, + processed_end_index=len(frame_paths) - 1, + warning=( + "video tracking unavailable; bbox fallback used with shot boundaries " + f"({type(exc).__name__})" + ), + ) + track_start, track_end = tracked_span( + track.frames, timestamps, seed_index ) - except Exception as exc: - track = sam.TrackResult( - tracker="bbox_fallback", - frames=[ - { - "frame_index": index, - "state": "tracked", - "confidence": 0.0, + if track_start is not None: + observed_track_start = ( + track_start + if observed_track_start is None + else min(observed_track_start, track_start) + ) + if track_end is not None: + observed_track_end = ( + track_end + if observed_track_end is None + else max(observed_track_end, track_end) + ) + observed_shot_start = min(observed_shot_start, shot_start) + observed_shot_end = max(observed_shot_end, shot_end) + for frame in track.frames: + index = int(frame.get("frame_index", -1)) + if 0 <= index < len(timestamps): + timestamp = timestamps[index] + observed_frames[timestamp] = { + "timestamp": timestamp, + "state": frame.get("state"), + "confidence": frame.get("confidence"), + "bbox": frame.get("bbox") or selection.bbox_json, } - for index in range(len(frame_paths)) - ], - processed_start_index=0, - processed_end_index=len(frame_paths) - 1, - warning=f"video tracking unavailable; bbox fallback used ({type(exc).__name__})", + if not _persistent_change(intent): + break + + tolerance = 1.1 / analysis_fps + first_ts = timestamps[0] + last_ts = timestamps[-1] + expand_left = ( + start > source_start + 1e-3 + and track_start is not None + and abs(track_start - first_ts) <= tolerance + and abs(shot_start - first_ts) <= tolerance + ) + expand_right = ( + end < bounded_end - 1e-3 + and track_end is not None + and abs(track_end - last_ts) <= tolerance + and abs(shot_end - last_ts) <= tolerance ) + if not expand_left and not expand_right: + break + span = end - start + max_window = min( + bounded_end - source_start, + MAX_CONTINUOUS_EDIT_SECONDS + CONTINUOUS_EDGE_PROBE_SECONDS, + ) + if span >= max_window - 1e-3: + if expand_left and expand_right: + break + if expand_right: + shift = min(span, selection.frame_ts - start) + next_end = min(bounded_end, end + shift) + next_start = max(source_start, next_end - max_window) + else: + shift = min(span, end - selection.frame_ts) + next_start = max(source_start, start - shift) + next_end = min(bounded_end, next_start + max_window) + else: + growth = min(span, max_window - span) + sides = int(expand_left) + int(expand_right) + per_side = growth / sides + next_start = ( + max(source_start, start - per_side) if expand_left else start + ) + next_end = min(bounded_end, end + per_side) if expand_right else end + if abs(next_start - start) < 1e-6 and abs(next_end - end) < 1e-6: + break + start, end = next_start, next_end + iteration += 1 - track_start, track_end = tracked_span(track.frames, timestamps, seed_index) + assert track is not None resolution = resolve_window_from_evidence( intent=intent, seed_ts=selection.frame_ts, duration=project.duration, analysis_start=start, analysis_end=end, - shot_start=shot_start, - shot_end=shot_end, - tracked_start=track_start, - tracked_end=track_end, - frames_inspected=len(timestamps), + shot_start=observed_shot_start, + shot_end=observed_shot_end, + tracked_start=observed_track_start, + tracked_end=observed_track_end, + frames_inspected=total_frames_inspected, explicit_core=explicit_core, - tracking_reached_budget=( - _persistent_change(intent) - and track_start is not None - and track_end is not None - and start > 0.0 - and end < project.duration - and abs(track_start - start) < 1 / ANALYSIS_FPS - and abs(track_end - end) < 1 / ANALYSIS_FPS - ), + tracking_reached_budget=False, + source_start=source_start, + source_end=bounded_end, + ) + if ( + _persistent_change(intent) + and resolution.edit_core.duration > MAX_CONTINUOUS_EDIT_SECONDS + 0.05 + ): + raise EditWindowLimitError(resolution.edit_core.duration) + resolution = resolution.model_copy( + update={ + "tracked_frames": [ + observed_frames[timestamp] for timestamp in sorted(observed_frames) + ] + } ) if track.warning: resolution.warnings.append(track.warning) diff --git a/apps/api/app/workers/generate_job.py b/apps/api/app/workers/generate_job.py index f5b97a8..bf6a8de 100644 --- a/apps/api/app/workers/generate_job.py +++ b/apps/api/app/workers/generate_job.py @@ -44,6 +44,7 @@ from app.db.session import AsyncSessionLocal from app.models.job import Job, Variant from app.models.project import Project +from app.schemas.edit_plan import EditIntent from app.services import ffmpeg, job_events, storage from app.services.continuity_validator import ( ContinuityReport, @@ -63,6 +64,7 @@ ) from app.services.generation_telemetry import build_local_flow_telemetry from app.services.local_compositor import composite_generated_target +from app.services.edit_prompt import planned_edit_prompt log = logging.getLogger("fiebatt.jobs.generate") @@ -99,23 +101,7 @@ def _provider_model(provider: str, edit_mode: str | None = None) -> str: }.get(provider, provider) -def _planned_edit_prompt(user_prompt: str, plan: dict[str, Any]) -> str: - """Keep the user's requirement while adding the grounded planner detail.""" - planned = str( - plan.get("prompt_for_video_edit") - or plan.get("prompt_for_runway") - or plan.get("prompt_for_veo") - or "" - ).strip() - requirement = user_prompt.strip() - if not planned or planned.casefold() == requirement.casefold(): - return requirement - return ( - "NON-NEGOTIABLE USER REQUIREMENT:\n" - f"{requirement}\n\n" - "GROUNDED VIDEO-EDIT INSTRUCTION:\n" - f"{planned}" - ) +_planned_edit_prompt = planned_edit_prompt async def _update_job(db: AsyncSession, job_id: str, **fields) -> None: @@ -550,6 +536,7 @@ async def _do_extract_end_frame() -> tuple[Path, bool]: mask_image_url: str | None = None subject_reference_path: str | None = None sam_available = False + sam_video_available = False if subject_frame_ok and bbox and not bbox_is_full_frame: try: sam_available = await sam_service.is_available() @@ -583,34 +570,56 @@ async def _do_extract_end_frame() -> tuple[Path, bool]: except Exception as exc: log.warning("SAM generation conditioning failed; using bbox crop", exc_info=True) await _emit(job_id, "sam_error", f"SAM localization failed; using bbox crop: {exc}") + sam_video_available = await sam_service.video_tracking_available() + + planned_intent: EditIntent | None = None + raw_planned_intent = payload.get("planned_intent") + if isinstance(raw_planned_intent, dict): + try: + planned_intent = EditIntent.model_validate(raw_planned_intent) + except Exception: + log.warning("job contains invalid planned intent", exc_info=True) await _emit( job_id, "plan_start", - "asking Gemini to structure the edit plan", + ( + "reusing the edit instruction chosen before window analysis" + if planned_intent is not None and planned_intent.grounded_edit is not None + else "asking Qwen to structure the edit plan" + ), user_prompt=prompt, bbox=bbox, duration=segment_duration, ) - # start Gemini plan while clip finishes in background + # New planned jobs already contain Qwen's grounded instruction. Legacy + # jobs still plan here so old clients remain usable. planning_frame_path = ( str(subject_frame_path) if subject_frame_ok else (conditioning_start_anchor or "") ) - plan_task = asyncio.create_task( - ai.gemini.plan_variants(prompt, bbox, planning_frame_path) + plan_task = ( + None + if planned_intent is not None and planned_intent.grounded_edit is not None + else asyncio.create_task( + ai.gemini.plan_variants(prompt, bbox, planning_frame_path) + ) ) clip_path, clip_url = await clip_task # should be done by now - try: - plans = await plan_task - except Exception as e: - log.exception("plan_variants failed") - async with AsyncSessionLocal() as db: - await _update_job(db, job_id, status="error", error=f"plan failed: {e}") - await _emit_terminal(job_id, "error", f"plan failed: {e}") - return + if plan_task is None: + assert planned_intent is not None and planned_intent.grounded_edit is not None + plans = [planned_intent.grounded_edit.model_dump(mode="json")] + else: + try: + plans = await plan_task + except Exception as e: + log.exception("plan_variants failed") + async with AsyncSessionLocal() as db: + await _update_job(db, job_id, status="error", error=f"plan failed: {e}") + await _emit_terminal(job_id, "error", f"plan failed: {e}") + return if not plans: async with AsyncSessionLocal() as db: @@ -842,7 +851,7 @@ async def maybe_composite( if ( not generation_window.adaptive or bbox_is_full_frame - or not sam_available + or not sam_video_available or not variant.url ): return variant diff --git a/apps/api/app/workers/local_chunk_job.py b/apps/api/app/workers/local_chunk_job.py new file mode 100644 index 0000000..92a3c5b --- /dev/null +++ b/apps/api/app/workers/local_chunk_job.py @@ -0,0 +1,322 @@ +"""Render and validate one long local edit as overlapping provider-sized chunks.""" +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +from sqlalchemy import select +from sqlalchemy.orm import selectinload + +from app.db.session import AsyncSessionLocal +from app.models.edit_plan import EditPlanRecord, GenerationChunk +from app.models.job import Job, Variant +from app.models.project import Project +from app.models.selection import SelectionArtifact +from app.schemas.edit_plan import EditIntent, LocalRangeResolution +from app.services import ffmpeg, job_events, storage +from app.services.edit_prompt import planned_edit_prompt +from app.services.generation_quality import final_semantic_quality +from app.services.generation_window import GenerationWindow +from app.services.global_chunk_execution import ( + PreviousChunk, + execute_global_chunk, + target_bbox, +) +from app.services.global_seam import assemble_global_occurrence +from app.workers.generate_job import ( + _provider_model, + _sample_variant_frames, + _score_variant_safe, +) + + +log = logging.getLogger("fiebatt.jobs.local_chunks") + + +@dataclass(slots=True) +class ExecutableChunk: + id: str + index: int + edit_start: float + edit_end: float + context_start: float + context_end: float + provider: str + payload_json: dict + output_url: str | None = None + + +async def _emit(job_id: str, stage: str, message: str, *, terminal: bool = False, **data: Any) -> None: + event: dict[str, Any] = { + "stage": stage, + "msg": message, + "ts": time.time(), + } + if terminal: + event["terminal"] = True + if data: + event["data"] = data + await job_events.publish(job_id, event) + + +async def _reference_subject( + project: Project, + selection: SelectionArtifact, +) -> Path: + if selection.subject_reference_url: + try: + return await storage.path_from_url(selection.subject_reference_url) + except Exception: + log.warning("stored subject reference unavailable; rebuilding it", exc_info=True) + frame_path, _ = storage.new_path("keyframes", "jpg") + await ffmpeg.extract_frame(project.video_path, selection.frame_ts, frame_path) + return await ffmpeg.crop_bbox_from_frame(frame_path, selection.bbox_json) + + +async def _fail( + job_id: str, + error: str, + *, + variant_id: str | None = None, + chunk_id: str | None = None, +) -> None: + message = error[:500] + async with AsyncSessionLocal() as db: + job = await db.get(Job, job_id) + if job is not None: + job.status = "error" + job.error = message + if variant_id: + variant = await db.get(Variant, variant_id) + if variant is not None: + variant.status = "error" + variant.error = message + if chunk_id: + chunk = await db.get(GenerationChunk, chunk_id) + if chunk is not None: + chunk.status = "error" + await db.commit() + await _emit(job_id, "error", message, terminal=True) + + +async def run(job_id: str) -> None: + variant_id: str | None = None + try: + async with AsyncSessionLocal() as db: + job = await db.get(Job, job_id) + if job is None: + return + project = await db.get(Project, job.project_id) + plan_id = str((job.payload or {}).get("plan_id") or "") + plan = ( + await db.execute( + select(EditPlanRecord) + .where(EditPlanRecord.id == plan_id) + .options(selectinload(EditPlanRecord.chunks)) + ) + ).scalar_one_or_none() + if project is None or plan is None: + raise ValueError("chunked edit source or plan is missing") + selection = await db.get(SelectionArtifact, plan.selection_id) + if selection is None or selection.project_id != project.id: + raise ValueError("chunked edit selection is missing") + chunks = list(plan.chunks) + if len(chunks) < 2: + raise ValueError("chunked edit requires at least two planned chunks") + resolution = LocalRangeResolution.model_validate(plan.range_json) + if resolution.edit_core.duration > 30.05: + raise ValueError( + "this edit covers more than 30 seconds; video rendering was not started" + ) + intent = EditIntent.model_validate(plan.intent_json) + grounded = ( + intent.grounded_edit.model_dump(mode="json") + if intent.grounded_edit is not None + else {} + ) + generation_prompt = planned_edit_prompt(plan.raw_prompt, grounded) + variant = Variant(job_id=job.id, index=0, status="processing") + db.add(variant) + await db.flush() + variant_id = variant.id + job.status = "processing" + job.error = None + payload = dict(job.payload or {}) + window = GenerationWindow( + core_start=resolution.edit_core.start_ts, + core_end=resolution.edit_core.end_ts, + context_start=resolution.generation_context.start_ts, + context_end=resolution.generation_context.end_ts, + adaptive=True, + ) + payload.update( + { + "execution_window": window.metadata(), + "generation_attempts": len(chunks), + "generated_seconds": sum( + chunk.context_end - chunk.context_start for chunk in chunks + ), + "provider_attempts": [chunk.provider for chunk in chunks], + "selected_provider": plan.provider, + "selected_model": _provider_model(plan.provider, "source_video"), + "selected_edit_mode": "source_video", + } + ) + job.payload = payload + await db.commit() + + await _emit( + job_id, + "chunk_plan", + f"rendering {len(chunks)} overlapping clips for the tracked edit window", + chunk_count=len(chunks), + ) + reference_subject = await _reference_subject(project, selection) + executable = [ + ExecutableChunk( + id=chunk.id, + index=chunk.index, + edit_start=chunk.edit_start, + edit_end=chunk.edit_end, + context_start=chunk.context_start, + context_end=chunk.context_end, + provider=chunk.provider, + payload_json=dict(chunk.payload_json or {}), + ) + for chunk in chunks + ] + + previous: PreviousChunk | None = None + for chunk in executable: + async with AsyncSessionLocal() as db: + row = await db.get(GenerationChunk, chunk.id) + if row is not None: + row.status = "processing" + await db.commit() + await _emit( + job_id, + "chunk_start", + f"rendering clip {chunk.index + 1} of {len(executable)}", + chunk_index=chunk.index, + provider=chunk.provider, + ) + try: + result = await execute_global_chunk( + project=project, + chunk=chunk, # type: ignore[arg-type] + prompt=generation_prompt, + reference_subject_path=reference_subject, + previous=previous, + ) + except Exception as exc: + await _fail( + job_id, + str(exc).strip() or type(exc).__name__, + variant_id=variant_id, + chunk_id=chunk.id, + ) + return + chunk.output_url = result.output_url + previous = PreviousChunk( + context_start=chunk.context_start, + context_end=chunk.context_end, + output_url=result.output_url, + ) + async with AsyncSessionLocal() as db: + row = await db.get(GenerationChunk, chunk.id) + if row is not None: + payload = dict(row.payload_json or {}) + payload["execution"] = result.metadata + payload["output_url"] = result.output_url + row.payload_json = payload + row.status = "generated" + await db.commit() + + occurrence = SimpleNamespace( + edit_start=resolution.generation_context.start_ts, + edit_end=resolution.generation_context.end_ts, + ) + assembly = await assemble_global_occurrence( + project=project, + occurrence=occurrence, # type: ignore[arg-type] + chunks=executable, # type: ignore[arg-type] + ) + frames = await _sample_variant_frames(assembly.output_url) + target_frames: list[str] = [] + media_duration = resolution.generation_context.duration + for index, frame in enumerate(frames): + crop_path, _ = storage.new_path("keyframes", "png") + sample_timestamp = ( + resolution.generation_context.start_ts + + media_duration * (index + 0.5) / len(frames) + ) + try: + sample_bbox = target_bbox( + {"track_frames": resolution.tracked_frames}, + sample_timestamp, + ) + except ValueError: + sample_bbox = selection.bbox_json + await ffmpeg.crop_bbox_from_frame( + frame, + sample_bbox, + crop_path, + ) + target_frames.append(str(crop_path)) + score = await _score_variant_safe( + frames, + generation_prompt, + target_frames=target_frames, + reference_target_path=str(reference_subject), + ) + quality = final_semantic_quality(score) + + async with AsyncSessionLocal() as db: + job = await db.get(Job, job_id) + variant = await db.get(Variant, variant_id) + if job is None or variant is None: + raise ValueError("chunked edit result record is missing") + payload = dict(job.payload or {}) + payload.update( + { + "generation_quality_state": quality.action.value, + "generation_quality_evidence": list(quality.evidence), + "chunk_seams": [seam.metadata() for seam in assembly.seams], + "continuity_validation": assembly.continuity, + } + ) + job.payload = payload + job.status = "done" + variant.status = "done" + variant.url = assembly.output_url + variant.description = grounded.get("description") or plan.raw_prompt + if isinstance(score, dict): + variant.visual_coherence = score.get("visual_coherence") + variant.prompt_adherence = score.get("prompt_adherence") + await db.commit() + + await _emit( + job_id, + "done", + ( + "generation complete with a quality hard-fail" + if quality.evidence + else "generation complete" + ), + terminal=True, + variant_url=assembly.output_url, + generation_quality_state=quality.action.value, + generation_quality_evidence=list(quality.evidence), + acceptance_blocked=bool(quality.evidence), + ) + except Exception as exc: + log.exception("chunked local generation failed") + await _fail( + job_id, + str(exc).strip() or type(exc).__name__, + variant_id=variant_id, + ) diff --git a/apps/api/tests/ai/test_adapters.py b/apps/api/tests/ai/test_adapters.py index b2ecbda..ab648d7 100644 --- a/apps/api/tests/ai/test_adapters.py +++ b/apps/api/tests/ai/test_adapters.py @@ -15,7 +15,7 @@ @pytest.mark.asyncio async def test_stub_plan_variants(): - """gemini.plan_variants returns 3 EditPlan dicts with expected fields.""" + """gemini.plan_variants returns one grounded edit instruction.""" from app.ai.services._stubs import gemini plans = await gemini.plan_variants( @@ -24,15 +24,27 @@ async def test_stub_plan_variants(): "/tmp/frame.png", ) - assert len(plans) == 3 + assert len(plans) == 1 for plan in plans: assert "description" in plan assert "tone" in plan assert "color_grading" in plan assert "region_emphasis" in plan - assert "prompt_for_runway" in plan - assert isinstance(plan["prompt_for_runway"], str) - assert len(plan["prompt_for_runway"]) > 0 + assert "prompt_for_video_edit" in plan + assert isinstance(plan["prompt_for_video_edit"], str) + assert len(plan["prompt_for_video_edit"]) > 0 + + +@pytest.mark.asyncio +async def test_stub_interpretation_separates_state_from_action(): + from app.ai.services._stubs import gemini + + bbox = {"x": 0.25, "y": 0.4, "w": 0.3, "h": 0.35} + state = await gemini.interpret_edit("make this ball pink", bbox, "") + action = await gemini.interpret_edit("make this ball bounce", bbox, "") + + assert state["decision"]["duration_policy"] == "continuous_occurrence" + assert action["decision"]["duration_policy"] == "bounded_action" @pytest.mark.asyncio diff --git a/apps/api/tests/ai/test_pipeline.py b/apps/api/tests/ai/test_pipeline.py index 0899844..6075ff5 100644 --- a/apps/api/tests/ai/test_pipeline.py +++ b/apps/api/tests/ai/test_pipeline.py @@ -1,7 +1,7 @@ """End-to-end pipeline test through stubs. Simulates the full fiebatt flow: - prompt -> edit plan -> generate 3 variants -> score -> accept -> entity search -> propagate + prompt -> edit plan -> generate -> score -> accept -> entity search -> propagate Runs entirely with stubs, no API keys needed. """ @@ -21,11 +21,11 @@ async def test_full_edit_loop(): prompt = "make this car red" bbox = {"x": 0.25, "y": 0.4, "w": 0.3, "h": 0.35} - # Step 2: gemini creates edit plan with 3 variants + # Step 2: Gemini creates the single grounded plan used by production. plans = await gemini.plan_variants(prompt, bbox, "/tmp/frame.png") - assert len(plans) == 3 + assert len(plans) == 1 - # Step 3: generate 3 variants in parallel (via stubs) + # Step 3: generate the planned result via stubs. variants = [] for plan in plans: result = await runway.generate("/tmp/clip.mp4", plan) @@ -33,7 +33,7 @@ async def test_full_edit_loop(): assert "description" in result variants.append(result) - assert len(variants) == 3 + assert len(variants) == 1 # Step 4: score each variant scores = [] @@ -46,7 +46,7 @@ async def test_full_edit_loop(): assert 1 <= score["prompt_adherence"] <= 10 scores.append(score) - assert len(scores) == 3 + assert len(scores) == 1 # Step 5: user picks variant 0 (best score) chosen = variants[0] diff --git a/apps/api/tests/ai/test_sam_tracking.py b/apps/api/tests/ai/test_sam_tracking.py index 88de850..5432ad0 100644 --- a/apps/api/tests/ai/test_sam_tracking.py +++ b/apps/api/tests/ai/test_sam_tracking.py @@ -90,3 +90,14 @@ def segment(worker_url: str, payload: dict) -> dict: assert base64.b64decode(captured["payload"]["image_b64"]) == b"frame" assert result.score == 0.9 assert (tmp_path / "frame.mask.png").read_bytes() == b"mask" + + +@pytest.mark.asyncio +async def test_huggingface_space_is_not_treated_as_video_tracker(monkeypatch): + monkeypatch.setattr( + sam, + "get_settings", + lambda: SimpleNamespace(vision_worker_url="https://segment.hf.space"), + ) + + assert await sam.video_tracking_available() is False diff --git a/apps/api/tests/test_cpu_tracking.py b/apps/api/tests/test_cpu_tracking.py new file mode 100644 index 0000000..71cabe8 --- /dev/null +++ b/apps/api/tests/test_cpu_tracking.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import cv2 +import numpy as np +import pytest + +from app.services.cpu_tracking import track_frames + + +def _moving_target_frames(tmp_path, *, count: int = 11, visible_until: int = 10): + paths = [] + for index in range(count): + frame = np.full((120, 200, 3), 28, dtype=np.uint8) + if index <= visible_until: + left = 40 + index * 5 + target = np.zeros((36, 36, 3), dtype=np.uint8) + target[::2, ::2] = (40, 220, 240) + target[1::2, 1::2] = (230, 60, 180) + frame[42:78, left:left + 36] = target + path = tmp_path / f"{index:03d}.jpg" + assert cv2.imwrite(str(path), frame) + paths.append(str(path)) + return paths + + +@pytest.mark.asyncio +async def test_cpu_tracker_follows_target_in_both_directions(tmp_path): + paths = _moving_target_frames(tmp_path) + result = await track_frames( + paths, + seed_frame_index=5, + bbox={"x": 65 / 200, "y": 42 / 120, "w": 36 / 200, "h": 36 / 120}, + ) + + states = {frame["frame_index"]: frame["state"] for frame in result.frames} + assert result.tracker == "opencv_mil" + assert all(states[index] == "tracked" for index in range(11)) + + +@pytest.mark.asyncio +async def test_cpu_tracker_stops_when_target_disappears(tmp_path): + paths = _moving_target_frames(tmp_path, visible_until=7) + result = await track_frames( + paths, + seed_frame_index=5, + bbox={"x": 65 / 200, "y": 42 / 120, "w": 36 / 200, "h": 36 / 120}, + ) + + states = {frame["frame_index"]: frame["state"] for frame in result.frames} + assert states[7] == "tracked" + assert any(state == "lost" for index, state in states.items() if index >= 8) diff --git a/apps/api/tests/test_dense_occurrence.py b/apps/api/tests/test_dense_occurrence.py index 7609395..595cd24 100644 --- a/apps/api/tests/test_dense_occurrence.py +++ b/apps/api/tests/test_dense_occurrence.py @@ -136,7 +136,7 @@ async def track(paths, **kwargs): ) assert result.passed is False - assert "real SAM2" in result.reason + assert "subject tracking" in result.reason def test_dense_tracks_merge_overlap_but_keep_reentry_separate(): diff --git a/apps/api/tests/test_edit_plan_api.py b/apps/api/tests/test_edit_plan_api.py index bed3490..531e2c3 100644 --- a/apps/api/tests/test_edit_plan_api.py +++ b/apps/api/tests/test_edit_plan_api.py @@ -11,6 +11,7 @@ from app.db.init import create_all # noqa: E402 from app.db.session import AsyncSessionLocal # noqa: E402 from app.main import app # noqa: E402 +from app.api.routes import generate as generate_route # noqa: E402 from app.models.project import Project # noqa: E402 from app.models.selection import SelectionArtifact # noqa: E402 from app.schemas.edit_plan import EditCore, GenerationContext, LocalRangeResolution # noqa: E402 @@ -105,10 +106,12 @@ async def test_create_and_get_non_generating_plan(client, owned_selection): assert body["generation_context"]["start_ts"] == 7.5 assert body["estimate"]["expected_generation_calls"] == 1 assert body["estimate"]["analysis_duration_ms"] >= 0.0 - assert body["adaptive_generation_enabled"] is False + assert body["adaptive_generation_enabled"] is True assert "fixture warning" in body["warnings"] - assert any("legacy fixed window" in warning for warning in body["warnings"]) + assert not any("legacy fixed window" in warning for warning in body["warnings"]) assert len(body["chunks"]) == 1 + assert body["intent"]["duration_policy"] == "bounded_action" + assert body["intent"]["grounded_edit"]["prompt_for_video_edit"] fetched = await client.get( f"/api/edit-plans/{body['plan_id']}", @@ -118,6 +121,106 @@ async def test_create_and_get_non_generating_plan(client, owned_selection): assert fetched.json()["plan_id"] == body["plan_id"] +@pytest.mark.asyncio +async def test_state_change_plan_covers_current_occurrence(client, owned_selection): + project_id, selection_id = owned_selection + response = await client.post( + "/api/edit-plans", + headers={"X-Session-Id": "plan-owner"}, + json={ + "project_id": project_id, + "selection_id": selection_id, + "prompt": "make this ball pink", + }, + ) + + assert response.status_code == 201, response.text + intent = response.json()["intent"] + assert intent["scope"] == "local" + assert intent["duration_policy"] == "continuous_occurrence" + + +@pytest.mark.asyncio +async def test_long_local_edit_is_split_into_provider_sized_chunks( + client, owned_selection, monkeypatch +): + project_id, selection_id = owned_selection + + async def resolve(*args, **kwargs): + core = EditCore(start_ts=3.0, end_ts=23.0) + return LocalRangeResolution( + edit_core=core, + generation_context=GenerationContext( + start_ts=2.25, + end_ts=23.75, + edit_core=core, + ), + occurrence_start=3.0, + occurrence_end=23.0, + analysis_start=2.0, + analysis_end=24.0, + frames_inspected=45, + tracked_frames=[ + { + "timestamp": timestamp, + "state": "tracked", + "confidence": 0.9, + "bbox": {"x": 0.2, "y": 0.1, "w": 0.3, "h": 0.7}, + } + for timestamp in range(2, 25) + ], + confidence=0.9, + ) + + monkeypatch.setattr(route, "resolve_local_range", resolve) + response = await client.post( + "/api/edit-plans", + headers={"X-Session-Id": "plan-owner"}, + json={ + "project_id": project_id, + "selection_id": selection_id, + "prompt": "make this ball pink", + }, + ) + + assert response.status_code == 201, response.text + body = response.json() + assert body["provider"] == "happyhorse" + assert body["estimate"]["expected_generation_calls"] == 2 + assert len(body["chunks"]) == 2 + assert all( + chunk["generation_context"]["end_ts"] + - chunk["generation_context"]["start_ts"] + <= 15.0 + for chunk in body["chunks"] + ) + assert body["chunks"][0]["edit_core"]["end_ts"] == body["chunks"][1]["edit_core"]["start_ts"] + + class FakeRunner: + def __init__(self): + self.submissions = [] + + def submit(self, job_id, factory): + self.submissions.append((job_id, factory)) + + runner = FakeRunner() + app.state.runner = runner + generated = await client.post( + "/api/generate", + headers={"X-Session-Id": "plan-owner"}, + json={ + "project_id": project_id, + "plan_id": body["plan_id"], + }, + ) + + assert generated.status_code == 200, generated.text + assert len(runner.submissions) == 1 + factory = runner.submissions[0][1] + closure_values = [cell.cell_contents for cell in factory.__closure__ or ()] + assert generate_route.local_chunk_job.run in closure_values + + @pytest.mark.asyncio async def test_plan_ownership_is_hidden(client, owned_selection): project_id, selection_id = owned_selection @@ -165,3 +268,106 @@ async def test_provider_constraints_fail_before_generation(client, owned_selecti ) assert response.status_code == 422 assert "exactly 4, 6, 8 seconds" in response.json()["detail"] + + +@pytest.mark.asyncio +async def test_over_thirty_second_plan_returns_render_limit_card_data( + client, owned_selection, monkeypatch +): + project_id, selection_id = owned_selection + async with AsyncSessionLocal() as db: + project = await db.get(Project, project_id) + project.duration = 50.0 + await db.commit() + + async def resolve(*args, **kwargs): + core = EditCore(start_ts=5.0, end_ts=36.0) + return LocalRangeResolution( + edit_core=core, + generation_context=GenerationContext( + start_ts=4.25, + end_ts=36.75, + edit_core=core, + ), + occurrence_start=5.0, + occurrence_end=36.0, + analysis_start=4.0, + analysis_end=37.0, + confidence=0.9, + ) + + monkeypatch.setattr(route, "resolve_local_range", resolve) + response = await client.post( + "/api/edit-plans", + headers={"X-Session-Id": "plan-owner"}, + json={ + "project_id": project_id, + "selection_id": selection_id, + "prompt": "make this ball pink", + }, + ) + + assert response.status_code == 422 + detail = response.json()["detail"] + assert detail["code"] == "edit_window_too_long" + assert detail["limit_seconds"] == 30.0 + assert detail["detected_seconds"] == 31.0 + + +@pytest.mark.asyncio +async def test_active_source_clip_must_contain_selection(client, owned_selection): + project_id, selection_id = owned_selection + response = await client.post( + "/api/edit-plans", + headers={"X-Session-Id": "plan-owner"}, + json={ + "project_id": project_id, + "selection_id": selection_id, + "prompt": "make this person jump", + "source_start_ts": 12.0, + "source_end_ts": 18.0, + }, + ) + + assert response.status_code == 422 + assert response.json()["detail"] == "selection is outside the active source clip" + + +@pytest.mark.asyncio +async def test_active_source_clip_bounds_reach_resolver( + client, owned_selection, monkeypatch +): + project_id, selection_id = owned_selection + + async def resolve(*args, **kwargs): + assert kwargs["source_start"] == 7.0 + assert kwargs["source_end"] == 13.0 + core = EditCore(start_ts=8.25, end_ts=11.75) + return LocalRangeResolution( + edit_core=core, + generation_context=GenerationContext( + start_ts=7.5, end_ts=12.5, edit_core=core + ), + occurrence_start=7.0, + occurrence_end=13.0, + analysis_start=7.0, + analysis_end=13.0, + frames_inspected=25, + confidence=0.9, + ) + + monkeypatch.setattr(route, "resolve_local_range", resolve) + response = await client.post( + "/api/edit-plans", + headers={"X-Session-Id": "plan-owner"}, + json={ + "project_id": project_id, + "selection_id": selection_id, + "prompt": "make this person jump", + "source_start_ts": 7.0, + "source_end_ts": 13.0, + }, + ) + + assert response.status_code == 201, response.text + assert response.json()["occurrence_start"] == 7.0 diff --git a/apps/api/tests/test_edit_scope.py b/apps/api/tests/test_edit_scope.py index bc07141..c5cb8f6 100644 --- a/apps/api/tests/test_edit_scope.py +++ b/apps/api/tests/test_edit_scope.py @@ -41,3 +41,17 @@ def test_local_accept_does_not_imply_global_discovery(): assert should_discover_occurrences(scope="local", explicitly_requested=False) is False assert should_discover_occurrences(scope="local", explicitly_requested=True) is True assert should_discover_occurrences(scope="all_occurrences", explicitly_requested=False) is True + + +def test_state_change_defaults_to_current_continuous_occurrence(): + result = plan_prompt_intent("make this ball pink") + + assert result.intent.scope == "local" + assert result.intent.duration_policy == "continuous_occurrence" + + +def test_action_defaults_to_bounded_local_window(): + result = plan_prompt_intent("make this ball bounce") + + assert result.intent.scope == "local" + assert result.intent.duration_policy == "bounded_action" diff --git a/apps/api/tests/test_ffmpeg_extract_frame.py b/apps/api/tests/test_ffmpeg_extract_frame.py index bbe8459..6428e2b 100644 --- a/apps/api/tests/test_ffmpeg_extract_frame.py +++ b/apps/api/tests/test_ffmpeg_extract_frame.py @@ -4,6 +4,8 @@ import subprocess from pathlib import Path +import cv2 + from app.services import ffmpeg @@ -27,3 +29,36 @@ def test_extract_frame_near_container_end(tmp_path: Path): assert result == output assert output.stat().st_size > 0 + + +def test_extract_sampled_frames_decodes_window_once(tmp_path: Path): + source = tmp_path / "source.mp4" + output_dir = tmp_path / "analysis" + subprocess.run( + [ + "ffmpeg", "-v", "error", "-y", + "-f", "lavfi", + "-i", "testsrc2=s=1280x720:r=24:d=3.0", + "-c:v", "libx264", + "-pix_fmt", "yuv420p", + str(source), + ], + check=True, + timeout=10, + ) + + paths, timestamps = asyncio.run( + ffmpeg.extract_sampled_frames( + source, + start_ts=0.5, + end_ts=2.5, + fps=4.0, + output_dir=output_dir, + ) + ) + + assert 8 <= len(paths) <= 10 + assert len(paths) == len(timestamps) + frame = cv2.imread(paths[0]) + assert frame is not None + assert frame.shape[1] == 640 diff --git a/apps/api/tests/test_global_chunk_execution.py b/apps/api/tests/test_global_chunk_execution.py index a6657ae..8d6c38c 100644 --- a/apps/api/tests/test_global_chunk_execution.py +++ b/apps/api/tests/test_global_chunk_execution.py @@ -177,9 +177,6 @@ async def publish(path, *, content_type): async def extract_frame(source, timestamp, output): return Path(output) - async def sam_available(): - return False - async def generate(clip_path, plan, **kwargs): calls.append((clip_path, plan, kwargs)) return {"url": clip_path, "description": "stub"} @@ -192,7 +189,6 @@ async def generate(clip_path, plan, **kwargs): lambda *args: (tmp_path / "target.jpg", "/target"), ) monkeypatch.setattr(execution.ffmpeg, "extract_frame", extract_frame) - monkeypatch.setattr(execution.sam, "is_available", sam_available) monkeypatch.setattr(execution.ai.runway, "generate", generate) project = SimpleNamespace(fps=30.0, video_path=tmp_path / "project.mp4") diff --git a/apps/api/tests/test_local_chunk_job.py b/apps/api/tests/test_local_chunk_job.py new file mode 100644 index 0000000..e098c42 --- /dev/null +++ b/apps/api/tests/test_local_chunk_job.py @@ -0,0 +1,220 @@ +from pathlib import Path + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app import models as _models # noqa: F401 +from app.db.base import Base +from app.models.edit_plan import EditPlanRecord, GenerationChunk +from app.models.job import Job, Variant +from app.models.project import Project +from app.models.selection import SelectionArtifact +from app.schemas.edit_plan import ( + EditCore, + EditIntent, + GenerationContext, + GroundedEditInstruction, + LocalRangeResolution, +) +from app.services.global_chunk_sequence import ChunkExecution +from app.services.global_seam import AssemblyResult +from app.workers import local_chunk_job + + +@pytest.mark.asyncio +async def test_chunk_worker_renders_in_order_and_publishes_one_variant( + tmp_path, + monkeypatch, +): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'chunks.db'}") + sessions = async_sessionmaker( + engine, + class_=AsyncSession, + expire_on_commit=False, + autoflush=False, + ) + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + + core = EditCore(start_ts=3.0, end_ts=23.0) + resolution = LocalRangeResolution( + edit_core=core, + generation_context=GenerationContext( + start_ts=2.25, + end_ts=23.75, + edit_core=core, + ), + occurrence_start=3.0, + occurrence_end=23.0, + analysis_start=2.0, + analysis_end=24.0, + confidence=0.9, + ) + intent = EditIntent( + raw_prompt="make this ball pink", + change_type="appearance", + duration_policy="continuous_occurrence", + grounded_edit=GroundedEditInstruction( + intent="appearance change", + description="pink ball", + region_emphasis="selected ball only", + prompt_for_video_edit="Change only the selected ball to pink.", + ), + ) + bbox = {"x": 0.2, "y": 0.1, "w": 0.3, "h": 0.3} + track_frames = [ + { + "timestamp": timestamp, + "state": "tracked", + "confidence": 0.9, + "bbox": bbox, + } + for timestamp in range(2, 25) + ] + async with sessions() as db: + project = Project( + session_id="owner", + video_path=str(tmp_path / "source.mp4"), + video_url="https://media.example/source.mp4", + duration=30.0, + fps=30.0, + width=1280, + height=720, + ) + db.add(project) + await db.flush() + selection = SelectionArtifact( + project_id=project.id, + frame_ts=10.0, + bbox_json=bbox, + contours_json=[], + mask_url="https://media.example/mask.png", + subject_reference_url="https://media.example/subject.png", + sam_score=0.9, + source_revision=project.video_url, + ) + db.add(selection) + await db.flush() + plan = EditPlanRecord( + project_id=project.id, + selection_id=selection.id, + raw_prompt=intent.raw_prompt, + scope="local", + intent_json=intent.model_dump(mode="json"), + range_json=resolution.model_dump(mode="json"), + estimate_json={}, + provider="happyhorse", + provider_reason="test", + warnings_json=[], + source_revision=project.video_url, + ) + db.add(plan) + await db.flush() + for index, values in enumerate( + [ + (3.0, 16.0, 2.25, 16.75), + (16.0, 23.0, 15.25, 23.75), + ] + ): + edit_start, edit_end, context_start, context_end = values + db.add( + GenerationChunk( + plan_id=plan.id, + index=index, + edit_start=edit_start, + edit_end=edit_end, + context_start=context_start, + context_end=context_end, + provider="happyhorse", + payload_json={ + "track_frames": track_frames, + "boundary_contract": { + "protect_source_before": index == 0, + "protect_source_after": index == 1, + "handoff_from_previous": index == 1, + "handoff_to_next": index == 0, + }, + }, + ) + ) + job = Job( + project_id=project.id, + kind="generate", + status="pending", + start_ts=3.0, + end_ts=23.0, + bbox_json=bbox, + prompt=intent.raw_prompt, + reference_frame_ts=10.0, + payload={"plan_id": plan.id}, + ) + db.add(job) + await db.commit() + job_id = job.id + + monkeypatch.setattr(local_chunk_job, "AsyncSessionLocal", sessions) + + async def reference_subject(project, selection): + return tmp_path / "subject.png" + + executions = [] + + async def execute(**kwargs): + chunk = kwargs["chunk"] + previous = kwargs["previous"] + executions.append((chunk.index, previous.output_url if previous else None)) + return ChunkExecution( + output_url=f"https://media.example/chunk-{chunk.index}.mp4", + metadata={"provider": chunk.provider}, + ) + + async def assemble(*, project, occurrence, chunks): + assert occurrence.edit_start == pytest.approx(2.25) + assert occurrence.edit_end == pytest.approx(23.75) + assert [chunk.output_url for chunk in chunks] == [ + "https://media.example/chunk-0.mp4", + "https://media.example/chunk-1.mp4", + ] + return AssemblyResult( + "https://media.example/final.mp4", + (), + {"entry": {"passed": True}, "exit": {"passed": True}}, + ) + + async def sample(url): + assert url == "https://media.example/final.mp4" + return [str(tmp_path / "frame.jpg")] + + async def crop(frame, bbox, output): + return Path(output) + + async def score(*args, **kwargs): + return {"visual_coherence": 9, "prompt_adherence": 9} + + monkeypatch.setattr(local_chunk_job, "_reference_subject", reference_subject) + monkeypatch.setattr(local_chunk_job, "execute_global_chunk", execute) + monkeypatch.setattr(local_chunk_job, "assemble_global_occurrence", assemble) + monkeypatch.setattr(local_chunk_job, "_sample_variant_frames", sample) + monkeypatch.setattr(local_chunk_job.ffmpeg, "crop_bbox_from_frame", crop) + monkeypatch.setattr(local_chunk_job, "_score_variant_safe", score) + + await local_chunk_job.run(job_id) + + assert executions == [ + (0, None), + (1, "https://media.example/chunk-0.mp4"), + ] + async with sessions() as db: + job = await db.get(Job, job_id) + variant = (await db.execute(select(Variant))).scalar_one() + chunks = ( + await db.execute(select(GenerationChunk).order_by(GenerationChunk.index)) + ).scalars().all() + assert job.status == "done" + assert job.payload["generation_quality_state"] == "pass" + assert variant.status == "done" + assert variant.url == "https://media.example/final.mp4" + assert [chunk.status for chunk in chunks] == ["generated", "generated"] + + await engine.dispose() diff --git a/apps/api/tests/test_local_range.py b/apps/api/tests/test_local_range.py index 2819866..4a7a987 100644 --- a/apps/api/tests/test_local_range.py +++ b/apps/api/tests/test_local_range.py @@ -49,6 +49,49 @@ def test_range_uses_two_sided_handles_without_covering_whole_occurrence(): assert result.generation_context.end_ts == pytest.approx(12.5) +def test_range_never_crosses_active_source_clip(): + intent = EditIntent( + raw_prompt="make this person jump", + change_type="motion", + estimated_action_seconds=3.5, + ) + result = resolve_window_from_evidence( + intent=intent, + seed_ts=11.5, + duration=30.0, + analysis_start=10.0, + analysis_end=13.0, + shot_start=8.0, + shot_end=15.0, + tracked_start=9.0, + tracked_end=14.0, + frames_inspected=13, + source_start=10.0, + source_end=13.0, + ) + + assert result.occurrence_start == pytest.approx(10.0) + assert result.occurrence_end == pytest.approx(13.0) + assert result.generation_context.start_ts == pytest.approx(10.0) + assert result.generation_context.end_ts == pytest.approx(13.0) + + +def test_analysis_window_respects_active_source_clip(): + intent = EditIntent( + raw_prompt="make this person jump", + change_type="motion", + estimated_action_seconds=3.5, + ) + + assert analysis_window( + intent, + 11.5, + 30.0, + source_start=10.0, + source_end=13.0, + ) == (10.0, 13.0) + + @pytest.mark.asyncio async def test_resolver_inspects_local_frames_and_reuses_cache(tmp_path): clear_local_range_cache() @@ -136,3 +179,156 @@ async def unavailable_tracker(*args, **kwargs): assert result.edit_core.duration == pytest.approx(3.5) assert any("bbox fallback used" in warning for warning in result.warnings) + + +@pytest.mark.asyncio +async def test_state_change_expands_outward_until_target_is_lost(tmp_path): + clear_local_range_cache() + source = tmp_path / "source.mp4" + source.write_bytes(b"unused") + project = SimpleNamespace(video_path=str(source), video_url="", duration=100.0) + selection = SimpleNamespace( + id="selection-state", + source_revision="revision-state", + frame_ts=50.0, + bbox_json={"x": 0.2, "y": 0.1, "w": 0.3, "h": 0.7}, + mask_url=None, + ) + intent = EditIntent( + raw_prompt="make this ball pink", + change_type="appearance", + duration_policy="continuous_occurrence", + ) + timestamps_by_path = {} + tracker_calls = 0 + + async def fake_extract(_source, timestamp, output): + timestamps_by_path[str(output)] = timestamp + cv2.imwrite(str(output), np.full((36, 64, 3), 80, dtype=np.uint8)) + return output + + async def fake_track(paths, **kwargs): + nonlocal tracker_calls + tracker_calls += 1 + frames = [] + for index, path in enumerate(paths): + timestamp = timestamps_by_path[path] + frames.append( + { + "frame_index": index, + "state": "tracked" if 40.0 <= timestamp <= 60.0 else "lost", + "confidence": 1.0, + } + ) + return TrackResult( + tracker="test", + frames=frames, + processed_start_index=0, + processed_end_index=len(frames) - 1, + ) + + result = await resolve_local_range( + project, + selection, + intent, + extract_frame=fake_extract, + track_frames=fake_track, + ) + + assert tracker_calls == 3 + assert result.analysis_start == pytest.approx(38.0) + assert result.analysis_end == pytest.approx(62.0) + assert result.occurrence_start == pytest.approx(40.0) + assert result.occurrence_end == pytest.approx(60.0) + assert result.generation_context.start_ts == pytest.approx(39.25) + assert result.generation_context.end_ts == pytest.approx(60.75) + + +@pytest.mark.asyncio +async def test_state_change_fails_closed_when_tracking_is_unavailable(tmp_path): + clear_local_range_cache() + source = tmp_path / "source.mp4" + source.write_bytes(b"unused") + project = SimpleNamespace(video_path=str(source), video_url="", duration=20.0) + selection = SimpleNamespace( + id="selection-state-failure", + source_revision="revision-state-failure", + frame_ts=10.0, + bbox_json={"x": 0.2, "y": 0.1, "w": 0.3, "h": 0.7}, + mask_url=None, + ) + intent = EditIntent( + raw_prompt="make this ball pink", + change_type="appearance", + duration_policy="continuous_occurrence", + ) + + async def fake_extract(_source, _timestamp, output): + cv2.imwrite(str(output), np.full((36, 64, 3), 80, dtype=np.uint8)) + return output + + async def unavailable_tracker(*args, **kwargs): + raise ConnectionError("tracker unavailable") + + with pytest.raises(ValueError, match="could not reliably track"): + await resolve_local_range( + project, + selection, + intent, + extract_frame=fake_extract, + track_frames=unavailable_tracker, + ) + + +@pytest.mark.asyncio +async def test_state_change_rejects_occurrences_over_thirty_seconds(tmp_path): + clear_local_range_cache() + source = tmp_path / "source.mp4" + source.write_bytes(b"unused") + project = SimpleNamespace(video_path=str(source), video_url="", duration=100.0) + selection = SimpleNamespace( + id="selection-long-state", + source_revision="revision-long-state", + frame_ts=50.0, + bbox_json={"x": 0.2, "y": 0.1, "w": 0.3, "h": 0.7}, + mask_url=None, + ) + intent = EditIntent( + raw_prompt="make this ball pink", + change_type="appearance", + duration_policy="continuous_occurrence", + ) + timestamps_by_path = {} + + async def fake_extract(_source, timestamp, output): + timestamps_by_path[str(output)] = timestamp + cv2.imwrite(str(output), np.full((36, 64, 3), 80, dtype=np.uint8)) + return output + + async def fake_track(paths, **kwargs): + return TrackResult( + tracker="test", + frames=[ + { + "frame_index": index, + "state": ( + "tracked" + if 30.0 <= timestamps_by_path[path] <= 70.0 + else "lost" + ), + "confidence": 1.0, + } + for index, path in enumerate(paths) + ], + processed_start_index=0, + processed_end_index=len(paths) - 1, + ) + + with pytest.raises(ValueError, match="more than 30 seconds"): + await resolve_local_range( + project, + selection, + intent, + extract_frame=fake_extract, + track_frames=fake_track, + ) diff --git a/apps/api/tests/test_pipeline.py b/apps/api/tests/test_pipeline.py index d687bbf..ccf7512 100644 --- a/apps/api/tests/test_pipeline.py +++ b/apps/api/tests/test_pipeline.py @@ -54,7 +54,7 @@ async def test_health(client: AsyncClient): body = res.json() assert body["ok"] is True assert body["ai_mode"] == "stub" - assert body["features"]["adaptive_edit_planning"] is False + assert body["features"]["adaptive_edit_planning"] is True assert body["features"]["global_edit_planning"] is False assert body["features"]["hard_failed_acceptance_override"] is False diff --git a/apps/api/tests/test_planned_generation_contract.py b/apps/api/tests/test_planned_generation_contract.py index c0a5863..fd5453f 100644 --- a/apps/api/tests/test_planned_generation_contract.py +++ b/apps/api/tests/test_planned_generation_contract.py @@ -35,8 +35,8 @@ def test_generate_request_rejects_partial_legacy_shape(): GenerateRequest(project_id="project-1", start_ts=1.0, end_ts=4.0) -def test_adaptive_generation_is_rollout_gated_by_default(): - assert Settings().adaptive_edit_planning is False +def test_adaptive_local_generation_is_enabled_by_default(): + assert Settings().adaptive_edit_planning is True assert Settings().global_edit_planning is False diff --git a/apps/web/src/components/editor-port/Inspector.tsx b/apps/web/src/components/editor-port/Inspector.tsx index ca3ddb4..4f1604a 100644 --- a/apps/web/src/components/editor-port/Inspector.tsx +++ b/apps/web/src/components/editor-port/Inspector.tsx @@ -129,6 +129,7 @@ function AiTab({ plan, planning, fallbackNotice, + proLimit, adaptivePlanningEnabled, busy, status, @@ -220,6 +221,12 @@ function AiTab({ {plan && variants.length === 0 ? ( ) : null} + {proLimit && variants.length === 0 ? ( + + ) : null} +
{plan.provider} - {plan.estimate.expected_generation_calls} call + {plan.estimate.expected_generation_calls} {plan.estimate.expected_generation_calls === 1 ? "call" : "calls"} {plan.estimate.expected_generated_seconds.toFixed(1)} generated seconds {plan.estimate.analysis_duration_ms.toFixed(0)}ms planning {plan.estimate.frames_inspected} frames inspected diff --git a/apps/web/src/components/editor-port/inspector.css b/apps/web/src/components/editor-port/inspector.css index 37e4117..8499ffe 100644 --- a/apps/web/src/components/editor-port/inspector.css +++ b/apps/web/src/components/editor-port/inspector.css @@ -112,6 +112,51 @@ .edit-plan__range button { padding: 6px 10px; border-radius: 6px; border: 1px solid var(--border); } .edit-plan__warning { margin: 7px 0 0; color: var(--warning, #d99a3e); font-size: 10px; } +.pro-limit { + position: relative; + overflow: hidden; + margin: 10px 10px 0; + padding: 15px; + border: 1px solid color-mix(in srgb, #a78bfa 52%, var(--border)); + border-radius: 12px; + background: + linear-gradient(145deg, color-mix(in srgb, #7c3aed 13%, var(--panel)), var(--panel)); + box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.05); +} +.pro-limit__glow { + position: absolute; + width: 130px; + height: 130px; + top: -80px; + right: -35px; + border-radius: 999px; + background: rgba(167, 139, 250, 0.24); + filter: blur(22px); + pointer-events: none; +} +.pro-limit__head { + position: relative; + display: flex; + align-items: center; + justify-content: space-between; + gap: 8px; +} +.pro-limit__badge { + padding: 4px 7px; + border: 1px solid rgba(196, 181, 253, 0.4); + border-radius: 999px; + color: #c4b5fd; + background: rgba(124, 58, 237, 0.14); + font-size: 9px; + letter-spacing: 0.08em; +} +.pro-limit__safe { color: var(--chrome-lo); font-size: 9px; } +.pro-limit h3 { position: relative; margin: 13px 0 6px; font-size: 15px; } +.pro-limit p { position: relative; margin: 0; color: var(--ink-dim); font-size: 11px; line-height: 1.5; } +.pro-limit ul { position: relative; margin: 12px 0; padding-left: 17px; color: var(--ink); font-size: 11px; line-height: 1.7; } +.pro-limit li::marker { color: #a78bfa; } +.pro-limit .pro-limit__hint { color: #c4b5fd; font-size: 9px; } + .pane__prompt { font-family: var(--f-ui); background: var(--recessed); diff --git a/apps/web/src/hooks/useGenerationSession.ts b/apps/web/src/hooks/useGenerationSession.ts index 554c4f2..597588e 100644 --- a/apps/web/src/hooks/useGenerationSession.ts +++ b/apps/web/src/hooks/useGenerationSession.ts @@ -14,6 +14,7 @@ import { type JobStreamEvent, type Variant, type BBox, + ApiError, } from "@/lib/api"; import type { Clip } from "@/stores/edl"; @@ -54,6 +55,10 @@ export function useGenerationSession({ const [plan, setPlan] = useState(null); const [planning, setPlanning] = useState(false); const [fallbackNotice, setFallbackNotice] = useState(null); + const [proLimit, setProLimit] = useState<{ + limitSeconds: number; + detectedSeconds: number | null; + } | null>(null); const [adaptivePlanningEnabled, setAdaptivePlanningEnabled] = useState(null); const [busy, setBusy] = useState(false); const [status, setStatus] = useState(""); @@ -65,6 +70,7 @@ export function useGenerationSession({ const jobIdRef = useRef(null); const generationTargetRef = useRef(null); const streamCtlRef = useRef(null); + const planBlockedRef = useRef(false); const canGenerate = !!clip && @@ -78,6 +84,8 @@ export function useGenerationSession({ function updatePrompt(value: string) { setPrompt(value); setFallbackNotice(null); + setProLimit(null); + planBlockedRef.current = false; if (plan && plan.intent.raw_prompt !== value.trim()) setPlan(null); } @@ -96,9 +104,11 @@ export function useGenerationSession({ setResult(null); jobIdRef.current = null; generationTargetRef.current = null; + planBlockedRef.current = false; setPlan(null); setPlanning(false); setFallbackNotice(null); + setProLimit(null); if (!keepPrompt) setPrompt(""); } @@ -135,6 +145,7 @@ export function useGenerationSession({ if (selectionId) { const planned = await preparePlan(undefined, true); if (planned) return true; + if (planBlockedRef.current) return false; } else { setFallbackNotice( "Precise selection unavailable. Rendering with legacy fixed window.", @@ -222,7 +233,9 @@ export function useGenerationSession({ allowLegacyFallback = false, ): Promise { if (!clip?.projectId || !selectionId || !prompt.trim()) return false; + const baseClip = sourceClip ?? clip; setPlanning(true); + planBlockedRef.current = false; setErr(null); setStatus("planning"); try { @@ -232,6 +245,8 @@ export function useGenerationSession({ prompt: prompt.trim(), explicit_start_ts: explicitRange?.start, explicit_end_ts: explicitRange?.end, + source_start_ts: baseClip.sourceStart, + source_end_ts: baseClip.sourceEnd, }); if (!next.adaptive_generation_enabled) { setPlan(null); @@ -247,8 +262,20 @@ export function useGenerationSession({ } setPlan(next); setFallbackNotice(null); + setProLimit(null); return true; } catch (error) { + if (error instanceof ApiError && error.code === "edit_window_too_long") { + planBlockedRef.current = true; + setFallbackNotice(null); + setProLimit({ + limitSeconds: Number(error.detail?.limit_seconds ?? 30), + detectedSeconds: error.detail?.detected_seconds == null + ? null + : Number(error.detail.detected_seconds), + }); + return false; + } if (allowLegacyFallback) { setFallbackNotice( "Adaptive planning unavailable. Rendering with legacy fixed window.", @@ -303,6 +330,7 @@ export function useGenerationSession({ plan, planning, fallbackNotice, + proLimit, adaptivePlanningEnabled, busy, status, diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index 156bbb2..70a9cd0 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -8,6 +8,18 @@ import { redirectToLogin } from "@/lib/auth"; const SESSION_KEY = "fiebatt.session_id"; +export class ApiError extends Error { + constructor( + public status: number, + message: string, + public code?: string, + public detail?: Record, + ) { + super(message); + this.name = "ApiError"; + } +} + export function getSessionId(): string { let sid = localStorage.getItem(SESSION_KEY); if (!sid) { @@ -33,8 +45,25 @@ async function request( throw new Error("401 Unauthorized"); } if (!res.ok) { - const text = await res.text().catch(() => ""); - throw new Error(`${res.status} ${res.statusText}: ${text}`); + const payload = await res.json().catch(() => null) as { + detail?: string | Record; + } | null; + const detail = payload?.detail; + if (detail && typeof detail === "object") { + const message = typeof detail.message === "string" + ? detail.message + : `${res.status} ${res.statusText}`; + throw new ApiError( + res.status, + message, + typeof detail.code === "string" ? detail.code : undefined, + detail, + ); + } + throw new ApiError( + res.status, + typeof detail === "string" ? detail : `${res.status} ${res.statusText}`, + ); } return res.json(); } @@ -166,6 +195,7 @@ export type EditPlanResp = { intent: { raw_prompt: string; change_type: "appearance" | "removal" | "replacement" | "motion" | "scene"; + duration_policy: "bounded_action" | "continuous_occurrence" | "explicit_range" | "all_occurrences"; action_phases: string[]; estimated_action_seconds: number; requires_recovery_motion: boolean; @@ -209,6 +239,8 @@ export type CreateEditPlanReq = { prompt: string; explicit_start_ts?: number; explicit_end_ts?: number; + source_start_ts?: number; + source_end_ts?: number; }; export type AcceptResp = {