initi worker working

.vscode/launch.json

@@ -13,5 +13,17 @@
         "python": "/home/robot_op/miniconda3/envs/stream3r/bin/python",
         "justMyCode": true
         },
+        {
+        "name": "Python: STream3R Worker",
+        "type": "debugpy",
+        "request": "launch",
+        "module": "stream3r.worker.main",
+        "args": ["--log-level", "INFO"],
+        "console": "integratedTerminal",
+        "cwd": "${workspaceFolder}",
+        "envFile": "${workspaceFolder}/.env",
+        "python": "/home/robot_op/miniconda3/envs/stream3r/bin/python",
+        "justMyCode": true
+        }
     ]
 }

design_docs/stream3r_api.md

@@ -0,0 +1,269 @@
+# STream3R API — Job Orchestration and Integration Plan
+
+## Executive Summary
+
+This document proposes a lightweight STream3R API service that wraps the async job system implemented by the RQ worker. The API exposes RESTful endpoints and Redis Stream subscriptions that allow upstream applications to submit reconstruction jobs, track progress, and retrieve completed artifacts. Responsibilities are split cleanly: the API handles request validation, persistence, and orchestration, while the GPU worker (documented in `design_docs/worker.md`) performs heavy inference and storage of artifacts.
+
+**Key outcomes**
+- Unified interface for both short `pose_pointmap` jobs and long-running `model_build` jobs.
+- Consistent job lifecycle backed by Postgres and Redis, mirroring the worker contract.
+- Environment-agnostic integration so other services can enqueue jobs or consume progress events without GPU access.
+
+---
+
+## 1. Scope & Goals
+
+### Goals
+- Provide a REST API for submitting jobs and querying job status or results.
+- Offer optional server-sent events (SSE) or WebSocket feeds for near-real-time updates on `pose_pointmap` jobs.
+- Enforce validation and idempotency for job submissions.
+- Expose artifacts written by the worker (S3 URLs, local paths) without re-hosting the files.
+
+### Non-Goals
+- Implement GPU inference or artifact generation (handled by RQ worker).
+- Manage long-term artifact retention or CDN delivery.
+- Provide fine-grained authorization beyond bearer token / API key patterns (left to integration).
+
+---
+
+## 2. Architecture Overview
+
+```
+Client ──HTTP──► STream3R API ──RQ Enqueue──► Redis Queue ─► Worker
+   │                          │                         │
+   │                          └──Postgres──────────────►│
+   │                          └──Redis Stream (events)◄─┘
+   │                          └──S3 (artifact URLs)◄───── Worker writes
+   │
+   └── Poll `/jobs/{id}` or Subscribe SSE/WebSocket for progress updates
+```
+
+Components:
+- **FastAPI** service (recommended) running behind an ASGI server.
+- **Redis**: shared with worker for queues and events.
+- **Postgres**: `stream3r_jobs` table is the canonical job record.
+- **S3/Backblaze** (or local storage): artifact URLs returned by worker.
+- **RQ worker** (implemented separately) executing jobs and updating state.
+
+---
+
+## 3. Endpoints
+
+### `POST /jobs`
+Submit a job for either `pose_pointmap` or `model_build`.
+
+**Request body**
+```json
+{
+  "job_type": "pose_pointmap",
+  "scene_id": "SCENE123",
+  "mode": "causal",
+  "streaming": true,
+  "frames": [
+    {"url": "https://.../frame_0000.jpg"},
+    {"path": "/data/captures/frame_0001.png"}
+  ],
+  "session_settings": {"prediction_mode": "Predicted Pointmap"},
+  "client_request_id": "optional-idempotency-key"
+}
+```
+
+**Behavior**
+- Validate payload (non-empty frames, supported job type, etc.).
+- If `client_request_id` is provided, search Postgres for an existing job with the same key to ensure idempotency.
+- Assign `job_id` (UUID) and enqueue the payload into the appropriate RQ queue (`pose_pointmap` or `model_build`).
+- Insert `stream3r_jobs` row with `status=queued`.
+- Return `202 Accepted` with job metadata:
+
+```json
+{
+  "job_id": "uuid",
+  "status": "queued",
+  "job_type": "pose_pointmap",
+  "scene_id": "SCENE123"
+}
+```
+
+### `GET /jobs/{job_id}`
+Fetch job state and artifact references from Postgres.
+
+**Response example** (`status=finished`)
+```json
+{
+  "job_id": "uuid",
+  "job_type": "model_build",
+  "scene_id": "SCENE123",
+  "status": "finished",
+  "created_at": "...",
+  "started_at": "...",
+  "completed_at": "...",
+  "result": {
+    "result_url": "s3://bucket/scene/SCENE123/stream3r/models/summary.json",
+    "model_dir": "s3://bucket/scene/SCENE123/stream3r/models/",
+    "artifacts": {
+      "scene_glb_url": "...",
+      "poses_url": "...",
+      "pointmaps": [ {"frame_id": "frame_0000", "url": "..."} ]
+    }
+  },
+  "error": null
+}
+```
+
+### `GET /jobs/{job_id}/events`
+Server-Sent Events endpoint bridging Redis Streams.
+
+- Uses `XREAD` on `stream3r:events` with `job_id` filter.
+- Suitable for browser or gateway consumers needing near-real-time progress.
+- Emits lines like:
+
+```
+event: progress
+data: {"progress": 60, "status": "progress"}
+
+event: finished
+data: {"result_url": "s3://..."}
+```
+
+Optionally provide a WebSocket variant if SSE is insufficient.
+
+### `GET /jobs`
+Paged listing/filtering (optional but useful for dashboards).
+
+Parameters: `scene_id`, `job_type`, `status`, pagination cursors.
+
+---
+
+## 4. Data Model & Persistence
+
+### Postgres (`stream3r_jobs`)
+
+The API is the authoritative owner of the job record. It should create and migrate the following schema during startup (extend with `client_request_id` if idempotency keys are required):
+
+```sql
+CREATE TABLE IF NOT EXISTS stream3r_jobs (
+  job_id         UUID PRIMARY KEY,
+  job_type       TEXT NOT NULL,           -- 'pose_pointmap' | 'model_build'
+  scene_id       TEXT NOT NULL,
+  status         TEXT NOT NULL,           -- 'queued' | 'started' | 'finished' | 'failed'
+  created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
+  started_at     TIMESTAMPTZ,
+  completed_at   TIMESTAMPTZ,
+  payload        JSONB,                   -- enqueue-time payload
+  result         JSONB,                   -- worker-published result bundle
+  error          TEXT,
+  client_request_id TEXT UNIQUE           -- optional idempotency key
+);
+
+CREATE INDEX IF NOT EXISTS stream3r_jobs_scene_id_idx ON stream3r_jobs(scene_id);
+CREATE INDEX IF NOT EXISTS stream3r_jobs_status_idx   ON stream3r_jobs(status);
+```
+
+### Redis
+
+- **Queues**: two RQ queues exist—`pose_pointmap` for latency-sensitive pose extraction and `model_build` for long reconstruction jobs. The API selects the queue based on `job_type`.
+- **Event Stream**: the worker pushes lifecycle updates to the Redis Stream `stream3r:events`. Every entry is a flat map with the following fields:
+
+```
+job_id, job_type, scene_id,
+status, progress, result_url, model_dir,
+error, ts
+```
+
+`status` takes `started`, `progress`, `finished`, or `failed`. `progress` is an integer percentage (0–100). `result_url` and `model_dir` mirror the URLs stored in Postgres when a job completes.
+
+### Artifact Storage Layout
+
+The worker persists artifacts to S3/Backblaze (or local storage) under a deterministic folder hierarchy. The API does not move files but should surface these URLs verbatim so consumers know where to fetch results:
+
+```
+scene/{scene_id}/stream3r/
+  results/
+    {job_id}.json                     # per-job result JSON (pose_pointmap)
+  models/
+    kv_cache.pt                       # serialized KV cache
+    predictions.npz                   # packed model outputs
+    session_settings.json             # runtime/config settings
+    selected_frames.json              # frame subset indices (optional)
+    scene.glb                         # fused 3D scene
+    poses.jsonl                       # per-frame extrinsics
+    summary.json                      # canonical model_build result JSON
+    pointmaps/
+      {frame_token}.npz               # per-frame world coords + confidence
+```
+
+`pose_pointmap` jobs typically populate `results/{job_id}.json` plus `pointmaps/`; `model_build` jobs populate the `models/` subtree. All URLs returned by the worker use this structure.
+
+---
+
+## 5. Job Lifecycle
+
+1. **Submit** (`POST /jobs`):
+   - Validate input, persist `queued` row, enqueue payload.
+   - Return `job_id`.
+2. **Worker processing**:
+   - Worker acquires GPU lock, runs inference, streams events, writes artifacts, and updates DB.
+3. **Status checks**:
+   - Clients poll `GET /jobs/{id}` or subscribe to `/jobs/{id}/events`.
+4. **Completion**:
+   - Job row contains `status=finished`, `result` JSON with URLs.
+   - API response is the source of truth for artifact discovery.
+5. **Failure**:
+   - Worker updates DB with `status=failed`, `error` string.
+   - API surfaces the error in `GET /jobs/{id}` and via events.
+
+---
+
+## 6. Request Validation Contracts
+
+`POST /jobs` validation rules:
+- `job_type` ∈ {`pose_pointmap`, `model_build`}.
+- `scene_id` non-empty string.
+- `frames` list size ≥ 1 (unless `frames_dir` provided).
+- Each frame entry must have exactly one of `url`, `path`, or `content` (base64 image string).
+- `mode` default `causal`; forbid `full` for streaming jobs to match worker behavior.
+- Optional numeric fields converted to `int/float` before enqueueing.
+- Enforce max frames (configurable) to avoid resource exhaustion.
+
+---
+
+## 7. Security & Authentication
+
+- Deploy behind an API gateway that injects `X-Client-Id` or similar metadata for auditing.
+- Support bearer token / API key auth via middleware; store hashed keys in Postgres if needed.
+- Restrict access to internal network when possible—as artifacts contain scene data.
+- Sanitize inbound URLs to prevent SSRF; optionally proxy downloads through a whitelist.
+
+---
+
+## 8. Observability & Operations
+
+- **Logging**: Structured logs capturing `job_id`, `scene_id`, `client_request_id`, and remote IP.
+- **Metrics**: Track enqueue latency, job duration (from DB timestamps), queue depth, event lag.
+- **Health checks**: `GET /healthz` verifying Redis and Postgres connectivity.
+- **Backpressure**: Before accepting a new job, check queue length; if above threshold, return `429 Too Many Requests`.
+- **Timeouts**: Configure HTTP request timeouts to avoid hanging on large payloads.
+
+---
+
+## 9. Deployment Considerations
+
+- Package as a standalone FastAPI app (e.g., `stream3r_api.main:app`).
+- Run under Uvicorn/Gunicorn with workers sized for I/O-bound traffic.
+- Configure service with the same environment variables as worker (`STREAM3R_REDIS_URL`, `STREAM3R_DB_DSN`, etc.).
+- Use infrastructure-as-code to provision Redis, Postgres, and S3 credentials shared with worker.
+
+---
+
+## 10. Future Enhancements
+
+- **Job cancellation**: Add `DELETE /jobs/{id}` to flag jobs for cancellation (requires worker support).
+- **Scene-level dashboards**: Aggregate artifacts from multiple jobs for a scene.
+- **Signed download URLs**: API could issue pre-signed URLs for public sharing, decoupled from worker credentials.
+- **Batch submissions**: Support uploading a tar/zip and asynchronously unpacking/validating frames.
+
+---
+
+**References**
+- `design_docs/worker.md` — Worker design and artifact contracts
+- `stream3r/worker/tasks.py` — Concrete payload fields exchanged with the API

