feat(api): accept a caller-supplied bbox to skip YOLO detection

api/README.md

@@ -9,10 +9,14 @@ Import as `temporal_model.api`. Depends on `temporal-model-core`.
 
 - `GET /health` — readiness + loaded model name/version.
 - `POST /predict` — body `{ "frames": ["<s3-key>", ...], "bucket": "<name>",
-  "roi_xyxyn": [x_min, y_min, x_max, y_max] }`
+  "roi_xyxyn": [x_min, y_min, x_max, y_max],
+  "bbox_xyxyn": [x_min, y_min, x_max, y_max], "bbox_confidence": 0.8 }`
   (ordered S3 keys; `bucket` optional, falls back to `S3_BUCKET`;
   `roi_xyxyn` optional normalized region of interest — tubes with no real
-  detection intersecting it are dropped before scoring);
+  detection intersecting it are dropped before scoring;
+  `bbox_xyxyn` optional caller-supplied detection — skips YOLO entirely, the
+  box becomes the only detection on every frame, mutually exclusive with
+  `roi_xyxyn`; `bbox_confidence` optional score stamped on it, default 1.0);
   returns `{ is_smoke, probability, model }` (`probability` = max kept-tube
   calibrated probability, `null` if uncalibrated).
   `POST /predict?verbose=true` adds a `details` block (decision, preprocessing,

api/src/temporal_model/api/app.py

@@ -144,7 +144,12 @@ async def predict(
                 )
 
             out = await runner.predict(
-                paths, roi=body.roi_xyxyn, timer=timer, profile=profile
+                paths,
+                roi=body.roi_xyxyn,
+                bbox=body.bbox_xyxyn,
+                bbox_confidence=body.bbox_confidence,
+                timer=timer,
+                profile=profile,
             )
 
             profiling = None

api/src/temporal_model/api/model_runner.py

@@ -122,6 +122,8 @@ async def predict(
         frame_paths: list[Path],
         *,
         roi: tuple[float, float, float, float] | None = None,
+        bbox: tuple[float, float, float, float] | None = None,
+        bbox_confidence: float = 1.0,
         timer: StageTimer | None = None,
         profile: dict[str, Any] | None = None,
     ) -> Any:
@@ -131,22 +133,53 @@ async def predict(
         cache is accessed by one prediction at a time. When ``timer``/``profile``
         are supplied, the ``detector`` stage is timed and cache counts recorded.
         ``roi`` is passed through to the core model untouched — the cache stays
-        full-frame (see the invariant in the ROI spec).
+        full-frame (see the invariant in the ROI spec). When ``bbox`` is set,
+        the detector and its cache are bypassed entirely: the box (stamped with
+        ``bbox_confidence``) is the only detection on every frame.
         """
         async with self._lock:
             return await run_in_threadpool(
-                self._predict_sync, frame_paths, roi, timer, profile
+                self._predict_sync,
+                frame_paths,
+                roi,
+                bbox,
+                bbox_confidence,
+                timer,
+                profile,
             )
 
     def _predict_sync(
         self,
         frame_paths: list[Path],
         roi: tuple[float, float, float, float] | None = None,
+        bbox: tuple[float, float, float, float] | None = None,
+        bbox_confidence: float = 1.0,
         timer: StageTimer | None = None,
         profile: dict[str, Any] | None = None,
     ) -> Any:
         started = time.perf_counter()
         frames = self._model.load_sequence(frame_paths)
+        if bbox is not None:
+            # Heavy core import deferred like _load_core_model.
+            from temporal_model.core.inference import (  # noqa: PLC0415
+                make_forced_detections,
+            )
+
+            forced = make_forced_detections(
+                frames, bbox_xyxyn=bbox, confidence=bbox_confidence
+            )
+            out = self._model.predict(
+                frames, frame_detections=forced, roi=roi, timer=timer
+            )
+            if profile is not None:
+                profile["n_frames"] = len(frames)
+                profile["forced_bbox"] = True
+            logger.info(
+                "predict: forced bbox, seq_len=%d, %.0fms",
+                len(frames),
+                (time.perf_counter() - started) * 1000.0,
+            )
+            return out
         resolved: dict[str, Any] = {}
         misses = []
         for f in frames:

api/src/temporal_model/api/schemas.py

@@ -8,7 +8,13 @@
 import re
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    field_validator,
+    model_validator,
+)
 
 from temporal_model.core.tubes import validate_roi
 
@@ -31,6 +37,18 @@ class PredictRequest(BaseModel):
     # detection intersecting it are dropped before scoring (see
     # docs/specs/2026-06-10-api-roi-design.md).
     roi_xyxyn: tuple[float, float, float, float] | None = None
+    # Optional caller-supplied detection box, normalized corners
+    # (x_min, y_min, x_max, y_max) — same xyxyn convention as roi_xyxyn. When
+    # set, the YOLO detector is skipped entirely and this box is taken as the
+    # only detection on every frame, yielding one full-length tube (see
+    # docs/specs/2026-06-11-api-forced-bbox-design.md). Mutually exclusive
+    # with roi_xyxyn: the bbox already pins where the smoke is.
+    bbox_xyxyn: tuple[float, float, float, float] | None = None
+    # Confidence stamped on the forced detections (e.g. the upstream
+    # detector's score). Gates nothing, but feeds the calibrator's
+    # mean-confidence feature, so it shifts the returned probability. Only
+    # meaningful alongside bbox_xyxyn.
+    bbox_confidence: float = Field(default=1.0, gt=0.0, le=1.0)
 
     @field_validator("frames")
     @classmethod
@@ -68,6 +86,31 @@ def _validate_roi(
             raise ValueError(f"roi_xyxyn: {e}") from e
         return v
 
+    @field_validator("bbox_xyxyn")
+    @classmethod
+    def _validate_bbox(
+        cls, v: tuple[float, float, float, float] | None
+    ) -> tuple[float, float, float, float] | None:
+        if v is None:
+            return v
+        # Same geometry rules as the ROI (shared core validator).
+        try:
+            validate_roi(v, name="bbox")
+        except ValueError as e:
+            raise ValueError(f"bbox_xyxyn: {e}") from e
+        return v
+
+    @model_validator(mode="after")
+    def _validate_bbox_combinations(self) -> "PredictRequest":
+        if self.bbox_xyxyn is not None and self.roi_xyxyn is not None:
+            raise ValueError(
+                "bbox_xyxyn and roi_xyxyn are mutually exclusive: the bbox is "
+                "the detection, an ROI filter on top of it is meaningless"
+            )
+        if self.bbox_xyxyn is None and "bbox_confidence" in self.model_fields_set:
+            raise ValueError("bbox_confidence requires bbox_xyxyn")
+        return self
+
 
 class FrameEntry(BaseModel):
     frame_idx: int

api/tests/test_app.py

@@ -66,9 +66,22 @@ def __init__(self, output=None, error=None):
         self._output = output
         self._error = error
         self.roi = None
-
-    async def predict(self, paths, *, roi=None, timer=None, profile=None):
+        self.bbox = None
+        self.bbox_confidence = None
+
+    async def predict(
+        self,
+        paths,
+        *,
+        roi=None,
+        bbox=None,
+        bbox_confidence=1.0,
+        timer=None,
+        profile=None,
+    ):
         self.roi = roi
+        self.bbox = bbox
+        self.bbox_confidence = bbox_confidence
         if self._error:
             raise self._error
         if timer is not None:
@@ -412,3 +425,60 @@ def test_predict_invalid_roi_is_400(client):
     body = r.json()
     assert body["code"] == "invalid_request"
     assert "roi_xyxyn" in body["detail"]
+
+
+def test_predict_passes_bbox_to_runner(client):
+    r = client.post(
+        "/predict",
+        json={
+            "frames": KEYS,
+            "bbox_xyxyn": [0.1, 0.2, 0.3, 0.4],
+            "bbox_confidence": 0.8,
+        },
+    )
+    assert r.status_code == 200
+    assert client.app.state.runner.bbox == (0.1, 0.2, 0.3, 0.4)
+    assert client.app.state.runner.bbox_confidence == 0.8
+
+
+def test_predict_bbox_confidence_defaults_to_one(client):
+    r = client.post(
+        "/predict", json={"frames": KEYS, "bbox_xyxyn": [0.1, 0.2, 0.3, 0.4]}
+    )
+    assert r.status_code == 200
+    assert client.app.state.runner.bbox_confidence == 1.0
+
+
+def test_predict_without_bbox_passes_none(client):
+    r = client.post("/predict", json={"frames": KEYS})
+    assert r.status_code == 200
+    assert client.app.state.runner.bbox is None
+
+
+def test_predict_invalid_bbox_is_400(client):
+    r = client.post(
+        "/predict", json={"frames": KEYS, "bbox_xyxyn": [0.3, 0.2, 0.1, 0.4]}
+    )
+    assert r.status_code == 400
+    body = r.json()
+    assert body["code"] == "invalid_request"
+    assert "bbox_xyxyn" in body["detail"]
+
+
+def test_predict_bbox_with_roi_is_400(client):
+    r = client.post(
+        "/predict",
+        json={
+            "frames": KEYS,
+            "bbox_xyxyn": [0.1, 0.2, 0.3, 0.4],
+            "roi_xyxyn": [0.0, 0.0, 1.0, 1.0],
+        },
+    )
+    assert r.status_code == 400
+    assert "mutually exclusive" in r.json()["detail"]
+
+
+def test_predict_bbox_confidence_without_bbox_is_400(client):
+    r = client.post("/predict", json={"frames": KEYS, "bbox_confidence": 0.9})
+    assert r.status_code == 400
+    assert "bbox_confidence" in r.json()["detail"]

api/tests/test_model_runner.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 from types import SimpleNamespace
 
+import pytest
 import yaml
 
 from temporal_model.api import model_runner as mr
@@ -121,6 +122,7 @@ def __init__(self):
         self.detect_calls: list[list[str]] = []
         self.predict_calls: list[set[str]] = []
         self.roi_calls: list[tuple | None] = []
+        self.frame_detections_calls: list[dict] = []
 
     def load_sequence(self, paths):
         return [
@@ -140,6 +142,7 @@ def detect(self, frames):
     def predict(self, frames, *, frame_detections=None, roi=None, timer=None):
         self.predict_calls.append(set(frame_detections or {}))
         self.roi_calls.append(roi)
+        self.frame_detections_calls.append(frame_detections or {})
         return SimpleNamespace(frame_ids=[f.frame_id for f in frames])
 
 
@@ -191,6 +194,47 @@ def test_predict_cache_disabled_detects_every_frame():
     assert model.detect_calls[1] == ["x_00", "x_01", "x_02"]  # full each call
 
 
+def test_predict_with_bbox_skips_detection():
+    model = _OrchestrationModel()
+    runner = ModelRunner(model, name="m", version="1", calibrated=True)
+    asyncio.run(
+        runner.predict(
+            ["c/x_00.jpg", "c/x_01.jpg"],
+            bbox=(0.1, 0.2, 0.5, 0.8),
+            bbox_confidence=0.7,
+        )
+    )
+
+    assert model.detect_calls == []
+    forced = model.frame_detections_calls[-1]
+    assert set(forced) == {"x_00", "x_01"}
+    det = forced["x_00"].detections[0]
+    assert (det.cx, det.cy, det.w, det.h) == pytest.approx((0.3, 0.5, 0.4, 0.6))
+    assert det.confidence == 0.7
+
+
+def test_predict_with_bbox_bypasses_cache():
+    model = _OrchestrationModel()
+    runner = ModelRunner(
+        model, name="m", version="1", calibrated=True, detection_cache_size=4096
+    )
+    asyncio.run(runner.predict(["c/x_00.jpg"], bbox=(0.1, 0.2, 0.3, 0.4)))
+    asyncio.run(runner.predict(["c/x_00.jpg"]))
+
+    # The forced run wrote nothing to the cache: the plain run re-detects.
+    assert model.detect_calls == [["x_00"]]
+
+
+def test_predict_with_bbox_records_profile_marker():
+    model = _OrchestrationModel()
+    runner = ModelRunner(model, name="m", version="1", calibrated=True)
+    profile: dict = {}
+    asyncio.run(
+        runner.predict(["c/x_00.jpg"], bbox=(0.1, 0.2, 0.3, 0.4), profile=profile)
+    )
+    assert profile == {"n_frames": 1, "forced_bbox": True}
+
+
 class _StubFrame:
     def __init__(self, fid):
         self.frame_id = fid

api/tests/test_schemas.py

@@ -227,6 +227,68 @@ def test_request_rejects_invalid_roi(roi):
         PredictRequest(frames=["a.jpg"], roi_xyxyn=roi)
 
 
+def test_request_bbox_defaults_to_none():
+    req = PredictRequest(frames=["a.jpg"])
+    assert req.bbox_xyxyn is None
+    assert req.bbox_confidence == 1.0
+
+
+def test_request_accepts_valid_bbox():
+    req = PredictRequest(
+        frames=["a.jpg"], bbox_xyxyn=[0.1, 0.2, 0.3, 0.4], bbox_confidence=0.7
+    )
+    assert req.bbox_xyxyn == (0.1, 0.2, 0.3, 0.4)
+    assert req.bbox_confidence == 0.7
+
+
+@pytest.mark.parametrize(
+    "bbox",
+    [
+        [-0.1, 0.2, 0.3, 0.4],  # out of range low
+        [0.1, 0.2, 0.3, 1.4],  # out of range high
+        [0.3, 0.2, 0.1, 0.4],  # x_min >= x_max
+        [0.1, 0.4, 0.3, 0.4],  # y_min >= y_max (zero height)
+        [0.1, 0.2, 0.3],  # too short
+        [0.1, 0.2, 0.3, 0.4, 0.5],  # too long
+        ["a", 0.2, 0.3, 0.4],  # non-numeric
+    ],
+)
+def test_request_rejects_invalid_bbox(bbox):
+    with pytest.raises(ValidationError):
+        PredictRequest(frames=["a.jpg"], bbox_xyxyn=bbox)
+
+
+@pytest.mark.parametrize("confidence", [0.0, -0.5, 1.5])
+def test_request_rejects_out_of_range_bbox_confidence(confidence):
+    with pytest.raises(ValidationError):
+        PredictRequest(
+            frames=["a.jpg"],
+            bbox_xyxyn=[0.1, 0.2, 0.3, 0.4],
+            bbox_confidence=confidence,
+        )
+
+
+def test_request_rejects_bbox_with_roi():
+    with pytest.raises(ValidationError, match="mutually exclusive"):
+        PredictRequest(
+            frames=["a.jpg"],
+            bbox_xyxyn=[0.1, 0.2, 0.3, 0.4],
+            roi_xyxyn=[0.0, 0.0, 1.0, 1.0],
+        )
+
+
+def test_request_rejects_bbox_confidence_without_bbox():
+    with pytest.raises(ValidationError, match="bbox_confidence requires"):
+        PredictRequest(frames=["a.jpg"], bbox_confidence=0.9)
+
+
+def test_request_allows_default_equal_bbox_confidence_only_implicitly():
+    # An explicit bbox_confidence is rejected without bbox_xyxyn even when it
+    # equals the default — explicitness, not the value, marks the client bug.
+    with pytest.raises(ValidationError, match="bbox_confidence requires"):
+        PredictRequest(frames=["a.jpg"], bbox_confidence=1.0)
+
+
 def test_verbose_details_map_num_tubes_outside_roi():
     details = _details([_tube(1, 0.9)])
     details["tubes"]["num_outside_roi"] = 3