Integrate capability-aware Apple Silicon runtime

backend/_routes/health.py

@@ -7,7 +7,7 @@
 
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
 
-from api_types import GpuInfoResponse, HealthResponse, MpsMemoryResponse
+from api_types import GpuInfoResponse, HealthResponse, MpsMemoryResponse, RuntimeTelemetryResponse
 from state import get_state_service
 from app_handler import AppHandler
 
@@ -29,6 +29,11 @@ def route_mps_memory(handler: AppHandler = Depends(get_state_service)) -> MpsMem
     return handler.health.get_mps_memory()
 
 
+@router.get("/api/runtime-telemetry", response_model=RuntimeTelemetryResponse)
+def route_runtime_telemetry(handler: AppHandler = Depends(get_state_service)) -> RuntimeTelemetryResponse:
+    return handler.health.get_runtime_telemetry()
+
+
 def _shutdown_process() -> None:
     os.kill(os.getpid(), signal.SIGTERM)
 

backend/api_types.py

@@ -90,8 +90,52 @@ class MpsMemoryResponse(BaseModel):
     recommended_max_mib: int | None = None
 
 
+RuntimeEngine: TypeAlias = Literal["torch", "mlx", "cloud"]
+
+
+class RuntimeProvenanceItem(BaseModel):
+    component: str
+    version: str
+    revision: str | None = None
+    source: str
+
+
 class RuntimePolicyResponse(BaseModel):
     force_api_generations: bool
+    fast_video_engine_preference: Literal["auto", "torch", "mlx"]
+    auto_fast_video_engine: Literal["torch", "mlx", "cloud"]
+    auto_selection_reason: str
+    execution_mode: Literal["eager", "low_ram", "unsupported"]
+    automatic_tiling: bool
+    mlx_model_source: str
+    mlx_model_variant: Literal["bf16", "q8"]
+    quality_warning: str | None = None
+    capability_engines: dict[str, RuntimeEngine]
+    provenance: list[RuntimeProvenanceItem]
+
+
+class RuntimeTelemetryResponse(BaseModel):
+    sampled_at: str
+    active_engine: RuntimeEngine | None = None
+    active_pipeline: str | None = None
+    process_rss_mib: int
+    system_total_mib: int
+    system_available_mib: int
+    mlx_active_mib: int | None = None
+    mlx_cache_mib: int | None = None
+    mlx_peak_mib: int | None = None
+    mlx_profile_status: Literal["running", "success", "error", "cancelled"] | None = None
+    mlx_profile_phase: str | None = None
+    mlx_profile_path: str | None = None
+    mlx_profile_sampled_at: str | None = None
+    mlx_runtime_identity: dict[str, object] | None = None
+    mps_allocated_mib: int | None = None
+    mps_driver_mib: int | None = None
+    mps_recommended_max_mib: int | None = None
+    local_metal_lease_status: Literal["idle", "waiting", "held"]
+    local_metal_lease_reason: str | None = None
+    local_metal_lease_waited_seconds: float = 0.0
+    local_metal_lease_owner: dict[str, object] | None = None
 
 
 class GenerationProgressResponse(BaseModel):
@@ -143,6 +187,8 @@ class SuggestGapPromptResponse(BaseModel):
 class GenerateVideoCompleteResponse(BaseModel):
     status: Literal["complete"]
     video_path: str
+    resolved_width: int | None = None
+    resolved_height: int | None = None
 
 
 class GenerateVideoCancelledResponse(BaseModel):

backend/app_handler.py

