diff --git a/Dockerfile b/Dockerfile
index 52ecfa7..549257f 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -64,6 +64,7 @@ COPY --from=builder --chown=app:app /opt/models /app/models
 COPY --from=builder --chown=app:app /build/models/download-manifest.json /app/models/download-manifest.json
 COPY --chown=app:app \
     face_api.py \
+    face_execution.py \
     face_config.py \
     face_models.py \
     face_pipeline.py \
diff --git a/README.md b/README.md
index fd4fa50..6ebc07e 100644
--- a/README.md
+++ b/README.md
@@ -178,6 +178,7 @@ const response = await fetch(`${apiBaseUrl}/api/v1/face/enrollments`, {
 | --- | --- | --- |
 | `FACE_SERVICE_TOKEN` | FastAPI와 Spring의 서버 secret | Spring→FastAPI 서비스 인증. 두 곳에 같은 값을 설정 |
 | `FACE_MATCH_THRESHOLD`, `FACE_MATCH_MARGIN` | FastAPI 환경 변수 | 인식 판정 기준. 실제 측정된 값을 사용 |
+| `FACE_SUCCESS_RECHECK_SECONDS` | FastAPI 환경 변수 | 성공한 얼굴 트랙 재인식 유예 시간. 기본 1초, `0`이면 비활성화 |
 | `FACE_AI_BASE_URL` | Spring 환경 변수 | FastAPI의 내부 주소 |
 | `PORT` | FastAPI 실행 환경 | 수신 포트. 기본값 `8000` |
 | `PUBLIC_ORIGIN` | Spring 환경 변수 | 로그인 완료 후 브라우저가 돌아갈 프론트 주소 |
@@ -191,14 +192,23 @@ const response = await fetch(`${apiBaseUrl}/api/v1/face/enrollments`, {
 ## 인식 지연 진단
 
 인식 세션은 후보 행렬을 최초 PUT에서 준비하고, 기존 3품질 프레임·쿨다운 조건을 통과한 얼굴에만
-임베딩을 계산합니다. 모델과 인식 임계값은 그대로 사용합니다. `face_api` logger를 DEBUG로 설정하면
-단계별 시간과 처리 개수만 기록합니다.
+임베딩을 계산합니다. 성공한 트랙은 최대 1초까지 반복 임베딩을 건너뛰며, 얼굴 이탈·저품질·위치/매칭
+모호성에는 즉시 재검사를 시작합니다. 이 동안에도 전체 후보 벡터와 인식 임계값은 그대로 유지합니다.
+동기 추론은 단일 작업자에서 실행하며 등록 영상은 프레임 단위로 추론 작업에 교차 제출됩니다.
+`face_api` logger를 DEBUG로 설정하면 단계별 시간과 처리 개수만 기록합니다.
 
 ```powershell
 python benchmarks/face_matching.py --students 1 50 200
 ```
 
+To measure the effect of suppressing repeated embeddings for a recently recognized
+track, run `python benchmarks/face_recheck_benchmark.py`. This uses synthetic data;
+the [measurement record](docs/reviews/face-recheck-benchmark-2026-10-13.md) reports
+latency and embedding counts, not real-face accuracy.
+
 이 벤치마크는 합성 데이터의 후보 비교만 측정합니다. 전체 카메라 지연이나 인식 정확도를 대신하지 않습니다.
+검출 비교는 선택된 승인 영상을 `benchmarks/face_detector_comparison.py`로 측정합니다. 합성 도형은
+실행·시간 측정에만 쓰며 얼굴 정확도 근거가 아닙니다.
 Spring의 시작 시각 기준 200ms 제한에는 V17 마이그레이션이 필요합니다.
 [구현·검증 결과와 배포 순서](docs/reviews/face-latency-2026-10-07.md)를 확인하세요.
 
diff --git a/benchmarks/face_detector_comparison.py b/benchmarks/face_detector_comparison.py
new file mode 100644
index 0000000..799a40f
--- /dev/null
+++ b/benchmarks/face_detector_comparison.py
@@ -0,0 +1,466 @@
+"""Experimental MediaPipe versus YuNet 2023mar FP32 comparison (never a runtime switch).
+
+Examples (run from the repository root):
+    .venv/Scripts/python.exe benchmarks/face_detector_comparison.py --download-yunet
+    .venv/Scripts/python.exe benchmarks/face_detector_comparison.py --video .local/approved.mp4
+    .venv/Scripts/python.exe benchmarks/face_detector_comparison.py --video .local/approved.mp4 \
+        --gallery-json .local/gallery.json --threshold 0.6 --margin 0.1
+
+Gallery JSON uses the existing AI session PUT payload: {"model": {...},
+"candidates": [{"student_id": "...", "vectors": [[...]]}]}. Supply the actual
+deployment threshold/margin explicitly. Do not commit or collect biometric inputs
+for this benchmark. It never copies local inputs or writes frames/vectors/results.
+The optional download retrieves only the publicly released detector model.
+
+Without --video, procedural drawings test execution and timing only. They cannot
+measure face accuracy. Even with a gallery, agreement with MediaPipe is not ground
+truth: labelled, approved validation is still needed before enabling a detector.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import platform
+import sys
+from collections.abc import Callable
+from dataclasses import asdict, dataclass
+from functools import cache
+from pathlib import Path
+from time import perf_counter_ns
+from urllib.request import urlopen
+
+import cv2
+import numpy as np
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from benchmarks.face_detection_modes import positive_int, synthetic_frame  # noqa: E402
+from face_models import FaceObservation, InferenceModels, ModelMetadata, _pose_bucket  # noqa: E402
+from face_pipeline import CandidateGallery, Match, normalize_vector, quality_issues  # noqa: E402
+
+YUNET_NAME = "face_detection_yunet_2023mar.onnx"
+YUNET_URL = (
+    "https://media.githubusercontent.com/media/opencv/opencv_zoo/main/"
+    f"models/face_detection_yunet/{YUNET_NAME}"
+)
+# Verified against the official repository's Git LFS pointer (not an INT8 variant).
+YUNET_SHA256 = "8f2383e4dd3cfbb4553ea8718107fc0423210dc964f9f4280604804ed2552fa4"
+YUNET_BYTES = 232589
+LANDMARK_REFERENCE = "https://docs.opencv.org/4.11.0/df/d20/classcv_1_1FaceDetectorYN.html"
+
+
+def ensure_yunet(path: Path, download: bool) -> None:
+    """Verify before OpenCV parses the model; never replace an existing file."""
+    if not path.is_file():
+        if not download:
+            raise ValueError("YuNet model is missing; run with --download-yunet")
+        with urlopen(YUNET_URL, timeout=30) as response:
+            content = response.read(YUNET_BYTES + 1)
+        if len(content) != YUNET_BYTES or hashlib.sha256(content).hexdigest() != YUNET_SHA256:
+            raise ValueError("downloaded YuNet model failed size/SHA-256 verification")
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with path.open("xb") as destination:
+            destination.write(content)
+    if path.stat().st_size != YUNET_BYTES:
+        raise ValueError("YuNet model has an unexpected size")
+    if hashlib.sha256(path.read_bytes()).hexdigest() != YUNET_SHA256:
+        raise ValueError("YuNet model has an unexpected SHA-256")
+
+
+def observation_from_yunet(row: np.ndarray, frame: np.ndarray) -> FaceObservation:
+    """Use YuNet landmarks with the existing 128x128 BGR 0095 alignment.
+
+    OpenCV documents right eye, left eye, nose, right mouth, left mouth: subject
+    right is image left for a frontal face, matching _align_face's target order.
+    YuNet eye locations and MediaPipe eye corners are NOT identical landmarks.
+    Its detector box also differs from the MediaPipe mesh envelope; we measure
+    resulting quality/embedding changes instead of treating them as compatible.
+    """
+    height, width = frame.shape[:2]
+    x, y, box_width, box_height = map(float, row[:4])
+    x0, y0 = max(0.0, x), max(0.0, y)
+    x1, y1 = min(float(width), x + box_width), min(float(height), y + box_height)
+    bbox = (x0 / width, y0 / height, max(0.0, x1 - x0) / width, max(0.0, y1 - y0) / height)
+    five = np.asarray(row[4:14], dtype=np.float32).reshape(5, 2)
+    crop = frame[int(y0):int(max(y0, y1)), int(x0):int(max(x0, x1))]
+    gray = None
+    try:
+        gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY) if crop.size else None
+        return FaceObservation(
+            bbox,
+            tuple((float(x), float(y)) for x, y in five),
+            None,
+            float(np.mean(gray)) if gray is not None else 0.0,
+            float(cv2.Laplacian(crop, cv2.CV_64F).var()) if crop.size else 0.0,
+            _pose_bucket(five),
+        )
+    finally:
+        if gray is not None:
+            gray.fill(0)
+
+
+class YuNetDetector:
+    def __init__(self, model_path: Path):
+        self.detector = cv2.FaceDetectorYN.create(
+            str(model_path), "", (640, 480), 0.9, 0.3, 5000,
+            cv2.dnn.DNN_BACKEND_OPENCV, cv2.dnn.DNN_TARGET_CPU,
+        )
+
+    def detect_faces(self, frame: np.ndarray) -> list[FaceObservation]:
+        self.detector.setInputSize((frame.shape[1], frame.shape[0]))
+        rows = None
+        try:
+            _, rows = self.detector.detect(frame)
+            if rows is None:
+                return []
+            # The service's MediaPipe configuration also has a maximum of 8 faces.
+            indices = np.argsort(-rows[:, 14], kind="stable")[:8]
+            return [observation_from_yunet(rows[index], frame) for index in indices]
+        finally:
+            if rows is not None:
+                rows.fill(0)
+
+
+def overlap(left: FaceObservation, right: FaceObservation) -> float:
+    x, y, w, h = left.bbox
+    a, b, c, d = right.bbox
+    intersection = max(0.0, min(x + w, a + c) - max(x, a)) * max(
+        0.0, min(y + h, b + d) - max(y, b)
+    )
+    union = w * h + c * d - intersection
+    return intersection / union if union > 0 else 0.0
+
+
+def match_faces(
+    baseline: list[FaceObservation], candidate: list[FaceObservation], minimum_iou: float = 0.3
+) -> tuple[tuple[int, int], ...]:
+    """Maximize paired count, then total IoU; never zip faces by horizontal order.
+
+    At most eight faces make an exact bit-mask assignment small. Each candidate
+    is used once. Missing/non-overlapping faces remain explicitly unmatched.
+    This establishes spatial correspondence, not identity or temporal tracking.
+    """
+    if len(baseline) > 8 or len(candidate) > 8:
+        raise ValueError("comparison supports at most eight faces per detector")
+    weights = [[overlap(left, right) for right in candidate] for left in baseline]
+
+    @cache
+    def assign(index: int, used: int) -> tuple[int, float, tuple[tuple[int, int], ...]]:
+        if index == len(baseline):
+            return 0, 0.0, ()
+        best = assign(index + 1, used)
+        for other, weight in enumerate(weights[index]):
+            if used & (1 << other) or weight < minimum_iou:
+                continue
+            count, total, pairs = assign(index + 1, used | (1 << other))
+            option = count + 1, total + weight, ((index, other), *pairs)
+            if option[:2] > best[:2]:
+                best = option
+        return best
+
+    return assign(0, 0)[2]
+
+
+def clear_json(value: object) -> None:
+    """Release decoded containers promptly; Python immutable scalars cannot be zeroed."""
+    if isinstance(value, dict):
+        for child in value.values():
+            clear_json(child)
+        value.clear()
+    elif isinstance(value, list):
+        for child in value:
+            clear_json(child)
+        value.clear()
+
+
+def load_gallery(path: Path) -> CandidateGallery:
+    """Validate the existing session payload without printing validation input."""
+    raw = bytearray()
+    payload = None
+    normalized: dict[str, list[np.ndarray]] = {}
+    try:
+        # Same 60 MiB body ceiling as the service, without retaining a bytes copy.
+        with path.open("rb") as stream:
+            raw = bytearray(stream.read(60 * 1024 * 1024 + 1))
+        if len(raw) > 60 * 1024 * 1024:
+            raise ValueError("gallery is too large")
+        payload = json.loads(raw)
+        if not isinstance(payload, dict) or payload.get("model") != asdict(ModelMetadata()):
+            raise ValueError("gallery model metadata must match existing 0095 FP32")
+        candidates = payload.get("candidates")
+        if not isinstance(candidates, list) or not 1 <= len(candidates) <= 200:
+            raise ValueError("gallery needs 1 to 200 candidates")
+        for item in candidates:
+            if not isinstance(item, dict):
+                raise ValueError("invalid gallery candidate")
+            student_id, vectors = item.get("student_id"), item.get("vectors")
+            if not isinstance(student_id, str) or not student_id or student_id in normalized:
+                raise ValueError("gallery candidate identifiers must be nonempty and unique")
+            if not isinstance(vectors, list) or not 1 <= len(vectors) <= 20:
+                raise ValueError("gallery candidates need 1 to 20 vectors")
+            normalized[student_id] = []
+            for vector in vectors:
+                normalized[student_id].append(normalize_vector(vector))
+        # Preserve the session PUT's first normalization, then gallery normalization.
+        return CandidateGallery.from_candidates(normalized)
+    finally:
+        raw[:] = b"\x00" * len(raw)
+        clear_json(payload)
+        for vectors in normalized.values():
+            for vector in vectors:
+                vector.fill(0)
+        normalized.clear()
+
+
+@dataclass
+class PipelineResult:
+    observations: list[FaceObservation]
+    vectors: list[np.ndarray | None]
+    decisions: list[Match | None]
+    detection_ns: int
+    total_ns: int
+
+    def clear(self) -> None:
+        for vector in self.vectors:
+            if vector is not None:
+                vector.fill(0)
+        self.vectors.clear()
+        self.observations.clear()
+        self.decisions.clear()
+
+
+def run_pipeline(
+    frame: np.ndarray,
+    detect: Callable[[np.ndarray], list[FaceObservation]],
+    models: InferenceModels,
+    gallery: CandidateGallery | None,
+    threshold: float | None,
+    margin: float | None,
+) -> PipelineResult:
+    result = PipelineResult([], [], [], 0, 0)
+    started = perf_counter_ns()
+    try:
+        result.observations = detect(frame)
+        result.detection_ns = perf_counter_ns() - started
+        for observation in result.observations:
+            if quality_issues(observation):
+                result.vectors.append(None)
+                result.decisions.append(None)
+                continue
+            vector = models.embed_face(frame, observation)
+            result.vectors.append(vector)
+            result.decisions.append(gallery.match(vector, threshold, margin) if gallery else None)
+        result.total_ns = perf_counter_ns() - started
+        return result
+    except BaseException:
+        result.clear()
+        raise
+
+
+def distribution(values: list[float]) -> dict[str, float | int | None]:
+    return {
+        "samples": len(values),
+        "median": float(np.median(values)) if values else None,
+        "p95": float(np.percentile(values, 95)) if values else None,
+    }
+
+
+def synthetic_scene(index: int, face_count: int) -> np.ndarray:
+    if face_count == 1:
+        return synthetic_frame(index)
+    columns, rows = (2, 2) if face_count == 4 else (4, 2)
+    canvas = np.full((480, 640, 3), 224, np.uint8)
+    width, height = 640 // columns, 480 // rows
+    drawing = None
+    resized = None
+    try:
+        for number in range(face_count):
+            drawing = synthetic_frame(index + number * 3)
+            resized = cv2.resize(drawing, (width, height), interpolation=cv2.INTER_AREA)
+            x, y = (number % columns) * width, (number // columns) * height
+            canvas[y:y + height, x:x + width] = resized
+            drawing.fill(0)
+            resized.fill(0)
+        return canvas
+    finally:
+        if drawing is not None:
+            drawing.fill(0)
+        if resized is not None:
+            resized.fill(0)
+
+
+def compare_stream(args, models, detector, gallery, synthetic_faces: int | None) -> dict:
+    capture = None
+    if args.video is not None:
+        if not args.video.is_file():
+            raise ValueError("benchmark video must be an existing local file")
+        capture = cv2.VideoCapture(str(args.video))
+        if not capture.isOpened():
+            capture.release()
+            raise ValueError("could not open benchmark video")
+    stage_times = {name: {"detect_geometry_quality_ms": [], "pipeline_ms": []} for name in ("mediapipe", "yunet")}
+    totals = {name: {"detected": 0, "embedded": 0} for name in stage_times}
+    paired = mismatch_frames = unmatched_baseline = unmatched_candidate = quality_disagreements = 0
+    decision_pairs = decision_agreements = different_known = 0
+    landmark_drifts: list[float] = []
+    cosines: list[float] = []
+    try:
+        for index in range(args.max_frames):
+            if capture is None:
+                frame = synthetic_scene(index, synthetic_faces)
+            else:
+                success, frame = capture.read()
+                if not success:
+                    break
+            results: dict[str, PipelineResult] = {}
+            try:
+                detectors = [("mediapipe", models.detect_faces), ("yunet", detector.detect_faces)]
+                for name, detect in detectors[::1 if index % 2 == 0 else -1]:
+                    results[name] = run_pipeline(frame, detect, models, gallery, args.threshold, args.margin)
+                if index < args.warmup:
+                    continue
+                for name, result in results.items():
+                    stage_times[name]["detect_geometry_quality_ms"].append(result.detection_ns / 1e6)
+                    stage_times[name]["pipeline_ms"].append(result.total_ns / 1e6)
+                    totals[name]["detected"] += len(result.observations)
+                    totals[name]["embedded"] += sum(vector is not None for vector in result.vectors)
+                baseline, candidate = results["mediapipe"], results["yunet"]
+                pairs = match_faces(baseline.observations, candidate.observations)
+                paired += len(pairs)
+                mismatch_frames += len(baseline.observations) != len(candidate.observations)
+                unmatched_baseline += len(baseline.observations) - len(pairs)
+                unmatched_candidate += len(candidate.observations) - len(pairs)
+                for left, right in pairs:
+                    first, second = baseline.observations[left], candidate.observations[right]
+                    delta = np.asarray(first.landmarks) - np.asarray(second.landmarks)
+                    # Aggregate RMS in pixels; landmark locations themselves are never output.
+                    landmark_drifts.append(float(np.sqrt(np.mean(np.sum(delta * delta, axis=1)))))
+                    delta.fill(0)
+                    one, two = baseline.vectors[left], candidate.vectors[right]
+                    quality_disagreements += (one is None) != (two is None)
+                    if one is not None and two is not None:
+                        cosines.append(float(np.clip(one @ two, -1.0, 1.0)))
+                    if gallery is not None:
+                        a, b = baseline.decisions[left], candidate.decisions[right]
+                        decision_pairs += 1
+                        decision_agreements += (a.status, a.student_id) == (b.status, b.student_id) if a and b else a is b
+                        different_known += bool(a and b and a.status == b.status == "KNOWN" and a.student_id != b.student_id)
+            finally:
+                for result in results.values():
+                    result.clear()
+                frame.fill(0)
+    finally:
+        if capture is not None:
+            capture.release()
+    measured = len(stage_times["mediapipe"]["pipeline_ms"])
+    if not measured:
+        raise ValueError("input did not contain enough frames after warmup")
+    metrics = {
+        name: {**{stage: distribution(values) for stage, values in times.items()}, **totals[name]}
+        for name, times in stage_times.items()
+    }
+    baseline_p95 = metrics["mediapipe"]["detect_geometry_quality_ms"]["p95"]
+    candidate_p95 = metrics["yunet"]["detect_geometry_quality_ms"]["p95"]
+    return {
+        "synthetic_drawings": synthetic_faces,
+        "measured_frames": measured,
+        "detectors": metrics,
+        "detection_p95_improvement_percent": 100 * (baseline_p95 - candidate_p95) / baseline_p95,
+        "face_count_mismatch_frames": mismatch_frames,
+        "spatially_paired_faces": paired,
+        "unmatched_baseline_faces": unmatched_baseline,
+        "unmatched_candidate_faces": unmatched_candidate,
+        "quality_disagreements": quality_disagreements,
+        "five_landmark_rms_drift_pixels": distribution(landmark_drifts),
+        "embedding_cosine": distribution(cosines),
+        "recognition": {
+            "evaluated_with_existing_gallery": gallery is not None,
+            "paired_decisions": decision_pairs,
+            "decision_agreements": decision_agreements,
+            "different_known_identity_pairs": different_known,
+            "false_accept_rate": None,
+            "false_reject_rate": None,
+        },
+        "activation_eligible": False,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--video", type=Path, help="approved local video only; no webcam/network inputs")
+    parser.add_argument("--gallery-json", type=Path, help="private existing 0095 session PUT payload")
+    parser.add_argument("--threshold", type=float)
+    parser.add_argument("--margin", type=float)
+    parser.add_argument("--download-yunet", action="store_true", help="download and verify public model if absent")
+    parser.add_argument("--yunet-model", type=Path, default=ROOT / "models" / YUNET_NAME)
+    parser.add_argument("--synthetic-faces", type=int, choices=[1, 4, 8], nargs="+", default=[1, 4, 8])
+    parser.add_argument("--warmup", type=positive_int, default=10)
+    parser.add_argument("--max-frames", type=positive_int, default=60)
+    args = parser.parse_args()
+    if args.max_frames <= args.warmup:
+        parser.error("--max-frames must exceed --warmup")
+    if args.gallery_json and (args.threshold is None or args.margin is None):
+        parser.error("gallery comparison requires explicit --threshold and --margin")
+    if args.gallery_json and args.video is None:
+        parser.error("gallery comparison requires an approved local --video")
+    if any(value is not None and not np.isfinite(value) for value in (args.threshold, args.margin)):
+        parser.error("threshold and margin must be finite")
+    gallery = None
+    models = None
+    try:
+        ensure_yunet(args.yunet_model, args.download_yunet)
+        if args.gallery_json is not None:
+            gallery = load_gallery(args.gallery_json)
+        models = InferenceModels(
+            str(ROOT / "models" / "face_landmarker.task"),
+            str(ROOT / "models" / "face-reidentification-retail-0095.xml"),
+            ModelMetadata(),
+            "f32",
+        )
+        detector = YuNetDetector(args.yunet_model)
+        measurements = [
+            compare_stream(args, models, detector, gallery, count)
+            for count in ([None] if args.video is not None else args.synthetic_faces)
+        ]
+        import mediapipe
+        import openvino
+
+        print(json.dumps({
+            "scope": "experimental_detector_quality_alignment_embedding_comparison",
+            "input_kind": "approved_local_video" if args.video else "synthetic_non_photographic",
+            "versions": {"python": platform.python_version(), "opencv": cv2.__version__, "mediapipe": mediapipe.__version__, "openvino": openvino.__version__},
+            "candidate_model": {"name": YUNET_NAME, "sha256": YUNET_SHA256, "size_bytes": YUNET_BYTES, "source": YUNET_URL, "landmark_reference": LANDMARK_REFERENCE},
+            "embedding_model": asdict(models.metadata),
+            "candidate_parameters": {"score_threshold": 0.9, "nms_threshold": 0.3, "top_k": 5000, "max_faces": 8, "input_resize": False, "target": "CPU_FP32"},
+            "threshold": args.threshold,
+            "margin": args.margin,
+            "warmup_frames_per_case": args.warmup,
+            "measurements": measurements,
+            "limitations": [
+                "Synthetic drawing counts are input counts, not labelled real faces or accuracy evidence.",
+                "Different detector boxes and eye landmark conventions can change quality and alignment.",
+                "Pipeline embeds every quality-passing face; HTTP, temporal gates and session queues are excluded.",
+                "Spatial IoU matching can be ambiguous for overlapping faces; it does not prove identity.",
+                "Gallery decisions show baseline agreement, not false acceptance/rejection without ground truth.",
+                "No activation: approved labelled accuracy validation and the 15 percent p95 gate remain required.",
+                "No frames, vectors, identities, landmarks, video paths or gallery paths are written or printed.",
+            ],
+        }, indent=2, allow_nan=False))
+    finally:
+        if models is not None:
+            models.close()
+        if gallery is not None:
+            gallery.clear()
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        # Library/JSON exceptions may contain private input or local file paths.
+        print(f"Benchmark failed ({type(exc).__name__}); verify local inputs/model files and arguments.", file=sys.stderr)
+        raise SystemExit(1) from None
diff --git a/benchmarks/face_recheck_benchmark.py b/benchmarks/face_recheck_benchmark.py
new file mode 100644
index 0000000..0f3f87d
--- /dev/null
+++ b/benchmarks/face_recheck_benchmark.py
@@ -0,0 +1,201 @@
+"""Compare repeated successful recognition with success recheck suppression.
+
+Only a procedural non-photographic frame is used. The exact embedding generated
+from that frame seeds one synthetic gallery entry so both lanes recognize it.
+The benchmark measures the existing API and prints only aggregate timings/counts.
+
+Example:
+    .venv/Scripts/python.exe benchmarks/face_recheck_benchmark.py
+"""
+
+from __future__ import annotations
+
+import argparse
+import io
+import json
+import logging
+import sys
+from pathlib import Path
+
+import cv2
+from fastapi.testclient import TestClient
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from face_api_synthetic import parse_timing_records, summarize, synthetic_candidates  # noqa: E402
+from face_detection_modes import positive_int, synthetic_frame  # noqa: E402
+
+from face_api import create_app  # noqa: E402
+from face_config import Settings  # noqa: E402
+from face_models import EMBEDDING_DIMENSION, MODEL_ID, ModelMetadata  # noqa: E402
+from face_pipeline import clear_observations, quality_issues  # noqa: E402
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--requests", type=positive_int, default=100, help="measured requests per lane")
+    parser.add_argument("--warmup", type=positive_int, default=20)
+    parser.add_argument("--students", type=positive_int, default=50)
+    parser.add_argument("--vectors", type=positive_int, default=20)
+    parser.add_argument("--seed", type=int, default=20261013)
+    args = parser.parse_args()
+    if args.warmup >= args.requests:
+        parser.error("--warmup must be less than --requests")
+    if args.students > 200 or args.vectors > 20:
+        parser.error("synthetic gallery is limited to 200 students and 20 vectors")
+
+    token = "synthetic-benchmark-token"
+    settings = Settings(service_token=token, match_threshold=0.6, match_margin=0.1)
+    metadata = ModelMetadata(version=settings.model_version)
+    candidate_vectors, candidates = synthetic_candidates(args.students, args.vectors, args.seed)
+    log_buffer = io.StringIO()
+    handler = logging.StreamHandler(log_buffer)
+    logger = logging.getLogger("face_api")
+    previous_level = logger.level
+    previous_handlers = logger.handlers[:]
+    previous_propagate = logger.propagate
+    logger.handlers = [handler]
+    logger.setLevel(logging.DEBUG)
+    logger.propagate = False
+    app = create_app(settings=settings)
+    lanes = (
+        ("recheck_disabled", "synthetic-recheck-disabled"),
+        ("recheck_enabled", "synthetic-recheck-enabled"),
+    )
+    headers = {"Authorization": f"Bearer {token}"}
+    frame = synthetic_frame(0)
+    encoded = None
+    probe = None
+    observations = []
+    try:
+        with TestClient(app) as client:
+            runtime = app.state.models
+            observations = runtime.detect_faces(frame)
+            eligible = [item for item in observations if not quality_issues(item)]
+            if not eligible:
+                raise RuntimeError("the procedural frame did not produce a quality face")
+            probe = runtime.embed_face(frame, eligible[0])
+            # CandidateGallery scores the mean of its best three templates.
+            candidate_vectors[0, :3] = probe
+            candidates[0]["vectors"] = candidate_vectors[0].tolist()
+            payload = {
+                "model": {
+                    "model_id": MODEL_ID,
+                    "version": metadata.version,
+                    "dimension": EMBEDDING_DIMENSION,
+                    "normalization": metadata.normalization,
+                },
+                "candidates": candidates,
+            }
+            for _, session_id in lanes:
+                response = client.put(
+                    f"/internal/v1/face/sessions/{session_id}", headers=headers, json=payload
+                )
+                response.raise_for_status()
+            app.state.sessions[lanes[0][1]].tracker.success_recheck_seconds = 0
+            app.state.sessions[lanes[1][1]].tracker.success_recheck_seconds = 1
+
+            success_counts = dict.fromkeys((name for name, _ in lanes), 0)
+            measured_statuses = {name: {} for name, _ in lanes}
+            for index in range(args.warmup + args.requests):
+                for name, session_id in lanes:
+                    success, encoded = cv2.imencode(".jpg", frame)
+                    if not success:
+                        raise RuntimeError("could not encode synthetic frame")
+                    response = client.post(
+                        f"/internal/v1/face/sessions/{session_id}/frames",
+                        headers={**headers, "Content-Type": "image/jpeg"},
+                        content=encoded.tobytes(),
+                    )
+                    response.raise_for_status()
+                    faces = response.json()["faces"]
+                    if not faces:
+                        raise RuntimeError("the procedural face disappeared during the benchmark")
+                    status = faces[0]["recognition"]["status"]
+                    if status == "KNOWN":
+                        success_counts[name] += 1
+                    if index >= args.warmup:
+                        measured_statuses[name][status] = measured_statuses[name].get(status, 0) + 1
+                    encoded.fill(0)
+            for _, session_id in lanes:
+                client.delete(f"/internal/v1/face/sessions/{session_id}", headers=headers).raise_for_status()
+    finally:
+        logger.handlers = previous_handlers
+        logger.setLevel(previous_level)
+        logger.propagate = previous_propagate
+        handler.close()
+        frame.fill(0)
+        if encoded is not None:
+            encoded.fill(0)
+        if probe is not None:
+            probe.fill(0)
+        clear_observations(observations)
+        candidate_vectors.fill(0)
+        for candidate in candidates:
+            for vector in candidate["vectors"]:
+                vector[:] = [0.0] * len(vector)
+
+    records = parse_timing_records(log_buffer.getvalue())
+    expected = 2 * (args.warmup + args.requests)
+    if len(records) != expected:
+        raise RuntimeError(f"expected {expected} timing records, received {len(records)}")
+    measured = {name: [] for name, _ in lanes}
+    for index, record in enumerate(records):
+        lane_index = index % len(lanes)
+        pair_index = index // len(lanes)
+        if pair_index >= args.warmup:
+            measured[lanes[lane_index][0]].append(record)
+
+    results = {}
+    for name, _ in lanes:
+        lane_records = measured[name]
+        stages = {}
+        for stage in ("admission", "queue", "decode", "detect", "embed", "match", "total"):
+            values = [float(record[stage]) for record in lane_records if stage in record]
+            if values:
+                stages[stage] = {**summarize(values), "samples": len(values)}
+        results[name] = {
+            "stages": stages,
+            "embedding_attempts": sum(int(record["embeddings"]) for record in lane_records),
+            "recognized_attempts_including_warmup": success_counts[name],
+            "measured_recognition_statuses": measured_statuses[name],
+        }
+
+    baseline = results["recheck_disabled"]
+    optimized = results["recheck_enabled"]
+    if optimized["embedding_attempts"] >= baseline["embedding_attempts"]:
+        raise RuntimeError("success recheck did not reduce embedding attempts")
+    if optimized["recognized_attempts_including_warmup"] == 0:
+        raise RuntimeError("known-face recognition was not exercised")
+    base_stages = baseline["stages"]
+    optimized_stages = optimized["stages"]
+    comparison = {
+        "embedding_attempt_reduction_percent": 100 * (
+            baseline["embedding_attempts"] - optimized["embedding_attempts"]
+        ) / baseline["embedding_attempts"],
+        "total_median_reduction_percent": 100 * (
+            base_stages["total"]["median_ms"] - optimized_stages["total"]["median_ms"]
+        ) / base_stages["total"]["median_ms"],
+        "total_p95_reduction_percent": 100 * (
+            base_stages["total"]["p95_ms"] - optimized_stages["total"]["p95_ms"]
+        ) / base_stages["total"]["p95_ms"],
+    }
+
+    print(json.dumps({
+        "scope": "synthetic_known_face_recheck_api_comparison",
+        "input_kind": "procedural_non_photographic_frame_with_matching_model_embedding",
+        "requests_per_lane": args.requests,
+        "warmup_per_lane": args.warmup,
+        "synthetic_students": args.students,
+        "synthetic_vectors_per_student": args.vectors,
+        "lanes": results,
+        "comparison_percent_reduction": comparison,
+        "note": "Measures repeated-known-face compute savings, not real face accuracy or field throughput.",
+    }, indent=2, allow_nan=False))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/compose.prod.yaml b/compose.prod.yaml
index d122e0a..5055fd2 100644
--- a/compose.prod.yaml
+++ b/compose.prod.yaml
@@ -12,5 +12,6 @@ services:
       FACE_MATCH_THRESHOLD: ${FACE_MATCH_THRESHOLD:-0.6}
       FACE_MATCH_MARGIN: ${FACE_MATCH_MARGIN:-0.1}
       FACE_INFERENCE_PRECISION_HINT: ${FACE_INFERENCE_PRECISION_HINT:-f32}
+      FACE_SUCCESS_RECHECK_SECONDS: ${FACE_SUCCESS_RECHECK_SECONDS:-1}
       FACE_SESSION_IDLE_SECONDS: 300
       FACE_SESSION_CLEANUP_INTERVAL_SECONDS: 60
diff --git a/docs/reviews/face-recheck-benchmark-2026-10-13.md b/docs/reviews/face-recheck-benchmark-2026-10-13.md
new file mode 100644
index 0000000..8277552
--- /dev/null
+++ b/docs/reviews/face-recheck-benchmark-2026-10-13.md
@@ -0,0 +1,32 @@
+# Repeated known-face benchmark — 2026-10-13
+
+[`face_recheck_benchmark.py`](../../benchmarks/face_recheck_benchmark.py) compares the
+real frame API with success rechecks disabled and enabled. It uses a procedural,
+non-photographic frame and seeds the synthetic gallery with the embedding generated
+from that same frame. Both lanes run interleaved in one process with the same 200
+students and 20 templates per student.
+
+Reproduce with:
+
+```powershell
+.\.venv\Scripts\python.exe benchmarks\face_recheck_benchmark.py --requests 120 --warmup 20 --students 200 --vectors 20
+```
+
+Two runs each used 20 warmups and 120 measured requests per lane:
+
+| Run | Recheck | Total median | Total p95 | Embedding attempts |
+| ---: | --- | ---: | ---: | ---: |
+| 1 | Disabled | 12.131 ms | 16.720 ms | 40 |
+| 1 | 1 second | 11.925 ms | 14.541 ms | 3 |
+| 2 | Disabled | 11.861 ms | 16.248 ms | 40 |
+| 2 | 1 second | 11.798 ms | 13.841 ms | 3 |
+
+The feature reduced embedding and gallery-match attempts by 92.5%. Total median
+latency changed by less than 2% across these runs, while p95 fell by 13.0% and
+14.8%. Face detection remained about 10.7 ms per frame and dominated total time.
+
+This measures repeated computation only. The synthetic frame is not representative
+of real faces and cannot establish recognition accuracy or field throughput. During
+the recheck window the API reports `NOT_ATTEMPTED` with no student ID; it does not
+replay the previous `KNOWN` result. Real labeled validation is needed before drawing
+accuracy conclusions or making any detector-resolution change.
diff --git a/face_api.py b/face_api.py
index 5ee8711..ac40f2d 100644
--- a/face_api.py
+++ b/face_api.py
@@ -7,7 +7,7 @@
 import time
 from collections.abc import Callable, Iterator
 from contextlib import asynccontextmanager, contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, field, replace
 from typing import Annotated, Literal
 
 import cv2
@@ -18,6 +18,7 @@
 from pydantic import BaseModel, Field
 
 from face_config import Settings
+from face_execution import InferenceExecutor
 from face_models import FaceObservation, InferenceModels, ModelMetadata
 from face_pipeline import (
     CandidateGallery,
@@ -67,6 +68,10 @@ def finish(self) -> None:
                 "outcome": self.outcome,
             })
 
+    def add_duration(self, stage: str, milliseconds: float) -> None:
+        if self.enabled:
+            self.durations[stage] = self.durations.get(stage, 0.0) + milliseconds
+
 
 class VectorModel(BaseModel):
     model_id: str
@@ -155,6 +160,253 @@ class Session:
     last_activity_at: float
     in_flight: int = 0
     closing: bool = False
+    frame_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
+
+
+class _EnrollmentWork:
+    """Decode sampled enrollment frames incrementally on the inference worker."""
+
+    def __init__(
+        self,
+        body: bytearray,
+        runtime: InferenceModels,
+        cfg: Settings,
+        metadata: ModelMetadata,
+    ) -> None:
+        self.body = body
+        self.runtime = runtime
+        self.cfg = cfg
+        self.metadata = metadata
+        self.av = __import__("av")
+        self.container = None
+        self.frames = None
+        self.targets: set[int] | None = None
+        self.frame_limit = max(1, min(cfg.enrollment_frames, 100))
+        self.vector_limit = max(1, min(cfg.representative_vectors, 20))
+        self.observations: list[FaceObservation] = []
+        self.per_track: dict[str, list[FaceObservation]] = {}
+        self.tracks = TrackManager()
+        self.decoded_count = 0
+        self.reviewed_count = 0
+        self.accepted_count = 0
+        self.done = False
+        self.vectors: list[np.ndarray] = []
+        self.stats: dict[str, object] = {}
+
+    def _open(self) -> None:
+        if self.container is not None:
+            return
+        try:
+            self.container = self.av.open(io.BytesIO(self.body), mode="r")
+            total = int(self.container.streams.video[0].frames or 0)
+        except (self.av.error.FFmpegError, EOFError, IndexError, ValueError) as exc:
+            raise FaceServiceError("INVALID_MEDIA", "video could not be decoded", 400) from exc
+        self.targets = _sample_indices(total, self.frame_limit) if total else None
+        self.frames = iter(_decoded_video_frames(self.container, self.av))
+
+    def step(self) -> bool:
+        """Inspect at most 20 decoded frames or one sampled frame per turn."""
+        if self.done:
+            return True
+        try:
+            self._open()
+            for _ in range(20):
+                if self.targets is None and self.reviewed_count >= self.frame_limit:
+                    self._finish()
+                    return True
+                try:
+                    frame = next(self.frames)
+                except StopIteration:
+                    self._finish()
+                    return True
+                index = self.decoded_count
+                self.decoded_count += 1
+                if self.targets is not None and index not in self.targets:
+                    continue
+                image = frame.to_ndarray(format="bgr24")
+                try:
+                    self.reviewed_count += 1
+                    found = self.runtime.detect_faces(image)
+                    base_index = len(self.observations)
+                    self.observations.extend(found)
+                    assigned = self.tracks.assign([item.bbox for item in found])
+                    for index, (observation, track) in enumerate(
+                        zip(found, assigned, strict=True)
+                    ):
+                        if not quality_issues(observation):
+                            embedding = self.runtime.embed_face(image, observation)
+                            observation = replace(observation, embedding=embedding)
+                            self.observations[base_index + index] = observation
+                            self.accepted_count += 1
+                        self.per_track.setdefault(track.track_id, []).append(observation)
+                finally:
+                    image.fill(0)
+                # Give queued camera frames a worker turn before decoding more.
+                return False
+            return False
+        except FaceServiceError:
+            raise
+        except Exception as exc:
+            raise FaceServiceError("INTERNAL_ERROR", "enrollment inference failed", 500) from exc
+
+    def _finish(self) -> None:
+        if self.done:
+            return
+        if len(self.per_track) > 1:
+            raise FaceServiceError("MULTIPLE_IDENTITIES", "enrollment must contain one face identity")
+        viable = {
+            track_id: items
+            for track_id, items in self.per_track.items()
+            if sum(item.embedding is not None and not quality_issues(item) for item in items)
+            >= self.vector_limit
+        }
+        if not viable:
+            if self.observations and all(
+                "LOW_LIGHT" in quality_issues(item) for item in self.observations
+            ):
+                raise FaceServiceError("LOW_LIGHT", "enrollment video is too dark")
+            raise FaceServiceError(
+                "INSUFFICIENT_QUALITY_FRAMES",
+                f"need {self.vector_limit} quality observations",
+            )
+        self.vectors = select_representatives(next(iter(viable.values())), self.vector_limit)
+        self.stats = {
+            "model": self.metadata.__dict__,
+            "reviewedFrames": self.reviewed_count,
+            "acceptedFrames": self.accepted_count,
+        }
+        self.done = True
+
+    def result(self) -> tuple[dict[str, object], list[np.ndarray]]:
+        if not self.done:
+            raise RuntimeError("enrollment work is not complete")
+        vectors, self.vectors = self.vectors, []
+        return self.stats, vectors
+
+    def close(self) -> None:
+        if self.frames is not None:
+            close = getattr(self.frames, "close", None)
+            if close is not None:
+                close()
+        if self.container is not None:
+            self.container.close()
+            self.container = None
+        clear_observations(self.observations)
+        _clear_vectors({"temporary": self.vectors})
+        self.vectors.clear()
+        self.per_track.clear()
+
+
+def _prepare_gallery(payload: SessionPayload, metadata: ModelMetadata) -> CandidateGallery:
+    candidates: dict[str, list[np.ndarray]] = {}
+    try:
+        incoming = ModelMetadata(
+            payload.model.model_id,
+            payload.model.version,
+            payload.model.dimension,
+            payload.model.normalization,
+        )
+        validate_model(incoming, metadata)
+        for item in payload.candidates:
+            if item.student_id in candidates:
+                raise FaceServiceError("MODEL_MISMATCH", "duplicate student IDs are not allowed")
+            candidates[item.student_id] = []
+            for vector in item.vectors:
+                candidates[item.student_id].append(normalize_vector(vector))
+        return CandidateGallery.from_candidates(candidates)
+    finally:
+        _clear_vectors(candidates)
+
+
+def _process_frame(
+    body: bytearray,
+    runtime: InferenceModels,
+    session: Session,
+    cfg: Settings,
+    frame_id: str,
+    timings: _FrameTimings,
+    queued_at_ns: int,
+) -> dict[str, object]:
+    image: np.ndarray | None = None
+    observations: list[FaceObservation] = []
+    try:
+        timings.add_duration("queue", (time.perf_counter_ns() - queued_at_ns) / 1_000_000)
+        with timings.measure("decode"):
+            image = cv2.imdecode(np.frombuffer(body, dtype=np.uint8), cv2.IMREAD_COLOR)
+        if image is None:
+            raise FaceServiceError("INVALID_MEDIA", "frame is not a valid image", 400)
+        with timings.measure("detect"):
+            observations = runtime.detect_faces(image)
+        timings.faces = len(observations)
+        height, width = image.shape[:2]
+        boxes = [_response_bbox(observation.bbox) for observation in observations]
+        landmarks = [
+            _response_landmarks(observation.landmarks, width, height)
+            for observation in observations
+        ]
+        tracks = session.tracker.assign(boxes)
+        results = []
+        for index, (observation, track) in enumerate(zip(observations, tracks, strict=True)):
+            issues = quality_issues(observation)
+            if issues:
+                session.tracker.reject_quality(track)
+                eligible = False
+            else:
+                eligible = session.tracker.observe_quality(track)
+            if (
+                not eligible
+                or cfg.match_threshold is None
+                or cfg.match_margin is None
+            ):
+                recognition = {
+                    "status": "NOT_ATTEMPTED",
+                    "studentId": None,
+                    "score": None,
+                    "margin": None,
+                }
+            else:
+                probe: np.ndarray | None = None
+                try:
+                    with timings.measure("embed"):
+                        timings.embeddings += 1
+                        probe = runtime.embed_face(image, observation)
+                    with timings.measure("match"):
+                        outcome = session.gallery.match(
+                            probe, cfg.match_threshold, cfg.match_margin
+                        )
+                finally:
+                    if probe is not None:
+                        probe.fill(0)
+                recognition = {
+                    "status": outcome.status,
+                    "studentId": outcome.student_id if outcome.status == "KNOWN" else None,
+                    "score": outcome.score,
+                    "margin": outcome.margin,
+                }
+                if outcome.status == "KNOWN":
+                    session.tracker.record_success(track)
+                else:
+                    session.tracker.record_failure(track)
+            results.append(
+                {
+                    "trackId": track.track_id,
+                    "bbox": boxes[index],
+                    "landmarks": landmarks[index],
+                    "quality": {
+                        "brightness": observation.brightness,
+                        "sharpness": observation.sharpness,
+                        "issues": issues,
+                    },
+                    "recognition": recognition,
+                    "attempts": track.failures,
+                    "qrRecommended": track.failures >= 4,
+                }
+            )
+        return {"frameId": frame_id, "faces": results}
+    finally:
+        if image is not None:
+            image.fill(0)
+        clear_observations(observations)
 
 
 def create_app(
@@ -169,6 +421,25 @@ def create_app(
     sessions_drained = asyncio.Event()
     sessions_drained.set()
     lock = asyncio.Lock()
+    frame_slots = asyncio.Semaphore(4)
+    enrollment_slots = asyncio.Semaphore(1)
+    requests_drained = asyncio.Event()
+    requests_drained.set()
+    active_requests = 0
+    stopping = False
+
+    def begin_request() -> None:
+        nonlocal active_requests
+        if stopping:
+            raise FaceServiceError("MODEL_NOT_READY", "service is shutting down", 503)
+        active_requests += 1
+        requests_drained.clear()
+
+    def end_request() -> None:
+        nonlocal active_requests
+        active_requests -= 1
+        if active_requests == 0:
+            requests_drained.set()
 
     def discard_session(session: Session) -> None:
         session.tracker.reset()
@@ -221,15 +492,25 @@ def release_session(session_id: str, session: Session) -> None:
 
     @asynccontextmanager
     async def lifespan(app: FastAPI):
+        nonlocal stopping
+        stopping = False
+        executor = InferenceExecutor()
         runtime = models
-        if runtime is None and cfg.landmarker_path.is_file() and cfg.embedding_model_path.is_file():
-            runtime = InferenceModels(
-                str(cfg.landmarker_path),
-                str(cfg.embedding_model_path),
-                metadata,
-                cfg.inference_precision_hint,
-            )
+        try:
+            if runtime is None and cfg.landmarker_path.is_file() and cfg.embedding_model_path.is_file():
+                runtime = await executor.run(
+                    InferenceModels,
+                    str(cfg.landmarker_path),
+                    str(cfg.embedding_model_path),
+                    metadata,
+                    cfg.inference_precision_hint,
+                    on_cancel=lambda loaded: loaded.close(),
+                )
+        except BaseException:
+            await executor.close()
+            raise
         app.state.models = runtime
+        app.state.executor = executor
         app.state.sessions = sessions
         app.state.retired_sessions = retired_sessions
         app.state.cleanup_expired_sessions = cleanup_expired_sessions
@@ -243,6 +524,7 @@ async def reap_expired_sessions() -> None:
         try:
             yield
         finally:
+            stopping = True
             cleanup_task.cancel()
             try:
                 await cleanup_task
@@ -251,9 +533,14 @@ async def reap_expired_sessions() -> None:
             for session in list(sessions.values()) + list(retired_sessions.values()):
                 retire_session(session)
             sessions.clear()
+            await requests_drained.wait()
             await sessions_drained.wait()
-            if runtime is not None and models is None:
-                runtime.close()
+            try:
+                if runtime is not None and models is None:
+                    await executor.run(runtime.close)
+            finally:
+                app.state.models = None
+                await executor.close()
 
     app = FastAPI(title="Dormitory Face Inference Service", version="0.1.0", lifespan=lifespan)
 
@@ -350,10 +637,23 @@ async def enroll(request: Request) -> EnrollmentResponse:
             raise handle_error(FaceServiceError("INVALID_MEDIA", "expected WebM or MP4 video", 400))
         body: bytearray | None = None
         vectors: list[np.ndarray] = []
+        admitted = False
+        work: _EnrollmentWork | None = None
         try:
-            body = await read_body(request)
-            stats, vectors = _extract_enrollment_vectors(body, runtime, cfg, metadata)
-            return {**stats, "vectors": [vector.tolist() for vector in vectors]}
+            begin_request()
+            admitted = True
+            async with enrollment_slots:
+                body = await read_body(request)
+                work = _EnrollmentWork(body, runtime, cfg, metadata)
+                try:
+                    # Yield the single inference worker between decoded video frames.
+                    while not await app.state.executor.run(work.step):
+                        pass
+                    stats, vectors = work.result()
+                    return {**stats, "vectors": [vector.tolist() for vector in vectors]}
+                finally:
+                    # run() drains native work before propagating cancellation.
+                    await app.state.executor.run(work.close)
         except FaceServiceError as exc:
             raise handle_error(exc) from exc
         except Exception as exc:
@@ -362,6 +662,8 @@ async def enroll(request: Request) -> EnrollmentResponse:
             _clear_vectors({"temporary": vectors})
             if body is not None:
                 body[:] = b"\x00" * len(body)
+            if admitted:
+                end_request()
 
     @app.put(
         "/internal/v1/face/sessions/{session_id}",
@@ -375,37 +677,37 @@ async def enroll(request: Request) -> EnrollmentResponse:
     async def put_session(
         session_id: Annotated[str, Path(min_length=1)], payload: SessionPayload
     ) -> SessionReadyResponse:
-        candidates: dict[str, list[np.ndarray]] = {}
         gallery: CandidateGallery | None = None
         installed = False
+        admitted = False
         try:
-            incoming = ModelMetadata(
-                payload.model.model_id,
-                payload.model.version,
-                payload.model.dimension,
-                payload.model.normalization,
+            if stopping:
+                raise FaceServiceError("MODEL_NOT_READY", "service is shutting down", 503)
+            begin_request()
+            admitted = True
+            gallery = await app.state.executor.run(
+                _prepare_gallery,
+                payload,
+                metadata,
+                on_cancel=lambda prepared: prepared.clear(),
             )
-            validate_model(incoming, metadata)
-            for item in payload.candidates:
-                if item.student_id in candidates:
-                    raise FaceServiceError("MODEL_MISMATCH", "duplicate student IDs are not allowed")
-                candidates[item.student_id] = []
-                for vector in item.vectors:
-                    candidates[item.student_id].append(normalize_vector(vector))
-            # Preserve the legacy PUT normalization followed by match normalization.
-            gallery = CandidateGallery.from_candidates(candidates)
             async with lock:
                 old = sessions.get(session_id)
                 if old is not None:
                     retire_session(old)
-                sessions[session_id] = Session(gallery, TrackManager(), monotonic())
+                sessions[session_id] = Session(
+                    gallery,
+                    TrackManager(success_recheck_seconds=cfg.success_recheck_seconds, monotonic=monotonic),
+                    monotonic(),
+                )
                 installed = True
         except FaceServiceError as exc:
             raise handle_error(exc) from exc
         finally:
-            _clear_vectors(candidates)
             if gallery is not None and not installed:
                 gallery.clear()
+            if admitted:
+                end_request()
         return SessionReadyResponse(status="ready")
 
     @app.post(
@@ -441,8 +743,7 @@ async def frame(
         timings = _FrameTimings()
         session: Session | None = None
         body: bytearray | None = None
-        image: np.ndarray | None = None
-        observations: list[FaceObservation] = []
+        admitted = False
         try:
             runtime: InferenceModels | None = app.state.models
             if runtime is None:
@@ -452,83 +753,47 @@ async def frame(
             session = acquire_session(session_id)
             if session is None:
                 raise FaceServiceError("SESSION_NOT_FOUND", "session is not active", 404)
-            with timings.measure("read"):
-                body = await read_body(request)
-            with timings.measure("decode"):
-                image = cv2.imdecode(np.frombuffer(body, dtype=np.uint8), cv2.IMREAD_COLOR)
-            if image is None:
-                raise FaceServiceError("INVALID_MEDIA", "frame is not a valid image", 400)
-            with timings.measure("detect"):
-                observations = runtime.detect_faces(image)
-            timings.faces = len(observations)
-            height, width = image.shape[:2]
-            boxes = [_response_bbox(observation.bbox) for observation in observations]
-            landmarks = [
-                _response_landmarks(observation.landmarks, width, height)
-                for observation in observations
-            ]
-            tracks = session.tracker.assign(boxes)
-            results = []
-            for index, (observation, track) in enumerate(zip(observations, tracks, strict=True)):
-                issues = quality_issues(observation)
-                if issues or not session.tracker.observe_quality(track):
-                    recognition = {"status": "NOT_ATTEMPTED", "studentId": None, "score": None, "margin": None}
-                elif cfg.match_threshold is None or cfg.match_margin is None:
-                    recognition = {"status": "NOT_ATTEMPTED", "studentId": None, "score": None, "margin": None}
-                else:
-                    probe: np.ndarray | None = None
-                    try:
-                        with timings.measure("embed"):
-                            timings.embeddings += 1
-                            probe = runtime.embed_face(image, observation)
-                        with timings.measure("match"):
-                            outcome = session.gallery.match(probe, cfg.match_threshold, cfg.match_margin)
-                    finally:
-                        if probe is not None:
-                            probe.fill(0)
-                    recognition = {
-                        "status": outcome.status,
-                        "studentId": outcome.student_id if outcome.status == "KNOWN" else None,
-                        "score": outcome.score,
-                        "margin": outcome.margin,
-                    }
-                    if outcome.status == "KNOWN":
-                        session.tracker.record_success(track)
-                    else:
-                        session.tracker.record_failure(track)
-                results.append(
-                    {
-                        "trackId": track.track_id,
-                        "bbox": boxes[index],
-                        "landmarks": landmarks[index],
-                        "quality": {
-                            "brightness": observation.brightness,
-                            "sharpness": observation.sharpness,
-                            "issues": issues,
-                        },
-                        "recognition": recognition,
-                        "attempts": track.failures,
-                        "qrRecommended": track.failures >= 4,
-                    }
-                )
+            begin_request()
+            admitted = True
+            admission_started_ns = time.perf_counter_ns()
+            async with frame_slots:
+                async with session.frame_lock:
+                    timings.add_duration(
+                        "admission", (time.perf_counter_ns() - admission_started_ns) / 1_000_000
+                    )
+                    if session.closing or sessions.get(session_id) is not session:
+                        timings.outcome = "retired"
+                        return {"frameId": x_frame_id, "faces": []}
+                    with timings.measure("read"):
+                        body = await read_body(request)
+                    queued_at_ns = time.perf_counter_ns()
+                    results = await app.state.executor.run(
+                        _process_frame,
+                        body,
+                        runtime,
+                        session,
+                        cfg,
+                        x_frame_id,
+                        timings,
+                        queued_at_ns,
+                    )
             # A streamed upload can resume after PUT/DELETE retired this session.
             if session.closing or sessions.get(session_id) is not session:
-                results = []
+                results["faces"] = []
             timings.outcome = "success"
-            return {"frameId": x_frame_id, "faces": results}
+            return results
         except FaceServiceError as exc:
             raise handle_error(exc) from exc
         except Exception as exc:
             raise handle_error(FaceServiceError("INTERNAL_ERROR", "frame inference failed", 500)) from exc
         finally:
             with timings.measure("cleanup"):
-                if image is not None:
-                    image.fill(0)
-                clear_observations(observations)
                 if body is not None:
                     body[:] = b"\x00" * len(body)
                 if session is not None:
                     release_session(session_id, session)
+                if admitted:
+                    end_request()
             timings.finish()
 
     @app.delete(
@@ -550,81 +815,6 @@ async def delete_session(
     return app
 
 
-def _extract_enrollment_vectors(
-    body: bytearray,
-    runtime: InferenceModels,
-    cfg: Settings,
-    metadata: ModelMetadata,
-) -> tuple[dict[str, object], list[np.ndarray]]:
-    container = None
-    observations: list[FaceObservation] = []
-    vectors: list[np.ndarray] = []
-    transferred = False
-    try:
-        av = __import__("av")
-        try:
-            container = av.open(io.BytesIO(body), mode="r")
-            stream = container.streams.video[0]
-            total = int(stream.frames or 0)
-        except (av.error.FFmpegError, EOFError, IndexError, ValueError) as exc:
-            raise FaceServiceError("INVALID_MEDIA", "video could not be decoded", 400) from exc
-        frame_limit = max(1, min(cfg.enrollment_frames, 100))
-        vector_limit = max(1, min(cfg.representative_vectors, 20))
-        targets = _sample_indices(total, frame_limit) if total else None
-        tracks = TrackManager()
-        per_track: dict[str, list[FaceObservation]] = {}
-        reviewed_count = 0
-        for index, frame in enumerate(_decoded_video_frames(container, av)):
-            if targets is not None and index not in targets:
-                continue
-            if targets is None and reviewed_count >= frame_limit:
-                break
-            image = frame.to_ndarray(format="bgr24")
-            try:
-                reviewed_count += 1
-                found = runtime.detect(image)
-                observations.extend(found)
-                assigned = tracks.assign([observation.bbox for observation in found])
-                for observation, track in zip(found, assigned):
-                    per_track.setdefault(track.track_id, []).append(observation)
-            finally:
-                image.fill(0)
-
-        # Enrollment is for one student. Never combine vectors from multiple faces.
-        if len(per_track) > 1:
-            raise FaceServiceError("MULTIPLE_IDENTITIES", "enrollment must contain one face identity")
-        viable = {
-            track_id: items
-            for track_id, items in per_track.items()
-            if sum(not quality_issues(item) for item in items) >= vector_limit
-        }
-        if not viable:
-            if observations and all("LOW_LIGHT" in quality_issues(item) for item in observations):
-                raise FaceServiceError("LOW_LIGHT", "enrollment video is too dark")
-            raise FaceServiceError(
-                "INSUFFICIENT_QUALITY_FRAMES",
-                f"need {vector_limit} quality observations",
-            )
-        vectors = select_representatives(next(iter(viable.values())), vector_limit)
-        stats = {
-            "model": metadata.__dict__,
-            "reviewedFrames": reviewed_count,
-            "acceptedFrames": sum(not quality_issues(item) for item in observations),
-        }
-        transferred = True
-        return stats, vectors
-    except FaceServiceError:
-        raise
-    except Exception as exc:
-        raise FaceServiceError("INTERNAL_ERROR", "enrollment inference failed", 500) from exc
-    finally:
-        if container is not None:
-            container.close()
-        clear_observations(observations)
-        if not transferred:
-            _clear_vectors({"temporary": vectors})
-
-
 def _decoded_video_frames(container, av):
     try:
         yield from container.decode(video=0)
diff --git a/face_config.py b/face_config.py
index ea50f61..29dda19 100644
--- a/face_config.py
+++ b/face_config.py
@@ -39,8 +39,13 @@ class Settings:
     inference_precision_hint: str = field(
         default_factory=lambda: os.getenv("FACE_INFERENCE_PRECISION_HINT", "f32").strip().lower()
     )
+    success_recheck_seconds: float = field(
+        default_factory=lambda: float(os.getenv("FACE_SUCCESS_RECHECK_SECONDS", "1"))
+    )
 
     def __post_init__(self) -> None:
+        if not isfinite(self.success_recheck_seconds) or not 0 <= self.success_recheck_seconds <= 1:
+            raise ValueError("FACE_SUCCESS_RECHECK_SECONDS must be between 0 and 1")
         if self.inference_precision_hint not in {"f32", "bf16"}:
             raise ValueError("FACE_INFERENCE_PRECISION_HINT must be f32 or bf16")
         if (
diff --git a/face_execution.py b/face_execution.py
new file mode 100644
index 0000000..e96754b
--- /dev/null
+++ b/face_execution.py
@@ -0,0 +1,105 @@
+"""Serialize native inference without blocking the HTTP event loop."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from functools import partial
+from typing import Any, TypeVar
+
+Result = TypeVar("Result")
+
+
+async def _drain(
+    future: asyncio.Future[Result],
+    on_cancel: Callable[[Result], None] | None = None,
+) -> Result:
+    """Finish accepted work before cancellation releases its input buffers.
+
+    A fresh shield is needed after every cancellation: the cancelled shield cannot
+    be reused, but the underlying future must never be cancelled or abandoned.
+    Cleanup callbacks run synchronously on the caller's event loop after native
+    work finishes, including when a second cancellation arrives while draining.
+    """
+    cancellation = None
+    while not future.done():
+        try:
+            await asyncio.shield(future)
+        except asyncio.CancelledError as error:
+            cancellation = error
+        except BaseException:
+            # Retrieve the native exception below, preserving request cancellation
+            # if that happened first. Calling result() also consumes the exception.
+            if not future.done():
+                raise
+
+    try:
+        result = future.result()
+    except BaseException:
+        if cancellation is not None:
+            raise cancellation from None
+        raise
+    if cancellation is not None:
+        try:
+            if on_cancel is not None:
+                on_cancel(result)
+        finally:
+            raise cancellation
+    return result
+
+
+class InferenceExecutor:
+    """One lifespan-owned native worker with FIFO submissions.
+
+    Callers bound admission and own input-buffer cleanup. Once run() is submitted,
+    its caller cannot leave run() until native code stops touching those buffers.
+    on_cancel disposes a successful result that cancellation prevents returning to
+    the caller (for example, a decoded image or an embedding).
+    """
+
+    def __init__(self) -> None:
+        self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="face-inference")
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._pending: set[asyncio.Future[Any]] = set()
+        self._closing = False
+        self._close_task: asyncio.Task[None] | None = None
+
+    def _current_loop(self) -> asyncio.AbstractEventLoop:
+        loop = asyncio.get_running_loop()
+        if self._loop is None:
+            self._loop = loop
+        elif self._loop is not loop:
+            raise RuntimeError("InferenceExecutor must be used from one event loop")
+        return loop
+
+    async def run(
+        self,
+        function: Callable[..., Result],
+        *args: Any,
+        on_cancel: Callable[[Result], None] | None = None,
+    ) -> Result:
+        loop = self._current_loop()
+        if self._closing:
+            raise RuntimeError("InferenceExecutor is closed")
+        future = loop.run_in_executor(self._executor, partial(function, *args))
+        self._pending.add(future)
+        try:
+            return await _drain(future, on_cancel)
+        finally:
+            self._pending.discard(future)
+
+    async def _drain_and_shutdown(self) -> None:
+        if self._pending:
+            await asyncio.gather(*tuple(self._pending), return_exceptions=True)
+        # Joining even an idle worker is blocking. Use the loop's separate default
+        # executor so the owned inference thread cannot try to join itself.
+        await asyncio.to_thread(self._executor.shutdown, wait=True, cancel_futures=False)
+
+    async def close(self) -> None:
+        """Reject new work, drain accepted jobs, and join the owned worker once."""
+        loop = self._current_loop()
+        if self._close_task is None:
+            self._closing = True
+            self._close_task = loop.create_task(self._drain_and_shutdown())
+        await _drain(self._close_task)
diff --git a/face_tracker.py b/face_tracker.py
index 232fdec..5af4cde 100644
--- a/face_tracker.py
+++ b/face_tracker.py
@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import time
+from collections.abc import Callable
 from dataclasses import dataclass, field
+from math import isfinite
 from uuid import uuid4
 
 
@@ -12,6 +15,7 @@ class Track:
     missed_frames: int = 0
     quality_observations: int = 0
     cooldown_frames: int = 0
+    success_recheck_at: float | None = None
 
 
 def iou(a: tuple[float, float, float, float], b: tuple[float, float, float, float]) -> float:
@@ -25,24 +29,59 @@ def iou(a: tuple[float, float, float, float], b: tuple[float, float, float, floa
 
 
 class TrackManager:
-    def __init__(self, max_missed: int = 6):
+    def __init__(
+        self,
+        max_missed: int = 6,
+        success_recheck_seconds: float = 1.0,
+        monotonic: Callable[[], float] = time.monotonic,
+    ):
+        if not isfinite(success_recheck_seconds) or not 0 <= success_recheck_seconds <= 1:
+            raise ValueError("success_recheck_seconds must be between 0 and 1")
         self.max_missed = max_missed
+        self.success_recheck_seconds = success_recheck_seconds
+        self.monotonic = monotonic
         self.tracks: list[Track] = []
 
     def assign(self, boxes: list[tuple[float, float, float, float]]) -> list[Track]:
         result: list[Track] = []
         unused = set(range(len(self.tracks)))
-        for box in boxes:
-            best_index = max(unused, key=lambda i: iou(self.tracks[i].bbox, box), default=None)
-            if best_index is not None and iou(self.tracks[best_index].bbox, box) >= 0.2:
+        # Snapshot all overlaps before mutating any bounding box. Continuity must
+        # be unambiguous in both directions, including already assigned tracks.
+        overlaps = [[iou(track.bbox, box) for track in self.tracks] for box in boxes]
+        for box_index, box in enumerate(boxes):
+            row = overlaps[box_index]
+            best_index = max(unused, key=lambda i: row[i], default=None)
+            if best_index is not None and row[best_index] >= 0.2:
                 track = self.tracks[best_index]
                 unused.remove(best_index)
+                if track.success_recheck_at is not None:
+                    overlap = row[best_index]
+                    other_track = max(
+                        (score for index, score in enumerate(row) if index != best_index),
+                        default=0.0,
+                    )
+                    other_box = max(
+                        (
+                            scores[best_index]
+                            for index, scores in enumerate(overlaps)
+                            if index != box_index
+                        ),
+                        default=0.0,
+                    )
+                    if (
+                        track.missed_frames
+                        or overlap < 0.5
+                        or overlap - other_track < 0.2
+                        or overlap - other_box < 0.2
+                    ):
+                        self.reject_quality(track)
             else:
                 track = Track()
             track.bbox, track.missed_frames = box, 0
             result.append(track)
         for index in unused:
             self.tracks[index].missed_frames += 1
+            self.reject_quality(self.tracks[index])
         self.tracks = [
             *result,
             *(
@@ -54,6 +93,7 @@ def assign(self, boxes: list[tuple[float, float, float, float]]) -> list[Track]:
         return result
 
     def record_failure(self, track: Track) -> int:
+        track.success_recheck_at = None
         track.failures += 1
         track.quality_observations = 0
         track.cooldown_frames = 5
@@ -63,12 +103,28 @@ def record_success(self, track: Track) -> None:
         track.failures = 0
         track.quality_observations = 0
         track.cooldown_frames = 0
+        track.success_recheck_at = (
+            self.monotonic() + self.success_recheck_seconds
+            if self.success_recheck_seconds
+            else None
+        )
+
+    def reject_quality(self, track: Track) -> None:
+        """Invalidate success reuse without changing legacy failure cooldowns."""
+        if track.success_recheck_at is not None:
+            track.success_recheck_at = None
+            track.quality_observations = 0
 
     def observe_quality(self, track: Track) -> bool:
         if track.cooldown_frames:
             track.cooldown_frames -= 1
             return False
-        track.quality_observations += 1
+        if track.success_recheck_at is None:
+            track.quality_observations += 1
+        else:
+            track.quality_observations = min(3, track.quality_observations + 1)
+        if track.success_recheck_at is not None and self.monotonic() < track.success_recheck_at:
+            return False
         return track.quality_observations >= 3
 
     def reset(self) -> None:
diff --git a/harness/check.test.mjs b/harness/check.test.mjs
index 470966b..67f3951 100644
--- a/harness/check.test.mjs
+++ b/harness/check.test.mjs
@@ -10,7 +10,7 @@ import { syncSkills } from './sync-skills.mjs';
 function fixture(t) {
   const temp = mkdtempSync(path.join(os.tmpdir(), 'dorm-harness-test-'));
   const repo = path.join(temp, 'repo');
-  const inputs = ['README.md', 'AGENTS.md', 'CLAUDE.md', 'package.json', '.gitignore', '.codex', '.claude', '.agents', 'docs', 'contracts', 'web', 'server', 'ai', 'infra', 'tests', 'harness', '.github', 'Dockerfile', '.dockerignore', 'models/download-manifest.json', 'face_api.py', 'face_models.py', 'face_pipeline.py', 'face_tracker.py', 'benchmarks'];
+  const inputs = ['README.md', 'AGENTS.md', 'CLAUDE.md', 'package.json', '.gitignore', '.codex', '.claude', '.agents', 'docs', 'contracts', 'web', 'server', 'ai', 'infra', 'tests', 'harness', '.github', 'Dockerfile', '.dockerignore', 'models/download-manifest.json', 'face_api.py', 'face_execution.py', 'face_models.py', 'face_pipeline.py', 'face_tracker.py', 'benchmarks'];
   mkdirSync(repo);
   const excluded = new Set(['node_modules', '.next', '.venv', '.git', 'target', 'build', 'dist', 'coverage', '.local', 'datasets', 'models', 'backups', 'auth.json', 'settings.local.json', 'sessions', 'log']);
   const safeInput = source => {
diff --git a/harness/hooks.mjs b/harness/hooks.mjs
index fe59b19..fb415d9 100644
--- a/harness/hooks.mjs
+++ b/harness/hooks.mjs
@@ -59,6 +59,7 @@ function contentFrom(payload) {
 function isAiIntegrationFile(filePath) {
   return filePath === 'face_api.py'
     || filePath === 'face_config.py'
+    || filePath === 'face_execution.py'
     || filePath === 'face_models.py'
     || filePath === 'face_pipeline.py'
     || filePath === 'face_tracker.py'
diff --git a/harness/hooks.test.mjs b/harness/hooks.test.mjs
index 906b546..e5f9946 100644
--- a/harness/hooks.test.mjs
+++ b/harness/hooks.test.mjs
@@ -57,3 +57,8 @@ test('asks AI work to be recorded in the integrated harness', () => {
   const notice = postToolNotice({ tool_name: 'Edit', tool_input: { file_path: 'face_pipeline.py' } });
   assert.match(notice, /harness:worklog/);
 });
+
+test('recognizes the isolated AI inference executor as AI integration code', () => {
+  const notice = postToolNotice({ tool_name: 'Edit', tool_input: { file_path: 'face_execution.py' } });
+  assert.match(notice, /harness:worklog/);
+});
diff --git a/harness/local-worklog.md b/harness/local-worklog.md
index e51fbf2..97ff955 100644
--- a/harness/local-worklog.md
+++ b/harness/local-worklog.md
@@ -37,3 +37,7 @@ npm run harness:worklog -- "작업 내용과 검증 결과"
 - 2026-10-03: FastAPI CPU 컨테이너 이미지와 모델 build-time SHA-256 검증 다운로드, readiness healthcheck, 로컬 env 예시 추가. Docker 이미지 빌드와 컨테이너 실행/readiness는 미실행.
 
 - 2026-10-03: README에 프론트→Spring 인증·얼굴 API 연결, 미설정 CORS 전제, AI 내부 배포·환경 변수·health·단일 worker 조건과 DevOps 순서 문서화. 문서 변경이며 테스트/배포 미실행.
+
+- 2026-10-13: AI inference performance: moved frame and sliced enrollment work to a lifespan-owned single worker; bounded frame admission; skipped repeat embeddings for recently recognized tracks with quality and tracking invalidation; added detector comparison tool without enabling a detector switch. Verified 172 AI tests, ruff, harness:check, compose config, benchmark CLI, and git diff check. No labeled face accuracy benchmark was run because no approved validation sample was supplied.
+
+- 2026-10-13: Repeated-known-frame API A/B, two runs: 200 synthetic students x20 vectors, 20 warmups/120 measured per lane. Recheck disabled/enabled produced 40/3 embedding attempts (92.5% fewer); total median 12.131/11.925ms and 11.861/11.798ms; p95 16.720/14.541ms and 16.248/13.841ms. Synthetic only; detection remained about 10.7ms and no real-face accuracy claim. See docs/reviews/face-recheck-benchmark-2026-10-13.md.
diff --git a/tests/face_service/test_api_latency.py b/tests/face_service/test_api_latency.py
index ff16c15..e93af48 100644
--- a/tests/face_service/test_api_latency.py
+++ b/tests/face_service/test_api_latency.py
@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import threading
 from contextlib import asynccontextmanager
 from pathlib import Path
 
@@ -67,6 +68,7 @@ def settings(**overrides):
         "service_token": "synthetic-service-token",
         "match_threshold": 0.5,
         "match_margin": 0.1,
+        "success_recheck_seconds": 0,
     }
     return Settings(**(values | overrides))
 
@@ -118,6 +120,36 @@ def test_only_eligible_third_good_observation_embeds_and_buffers_are_cleared():
         assert app.state.sessions[SESSION].tracker.tracks[0].quality_observations == 0
 
 
+def test_recently_recognized_track_skips_embedding_until_recheck_deadline():
+    now = [10.0]
+    models = MetadataModels()
+    app = create_app(settings(success_recheck_seconds=1), models, monotonic=lambda: now[0])
+    image = encoded_image()
+    with TestClient(app, headers=AUTH) as client:
+        install_session(client)
+        for _ in range(3):
+            face = send_frame(client, image).json()["faces"][0]
+        assert face["recognition"]["status"] == "KNOWN"
+        assert len(models.probes) == 1
+
+        for expected_quality_count in (1, 2, 3):
+            face = send_frame(client, image).json()["faces"][0]
+            assert face["recognition"] == {
+                "status": "NOT_ATTEMPTED",
+                "studentId": None,
+                "score": None,
+                "margin": None,
+            }
+            assert app.state.sessions[SESSION].tracker.tracks[0].quality_observations == expected_quality_count
+        assert len(models.probes) == 1
+
+        now[0] += 1
+        face = send_frame(client, image).json()["faces"][0]
+        assert face["recognition"]["status"] == "KNOWN"
+        assert len(models.probes) == 2
+        assert all(not probe.any() for probe in models.probes)
+
+
 def test_poor_frames_do_not_advance_quality_or_decrement_failure_cooldown():
     models = MetadataModels(vector_indices=(1,))
     app = create_app(settings(), models)
@@ -319,7 +351,7 @@ def test_debug_timings_contain_only_fixed_stage_durations_counts_and_outcome(cap
                 assert send_frame(client, encoded_image()).status_code == 200
     records = [record for record in caplog.records if record.name == "face_api"]
     assert len(records) == 3
-    stages = {"read", "decode", "detect", "embed", "match", "cleanup", "total"}
+    stages = {"admission", "read", "queue", "decode", "detect", "embed", "match", "cleanup", "total"}
     for index, record in enumerate(records):
         assert record.msg == "face_frame_timing_ms %s"
         fields = record.args
@@ -451,6 +483,58 @@ async def streamed_body():
     asyncio.run(scenario())
 
 
+def test_cancelled_frame_drains_native_inference_before_scrubbing_buffers(monkeypatch):
+    buffers = []
+
+    class RetainedBuffer(bytearray):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            buffers.append(self)
+
+    class BlockingModels(MetadataModels):
+        def __init__(self):
+            super().__init__()
+            self.entered = threading.Event()
+            self.unblock = threading.Event()
+            self.worker_thread = None
+
+        def detect_faces(self, image):
+            self.worker_thread = threading.get_ident()
+            self.entered.set()
+            if not self.unblock.wait(3):
+                raise RuntimeError("synthetic detector unblock timed out")
+            return super().detect_faces(image)
+
+    monkeypatch.setattr(face_api, "bytearray", RetainedBuffer, raising=False)
+
+    async def scenario():
+        models = BlockingModels()
+        app = create_app(settings(), models)
+        async with async_client(app) as client:
+            assert (await client.put(f"/internal/v1/face/sessions/{SESSION}", json=payload())).status_code == 200
+            session = app.state.sessions[SESSION]
+            task = asyncio.create_task(async_frame(client, encoded_image()))
+            try:
+                entered = await asyncio.wait_for(asyncio.to_thread(models.entered.wait, 3), timeout=4)
+                assert entered
+                assert models.worker_thread != threading.get_ident()
+                assert buffers and any(buffers[0]) and session.in_flight == 1
+                await asyncio.sleep(0)
+                task.cancel()
+                await asyncio.sleep(0.02)
+                assert not task.done()
+                assert any(buffers[0]) and session.in_flight == 1
+            finally:
+                models.unblock.set()
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(task, timeout=3)
+            assert session.in_flight == 0
+            assert buffers and all(not any(buffer) for buffer in buffers)
+            assert models.images and all(not image.any() for image in models.images)
+
+    asyncio.run(scenario())
+
+
 @pytest.mark.parametrize("invalid", ["duplicate", "zero"])
 def test_rejected_session_payload_preserves_current_gallery_and_scrubs_partial_vectors(monkeypatch, invalid):
     models = MetadataModels()
diff --git a/tests/face_service/test_config.py b/tests/face_service/test_config.py
index f01e2c7..c88e695 100644
--- a/tests/face_service/test_config.py
+++ b/tests/face_service/test_config.py
@@ -41,3 +41,22 @@ def test_inference_precision_hint_reads_environment(monkeypatch):
 def test_unknown_inference_precision_hint_is_rejected():
     with pytest.raises(ValueError, match="FACE_INFERENCE_PRECISION_HINT"):
         Settings(inference_precision_hint="fp16")
+
+
+def test_success_recheck_defaults_to_one_second(monkeypatch):
+    monkeypatch.delenv("FACE_SUCCESS_RECHECK_SECONDS", raising=False)
+
+    assert Settings().success_recheck_seconds == 1.0
+
+
+@pytest.mark.parametrize("seconds", ["0", "0.25", "1"])
+def test_success_recheck_reads_environment(monkeypatch, seconds):
+    monkeypatch.setenv("FACE_SUCCESS_RECHECK_SECONDS", seconds)
+
+    assert Settings().success_recheck_seconds == float(seconds)
+
+
+@pytest.mark.parametrize("seconds", [-0.1, 1.1, float("inf"), float("-inf"), float("nan")])
+def test_invalid_success_recheck_seconds_are_rejected(seconds):
+    with pytest.raises(ValueError, match="FACE_SUCCESS_RECHECK_SECONDS"):
+        Settings(success_recheck_seconds=seconds)
diff --git a/tests/face_service/test_execution.py b/tests/face_service/test_execution.py
new file mode 100644
index 0000000..06c3b8d
--- /dev/null
+++ b/tests/face_service/test_execution.py
@@ -0,0 +1,282 @@
+from __future__ import annotations
+
+import asyncio
+import threading
+
+import pytest
+
+from face_execution import InferenceExecutor
+
+
+def test_native_work_runs_in_fifo_order_on_one_thread_without_blocking_event_loop():
+    async def scenario():
+        executor = InferenceExecutor()
+        loop = asyncio.get_running_loop()
+        started = asyncio.Event()
+        release = threading.Event()
+        order = []
+        threads = []
+
+        def work(value):
+            threads.append(threading.current_thread())
+            order.append(value)
+            if value == 0:
+                loop.call_soon_threadsafe(started.set)
+                assert release.wait(timeout=5)
+            return value
+
+        tasks = [asyncio.create_task(executor.run(work, 0))]
+        try:
+            await asyncio.wait_for(started.wait(), timeout=3)
+            for value in (1, 2):
+                tasks.append(asyncio.create_task(executor.run(work, value)))
+                await asyncio.sleep(0)
+            # Reaching this point while native work is still blocked also verifies
+            # HTTP/heartbeat coroutines retain access to the event loop.
+            assert order == [0]
+            assert not any(task.done() for task in tasks)
+            assert len(executor._pending) == 3
+            release.set()
+            assert await asyncio.gather(*tasks) == [0, 1, 2]
+            assert order == [0, 1, 2]
+            assert len(set(threads)) == 1
+            assert threads[0] is not threading.current_thread()
+            assert not executor._pending
+        finally:
+            release.set()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            await executor.close()
+        assert not threads[0].is_alive()
+
+    asyncio.run(scenario())
+
+
+@pytest.mark.parametrize("repeat_cancel", [False, True])
+@pytest.mark.parametrize("native_failure", [False, True])
+def test_cancellation_drains_native_work_before_result_and_input_cleanup(
+    repeat_cancel, native_failure
+):
+    async def scenario():
+        executor = InferenceExecutor()
+        loop = asyncio.get_running_loop()
+        started = asyncio.Event()
+        release = threading.Event()
+        source = bytearray(b"synthetic-input")
+        output = bytearray(b"synthetic-output")
+        cleanup_calls = []
+        worker_reads = []
+
+        def work():
+            loop.call_soon_threadsafe(started.set)
+            assert release.wait(timeout=5)
+            worker_reads.append(bytes(source))
+            if native_failure:
+                raise ValueError("synthetic native failure")
+            return output
+
+        def cleanup(result):
+            assert result is output
+            cleanup_calls.append(threading.current_thread())
+            result[:] = b"\0" * len(result)
+
+        async def request():
+            try:
+                await executor.run(work, on_cancel=cleanup)
+            finally:
+                source[:] = b"\0" * len(source)
+
+        task = asyncio.create_task(request())
+        try:
+            await asyncio.wait_for(started.wait(), timeout=3)
+            task.cancel()
+            await asyncio.sleep(0)
+            if repeat_cancel:
+                task.cancel()
+                await asyncio.sleep(0)
+            assert not task.done()
+            assert source == b"synthetic-input"
+            assert output == b"synthetic-output"
+            assert not cleanup_calls
+            release.set()
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(task, timeout=3)
+            assert worker_reads == [b"synthetic-input"]
+            assert not any(source)
+            assert cleanup_calls == ([] if native_failure else [threading.current_thread()])
+            assert bool(any(output)) is native_failure
+            assert not executor._pending
+            # Cancellation and native failures do not poison the single worker.
+            assert await executor.run(lambda: "next") == "next"
+        finally:
+            release.set()
+            await asyncio.gather(task, return_exceptions=True)
+            await executor.close()
+
+    asyncio.run(scenario())
+
+
+def test_cancelled_queued_job_still_finishes_and_disposes_its_result():
+    async def scenario():
+        executor = InferenceExecutor()
+        loop = asyncio.get_running_loop()
+        started = asyncio.Event()
+        release = threading.Event()
+        completed = []
+        cleaned = []
+
+        def blocked():
+            loop.call_soon_threadsafe(started.set)
+            assert release.wait(timeout=5)
+
+        def queued():
+            completed.append(True)
+            return None
+
+        first = asyncio.create_task(executor.run(blocked))
+        second = None
+        try:
+            await asyncio.wait_for(started.wait(), timeout=3)
+            second = asyncio.create_task(executor.run(queued, on_cancel=cleaned.append))
+            await asyncio.sleep(0)
+            second.cancel()
+            await asyncio.sleep(0)
+            assert not second.done() and not completed and not cleaned
+            release.set()
+            assert await first is None
+            with pytest.raises(asyncio.CancelledError):
+                await second
+            assert completed == [True] and cleaned == [None]
+        finally:
+            release.set()
+            await asyncio.gather(
+                *([first, second] if second is not None else [first]), return_exceptions=True
+            )
+            await executor.close()
+
+    asyncio.run(scenario())
+
+
+def test_native_exceptions_propagate_and_normal_results_do_not_call_cancel_cleanup():
+    async def scenario():
+        executor = InferenceExecutor()
+        cleaned = []
+
+        def fail():
+            raise ValueError("synthetic native failure")
+
+        try:
+            with pytest.raises(ValueError, match="synthetic native failure"):
+                await executor.run(fail, on_cancel=cleaned.append)
+            assert await executor.run(lambda: None, on_cancel=cleaned.append) is None
+            assert not cleaned and not executor._pending
+        finally:
+            await executor.close()
+
+    asyncio.run(scenario())
+
+
+def test_result_cleanup_failure_preserves_request_cancellation():
+    async def scenario():
+        executor = InferenceExecutor()
+        loop = asyncio.get_running_loop()
+        started = asyncio.Event()
+        release = threading.Event()
+        cleaned = []
+
+        def blocked():
+            loop.call_soon_threadsafe(started.set)
+            assert release.wait(timeout=5)
+            return "result"
+
+        def cleanup(result):
+            cleaned.append(result)
+            raise ValueError("synthetic cleanup failure")
+
+        task = asyncio.create_task(executor.run(blocked, on_cancel=cleanup))
+        try:
+            await asyncio.wait_for(started.wait(), timeout=3)
+            task.cancel()
+            await asyncio.sleep(0)
+            release.set()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+            assert cleaned == ["result"]
+            assert not executor._pending
+        finally:
+            release.set()
+            await asyncio.gather(task, return_exceptions=True)
+            await executor.close()
+
+    asyncio.run(scenario())
+
+
+def test_close_rejects_submissions_drains_accepted_jobs_and_is_repeatable_when_cancelled():
+    async def scenario():
+        executor = InferenceExecutor()
+        loop = asyncio.get_running_loop()
+        started = asyncio.Event()
+        release = threading.Event()
+        order = []
+        threads = []
+
+        def work(value):
+            threads.append(threading.current_thread())
+            order.append(value)
+            if value == 0:
+                loop.call_soon_threadsafe(started.set)
+                assert release.wait(timeout=5)
+            return value
+
+        jobs = [asyncio.create_task(executor.run(work, 0))]
+        closes = []
+        try:
+            await asyncio.wait_for(started.wait(), timeout=3)
+            jobs.append(asyncio.create_task(executor.run(work, 1)))
+            await asyncio.sleep(0)
+            closes.append(asyncio.create_task(executor.close()))
+            await asyncio.sleep(0)
+            with pytest.raises(RuntimeError, match="closed"):
+                await executor.run(work, 2)
+            closes.append(asyncio.create_task(executor.close()))
+            await asyncio.sleep(0)
+            closes[0].cancel()
+            await asyncio.sleep(0)
+            closes[0].cancel()
+            await asyncio.sleep(0)
+            assert order == [0]
+            assert not any(task.done() for task in jobs + closes)
+            release.set()
+            assert await asyncio.gather(*jobs) == [0, 1]
+            with pytest.raises(asyncio.CancelledError):
+                await closes[0]
+            await closes[1]
+            await executor.close()
+            assert order == [0, 1]
+            assert not any(thread.is_alive() for thread in threads)
+            assert not executor._pending
+            with pytest.raises(RuntimeError, match="closed"):
+                await executor.run(work, 3)
+        finally:
+            release.set()
+            await asyncio.gather(*jobs, *closes, return_exceptions=True)
+            await executor.close()
+
+    asyncio.run(scenario())
+
+
+def test_close_without_submissions_and_cross_loop_use():
+    executor = InferenceExecutor()
+
+    async def first_loop():
+        await executor.close()
+        await executor.close()
+
+    asyncio.run(first_loop())
+
+    async def another_loop():
+        with pytest.raises(RuntimeError, match="one event loop"):
+            await executor.run(lambda: None)
+        with pytest.raises(RuntimeError, match="one event loop"):
+            await executor.close()
+
+    asyncio.run(another_loop())
diff --git a/tests/face_service/test_tracker.py b/tests/face_service/test_tracker.py
new file mode 100644
index 0000000..58fd234
--- /dev/null
+++ b/tests/face_service/test_tracker.py
@@ -0,0 +1,253 @@
+from dataclasses import dataclass
+
+import pytest
+
+from face_tracker import TrackManager
+
+BOX = (0.0, 0.0, 100.0, 100.0)
+DISTANT_BOX = (300.0, 0.0, 100.0, 100.0)
+
+
+@dataclass
+class Clock:
+    now: float = 0.0
+
+    def __call__(self):
+        return self.now
+
+
+@pytest.fixture
+def clock():
+    return Clock()
+
+
+@pytest.fixture
+def manager(clock):
+    return TrackManager(monotonic=clock)
+
+
+def test_first_attempt_still_requires_three_quality_observations(manager):
+    track = manager.assign([BOX])[0]
+
+    assert [manager.observe_quality(track) for _ in range(3)] == [False, False, True]
+
+
+def test_success_suppresses_attempts_without_extending_deadline(manager, clock):
+    track = manager.assign([BOX])[0]
+    manager.record_success(track)
+    deadline = track.success_recheck_at
+
+    for now in (0.1, 0.2, 0.3, 0.9, 0.999):
+        clock.now = now
+        assert manager.assign([BOX])[0] is track
+        assert not manager.observe_quality(track)
+        assert track.success_recheck_at == deadline
+        assert track.failures == 0
+
+    assert track.quality_observations == 3
+    clock.now = 1.0
+    assert manager.observe_quality(track)
+
+
+def test_expiry_alone_does_not_replace_three_quality_observations(manager, clock):
+    track = manager.assign([BOX])[0]
+    manager.record_success(track)
+    clock.now = 2.0
+
+    assert [manager.observe_quality(track) for _ in range(3)] == [False, False, True]
+
+
+def test_next_actual_success_starts_a_new_deadline(manager, clock):
+    track = manager.assign([BOX])[0]
+    manager.record_failure(track)
+    manager.record_success(track)
+    assert track.failures == track.quality_observations == track.cooldown_frames == 0
+
+    clock.now = 1.25
+    assert [manager.observe_quality(track) for _ in range(3)] == [False, False, True]
+    manager.record_success(track)
+
+    assert track.success_recheck_at == 2.25
+    assert track.quality_observations == 0
+    assert not manager.observe_quality(track)
+
+
+def test_disabling_success_recheck_preserves_legacy_quality_and_misses(clock):
+    manager = TrackManager(success_recheck_seconds=0, monotonic=clock)
+    track = manager.assign([BOX])[0]
+    manager.record_success(track)
+    assert track.success_recheck_at is None
+    assert not manager.observe_quality(track)
+
+    manager.reject_quality(track)
+    manager.assign([])
+    assert manager.assign([(50.0, 0.0, 100.0, 100.0)])[0] is track
+    assert track.quality_observations == 1
+    assert not manager.observe_quality(track)
+    assert manager.observe_quality(track)
+
+
+def test_one_missed_frame_invalidates_success_even_if_track_survives(manager):
+    track = manager.assign([BOX])[0]
+    manager.record_success(track)
+    manager.observe_quality(track)
+
+    manager.assign([])
+
+    assert track.missed_frames == 1
+    assert track.success_recheck_at is None
+    assert track.quality_observations == 0
+    assert manager.assign([BOX])[0] is track
+    assert [manager.observe_quality(track) for _ in range(3)] == [False, False, True]
+
+
+@pytest.mark.parametrize("after_expiry", [False, True])
+def test_low_quality_invalidates_success_and_quality_count(manager, clock, after_expiry):
+    track = manager.assign([BOX])[0]
+    manager.record_success(track)
+    for _ in range(3):
+        manager.observe_quality(track)
+    if after_expiry:
+        clock.now = 2.0
+
+    manager.reject_quality(track)
+
+    assert track.success_recheck_at is None
+    assert track.quality_observations == 0
+    assert [manager.observe_quality(track) for _ in range(3)] == [False, False, True]
+
+
+@pytest.mark.parametrize("failed", [False, True])
+def test_low_quality_preserves_legacy_state_without_prior_success(manager, failed):
+    track = manager.assign([BOX])[0]
+    if failed:
+        manager.record_failure(track)
+    manager.observe_quality(track)
+    previous = (track.failures, track.quality_observations, track.cooldown_frames)
+
+    manager.reject_quality(track)
+    manager.assign([])
+
+    assert (track.failures, track.quality_observations, track.cooldown_frames) == previous
+
+
+@pytest.mark.parametrize("after_expiry", [False, True])
+def test_abrupt_movement_invalidates_success_without_replacing_track(manager, clock, after_expiry):
+    track = manager.assign([BOX])[0]
+    manager.record_success(track)
+    for _ in range(3):
+        manager.observe_quality(track)
+    if after_expiry:
+        clock.now = 2.0
+
+    assert manager.assign([(50.0, 0.0, 100.0, 100.0)])[0] is track
+
+    assert track.success_recheck_at is None
+    assert track.quality_observations == 0
+    assert not manager.observe_quality(track)
+
+
+def test_current_face_ambiguous_between_previous_tracks_invalidates_success(manager):
+    tracks = manager.assign([BOX, (20.0, 0.0, 100.0, 100.0)])
+    for track in tracks:
+        manager.record_success(track)
+        manager.observe_quality(track)
+
+    matched = manager.assign([(10.0, 0.0, 100.0, 100.0)])[0]
+
+    assert matched in tracks
+    assert all(track.success_recheck_at is None for track in tracks)
+    assert all(track.quality_observations == 0 for track in tracks)
+
+
+def test_previous_track_ambiguous_between_current_faces_invalidates_success(manager):
+    track = manager.assign([BOX])[0]
+    manager.record_success(track)
+    manager.observe_quality(track)
+
+    matches = manager.assign([BOX, (10.0, 0.0, 100.0, 100.0)])
+
+    assert matches[0] is track
+    assert matches[1] is not track
+    assert all(match.success_recheck_at is None for match in matches)
+    assert track.quality_observations == 0
+
+
+def test_greedy_second_choice_cannot_reuse_success_from_a_nonbest_track(manager):
+    tracks = manager.assign([BOX, (20.0, 0.0, 100.0, 100.0)])
+    for track in tracks:
+        manager.record_success(track)
+        manager.observe_quality(track)
+
+    matches = manager.assign([BOX, (1.0, 0.0, 100.0, 100.0)])
+
+    assert matches == tracks
+    assert all(track.success_recheck_at is None for track in tracks)
+    assert all(track.quality_observations == 0 for track in tracks)
+
+
+def test_new_face_is_eligible_independently_of_successful_face(manager):
+    known = manager.assign([BOX])[0]
+    manager.record_success(known)
+    matches = manager.assign([BOX, DISTANT_BOX])
+
+    assert matches[0] is known
+    assert not manager.observe_quality(known)
+    assert matches[1].success_recheck_at is None
+    assert [manager.observe_quality(matches[1]) for _ in range(3)] == [False, False, True]
+
+
+def test_multiple_successful_faces_keep_independent_deadlines_and_tracking(manager, clock):
+    first, second = manager.assign([BOX, DISTANT_BOX])
+    manager.record_success(first)
+    clock.now = 0.25
+    manager.record_success(second)
+
+    assert manager.assign([DISTANT_BOX, BOX]) == [second, first]
+    for _ in range(3):
+        assert not manager.observe_quality(first)
+        assert not manager.observe_quality(second)
+
+    clock.now = 1.0
+    assert manager.observe_quality(first)
+    assert not manager.observe_quality(second)
+    assert second.success_recheck_at == 1.25
+
+
+def test_failure_clears_success_and_keeps_five_good_frames_before_three_observations(manager):
+    track = manager.assign([BOX])[0]
+    manager.record_success(track)
+
+    assert manager.record_failure(track) == 1
+    assert track.success_recheck_at is None
+    assert [manager.observe_quality(track) for _ in range(8)] == [False] * 7 + [True]
+    assert manager.record_failure(track) == 2
+
+
+def test_reset_and_new_tracks_do_not_inherit_success(manager):
+    old = manager.assign([BOX])[0]
+    manager.record_success(old)
+    manager.reset()
+
+    new = manager.assign([BOX])[0]
+
+    assert new.track_id != old.track_id
+    assert new.success_recheck_at is None
+    assert [manager.observe_quality(new) for _ in range(3)] == [False, False, True]
+
+
+def test_out_of_range_face_creates_new_track_without_success(manager):
+    old = manager.assign([BOX])[0]
+    manager.record_success(old)
+
+    new = manager.assign([DISTANT_BOX])[0]
+
+    assert new is not old
+    assert old.success_recheck_at is None
+    assert new.success_recheck_at is None
+
+
+@pytest.mark.parametrize("seconds", [-0.1, 1.1, float("inf"), float("-inf"), float("nan")])
+def test_invalid_recheck_settings_are_rejected(seconds):
+    with pytest.raises(ValueError, match="success_recheck_seconds"):
+        TrackManager(success_recheck_seconds=seconds)