design_docs/worker.md

@@ -0,0 +1,300 @@
+# STream3R — Jobs, Events, and Storage Design
+
+## **Executive Summary**
+
+The **STream3R Job System** provides an asynchronous GPU job orchestration layer for 3D scene reconstruction and perception tasks.  
+It standardizes how **pose and world-coordinate extraction** and **scene model building** are executed, stored, and tracked across services.
+
+### **Primary Goals**
+
+- **Asynchronous GPU processing:**  
+  All heavy inference runs on background RQ workers; FastAPI services only enqueue jobs and monitor progress.
+- **Unified observability:**  
+  - **Redis Streams** for job lifecycle events and progress (`stream3r:events`)  
+  - **Postgres (`stream3r_jobs`)** as the canonical job record  
+  - **S3/Backblaze** for durable artifacts and results
+- **Two calling modes:**  
+  - `get_pose_and_world_coords` → **Streams-based (Option A)** for near-real-time updates  
+  - `create_model` → **Polling (Option B)** for long-running model generation
+- **Consistent storage under** `/scene/{scene_id}/stream3r/`, containing:
+  - `kv_cache.pt` — serialized key/value cache state  
+  - `predictions.npz` — packed outputs from the model build  
+  - `session_settings.json` — runtime/config parameters  
+  - `selected_frames.json` — frame subset selection  
+  - `scene.glb` — final assembled scene model  
+  - `poses.jsonl` — per-frame extrinsics (camera poses)  
+  - `pointmaps/*.npz` — per-frame world coordinates + confidence maps  
+
+### **Key Outcomes**
+- Clean separation of API ↔ GPU worker responsibilities  
+- Event-driven feedback for quick jobs; reliable polling for long ones  
+- Durable, versioned scene data under a unified layout  
+- End-to-end traceability of all STream3R jobs via Redis + Postgres + S3  
+
+---
+
+## 1. Queues, Streams, and Locks
+
+| Component | Purpose | Notes |
+|------------|----------|-------|
+| `pose_pointmap` | RQ queue for latency-sensitive `pose_pointmap` jobs |  |
+| `model_build` | RQ queue for long `model_build` jobs |  |
+| `stream3r:events` | Redis Stream for all job events (`started`, `progress`, `finished`, `failed`) | trimmed periodically |
+| `gpu:lock` | Redis lock ensuring single GPU job at a time per machine |  |
+
+Each Stream event is a flat map of strings:
+```
+
+job_id, job_type, scene_id,
+status, progress, result_url, model_dir,
+error, ts
+
+```
+
+---
+
+## 2. S3 / Backblaze Storage Layout
+
+All STream3R artifacts live under a **scene folder**:
+
+```
+
+s3://<bucket>/scene/{scene_id}/stream3r/
+results/
+{job_id}.json                     # per-job result JSON (pose_pointmap)
+models/
+kv_cache.pt                       # serialized KV cache
+predictions.npz                   # packed model outputs
+session_settings.json             # runtime/config settings
+selected_frames.json              # frame subset indices
+scene.glb                         # fused 3D scene
+poses.jsonl                       # per-frame extrinsics
+summary.json                      # canonical model_build result JSON
+pointmaps/
+{frame_token}.npz                 # per-frame world_coords + confidence
+
+````
+
+**Key Result URLs**
+- Pose/pointmap job → `s3://.../scene/{scene_id}/stream3r/results/{job_id}.json`  
+- Model build job → `s3://.../scene/{scene_id}/stream3r/models/summary.json`
+
+---
+
+## 3. Database: `stream3r_jobs`
+
+Canonical job table in Postgres.
+
+```sql
+CREATE TABLE IF NOT EXISTS stream3r_jobs (
+  job_id         UUID PRIMARY KEY,
+  job_type       TEXT NOT NULL,           -- 'pose_pointmap' | 'model_build'
+  scene_id       TEXT NOT NULL,
+  status         TEXT NOT NULL,           -- 'queued' | 'started' | 'finished' | 'failed'
+  created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
+  started_at     TIMESTAMPTZ,
+  completed_at   TIMESTAMPTZ,
+  payload        JSONB,                   -- enqueue-time payload
+  result         JSONB,                   -- URLs / metrics
+  error          TEXT
+);
+
+CREATE INDEX IF NOT EXISTS stream3r_jobs_scene_id_idx ON stream3r_jobs(scene_id);
+CREATE INDEX IF NOT EXISTS stream3r_jobs_status_idx   ON stream3r_jobs(status);
+````
+
+**Upsert pattern:**
+
+* Insert on enqueue (`queued`)
+* Update on start → `started`
+* Update on finish → `finished`, add `result`
+* Update on failure → `failed`, add `error`
+
+---
+
+## 4. Result JSON Schemas
+
+### a. Pose + World Coords (per-frame)
+
+`s3://…/scene/{scene_id}/stream3r/results/{job_id}.json`
+
+```json
+{
+  "job_id": "uuid",
+  "job_type": "pose_pointmap",
+  "scene_id": "SCENE123",
+  "artifacts": {
+    "pointmap_url": "s3://.../scene/SCENE123/stream3r/pointmaps/frame_000010.npz"
+  },
+  "pose": { "R": [[...]], "t": [x, y, z] },
+  "intrinsics": { "fx":..., "fy":..., "cx":..., "cy":... },
+  "metrics": { "runtime_s": 1.23 },
+  "stream3r": { "cfg": "configs/stream3r_base.yaml", "commit": "<git_sha>" }
+}
+```
+
+### b. Model Build (scene-level)
+
+`s3://…/scene/{scene_id}/stream3r/models/summary.json`
+
+```json
+{
+  "job_id": "uuid",
+  "job_type": "model_build",
+  "scene_id": "SCENE123",
+  "artifacts": {
+    "model_dir":        "s3://.../scene/SCENE123/stream3r/models/",
+    "kv_cache":         "s3://.../scene/SCENE123/stream3r/models/kv_cache.pt",
+    "predictions":      "s3://.../scene/SCENE123/stream3r/models/predictions.npz",
+    "session_settings": "s3://.../scene/SCENE123/stream3r/models/session_settings.json",
+    "selected_frames":  "s3://.../scene/SCENE123/stream3r/models/selected_frames.json",
+    "scene_glb":        "s3://.../scene/SCENE123/stream3r/models/scene.glb",
+    "poses_jsonl":      "s3://.../scene/SCENE123/stream3r/models/poses.jsonl"
+  },
+  "metrics": { "frames": 128, "runtime_s": 42.3 },
+  "stream3r": { "cfg": "configs/stream3r_base.yaml", "commit": "<git_sha>" }
+}
+```
+
+---
+
+## 5. Caller API Responsibilities
+
+### `get_pose_and_world_coords` → **Option A (Streams)**
+
+1. Enqueue job → get `job_id`
+2. `XREAD BLOCK` on `stream3r:events` until `status=finished`
+3. On finish:
+
+   * Fetch `result_url`
+   * Load JSON → retrieve `pose`, `intrinsics`, and `pointmap_url`
+   * Download `.npz` to get `world_coords` + `confidence`
+
+### `create_model` → **Option B (Polling)**
+
+1. Enqueue job → return `job_id` immediately
+2. Periodically poll `GET /jobs/{job_id}`
+3. On `finished`:
+
+   * Read `result` with `result_url` + `model_dir`
+   * Download `summary.json` and listed model files
+
+---
+
+## 6. Worker Event & Persistence Flow
+
+1. **Acquire GPU lock**
+2. **Emit** `started`
+3. **Upsert** DB row (`stream3r_jobs`)
+4. **Run inference**, emitting `progress` events (every N frames)
+5. **Save** artifacts to S3:
+
+   * `pointmaps/*.npz` with `{xyz, conf}`
+   * `poses.jsonl`
+   * Model outputs listed above
+6. **Write** result JSON → emit `finished`
+7. **Update** DB row → `status=finished, result=…`
+8. On error → emit `failed`, update DB
+
+---
+
+## 7. Example Event Payloads (Redis Stream)
+
+**Started**
+
+```
+job_id=uuid
+job_type=pose_pointmap
+scene_id=SCENE123
+status=started
+progress=1
+ts=1730312345.12
+```
+
+**Progress**
+
+```
+job_id=uuid
+job_type=model_build
+scene_id=SCENE123
+status=progress
+progress=40
+ts=1730312456.22
+```
+
+**Finished**
+
+```
+job_id=uuid
+job_type=model_build
+scene_id=SCENE123
+status=finished
+progress=100
+result_url=s3://bucket/scene/SCENE123/stream3r/models/summary.json
+model_dir=s3://bucket/scene/SCENE123/stream3r/models/
+ts=1730312567.33
+```
+
+**Failed**
+
+```
+job_id=uuid
+job_type=pose_pointmap
+scene_id=SCENE123
+status=failed
+error=RuntimeError: CUDA OOM
+ts=1730312570.00
+```
+
+---
+
+## 8. Operational Guidelines
+
+| Concern                    | Best Practice                                                   |
+| -------------------------- | --------------------------------------------------------------- |
+| **GPU Safety**             | Use `gpu:lock` to serialize jobs per GPU                        |
+| **Redis Stream retention** | `XTRIM stream3r:events MAXLEN ~50000`                           |
+| **Durability**             | All artifacts and summaries must persist to S3/Backblaze        |
+| **DB Reliability**         | Upsert on each transition; retry writes if DB unavailable       |
+| **Idempotency**            | Support caller-supplied `job_id` or `request_id`                |
+| **Security**               | Keep Redis internal; use signed or private S3 URLs              |
+| **Backpressure**           | Enqueueing API should reject (`429`) when queue depth too large |
+
+---
+
+## 9. End-to-End Flows
+
+### 🔹 Pose + World Coords (short job)
+
+1. API enqueues job → returns `job_id`
+2. Client subscribes via Redis Stream (blocking XREAD)
+3. Worker runs inference → writes `pointmap.npz` + `result.json`
+4. Worker emits `finished` → client downloads results
+
+### 🔹 Model Build (long job)
+
+1. API enqueues → returns `job_id`
+2. Client polls `GET /jobs/{id}` or DB row
+3. Worker fuses frames → writes full scene model files
+4. Worker updates DB + emits `finished`
+5. Client retrieves `summary.json` + artifacts under `/scene/{scene_id}/stream3r/models/`
+
+---
+
+## 10. Summary
+
+| Component                      | Responsibility                          | Persistence                     |
+| ------------------------------ | --------------------------------------- | ------------------------------- |
+| **FastAPI API**                | Enqueue jobs, expose `/jobs/{id}`       | DB (via worker), Redis (events) |
+| **GPU Worker**                 | Execute STream3R inference, emit events | S3/Backblaze, DB                |
+| **Redis Streams**              | Event bus for progress + completion     | ephemeral                       |
+| **Postgres (`stream3r_jobs`)** | Canonical job record                    | durable                         |
+| **S3/Backblaze /scene/**       | Scene artifacts, model data             | durable                         |
+
+---
+
+**Outcome:**
+This design provides an **asynchronous, event-driven, and durable** framework for managing STream3R GPU jobs, with standardized scene storage, traceable job metadata, and clear integration points for both real-time and long-running workflows.
+
+```
+```

requirements.txt

@@ -43,10 +43,17 @@ pyglet<2
 huggingface-hub[torch]>=0.22
 spaces
 
+# --------- worker --------- #
+redis
+rq
+boto3
+psycopg2-binary
+requests
+
 # --------- eval --------- #
 accelerate
 evo
 
 # --------- demo --------- #
 gradio==5.17.1
-onnxruntime
+onnxruntime

stream3r/models/components/utils/geometry.py

@@ -32,8 +32,14 @@ def unproject_depth_map_to_point_map(
 
     world_points_list = []
     for frame_idx in range(depth_map.shape[0]):
+        intrinsic = intrinsics_cam[frame_idx]
+        if intrinsic.shape[-2:] != (3, 3):
+            intrinsic = intrinsic.reshape(-1, 3, 3)[0]
+        extrinsic = extrinsics_cam[frame_idx]
+        if extrinsic.shape[-2:] != (3, 4):
+            extrinsic = extrinsic.reshape(-1, 3, 4)[0]
         cur_world_points, _, _ = depth_to_world_coords_points(
-            depth_map[frame_idx].squeeze(-1), extrinsics_cam[frame_idx], intrinsics_cam[frame_idx]
+            depth_map[frame_idx].squeeze(-1), extrinsic, intrinsic
         )
         world_points_list.append(cur_world_points)
     world_points_array = np.stack(world_points_list, axis=0)

stream3r/worker/__init__.py

@@ -0,0 +1,8 @@
+"""Worker utilities for running STream3R jobs via RQ."""
+
+from .tasks import model_build_job, pose_pointmap_job
+
+__all__ = [
+    "pose_pointmap_job",
+    "model_build_job",
+]

stream3r/worker/config.py

@@ -0,0 +1,213 @@
+"""Configuration helpers for the STream3R RQ worker."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import os
+from pathlib import Path
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    value = value.strip().lower()
+    if value in {"1", "true", "yes", "y", "on"}:
+        return True
+    if value in {"0", "false", "no", "n", "off"}:
+        return False
+    return default
+
+
+def _env_bool_any(default: bool, *names: str) -> bool:
+    for name in names:
+        if not name:
+            continue
+        value = os.getenv(name)
+        if value is None:
+            continue
+        value = value.strip().lower()
+        if value in {"1", "true", "yes", "y", "on"}:
+            return True
+        if value in {"0", "false", "no", "n", "off"}:
+            return False
+    return default
+
+
+def _env_int(name: str, default: int) -> int:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    try:
+        return int(value)
+    except ValueError:
+        return default
+
+
+def _env_value(primary: str, *aliases: str, default: str | None = None) -> str | None:
+    """Best-effort environment lookup with fallbacks for legacy names."""
+
+    names = (primary, *aliases)
+    for name in names:
+        if not name:
+            continue
+        value = os.getenv(name)
+        if value:
+            return value
+    return default
+
+
+@dataclass(slots=True)
+class WorkerSettings:
+    """Runtime configuration derived from environment variables."""
+
+    redis_url: str = "redis://localhost:6379/0"
+    redis_events_stream: str = "stream3r:events"
+    redis_stream_maxlen: int = 50_000
+    redis_healthcheck_interval: int = 30
+
+    pose_queue: str = "pose_pointmap"
+    model_queue: str = "model_build"
+
+    gpu_lock_key: str = "gpu:lock"
+    gpu_lock_timeout: int = 3600
+    gpu_lock_blocking_timeout: int = 600
+
+    storage_prefix: str = "scene"
+    s3_bucket: str | None = None
+    s3_endpoint_url: str | None = None
+    s3_region: str | None = None
+    s3_profile: str | None = None
+    s3_force_path_style: bool = False
+    aws_access_key_id: str | None = None
+    aws_secret_access_key: str | None = None
+    aws_session_token: str | None = None
+    local_storage_root: Path = field(default_factory=lambda: Path("storage"))
+
+    db_dsn: str | None = None
+    job_table: str = "stream3r_jobs"
+
+    model_id: str = "yslan/STream3R"
+    model_revision: str | None = None
+    model_device_preference: str | None = None
+    model_dtype: str | None = None
+
+    default_mode: str = "window"
+    default_streaming: bool = True
+    download_workers: int = 4
+
+    worker_name: str | None = None
+
+    session_cache_filename: str = "kv_cache.pt"
+    predictions_filename: str = "predictions.npz"
+    poses_filename: str = "poses.jsonl"
+    result_filename: str = "summary.json"
+    selected_frames_filename: str = "selected_frames.json"
+    scene_glb_filename: str = "scene.glb"
+    session_settings_filename: str = "session_settings.json"
+    summary_results_filename: str = "summary.json"
+
+    pointmap_dir: str = "pointmaps"
+    models_dir: str = "models"
+    results_dir: str = "results"
+
+    scene_media_api_base_url: str | None = None
+    scene_media_api_token: str | None = None
+    scene_media_page_size: int = 200
+    stream_window_size: int = 10
+    max_frames_per_job: int = 10
+
+    @classmethod
+    def from_env(cls) -> "WorkerSettings":
+        base = cls()
+
+        kwargs: dict[str, object] = {
+            "redis_url": _env_value("STREAM3R_REDIS_URL", "REDIS_URL", default=base.redis_url),
+            "redis_events_stream": os.getenv("STREAM3R_EVENTS_STREAM", base.redis_events_stream),
+            "redis_stream_maxlen": _env_int("STREAM3R_EVENTS_MAXLEN", base.redis_stream_maxlen),
+            "redis_healthcheck_interval": _env_int(
+                "STREAM3R_REDIS_HEALTHCHECK", base.redis_healthcheck_interval
+            ),
+            "pose_queue": os.getenv("STREAM3R_QUEUE_POSE", base.pose_queue),
+            "model_queue": os.getenv("STREAM3R_QUEUE_MODEL", base.model_queue),
+            "gpu_lock_key": os.getenv("STREAM3R_GPU_LOCK_KEY", base.gpu_lock_key),
+            "gpu_lock_timeout": _env_int("STREAM3R_GPU_LOCK_TIMEOUT", base.gpu_lock_timeout),
+            "gpu_lock_blocking_timeout": _env_int(
+                "STREAM3R_GPU_LOCK_BLOCK", base.gpu_lock_blocking_timeout
+            ),
+            "storage_prefix": os.getenv("STREAM3R_STORAGE_PREFIX", base.storage_prefix),
+            "s3_bucket": _env_value("STREAM3R_STORAGE_BUCKET", "S3_BUCKET", default=base.s3_bucket) or None,
+            "s3_endpoint_url": _env_value("STREAM3R_S3_ENDPOINT", "S3_ENDPOINT", default=base.s3_endpoint_url) or None,
+            "s3_region": _env_value("STREAM3R_S3_REGION", "AWS_REGION", default=base.s3_region) or None,
+            "s3_profile": os.getenv("STREAM3R_S3_PROFILE", base.s3_profile or "") or None,
+            "s3_force_path_style": _env_bool_any(
+                base.s3_force_path_style,
+                "STREAM3R_S3_FORCE_PATH",
+                "S3_FORCE_PATH_STYLE",
+            ),
+            "aws_access_key_id": _env_value("AWS_ACCESS_KEY_ID", default=base.aws_access_key_id) or None,
+            "aws_secret_access_key": _env_value("AWS_SECRET_ACCESS_KEY", default=base.aws_secret_access_key) or None,
+            "aws_session_token": _env_value("AWS_SESSION_TOKEN", default=base.aws_session_token) or None,
+            "local_storage_root": Path(
+                os.getenv("STREAM3R_LOCAL_STORAGE", str(base.local_storage_root))
+            ).resolve(),
+            "db_dsn": _env_value("STREAM3R_DB_DSN", "DATABASE_URL", default=base.db_dsn) or None,
+            "job_table": os.getenv("STREAM3R_JOBS_TABLE", base.job_table),
+            "model_id": os.getenv("STREAM3R_MODEL_ID", base.model_id),
+            "model_revision": os.getenv("STREAM3R_MODEL_REVISION", base.model_revision or "") or None,
+            "model_device_preference": os.getenv(
+                "STREAM3R_MODEL_DEVICE", base.model_device_preference or ""
+            )
+            or None,
+            "model_dtype": os.getenv("STREAM3R_MODEL_DTYPE", base.model_dtype or "") or None,
+            "default_mode": os.getenv("STREAM3R_DEFAULT_MODE", base.default_mode),
+            "default_streaming": _env_bool(
+                "STREAM3R_DEFAULT_STREAMING", base.default_streaming
+            ),
+            "download_workers": _env_int("STREAM3R_DOWNLOAD_WORKERS", base.download_workers),
+            "worker_name": os.getenv("STREAM3R_WORKER_NAME", base.worker_name or "") or None,
+            "session_cache_filename": os.getenv(
+                "STREAM3R_SESSION_CACHE", base.session_cache_filename
+            ),
+            "predictions_filename": os.getenv(
+                "STREAM3R_PREDICTIONS_FILE", base.predictions_filename
+            ),
+            "poses_filename": os.getenv("STREAM3R_POSES_FILE", base.poses_filename),
+            "result_filename": os.getenv("STREAM3R_RESULT_FILE", base.result_filename),
+            "selected_frames_filename": os.getenv(
+                "STREAM3R_SELECTED_FRAMES_FILE", base.selected_frames_filename
+            ),
+            "scene_glb_filename": os.getenv("STREAM3R_SCENE_GLB_FILE", base.scene_glb_filename),
+            "session_settings_filename": os.getenv(
+                "STREAM3R_SESSION_SETTINGS_FILE", base.session_settings_filename
+            ),
+            "summary_results_filename": os.getenv(
+                "STREAM3R_SUMMARY_FILE", base.summary_results_filename
+            ),
+            "pointmap_dir": os.getenv("STREAM3R_POINTMAP_DIR", base.pointmap_dir),
+            "models_dir": os.getenv("STREAM3R_MODELS_DIR", base.models_dir),
+            "results_dir": os.getenv("STREAM3R_RESULTS_DIR", base.results_dir),
+            "scene_media_api_base_url": _env_value(
+                "STREAM3R_MEDIA_API_BASE_URL",
+                "API_BASE_URL",
+                default=base.scene_media_api_base_url,
+            )
+            or None,
+            "scene_media_api_token": _env_value(
+                "STREAM3R_MEDIA_API_TOKEN",
+                "MEDIA_API_TOKEN",
+                default=base.scene_media_api_token,
+            )
+            or None,
+            "scene_media_page_size": _env_int(
+                "STREAM3R_MEDIA_PAGE_SIZE", base.scene_media_page_size
+            ),
+            "stream_window_size": _env_int(
+                "STREAM3R_WINDOW_SIZE", base.stream_window_size
+            ),
+            "max_frames_per_job": _env_int(
+                "STREAM3R_MAX_FRAMES", base.max_frames_per_job
+            ),
+        }
+
+        return cls(**kwargs)

stream3r/worker/db.py

@@ -0,0 +1,170 @@
+"""Database helpers for persisting job metadata."""
+
+from __future__ import annotations
+
+import json
+import logging
+from contextlib import contextmanager
+from typing import Iterator, Mapping
+
+try:  # Optional dependency
+    import psycopg2
+    from psycopg2.extensions import connection as PGConnection
+    from psycopg2.extras import Json
+except ModuleNotFoundError:  # pragma: no cover - exercised when psycopg2 missing
+    psycopg2 = None  # type: ignore[assignment]
+    PGConnection = None  # type: ignore[assignment]
+    Json = None  # type: ignore[assignment]
+
+from .config import WorkerSettings
+
+
+logger = logging.getLogger(__name__)
+
+
+SCHEMA_SQL = """
+CREATE TABLE IF NOT EXISTS {table_name} (
+  job_id         UUID PRIMARY KEY,
+  job_type       TEXT NOT NULL,
+  scene_id       TEXT NOT NULL,
+  status         TEXT NOT NULL,
+  created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
+  started_at     TIMESTAMPTZ,
+  completed_at   TIMESTAMPTZ,
+  payload        JSONB,
+  result         JSONB,
+  error          TEXT
+);
+
+CREATE INDEX IF NOT EXISTS {table_name}_scene_id_idx ON {table_name}(scene_id);
+CREATE INDEX IF NOT EXISTS {table_name}_status_idx   ON {table_name}(status);
+"""
+
+
+class DatabaseError(RuntimeError):
+    """Raised when database operations fail."""
+
+
+class BaseDatabaseClient:
+    """Small interface for database persistence."""
+
+    def ensure_schema(self) -> None:  # pragma: no cover - noop implementation
+        raise NotImplementedError
+
+    def upsert_job(
+        self,
+        *,
+        job_id: str,
+        job_type: str,
+        scene_id: str,
+        status: str,
+        payload: Mapping[str, object] | None = None,
+        result: Mapping[str, object] | None = None,
+        error: str | None = None,
+    ) -> None:
+        raise NotImplementedError
+
+    def close(self) -> None:  # pragma: no cover - noop implementation
+        pass
+
+
+class NoopDatabaseClient(BaseDatabaseClient):
+    """Fallback when no database configuration is provided."""
+
+    def ensure_schema(self) -> None:  # pragma: no cover - nothing to do
+        logger.debug("Database is disabled; skipping schema creation")
+
+    def upsert_job(
+        self,
+        *,
+        job_id: str,
+        job_type: str,
+        scene_id: str,
+        status: str,
+        payload: Mapping[str, object] | None = None,
+        result: Mapping[str, object] | None = None,
+        error: str | None = None,
+    ) -> None:
+        logger.debug(
+            "Noop DB: job_id=%s job_type=%s scene_id=%s status=%s", job_id, job_type, scene_id, status
+        )
+
+
+class DatabaseClient(BaseDatabaseClient):
+    """Postgres implementation using psycopg2."""
+
+    def __init__(self, settings: WorkerSettings):
+        if psycopg2 is None:  # pragma: no cover - optional dependency guard
+            raise DatabaseError("psycopg2-binary is required for database support")
+
+        self.settings = settings
+
+    @contextmanager
+    def _connect(self) -> Iterator[PGConnection]:
+        conn = psycopg2.connect(self.settings.db_dsn)  # type: ignore[arg-type]
+        try:
+            conn.autocommit = True
+            yield conn
+        finally:
+            conn.close()
+
+    def ensure_schema(self) -> None:
+        table_name = self.settings.job_table
+        with self._connect() as conn:
+            with conn.cursor() as cur:
+                cur.execute(SCHEMA_SQL.format(table_name=table_name))
+
+    def upsert_job(
+        self,
+        *,
+        job_id: str,
+        job_type: str,
+        scene_id: str,
+        status: str,
+        payload: Mapping[str, object] | None = None,
+        result: Mapping[str, object] | None = None,
+        error: str | None = None,
+    ) -> None:
+        table = self.settings.job_table
+        payload_json = Json(payload) if payload is not None and Json is not None else None
+        result_json = Json(result) if result is not None and Json is not None else None
+
+        with self._connect() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f"""
+                    INSERT INTO {table} (job_id, job_type, scene_id, status, payload, result, error, started_at, completed_at)
+                    VALUES (%s, %s, %s, %s, %s, %s, %s,
+                            CASE WHEN %s = 'started' THEN now() ELSE NULL END,
+                            CASE WHEN %s IN ('finished', 'failed') THEN now() ELSE NULL END)
+                    ON CONFLICT (job_id)
+                    DO UPDATE SET
+                        job_type = EXCLUDED.job_type,
+                        scene_id = EXCLUDED.scene_id,
+                        status = EXCLUDED.status,
+                        payload = COALESCE(EXCLUDED.payload, {table}.payload),
+                        result = COALESCE(EXCLUDED.result, {table}.result),
+                        error = EXCLUDED.error,
+                        started_at = COALESCE({table}.started_at, EXCLUDED.started_at),
+                        completed_at = COALESCE({table}.completed_at, EXCLUDED.completed_at)
+                    """,
+                    (
+                        job_id,
+                        job_type,
+                        scene_id,
+                        status,
+                        payload_json,
+                        result_json,
+                        error,
+                        status,
+                        status,
+                    ),
+                )
+
+
+def create_database_client(settings: WorkerSettings) -> BaseDatabaseClient:
+    """Factory that returns a database client or a noop fallback."""
+
+    if not settings.db_dsn:
+        return NoopDatabaseClient()
+    return DatabaseClient(settings)

stream3r/worker/main.py

@@ -0,0 +1,59 @@
+"""CLI entrypoint to launch an RQ worker for STream3R jobs."""
+
+from __future__ import annotations
+
+import argparse
+import logging
+from typing import Sequence
+
+from rq import Queue, Worker
+
+from .config import WorkerSettings
+from .runtime import get_runtime
+
+
+logger = logging.getLogger(__name__)
+
+
+def _parse_args(default_queues: Sequence[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run the STream3R RQ worker")
+    parser.add_argument(
+        "--queue",
+        "--queues",
+        dest="queues",
+        action="append",
+        help="Queue names to listen to (can be repeated)",
+    )
+    parser.add_argument(
+        "--burst",
+        action="store_true",
+        help="Run in burst mode (exit when queues are empty)",
+    )
+    parser.add_argument(
+        "--log-level",
+        default="INFO",
+        help="Logging level",
+    )
+    args = parser.parse_args()
+    if not args.queues:
+        args.queues = list(default_queues)
+    return args
+
+
+def main() -> None:
+    settings = WorkerSettings.from_env()
+    args = _parse_args([settings.pose_queue, settings.model_queue])
+    logging.basicConfig(level=getattr(logging, str(args.log_level).upper(), logging.INFO))
+
+    runtime = get_runtime()
+
+    queues = [Queue(name, connection=runtime.redis) for name in args.queues]
+    for queue in queues:
+        logger.info("Listening on queue '%s'", queue.name)
+
+    worker = Worker(queues, name=settings.worker_name)
+    worker.work(burst=args.burst)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

stream3r/worker/pipeline.py

@@ -0,0 +1,144 @@
+"""Inference pipeline utilities reused by worker jobs."""
+
+from __future__ import annotations
+
+from contextlib import nullcontext
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Iterable, Mapping
+
+import numpy as np
+import torch
+
+from stream3r.models.components.utils.load_fn import load_and_preprocess_images
+from stream3r.models.components.utils.pose_enc import pose_encoding_to_extri_intri
+from stream3r.stream_session import StreamSession
+
+from .runtime import WorkerRuntime
+
+
+ProgressCallback = Callable[[int, int], None]
+
+
+@dataclass
+class InferenceResult:
+    predictions: dict[str, np.ndarray]
+    total_frames: int
+    cache_path: Path | None
+
+
+def _to_numpy(payload):
+    if isinstance(payload, torch.Tensor):
+        return payload.detach().cpu().numpy()
+    if isinstance(payload, dict):
+        return {k: _to_numpy(v) for k, v in payload.items()}
+    if isinstance(payload, (list, tuple)):
+        converted = [_to_numpy(item) for item in payload]
+        return type(payload)(converted)
+    return payload
+
+
+def run_stream3r_inference(
+    *,
+    runtime: WorkerRuntime,
+    image_paths: Iterable[Path],
+    mode: str,
+    streaming: bool,
+    cache_output_path: Path | None,
+    progress_cb: ProgressCallback | None = None,
+    window_size: int | None = None,
+) -> InferenceResult:
+    """Execute STream3R inference for the provided frames."""
+
+    image_list = [Path(p) for p in image_paths]
+    if not image_list:
+        raise ValueError("No images provided to inference pipeline")
+
+    model = runtime.get_model()
+    device = runtime.model_device()
+
+    images = load_and_preprocess_images([str(path) for path in image_list])
+    total_frames = images.shape[0]
+
+    autocast_dtype = runtime.autocast_dtype()
+    autocast_ctx = (
+        torch.autocast(device_type=device.type, dtype=autocast_dtype)
+        if device.type == "cuda"
+        else nullcontext()
+    )
+
+    predictions: Mapping[str, torch.Tensor]
+    cache_path: Path | None = None
+
+    model.eval()
+
+    if window_size is not None and window_size <= 0:
+        window_size = None
+
+    with torch.no_grad():
+        if streaming:
+            session_kwargs = {"mode": mode}
+            if window_size is not None:
+                session_kwargs["window_size"] = window_size
+            session = StreamSession(model, **session_kwargs)
+            session.clear()
+            for idx in range(total_frames):
+                frame = images[idx : idx + 1].to(device)
+                with autocast_ctx:
+                    session.forward_stream(frame)
+                if progress_cb is not None:
+                    progress_cb(idx + 1, total_frames)
+
+            if cache_output_path is not None:
+                session.save_cache(str(cache_output_path))
+                cache_path = cache_output_path
+
+            predictions = session.get_all_predictions()
+        else:
+            inputs = images.to(device)
+            with autocast_ctx:
+                predictions = model(inputs, mode=mode)
+            if progress_cb is not None:
+                progress_cb(total_frames, total_frames)
+
+    predictions = dict(predictions)
+
+    # Augment predictions with pose matrices and world coordinates
+    height, width = images.shape[-2:]
+
+    pose_enc = predictions.get("pose_enc")
+    if pose_enc is None:
+        raise RuntimeError("Model predictions missing 'pose_enc'")
+
+    if not isinstance(pose_enc, torch.Tensor):
+        pose_enc = torch.as_tensor(pose_enc)
+
+    if pose_enc.dim() == 2:  # streaming cache might drop batch dim
+        pose_enc = pose_enc.unsqueeze(0)
+
+    extrinsic, intrinsic = pose_encoding_to_extri_intri(pose_enc, (height, width))
+    predictions["extrinsic"] = extrinsic
+    predictions["intrinsic"] = intrinsic
+
+    for key, value in list(predictions.items()):
+        if isinstance(value, torch.Tensor):
+            predictions[key] = value
+
+    predictions_np = {key: _to_numpy(value) for key, value in predictions.items()}
+
+    pose_enc_np = predictions_np.pop("pose_enc", None)
+    if pose_enc_np is not None and pose_enc_np.ndim >= 3:
+        predictions_np["pose_enc"] = pose_enc_np
+
+    # Remove batch dimension if present
+    for key, value in list(predictions_np.items()):
+        if isinstance(value, np.ndarray) and value.shape[0] == 1:
+            predictions_np[key] = np.squeeze(value, axis=0)
+
+    torch.cuda.empty_cache()
+
+    return InferenceResult(
+        predictions=predictions_np,
+        total_frames=total_frames,
+        cache_path=cache_path,
+    )

stream3r/worker/runtime.py

@@ -0,0 +1,159 @@
+"""Runtime registry for shared worker resources."""
+
+from __future__ import annotations
+
+import logging
+from contextlib import contextmanager
+from threading import Lock
+from typing import Any, Dict, Mapping
+
+import redis
+import torch
+
+from stream3r.models.stream3r import STream3R
+
+from .config import WorkerSettings
+from .db import BaseDatabaseClient, create_database_client
+from .storage import StorageClient, create_storage_client
+
+
+logger = logging.getLogger(__name__)
+
+
+class WorkerRuntime:
+    """Holds shared state reused across RQ jobs."""
+
+    def __init__(self, settings: WorkerSettings):
+        self.settings = settings
+        self._redis = redis.Redis.from_url(
+            settings.redis_url,
+            decode_responses=False,
+            health_check_interval=settings.redis_healthcheck_interval,
+        )
+        self.storage: StorageClient = create_storage_client(settings)
+        self.db: BaseDatabaseClient = create_database_client(settings)
+
+        try:
+            self.db.ensure_schema()
+        except Exception as exc:  # pragma: no cover - depends on external DB
+            logger.warning("Failed to ensure job schema: %s", exc)
+
+        self._model: STream3R | None = None
+        self._model_lock = Lock()
+        self._device: torch.device | None = None
+        self._autocast_dtype: torch.dtype | None = None
+
+    # -----------------------------------------------------------------
+    # Redis helpers
+    # -----------------------------------------------------------------
+    @property
+    def redis(self) -> redis.Redis:
+        return self._redis
+
+    def emit_event(self, payload: Mapping[str, Any]) -> None:
+        try:
+            stream = self.settings.redis_events_stream
+            data = {k: str(v) for k, v in payload.items() if v is not None}
+            maxlen = self.settings.redis_stream_maxlen
+            self._redis.xadd(stream, data, maxlen=maxlen, approximate=True)
+        except redis.RedisError as exc:  # pragma: no cover - depends on Redis
+            logger.warning("Failed to emit event to Redis: %s", exc)
+
+    @contextmanager
+    def gpu_lock(self) -> Any:
+        lock = self._redis.lock(
+            self.settings.gpu_lock_key,
+            timeout=self.settings.gpu_lock_timeout,
+            blocking_timeout=self.settings.gpu_lock_blocking_timeout,
+        )
+        acquired = False
+        try:
+            acquired = lock.acquire(blocking=True)
+            if not acquired:
+                raise TimeoutError("Timed out waiting for GPU lock")
+            yield
+        finally:
+            if acquired:
+                try:
+                    lock.release()
+                except redis.RedisError:  # pragma: no cover - depends on Redis
+                    logger.debug("GPU lock already released")
+
+    # -----------------------------------------------------------------
+    # Model helpers
+    # -----------------------------------------------------------------
+    def _resolve_device(self) -> torch.device:
+        if self._device is not None:
+            return self._device
+
+        preference = self.settings.model_device_preference
+        if preference:
+            try:
+                device = torch.device(preference)
+            except (ValueError, RuntimeError):
+                logger.warning("Unknown device preference '%s', falling back to auto", preference)
+                device = None
+        else:
+            device = None
+
+        if device is None:
+            device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+        self._device = device
+        return device
+
+    def _resolve_autocast_dtype(self) -> torch.dtype:
+        if self._autocast_dtype is not None:
+            return self._autocast_dtype
+
+        dtype_name = self.settings.model_dtype
+        if dtype_name:
+            try:
+                self._autocast_dtype = getattr(torch, dtype_name)
+                return self._autocast_dtype
+            except AttributeError:
+                logger.warning("Unsupported dtype '%s', using default", dtype_name)
+
+        if torch.cuda.is_available() and torch.cuda.get_device_capability()[0] >= 8:
+            self._autocast_dtype = torch.bfloat16
+        else:
+            self._autocast_dtype = torch.float16 if torch.cuda.is_available() else torch.float32
+        return self._autocast_dtype
+
+    def get_model(self) -> STream3R:
+        with self._model_lock:
+            if self._model is None:
+                logger.info("Loading STream3R model '%s'", self.settings.model_id)
+                model = STream3R.from_pretrained(
+                    self.settings.model_id,
+                    revision=self.settings.model_revision,
+                )
+                device = self._resolve_device()
+                model.to(device)
+                model.eval()
+                self._model = model
+            return self._model
+
+    def model_device(self) -> torch.device:
+        return self._resolve_device()
+
+    def autocast_dtype(self) -> torch.dtype:
+        return self._resolve_autocast_dtype()
+
+    # -----------------------------------------------------------------
+    def close(self) -> None:
+        try:
+            self.db.close()
+        except AttributeError:
+            pass
+
+
+_RUNTIME: WorkerRuntime | None = None
+
+
+def get_runtime() -> WorkerRuntime:
+    global _RUNTIME
+    if _RUNTIME is None:
+        settings = WorkerSettings.from_env()
+        _RUNTIME = WorkerRuntime(settings)
+    return _RUNTIME

stream3r/worker/storage.py

@@ -0,0 +1,180 @@
+"""Storage backends for persisting STream3R job artifacts."""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+from typing import Mapping
+
+try:  # Lazy import to keep optional dependency
+    import boto3
+    from botocore.client import Config as Boto3Config
+    from botocore.exceptions import BotoCoreError, ClientError
+except ModuleNotFoundError:  # pragma: no cover - fallback when boto3 missing
+    boto3 = None  # type: ignore[assignment]
+    Boto3Config = None  # type: ignore[assignment]
+    BotoCoreError = ClientError = Exception  # type: ignore[assignment]
+
+from .config import WorkerSettings
+
+
+class StorageError(RuntimeError):
+    """Raised when artifact persistence fails."""
+
+
+class StorageClient:
+    """Abstract base class providing a minimal upload interface."""
+
+    def __init__(self, settings: WorkerSettings):
+        self.settings = settings
+
+    # --- key builders -------------------------------------------------
+    def build_key(self, scene_id: str, *parts: str) -> str:
+        components = [str(self.settings.storage_prefix), str(scene_id), "stream3r"]
+        for part in parts:
+            if not part:
+                continue
+            components.append(str(part).strip("/"))
+        return "/".join(components)
+
+    def build_uri(self, key: str) -> str:
+        raise NotImplementedError
+
+    # --- upload primitives -------------------------------------------
+    def upload_file(self, local_path: Path, key: str, *, content_type: str | None = None) -> str:
+        raise NotImplementedError
+
+    def upload_bytes(self, data: bytes, key: str, *, content_type: str | None = None) -> str:
+        raise NotImplementedError
+
+    def upload_json(self, payload: Mapping[str, object], key: str) -> str:
+        data = json.dumps(payload, allow_nan=False).encode("utf-8")
+        return self.upload_bytes(data, key, content_type="application/json")
+
+    def download_to_path(self, key: str, destination: Path) -> Path:
+        """Download an object identified by *key* into *destination*."""
+
+        raise NotImplementedError
+
+    # --- helpers ------------------------------------------------------
+    def ensure_dir(self, scene_id: str, *parts: str) -> str:
+        """Create a logical directory path under the storage prefix."""
+        key = self.build_key(scene_id, *parts)
+        if not key.endswith("/"):
+            key = f"{key}/"
+        return key
+
+
+class S3StorageClient(StorageClient):
+    """S3-compatible storage backend."""
+
+    def __init__(self, settings: WorkerSettings):
+        if boto3 is None:  # pragma: no cover - guarded by optional dependency
+            raise StorageError("boto3 is required for S3 storage but is not installed")
+
+        super().__init__(settings)
+
+        session_kwargs: dict[str, object] = {}
+        if settings.s3_profile:
+            session_kwargs["profile_name"] = settings.s3_profile
+        if settings.s3_region:
+            session_kwargs["region_name"] = settings.s3_region
+        if settings.aws_access_key_id and settings.aws_secret_access_key:
+            session_kwargs["aws_access_key_id"] = settings.aws_access_key_id
+            session_kwargs["aws_secret_access_key"] = settings.aws_secret_access_key
+        if settings.aws_session_token:
+            session_kwargs["aws_session_token"] = settings.aws_session_token
+
+        session = boto3.session.Session(**session_kwargs)
+        config = None
+        if settings.s3_force_path_style and Boto3Config is not None:
+            config = Boto3Config(s3={"addressing_style": "path"})
+
+        self._client = session.client(
+            "s3",
+            endpoint_url=settings.s3_endpoint_url,
+            config=config,
+        )
+
+        if not settings.s3_bucket:
+            raise StorageError("STREAM3R_STORAGE_BUCKET is required for S3 storage")
+
+    def build_uri(self, key: str) -> str:
+        bucket = self.settings.s3_bucket
+        return f"s3://{bucket}/{key}"
+
+    def upload_file(self, local_path: Path, key: str, *, content_type: str | None = None) -> str:
+        extra_args = {"ContentType": content_type} if content_type else None
+        try:
+            self._client.upload_file(str(local_path), self.settings.s3_bucket, key, ExtraArgs=extra_args)
+        except (BotoCoreError, ClientError) as exc:  # pragma: no cover - network side effects
+            raise StorageError(f"Failed to upload {local_path} to {key}: {exc}") from exc
+        return self.build_uri(key)
+
+    def upload_bytes(self, data: bytes, key: str, *, content_type: str | None = None) -> str:
+        extra_args = {"ContentType": content_type} if content_type else None
+        try:
+            self._client.put_object(
+                Bucket=self.settings.s3_bucket,
+                Key=key,
+                Body=data,
+                ContentType=extra_args.get("ContentType") if extra_args else None,
+            )
+        except (BotoCoreError, ClientError) as exc:  # pragma: no cover - network side effects
+            raise StorageError(f"Failed to upload payload to {key}: {exc}") from exc
+        return self.build_uri(key)
+
+    def download_to_path(self, key: str, destination: Path) -> Path:
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            object_key = str(key).lstrip("/")
+            self._client.download_file(self.settings.s3_bucket, object_key, str(destination))
+        except (BotoCoreError, ClientError) as exc:  # pragma: no cover - network side effects
+            raise StorageError(
+                f"Failed to download {object_key} from bucket {self.settings.s3_bucket} to {destination}: {exc}"
+            ) from exc
+        return destination
+
+
+class LocalStorageClient(StorageClient):
+    """On-disk storage backend for development and testing."""
+
+    def __init__(self, settings: WorkerSettings):
+        super().__init__(settings)
+        self.root = settings.local_storage_root
+        self.root.mkdir(parents=True, exist_ok=True)
+
+    def _resolve(self, key: str) -> Path:
+        path = self.root.joinpath(*key.split("/"))
+        path.parent.mkdir(parents=True, exist_ok=True)
+        return path
+
+    def build_uri(self, key: str) -> str:
+        return str(self._resolve(key))
+
+    def upload_file(self, local_path: Path, key: str, *, content_type: str | None = None) -> str:  # noqa: ARG002
+        destination = self._resolve(key)
+        shutil.copyfile(local_path, destination)
+        return str(destination)
+
+    def upload_bytes(self, data: bytes, key: str, *, content_type: str | None = None) -> str:  # noqa: ARG002
+        destination = self._resolve(key)
+        destination.write_bytes(data)
+        return str(destination)
+
+    def download_to_path(self, key: str, destination: Path) -> Path:
+        source = self._resolve(key)
+        if not source.exists():
+            raise StorageError(f"Local object not found for key: {key}")
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copyfile(source, destination)
+        return destination
+
+
+def create_storage_client(settings: WorkerSettings) -> StorageClient:
+    """Instantiate the appropriate storage backend."""
+
+    if settings.s3_bucket:
+        return S3StorageClient(settings)
+    return LocalStorageClient(settings)

stream3r/worker/tasks.py

@@ -0,0 +1,1036 @@
+"""RQ job entrypoints for STream3R worker."""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+import re
+import shutil
+import tempfile
+import traceback
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable, Mapping
+
+import numpy as np
+import requests
+from rq import get_current_job
+
+from stream3r.utils.visual_utils import predictions_to_glb
+
+from .pipeline import InferenceResult, run_stream3r_inference
+from .runtime import WorkerRuntime, get_runtime
+
+
+logger = logging.getLogger(__name__)
+
+IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".bmp", ".webp"}
+_SAFE_CHARS = re.compile(r"[^0-9A-Za-z_-]")
+
+
+def _as_bool(value: Any, default: bool) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        lowered = value.strip().lower()
+        if lowered in {"1", "true", "yes", "y", "on"}:
+            return True
+        if lowered in {"0", "false", "no", "n", "off"}:
+            return False
+    return default
+
+
+def _as_int(value: Any, default: int) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+@dataclass(slots=True)
+class FrameRecord:
+    index: int
+    frame_id: str
+    path: Path
+    source: str | None = None
+    timestamp: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+class ProgressTracker:
+    """Aggregates frame progress to percentage updates."""
+
+    def __init__(self, runtime: WorkerRuntime, job_meta: Mapping[str, str | None]):
+        self.runtime = runtime
+        self.job_meta = job_meta
+        self.last_value = -1
+
+    def __call__(self, processed: int, total: int) -> None:
+        if total <= 0:
+            return
+        percent = int(round((processed / total) * 100))
+        percent = max(0, min(100, percent))
+        if percent == self.last_value:
+            return
+        self.last_value = percent
+        payload = {
+            **self.job_meta,
+            "status": "progress",
+            "progress": percent,
+            "ts": datetime.now(timezone.utc).timestamp(),
+        }
+        runtime_emit(self.runtime, payload)
+
+
+def runtime_emit(runtime: WorkerRuntime, payload: Mapping[str, Any]) -> None:
+    runtime.emit_event(payload)
+
+
+def _slugify(value: str, fallback: str) -> str:
+    candidate = _SAFE_CHARS.sub("_", value).strip("_")
+    if not candidate:
+        candidate = fallback
+    return candidate[:128]
+
+
+def _is_url(value: str) -> bool:
+    return value.startswith("http://") or value.startswith("https://")
+
+
+def _download_to_path(url: str, destination: Path) -> None:
+    response = requests.get(url, stream=True, timeout=60)
+    response.raise_for_status()
+    with destination.open("wb") as handle:
+        for chunk in response.iter_content(chunk_size=1 << 16):
+            if chunk:
+                handle.write(chunk)
+
+
+def _write_base64(content: str, destination: Path) -> None:
+    data = base64.b64decode(content)
+    destination.write_bytes(data)
+
+
+def _resolve_frame_entry(entry: Any, *, index: int, dest_dir: Path) -> FrameRecord:
+    metadata: dict[str, Any] = {}
+    timestamp = None
+    source = None
+    dest_dir.mkdir(parents=True, exist_ok=True)
+
+    if isinstance(entry, str):
+        if _is_url(entry):
+            source = entry
+            frame_id = _slugify(Path(entry).stem or f"frame_{index:06d}", f"frame_{index:06d}")
+            destination = dest_dir / f"{frame_id}.jpg"
+            _download_to_path(entry, destination)
+        else:
+            path = Path(entry)
+            if not path.exists():
+                raise FileNotFoundError(f"Frame path does not exist: {entry}")
+            frame_id = _slugify(path.stem, f"frame_{index:06d}")
+            destination = dest_dir / path.name
+            shutil.copy2(path, destination)
+    elif isinstance(entry, Mapping):
+        frame_id = _slugify(str(entry.get("frame_id") or entry.get("id") or f"frame_{index:06d}"), f"frame_{index:06d}")
+        timestamp = entry.get("timestamp")
+        metadata = {k: v for k, v in entry.items() if k not in {"path", "url", "content", "frame_id", "id", "timestamp"}}
+
+        if path := entry.get("path") or entry.get("local_path"):
+            path = Path(path)
+            if not path.exists():
+                raise FileNotFoundError(f"Frame path does not exist: {path}")
+            destination = dest_dir / (path.name if path.suffix else f"{frame_id}.jpg")
+            shutil.copy2(path, destination)
+        elif url := entry.get("url"):
+            source = url
+            suffix = Path(url).suffix or ".jpg"
+            destination = dest_dir / f"{frame_id}{suffix}"
+            _download_to_path(url, destination)
+        elif content := entry.get("content"):
+            destination = dest_dir / f"{frame_id}.jpg"
+            _write_base64(content, destination)
+        else:
+            raise ValueError("Frame entry must include 'path', 'url', or 'content'")
+    else:
+        raise TypeError("Unsupported frame specification")
+
+    if destination.suffix.lower() not in IMAGE_EXTENSIONS:
+        destination = destination.with_suffix(".png")
+
+    return FrameRecord(
+        index=index,
+        frame_id=_slugify(destination.stem, f"frame_{index:06d}"),
+        path=destination,
+        source=source,
+        timestamp=timestamp,
+        metadata=metadata,
+    )
+
+
+def _collect_frames(
+    runtime: WorkerRuntime,
+    scene_id: str,
+    payload: Mapping[str, Any],
+    tmp_dir: Path,
+) -> list[FrameRecord]:
+    frames_dir = tmp_dir / "frames"
+    frames_payload = payload.get("frames") or []
+    frame_limit = runtime.settings.max_frames_per_job
+
+    records: list[FrameRecord] = []
+    if frames_payload:
+        for entry in frames_payload:
+            if frame_limit and frame_limit > 0 and len(records) >= frame_limit:
+                break
+            records.append(_resolve_frame_entry(entry, index=len(records), dest_dir=frames_dir))
+    else:
+        directory = payload.get("frames_dir") or payload.get("images_dir")
+        if directory:
+            directory_path = Path(directory)
+            if not directory_path.is_dir():
+                raise ValueError(f"frames_dir does not exist: {directory}")
+            for idx, file in enumerate(sorted(directory_path.iterdir())):
+                if file.suffix.lower() not in IMAGE_EXTENSIONS:
+                    continue
+                if frame_limit and frame_limit > 0 and len(records) >= frame_limit:
+                    break
+                destination = frames_dir / file.name
+                shutil.copy2(file, destination)
+                records.append(
+                    FrameRecord(
+                        index=len(records),
+                        frame_id=_slugify(file.stem, f"frame_{idx:06d}"),
+                        path=destination,
+                    )
+                )
+
+    if not records:
+        records = _collect_frames_from_scene_media(runtime, scene_id, frames_dir)
+
+    if not records:
+        raise ValueError(f"No valid frames found for scene '{scene_id}'")
+
+    limit = runtime.settings.max_frames_per_job
+    if limit and limit > 0 and len(records) > limit:
+        records = records[:limit]
+
+    for new_idx, record in enumerate(records):
+        if record.index != new_idx:
+            record.index = new_idx
+
+    return records
+
+
+def _sanitize_payload(payload: Mapping[str, Any]) -> dict[str, Any]:
+    result = dict(payload)
+    frames = result.pop("frames", None)
+    if frames is not None:
+        result["frame_count"] = len(frames)
+    if "frames_dir" in result:
+        result["frames_dir"] = str(result["frames_dir"])
+    return result
+
+
+def _prepare_session_settings(
+    payload: Mapping[str, Any],
+    *,
+    mode: str,
+    streaming: bool,
+    frame_records: list[FrameRecord],
+    window_size: int | None = None,
+) -> dict[str, Any]:
+    base_settings = payload.get("session_settings") or {}
+    session_settings = dict(base_settings)
+    session_settings.update(
+        {
+            "mode": mode,
+            "streaming": streaming,
+            "frame_count": len(frame_records),
+        }
+    )
+    window_setting = window_size if window_size is not None else payload.get("window_size")
+    if window_setting:
+        try:
+            session_settings["window_size"] = int(window_setting)
+        except (TypeError, ValueError):
+            pass
+    return session_settings
+
+
+def _collect_frames_from_scene_media(
+    runtime: WorkerRuntime,
+    scene_id: str,
+    dest_dir: Path,
+) -> list[FrameRecord]:
+    base_url = runtime.settings.scene_media_api_base_url
+    if not base_url:
+        raise ValueError(
+            "Scene media API base URL is not configured. Set API_BASE_URL"
+        )
+
+    base_url = base_url.rstrip("/")
+    dest_dir.mkdir(parents=True, exist_ok=True)
+
+    per_page = runtime.settings.scene_media_page_size
+    if per_page <= 0:
+        per_page = 100
+    per_page = max(1, min(per_page, 1000))
+    frame_limit = runtime.settings.max_frames_per_job
+
+    headers = {}
+    token = runtime.settings.scene_media_api_token
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
+    url = f"{base_url}/scenes/{scene_id}/media"
+    session = requests.Session()
+    records: list[FrameRecord] = []
+    offset = 0
+
+    while True:
+        if frame_limit and frame_limit > 0 and len(records) >= frame_limit:
+            break
+
+        request_limit = per_page
+        if frame_limit and frame_limit > 0:
+            remaining = frame_limit - len(records)
+            if remaining <= 0:
+                break
+            request_limit = min(request_limit, remaining)
+
+        params = {
+            "limit": request_limit,
+            "offset": offset,
+            "media_type": "image",
+        }
+        try:
+            response = session.get(url, params=params, headers=headers, timeout=30)
+            response.raise_for_status()
+        except requests.RequestException as exc:
+            raise RuntimeError(f"Failed to fetch media for scene '{scene_id}': {exc}") from exc
+
+        data = response.json()
+        items = data.get("items") or []
+        if not items:
+            break
+
+        for item in items:
+            if frame_limit and frame_limit > 0 and len(records) >= frame_limit:
+                break
+            file_key = item.get("file")
+            if not file_key:
+                continue
+
+            idx = len(records)
+            source_path = Path(str(file_key))
+            suffix = source_path.suffix if source_path.suffix else ".png"
+            frame_id = _slugify(source_path.stem or f"frame_{idx:06d}", f"frame_{idx:06d}")
+            destination = dest_dir / f"{frame_id}{suffix}"
+
+            try:
+                runtime.storage.download_to_path(str(file_key), destination)
+            except Exception as exc:  # pragma: no cover - download depends on external storage
+                raise RuntimeError(f"Failed to download media '{file_key}' for scene '{scene_id}': {exc}") from exc
+
+            records.append(
+                FrameRecord(
+                    index=idx,
+                    frame_id=frame_id,
+                    path=destination,
+                    source=str(file_key),
+                    timestamp=item.get("captured_at"),
+                    metadata={
+                        "media_id": item.get("id"),
+                        "media_type": item.get("media_type"),
+                    },
+                )
+            )
+
+        if len(items) < request_limit:
+            break
+        offset += request_limit
+
+    return records
+
+
+def _pose_confidence(predictions: Mapping[str, np.ndarray]) -> np.ndarray | None:
+    if "world_points_conf" in predictions:
+        return np.asarray(predictions["world_points_conf"], dtype=np.float32)
+    if "depth_conf" in predictions:
+        return np.asarray(predictions["depth_conf"], dtype=np.float32)
+    return None
+
+
+def _save_pointmaps(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    predictions: Mapping[str, np.ndarray],
+    frame_records: list[FrameRecord],
+    temp_dir: Path,
+) -> dict[str, Any]:
+    world_points = predictions.get("world_points")
+    if world_points is None:
+        world_points = predictions.get("world_points_from_depth")
+    if world_points is None:
+        raise RuntimeError("Predictions missing world points")
+
+    world_points = np.asarray(world_points)
+    confidence = _pose_confidence(predictions)
+    if confidence is None:
+        confidence = np.ones(world_points.shape[:-1], dtype=np.float32)
+
+    local_dir = temp_dir / "pointmaps"
+    local_dir.mkdir(parents=True, exist_ok=True)
+
+    entries: list[dict[str, Any]] = []
+    for record in frame_records:
+        idx = record.index
+        filename = f"{record.frame_id}.npz"
+        local_file = local_dir / filename
+        np.savez(
+            local_file,
+            xyz=np.asarray(world_points[idx], dtype=np.float32),
+            confidence=np.asarray(confidence[idx], dtype=np.float32),
+        )
+        key = runtime.storage.build_key(scene_id, runtime.settings.pointmap_dir, filename)
+        uri = runtime.storage.upload_file(local_file, key, content_type="application/octet-stream")
+        entries.append(
+            {
+                "frame_id": record.frame_id,
+                "frame_index": record.index,
+                "url": uri,
+                "timestamp": record.timestamp,
+            }
+        )
+
+    directory_uri = runtime.storage.build_uri(
+        runtime.storage.build_key(scene_id, runtime.settings.pointmap_dir)
+    )
+
+    return {
+        "pointmaps": entries,
+        "pointmap_dir": directory_uri,
+    }
+
+
+def _write_poses_jsonl(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    job_id: str,
+    predictions: Mapping[str, np.ndarray],
+    frame_records: list[FrameRecord],
+    temp_dir: Path,
+) -> str:
+    extrinsic = np.asarray(predictions.get("extrinsic"))
+    intrinsic = predictions.get("intrinsic")
+    if intrinsic is not None:
+        intrinsic = np.asarray(intrinsic)
+
+    local_file = temp_dir / "poses.jsonl"
+    with local_file.open("w", encoding="utf-8") as handle:
+        for record in frame_records:
+            idx = record.index
+            payload = {
+                "job_id": job_id,
+                "scene_id": scene_id,
+                "frame_id": record.frame_id,
+                "frame_index": record.index,
+                "extrinsic": extrinsic[idx].tolist(),
+            }
+            if intrinsic is not None:
+                payload["intrinsic"] = intrinsic[idx].tolist()
+            if record.timestamp is not None:
+                payload["timestamp"] = record.timestamp
+            if record.source is not None:
+                payload["source"] = record.source
+            if record.metadata:
+                payload["metadata"] = record.metadata
+            handle.write(json.dumps(payload))
+            handle.write("\n")
+
+    key = runtime.storage.build_key(
+        scene_id,
+        runtime.settings.models_dir,
+        runtime.settings.poses_filename,
+    )
+    return runtime.storage.upload_file(local_file, key, content_type="application/json")
+
+
+def _upload_cache(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    cache_path: Path | None,
+) -> str | None:
+    if cache_path is None or not cache_path.exists():
+        return None
+    key = runtime.storage.build_key(
+        scene_id,
+        runtime.settings.models_dir,
+        runtime.settings.session_cache_filename,
+    )
+    return runtime.storage.upload_file(cache_path, key, content_type="application/octet-stream")
+
+
+def _write_predictions_npz(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    predictions: Mapping[str, np.ndarray],
+    temp_dir: Path,
+) -> str:
+    payload = {k: v for k, v in predictions.items() if isinstance(v, np.ndarray)}
+    local_file = temp_dir / runtime.settings.predictions_filename
+    np.savez(local_file, **payload)
+    key = runtime.storage.build_key(
+        scene_id,
+        runtime.settings.models_dir,
+        runtime.settings.predictions_filename,
+    )
+    return runtime.storage.upload_file(local_file, key, content_type="application/octet-stream")
+
+
+def _write_session_settings(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    session_settings: Mapping[str, Any],
+    temp_dir: Path,
+) -> str:
+    local_file = temp_dir / runtime.settings.session_settings_filename
+    local_file.write_text(json.dumps(session_settings, indent=2), encoding="utf-8")
+    key = runtime.storage.build_key(
+        scene_id,
+        runtime.settings.models_dir,
+        runtime.settings.session_settings_filename,
+    )
+    return runtime.storage.upload_file(local_file, key, content_type="application/json")
+
+
+def _write_selected_frames(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    selected_frames: list[dict[str, Any]],
+    top_k: int,
+    temp_dir: Path,
+) -> str | None:
+    if not selected_frames:
+        return None
+    local_file = temp_dir / runtime.settings.selected_frames_filename
+    payload = {"top_k": top_k, "frames": selected_frames}
+    local_file.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+    key = runtime.storage.build_key(
+        scene_id,
+        runtime.settings.models_dir,
+        runtime.settings.selected_frames_filename,
+    )
+    return runtime.storage.upload_file(local_file, key, content_type="application/json")
+
+
+def _compute_selected_frames(
+    predictions: Mapping[str, np.ndarray],
+    frame_records: list[FrameRecord],
+    top_k: int,
+) -> list[dict[str, Any]]:
+    if top_k <= 0:
+        return []
+    confidence = _pose_confidence(predictions)
+    if confidence is None:
+        return []
+    scores = confidence.reshape(confidence.shape[0], -1).mean(axis=1)
+    indices = np.argsort(scores)[::-1][:top_k]
+    result = []
+    for idx in indices:
+        record = frame_records[int(idx)]
+        result.append(
+            {
+                "frame_id": record.frame_id,
+                "frame_index": record.index,
+                "score": float(scores[idx]),
+            }
+        )
+    return result
+
+
+def _save_scene_glb(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    predictions: Mapping[str, np.ndarray],
+    temp_dir: Path,
+    payload: Mapping[str, Any],
+) -> str:
+    local_file = temp_dir / runtime.settings.scene_glb_filename
+    scene = predictions_to_glb(
+        dict(predictions),
+        conf_thres=float(payload.get("conf_thres", 3.0)),
+        filter_by_frames=payload.get("frame_filter", "All"),
+        mask_black_bg=_as_bool(payload.get("mask_black_bg"), False),
+        mask_white_bg=_as_bool(payload.get("mask_white_bg"), False),
+        show_cam=_as_bool(payload.get("show_cam"), True),
+        mask_sky=_as_bool(payload.get("mask_sky"), False),
+        target_dir=str(temp_dir),
+        prediction_mode=payload.get("prediction_mode", "Predicted Pointmap"),
+    )
+    scene.export(file_obj=str(local_file))
+    key = runtime.storage.build_key(
+        scene_id,
+        runtime.settings.models_dir,
+        runtime.settings.scene_glb_filename,
+    )
+    return runtime.storage.upload_file(local_file, key, content_type="model/gltf-binary")
+
+
+def _write_summary_json(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    summary: Mapping[str, Any],
+    temp_dir: Path,
+) -> str:
+    filename = runtime.settings.result_filename
+    local_file = temp_dir / filename
+    local_file.write_text(json.dumps(summary, indent=2), encoding="utf-8")
+    key = runtime.storage.build_key(
+        scene_id,
+        runtime.settings.models_dir,
+        filename,
+    )
+    return runtime.storage.upload_file(local_file, key, content_type="application/json")
+
+
+def _upload_result_record(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    job_id: str,
+    payload: Mapping[str, Any],
+) -> str:
+    local = json.dumps(payload, indent=2).encode("utf-8")
+    key = runtime.storage.build_key(
+        scene_id,
+        runtime.settings.results_dir,
+        f"{job_id}.json",
+    )
+    return runtime.storage.upload_bytes(local, key, content_type="application/json")
+
+
+def _model_dir_uri(runtime: WorkerRuntime, scene_id: str) -> str:
+    return runtime.storage.build_uri(
+        runtime.storage.build_key(scene_id, runtime.settings.models_dir)
+    )
+
+
+def _generate_core_outputs(
+    *,
+    runtime: WorkerRuntime,
+    scene_id: str,
+    job_id: str,
+    predictions: Mapping[str, np.ndarray],
+    frame_records: list[FrameRecord],
+    inference: InferenceResult,
+    session_settings: Mapping[str, Any],
+    temp_dir: Path,
+) -> dict[str, Any]:
+    pointmap_info = _save_pointmaps(
+        runtime=runtime,
+        scene_id=scene_id,
+        predictions=predictions,
+        frame_records=frame_records,
+        temp_dir=temp_dir,
+    )
+
+    poses_url = _write_poses_jsonl(
+        runtime=runtime,
+        scene_id=scene_id,
+        job_id=job_id,
+        predictions=predictions,
+        frame_records=frame_records,
+        temp_dir=temp_dir,
+    )
+
+    cache_url = _upload_cache(
+        runtime=runtime,
+        scene_id=scene_id,
+        cache_path=inference.cache_path,
+    )
+
+    predictions_url = _write_predictions_npz(
+        runtime=runtime,
+        scene_id=scene_id,
+        predictions=predictions,
+        temp_dir=temp_dir,
+    )
+
+    session_settings_url = _write_session_settings(
+        runtime=runtime,
+        scene_id=scene_id,
+        session_settings=session_settings,
+        temp_dir=temp_dir,
+    )
+
+    extrinsic = np.asarray(predictions.get("extrinsic"))
+    intrinsic = predictions.get("intrinsic")
+    if intrinsic is not None:
+        intrinsic = np.asarray(intrinsic)
+
+    frames_payload: list[dict[str, Any]] = []
+    for entry in pointmap_info["pointmaps"]:
+        idx = entry["frame_index"]
+        frame = frame_records[idx]
+        frame_payload = {
+            "frame_id": frame.frame_id,
+            "frame_index": frame.index,
+            "pointmap_url": entry["url"],
+            "extrinsic": extrinsic[idx].tolist(),
+        }
+        if intrinsic is not None:
+            frame_payload["intrinsic"] = intrinsic[idx].tolist()
+        if frame.timestamp is not None:
+            frame_payload["timestamp"] = frame.timestamp
+        if frame.source is not None:
+            frame_payload["source"] = frame.source
+        frames_payload.append(frame_payload)
+
+    artifacts = {
+        "poses_url": poses_url,
+        "pointmap_dir": pointmap_info["pointmap_dir"],
+        "pointmaps": pointmap_info["pointmaps"],
+        "predictions_url": predictions_url,
+        "session_settings_url": session_settings_url,
+    }
+    if cache_url:
+        artifacts["kv_cache_url"] = cache_url
+
+    return {
+        "artifacts": artifacts,
+        "frames": frames_payload,
+    }
+
+
+def _handle_pose_pointmap(
+    *,
+    runtime: WorkerRuntime,
+    payload: Mapping[str, Any],
+    mode: str,
+    streaming: bool,
+    job_id: str,
+    scene_id: str,
+    frame_records: list[FrameRecord],
+    inference: InferenceResult,
+    session_settings: Mapping[str, Any],
+    temp_dir: Path,
+) -> dict[str, Any]:
+    predictions = inference.predictions
+    core = _generate_core_outputs(
+        runtime=runtime,
+        scene_id=scene_id,
+        job_id=job_id,
+        predictions=predictions,
+        frame_records=frame_records,
+        inference=inference,
+        session_settings=session_settings,
+        temp_dir=temp_dir,
+    )
+
+    result_payload = {
+        "job_id": job_id,
+        "job_type": "pose_pointmap",
+        "scene_id": scene_id,
+        "mode": mode,
+        "streaming": streaming,
+        "frame_count": inference.total_frames,
+        "created_at": datetime.now(timezone.utc).isoformat(),
+        "artifacts": core["artifacts"],
+        "frames": core["frames"],
+    }
+
+    result_url = _upload_result_record(
+        runtime=runtime,
+        scene_id=scene_id,
+        job_id=job_id,
+        payload=result_payload,
+    )
+    result_payload["result_url"] = result_url
+    result_payload["model_dir"] = _model_dir_uri(runtime, scene_id)
+
+    return result_payload
+
+
+JobHandler = Callable[..., dict[str, Any]]
+
+
+def _execute_job(job_type: str, payload: Mapping[str, Any], handler: JobHandler) -> dict[str, Any]:
+    runtime = get_runtime()
+    job = get_current_job()
+    payload = dict(payload)
+
+    job_id = str(payload.get("job_id") or (job.id if job else uuid.uuid4()))
+    scene_id = payload.get("scene_id")
+    if not scene_id:
+        raise ValueError("Job payload is missing 'scene_id'")
+
+    payload.setdefault("job_type", job_type)
+    payload.setdefault("scene_id", scene_id)
+
+    mode = payload.get("mode") or runtime.settings.default_mode
+    streaming = _as_bool(payload.get("streaming"), runtime.settings.default_streaming)
+    window_size: int | None = None
+
+    if mode == "window":
+        streaming = True
+        payload["streaming"] = True
+        window_candidate = payload.get("window_size") or runtime.settings.stream_window_size
+        try:
+            window_size = int(window_candidate) if window_candidate else None
+        except (TypeError, ValueError):
+            window_size = runtime.settings.stream_window_size or None
+        if window_size and window_size > 0:
+            payload["window_size"] = window_size
+        else:
+            window_size = None
+
+    payload["mode"] = mode
+    timeout_override = payload.get("timeout")
+    if timeout_override is not None:
+        try:
+            job.timeout = int(timeout_override)
+        except (TypeError, ValueError):
+            pass
+
+    # Default to 15 minutes if no timeout already applied
+    if job.timeout is None:
+        job.timeout = 15 * 60
+
+    sanitized_payload = _sanitize_payload(payload)
+
+    job_meta = {
+        "job_id": job_id,
+        "job_type": job_type,
+        "scene_id": scene_id,
+    }
+
+    runtime.db.upsert_job(
+        job_id=job_id,
+        job_type=job_type,
+        scene_id=scene_id,
+        status="started",
+        payload=sanitized_payload,
+    )
+
+    runtime_emit(
+        runtime,
+        {
+            **job_meta,
+            "status": "started",
+            "progress": 0,
+            "ts": datetime.now(timezone.utc).timestamp(),
+        },
+    )
+
+    try:
+        with runtime.gpu_lock():
+            with tempfile.TemporaryDirectory(prefix=f"stream3r_{job_id}_") as tmp_dir:
+                temp_path = Path(tmp_dir)
+                frame_records = _collect_frames(runtime, scene_id, payload, temp_path)
+                cache_path = temp_path / runtime.settings.session_cache_filename if streaming else None
+
+                tracker = ProgressTracker(runtime, job_meta)
+                inference = run_stream3r_inference(
+                    runtime=runtime,
+                    image_paths=[record.path for record in frame_records],
+                    mode=mode,
+                    streaming=streaming,
+                    cache_output_path=cache_path,
+                    progress_cb=tracker,
+                    window_size=window_size if streaming and mode == "window" else None,
+                )
+
+                session_settings = _prepare_session_settings(
+                    payload,
+                    mode=mode,
+                    streaming=streaming,
+                    frame_records=frame_records,
+                    window_size=window_size,
+                )
+
+                result_payload = handler(
+                    runtime=runtime,
+                    payload=payload,
+                    mode=mode,
+                    streaming=streaming,
+                    job_id=job_id,
+                    scene_id=scene_id,
+                    frame_records=frame_records,
+                    inference=inference,
+                    session_settings=session_settings,
+                    temp_dir=temp_path,
+                )
+
+    except Exception as exc:
+        error_text = traceback.format_exc()
+        runtime.db.upsert_job(
+            job_id=job_id,
+            job_type=job_type,
+            scene_id=scene_id,
+            status="failed",
+            error=error_text,
+        )
+        runtime_emit(
+            runtime,
+            {
+                **job_meta,
+                "status": "failed",
+                "ts": datetime.now(timezone.utc).timestamp(),
+                "error": str(exc),
+            },
+        )
+        logger.exception("Job %s failed", job_id)
+        raise
+
+    runtime.db.upsert_job(
+        job_id=job_id,
+        job_type=job_type,
+        scene_id=scene_id,
+        status="finished",
+        result=result_payload,
+    )
+
+    runtime_emit(
+        runtime,
+        {
+            **job_meta,
+            "status": "finished",
+            "progress": 100,
+            "result_url": result_payload.get("result_url"),
+            "model_dir": result_payload.get("model_dir"),
+            "ts": datetime.now(timezone.utc).timestamp(),
+        },
+    )
+
+    return result_payload
+
+
+def pose_pointmap_job(payload: Mapping[str, Any]) -> dict[str, Any]:
+    """Process a pose + pointmap job."""
+
+    return _execute_job("pose_pointmap", payload, _handle_pose_pointmap)
+
+
+def model_build_job(payload: Mapping[str, Any]) -> dict[str, Any]:
+    """Process a full model build job."""
+
+    return _execute_job("model_build", payload, _handle_model_build)
+
+
+def _handle_model_build(
+    *,
+    runtime: WorkerRuntime,
+    payload: Mapping[str, Any],
+    mode: str,
+    streaming: bool,
+    job_id: str,
+    scene_id: str,
+    frame_records: list[FrameRecord],
+    inference: InferenceResult,
+    session_settings: Mapping[str, Any],
+    temp_dir: Path,
+) -> dict[str, Any]:
+    predictions = inference.predictions
+
+    core = _generate_core_outputs(
+        runtime=runtime,
+        scene_id=scene_id,
+        job_id=job_id,
+        predictions=predictions,
+        frame_records=frame_records,
+        inference=inference,
+        session_settings=session_settings,
+        temp_dir=temp_dir,
+    )
+
+    artifacts = dict(core["artifacts"])
+
+    top_k = _as_int(payload.get("top_k_frames") or payload.get("top_k"), 0)
+    selected_frames = _compute_selected_frames(predictions, frame_records, top_k)
+    selected_frames_url = _write_selected_frames(
+        runtime=runtime,
+        scene_id=scene_id,
+        selected_frames=selected_frames,
+        top_k=top_k,
+        temp_dir=temp_dir,
+    )
+    if selected_frames_url:
+        artifacts["selected_frames_url"] = selected_frames_url
+
+    scene_glb_url = _save_scene_glb(
+        runtime=runtime,
+        scene_id=scene_id,
+        predictions=predictions,
+        temp_dir=temp_dir,
+        payload=payload,
+    )
+    artifacts["scene_glb_url"] = scene_glb_url
+
+    summary_payload = {
+        "job_id": job_id,
+        "job_type": "model_build",
+        "scene_id": scene_id,
+        "frame_count": inference.total_frames,
+        "created_at": datetime.now(timezone.utc).isoformat(),
+        "artifacts": artifacts,
+        "selected_frames": selected_frames,
+        "parameters": {
+            "mode": mode,
+            "streaming": streaming,
+            "conf_thres": float(payload.get("conf_thres", 3.0)),
+            "frame_filter": payload.get("frame_filter", "All"),
+            "mask_black_bg": _as_bool(payload.get("mask_black_bg"), False),
+            "mask_white_bg": _as_bool(payload.get("mask_white_bg"), False),
+            "show_cam": _as_bool(payload.get("show_cam"), True),
+            "mask_sky": _as_bool(payload.get("mask_sky"), False),
+            "prediction_mode": payload.get("prediction_mode", "Predicted Pointmap"),
+        },
+    }
+
+    summary_url = _write_summary_json(
+        runtime=runtime,
+        scene_id=scene_id,
+        summary=summary_payload,
+        temp_dir=temp_dir,
+    )
+    artifacts["summary_url"] = summary_url
+
+    result_record = dict(summary_payload)
+    result_record["result_url"] = summary_url
+    result_record_url = _upload_result_record(
+        runtime=runtime,
+        scene_id=scene_id,
+        job_id=job_id,
+        payload=result_record,
+    )
+
+    result_payload = {
+        "job_id": job_id,
+        "job_type": "model_build",
+        "scene_id": scene_id,
+        "mode": mode,
+        "streaming": streaming,
+        "frame_count": inference.total_frames,
+        "created_at": summary_payload["created_at"],
+        "artifacts": artifacts,
+        "frames": core["frames"],
+        "selected_frames": selected_frames,
+        "summary_url": summary_url,
+        "result_url": summary_url,
+        "result_record_url": result_record_url,
+        "model_dir": _model_dir_uri(runtime, scene_id),
+    }
+
+    return result_payload

worker/__init__.py

@@ -0,0 +1 @@
+"""Compatibility shim package for legacy job import paths."""

worker/stream3r/__init__.py

@@ -0,0 +1 @@
+"""Compatibility namespace for legacy worker module paths."""

worker/stream3r/jobs.py

@@ -0,0 +1,63 @@
+"""Legacy job dispatch entrypoints for compatibility with existing queues."""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Mapping
+
+from stream3r.worker.tasks import model_build_job, pose_pointmap_job
+
+
+_HANDLERS: dict[str, Callable[[Mapping[str, Any]], Any]] = {
+    "pose_pointmap": pose_pointmap_job,
+    "model_build": model_build_job,
+}
+
+
+def handle_job(*args: Any, **kwargs: Any) -> Any:
+    """Dispatch jobs enqueued with the legacy `worker.stream3r.jobs.handle_job` path.
+
+    Supports the following invocation patterns:
+
+    - ``handle_job(payload)`` where ``payload`` is a mapping containing ``job_type``.
+    - ``handle_job(job_type, payload)`` matching older enqueue signatures.
+    - ``handle_job(job_type=..., payload=...)`` keyword usage.
+    """
+
+    job_type: str | None = None
+    payload: Mapping[str, Any] | None = None
+
+    if args:
+        if isinstance(args[0], Mapping) and "job_type" in args[0]:
+            payload = args[0]
+            job_type = str(payload.get("job_type"))
+        else:
+            job_type = str(args[0]) if args else None
+            if len(args) > 1 and isinstance(args[1], Mapping):
+                payload = args[1]
+
+    if "job_type" in kwargs and not job_type:
+        job_type = str(kwargs["job_type"])
+    if "payload" in kwargs and payload is None:
+        candidate = kwargs["payload"]
+        if isinstance(candidate, Mapping):
+            payload = candidate
+
+    if payload is None and isinstance(args[0], Mapping):
+        payload = args[0]
+        job_type = str(payload.get("job_type")) if payload else job_type
+
+    if payload is None:
+        raise ValueError("handle_job requires a payload mapping")
+
+    if not job_type:
+        job_type = str(payload.get("job_type", "")).strip()
+
+    if not job_type:
+        raise ValueError("handle_job payload is missing 'job_type'")
+
+    handler = _HANDLERS.get(job_type)
+    if handler is None:
+        raise ValueError(f"Unsupported job_type '{job_type}'")
+
+    return handler(payload)
+