@@ -68,6 +68,7 @@ def __init__(
         ltx_api_client: LTXAPIClient,
         zit_api_client: ZitAPIClient,
         fast_video_pipeline_class: type[FastVideoPipeline],
+        mlx_fast_video_pipeline_class: type[FastVideoPipeline],
         image_generation_pipeline_class: type[ImageGenerationPipeline],
         ic_lora_pipeline_class: type[IcLoraPipeline],
         depth_processor_pipeline_class: type[DepthProcessorPipeline],
@@ -88,6 +89,7 @@ def __init__(
         self.ltx_api_client = ltx_api_client
         self.zit_api_client = zit_api_client
         self.fast_video_pipeline_class = fast_video_pipeline_class
+        self.mlx_fast_video_pipeline_class = mlx_fast_video_pipeline_class
         self.image_generation_pipeline_class = image_generation_pipeline_class
         self.ic_lora_pipeline_class = ic_lora_pipeline_class
         self.depth_processor_pipeline_class = depth_processor_pipeline_class
@@ -160,6 +162,7 @@ def __init__(
             text_handler=self.text,
             gpu_cleaner=gpu_cleaner,
             fast_video_pipeline_class=fast_video_pipeline_class,
+            mlx_fast_video_pipeline_class=mlx_fast_video_pipeline_class,
             image_generation_pipeline_class=image_generation_pipeline_class,
             ic_lora_pipeline_class=ic_lora_pipeline_class,
             depth_processor_pipeline_class=depth_processor_pipeline_class,
@@ -273,6 +276,7 @@ class ServiceBundle:
     ltx_api_client: LTXAPIClient
     zit_api_client: ZitAPIClient
     fast_video_pipeline_class: type[FastVideoPipeline]
+    mlx_fast_video_pipeline_class: type[FastVideoPipeline]
     image_generation_pipeline_class: type[ImageGenerationPipeline]
     ic_lora_pipeline_class: type[IcLoraPipeline]
     depth_processor_pipeline_class: type[DepthProcessorPipeline]
@@ -285,6 +289,7 @@ class ServiceBundle:
 def build_default_service_bundle(config: RuntimeConfig) -> ServiceBundle:
     """Build real runtime services with lazy heavy imports isolated from tests."""
     from services.fast_video_pipeline.ltx_fast_video_pipeline import LTXFastVideoPipeline
+    from services.fast_video_pipeline.mlx_fast_video_pipeline import MLXFastVideoPipeline
     from services.zit_api_client.zit_api_client_impl import ZitAPIClientImpl
     from services.gpu_cleaner.torch_cleaner import TorchCleaner
     from services.gpu_info.gpu_info_impl import GpuInfoImpl
@@ -323,6 +328,7 @@ def build_default_service_bundle(config: RuntimeConfig) -> ServiceBundle:
         ltx_api_client=LTXAPIClientImpl(http=http, ltx_api_base_url=config.ltx_api_base_url),
         zit_api_client=ZitAPIClientImpl(http=http),
         fast_video_pipeline_class=LTXFastVideoPipeline,
+        mlx_fast_video_pipeline_class=MLXFastVideoPipeline,
         image_generation_pipeline_class=ZitImageGenerationPipeline,
         ic_lora_pipeline_class=LTXIcLoraPipeline,
         depth_processor_pipeline_class=MidasDPTPipeline,
@@ -354,6 +360,7 @@ def build_initial_state(
         ltx_api_client=bundle.ltx_api_client,
         zit_api_client=bundle.zit_api_client,
         fast_video_pipeline_class=bundle.fast_video_pipeline_class,
+        mlx_fast_video_pipeline_class=bundle.mlx_fast_video_pipeline_class,
         image_generation_pipeline_class=bundle.image_generation_pipeline_class,
         ic_lora_pipeline_class=bundle.ic_lora_pipeline_class,
         depth_processor_pipeline_class=bundle.depth_processor_pipeline_class,

backend/export_openapi_schema.py

@@ -64,6 +64,7 @@ def _build_schema() -> dict[str, object]:
             ltx_api_client=cast(Any, fake.ltx_api_client),
             zit_api_client=cast(Any, fake.zit_api_client),
             fast_video_pipeline_class=cast(Any, type(fake.fast_video_pipeline)),
+            mlx_fast_video_pipeline_class=cast(Any, type(fake.fast_video_pipeline)),
             image_generation_pipeline_class=cast(Any, type(fake.image_generation_pipeline)),
             ic_lora_pipeline_class=cast(Any, type(fake.ic_lora_pipeline)),
             depth_processor_pipeline_class=cast(Any, type(fake.depth_processor_pipeline)),

backend/handlers/extend_handler.py

@@ -164,16 +164,21 @@ def _run_local_extend(
             extend_frames = self._duration_to_extend_frames(duration, fps)
             target_width, target_height = resolve_target_resolution(resolution, source_width, source_height)
 
-            try:
-                self._text.prepare_text_encoding(prompt, enhance_prompt=False)
-            except RuntimeError as exc:
-                raise HTTPError(400, str(exc)) from exc
-
             generation_id = uuid.uuid4().hex[:8]
             seed = self._resolve_seed()
             output_path = self.config.outputs_dir / f"extend_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{generation_id}.mp4"
 
+            lease = None
             try:
+                lease = self._generation.acquire_local_metal_lease(
+                    generation_id=generation_id,
+                    workload="extend",
+                    reason="Torch local extend generation",
+                )
+                try:
+                    self._text.prepare_text_encoding(prompt, enhance_prompt=False)
+                except RuntimeError as exc:
+                    raise HTTPError(400, str(exc)) from exc
                 pipeline_state = self._pipelines.load_retake_pipeline(distilled=True)
                 self._generation.start_generation(generation_id)
                 self._generation.update_progress("loading_model", 5, 0, 1)
@@ -212,6 +217,9 @@ def _run_local_extend(
                 raise HTTPError(500, f"Generation error: {exc}") from exc
             finally:
                 self._text.clear_api_embeddings()
+                if lease is not None:
+                    self._pipelines.cleanup_runtime_caches()
+                    lease.close()
 
     @staticmethod
     def _duration_to_extend_frames(duration: float, fps: float) -> int:

backend/handlers/generation_handler.py

@@ -17,7 +17,9 @@
     GenerationProgressResponse,
 )
 from handlers.base import StateHandlerBase, with_state_lock
+from services.fast_video_pipeline.mlx_fast_video_pipeline import cancel_active_mlx_sidecar
 from services.patches import diffusion_stage_cache
+from services.local_metal_lease import LocalMetalLeaseHandle, local_metal_lease
 from state.app_state_types import (
     ApiGeneration,
     AppState,
@@ -63,11 +65,29 @@ def try_reserve_generation_start(self) -> bool:
             logger.info("Generation start reservation denied: another reservation is still active")
             return False
         self.state.generation_starting_since = time.monotonic()
+        self.state.generation_starting_id = None
+        self.state.generation_starting_phase = "starting"
+        self.state.generation_start_cancelled = False
         return True
 
     @with_state_lock
     def release_generation_start_reservation(self) -> None:
         self.state.generation_starting_since = None
+        self.state.generation_starting_id = None
+        self.state.generation_starting_phase = "starting"
+        self.state.generation_start_cancelled = False
+
+    @with_state_lock
+    def annotate_generation_start(self, generation_id: str, phase: str) -> None:
+        if self.state.generation_starting_since is None:
+            return
+        self.state.generation_starting_id = generation_id
+        self.state.generation_starting_phase = phase
+
+    @with_state_lock
+    def update_generation_start_phase(self, phase: str) -> None:
+        if self.state.generation_starting_since is not None:
+            self.state.generation_starting_phase = phase
 
     @contextmanager
     def reserved_generation_start(self) -> Iterator[None]:
@@ -88,19 +108,82 @@ def reserved_generation_start(self) -> Iterator[None]:
         finally:
             self.release_generation_start_reservation()
 
+    @contextmanager
+    def hold_local_metal_lease(
+        self,
+        *,
+        generation_id: str,
+        workload: str,
+        reason: str,
+    ) -> Iterator[None]:
+        """Hold the cross-product Metal lease before any local model load."""
+        self.annotate_generation_start(generation_id, "waiting_for_local_accelerator")
+
+        def _on_wait(waited: float, owner: dict[str, object] | None) -> None:
+            owner_product = owner.get("product") if owner else "another LTX app"
+            owner_pid = owner.get("pid") if owner else "?"
+            self.update_generation_start_phase(
+                f"waiting_for_local_accelerator:{owner_product}:pid={owner_pid}:"
+                f"{waited:.1f}s"
+            )
+
+        with local_metal_lease(
+            job_id=generation_id,
+            workload=workload,
+            reason=reason,
+            is_cancelled=self.is_generation_cancelled,
+            on_wait=_on_wait,
+        ):
+            self.update_generation_start_phase("loading_model")
+            yield
+
+    def acquire_local_metal_lease(
+        self,
+        *,
+        generation_id: str,
+        workload: str,
+        reason: str,
+    ) -> LocalMetalLeaseHandle:
+        """Acquire an explicit lease handle; caller closes it after GPU cleanup."""
+        self.annotate_generation_start(generation_id, "waiting_for_local_accelerator")
+
+        def _on_wait(waited: float, owner: dict[str, object] | None) -> None:
+            owner_product = owner.get("product") if owner else "another LTX app"
+            owner_pid = owner.get("pid") if owner else "?"
+            self.update_generation_start_phase(
+                f"waiting_for_local_accelerator:{owner_product}:pid={owner_pid}:"
+                f"{waited:.1f}s"
+            )
+
+        handle = LocalMetalLeaseHandle(
+            local_metal_lease(
+                job_id=generation_id,
+                workload=workload,
+                reason=reason,
+                is_cancelled=self.is_generation_cancelled,
+                on_wait=_on_wait,
+            )
+        )
+        self.update_generation_start_phase("loading_model")
+        return handle
+
     @with_state_lock
     def start_generation(self, generation_id: str) -> None:
         if self.is_generation_running():
             raise RuntimeError("Generation already in progress")
+        if self.state.generation_start_cancelled:
+            raise RuntimeError("Generation was cancelled before model loading completed")
         if self.state.gpu_slot is None:
             raise RuntimeError("No active GPU pipeline")
         self.state.generation_starting_since = None
-
-        # EXPERIMENTAL: push the live Settings toggle, then drop any transformer
-        # cached from the previous generation before this one starts -- otherwise
-        # it stays resident while this generation's own text encoder/VAE/etc.
-        # build, double-booking VRAM. See that module's GENERATION-SCOPED
-        # docstring section for the RTX 5090 repro (~42GB reported on a 32GB card).
+        self.state.generation_starting_id = None
+        self.state.generation_starting_phase = "starting"
+        self.state.generation_start_cancelled = False
+
+        # Push the live Settings toggle, then drop any transformer left by a
+        # cancelled/failed generation before this one starts. A normal two-stage
+        # generation retires the cache immediately after the reuse hit, before VAE
+        # decode; this is the defensive backstop. See diffusion_stage_cache.py.
         diffusion_stage_cache.set_enabled(self.state.app_settings.diffusion_stage_cache_enabled)
         diffusion_stage_cache.evict()
 
@@ -210,6 +293,8 @@ def _generation_for_polling(self) -> GenerationState | None:
 
     @with_state_lock
     def is_generation_cancelled(self) -> bool:
+        if self.state.generation_start_cancelled:
+            return True
         match self._active_generation_state():
             case (_, GenerationCancelled()):
                 return True
@@ -236,10 +321,18 @@ def update_progress(
 
     @with_state_lock
     def cancel_generation(self) -> CancelResponse:
+        if self.state.generation_starting_since is not None:
+            self.state.generation_start_cancelled = True
+            generation_id = self.state.generation_starting_id or "starting"
+            return CancelCancellingResponse(status="cancelling", id=generation_id)
         running_generation = self._running_generation()
         if running_generation is not None:
             slot, running = running_generation
             self._set_generation_state(slot, GenerationCancelled(id=running.id))
+            # Torch pipelines observe the state cooperatively. MLX is isolated
+            # in a fresh child process, so cancellation tears down its whole
+            # process group and releases unified memory deterministically.
+            cancel_active_mlx_sidecar()
             return CancelCancellingResponse(status="cancelling", id=running.id)
 
         cancelled_generation = self._cancelled_generation()
@@ -265,6 +358,9 @@ def fail_generation(self, error: str) -> None:
         # happened before start_generation()/start_api_generation() ever ran (e.g. pipeline load
         # itself threw) — that path never touches generation_starting_since otherwise.
         self.state.generation_starting_since = None
+        self.state.generation_starting_id = None
+        self.state.generation_starting_phase = "starting"
+        self.state.generation_start_cancelled = False
         running_generation = self._running_generation()
         if running_generation is not None:
             slot, running = running_generation
@@ -286,6 +382,16 @@ def get_generation_progress(self) -> GenerationProgressResponse:
         # until start_generation() overwrites it. Matching on gen first would report that stale
         # terminal state instead of "starting" for every generation after the first.
         if self.state.generation_starting_since is not None:
+            generation_id = self.state.generation_starting_id
+            if self.state.generation_start_cancelled:
+                return GenerationProgressResponse(
+                    status="cancelled",
+                    phase="cancelled",
+                    progress=0,
+                    currentStep=0,
+                    totalSteps=0,
+                    id=generation_id,
+                )
             # Reserved (try_reserve_generation_start succeeded) but pipeline load hasn't
             # finished, so start_generation() hasn't run and there's no real id yet.
             # Still report "running": a client polling for "is anything busy right now"
@@ -295,10 +401,11 @@ def get_generation_progress(self) -> GenerationProgressResponse:
             # docstring for why this window needed closing on the write side too.
             return GenerationProgressResponse(
                 status="running",
-                phase="starting",
+                phase=self.state.generation_starting_phase,
                 progress=0,
                 currentStep=0,
                 totalSteps=0,
+                id=generation_id,
             )
 
         gen = self._generation_for_polling()