feat(backend): enhance inference pipeline with GLM logic and structured outputs

inference.py

@@ -7,7 +7,7 @@ import os
 
 import logging
 import time
-from threading import RLock, Thread
+from threading import Event, RLock, Thread
 from queue import Queue, PriorityQueue, Full, Empty
 from typing import Any, Dict, List, Optional, Sequence, Tuple
 
@@ -15,15 +15,15 @@ import cv2
 import numpy as np
 import torch
 from concurrent.futures import ThreadPoolExecutor
-from threading import RLock
 from models.detectors.base import ObjectDetector
 from models.model_loader import load_detector, load_detector_on_device
 from models.segmenters.model_loader import load_segmenter, load_segmenter_on_device
 from models.depth_estimators.model_loader import load_depth_estimator, load_depth_estimator_on_device
 from models.depth_estimators.base import DepthEstimator
-from utils.video import extract_frames, write_video, VideoReader, VideoWriter, AsyncVideoReader, StreamingVideoWriter
-from utils.gpt_reasoning import estimate_threat_gpt, encode_frame_to_b64
-from utils.relevance import evaluate_relevance, evaluate_relevance_llm
+from utils.video import extract_frames, write_video, VideoReader, VideoWriter, StreamingVideoWriter
+from utils.relevance import evaluate_relevance
+from utils.enrichment import run_enrichment
+from utils.schemas import AssessmentStatus
 from jobs.storage import set_track_data
 import tempfile
 import json as json_module
@@ -781,7 +781,7 @@ def process_first_frame(
                     "bbox": [int(c) for c in box],
                     "score": float(seg_result.scores[idx]) if seg_result.scores is not None and idx < len(seg_result.scores) else 1.0,
                     "track_id": f"T{idx + 1:02d}",
-                    "assessment_status": "UNASSESSED",
+                    "assessment_status": AssessmentStatus.UNASSESSED,
                 })
         return processed, detections
 
@@ -791,7 +791,7 @@ def process_first_frame(
 
     # Tag all detections as unassessed — GPT runs later in enrichment thread
     for det in detections:
-        det["assessment_status"] = "UNASSESSED"
+        det["assessment_status"] = AssessmentStatus.UNASSESSED
 
     return processed, detections
 
@@ -1067,7 +1067,7 @@ def run_inference(
     # --- GPT Enrichment Thread (non-blocking) ---
     # Runs LLM relevance + GPT threat assessment off the writer's critical path.
     gpt_enrichment_queue = Queue(maxsize=4)
-    _relevance_refined = [False]  # mutable container for thread-safe sharing
+    _relevance_refined = Event()
 
     def enrichment_thread_fn(tracker_ref):
         """Dedicated thread for GPT/LLM calls. Receives work from writer, injects results via tracker."""
@@ -1077,64 +1077,13 @@ def run_inference(
                 break  # Sentinel — shutdown
             frame_idx, frame_data, gpt_dets, ms = item
             try:
-                # LLM post-filter (LLM_EXTRACTED mode)
-                if ms and ms.parse_mode == "LLM_EXTRACTED":
-                    unique_labels = list({
-                        d.get("label", "").lower()
-                        for d in gpt_dets if d.get("label")
-                    })
-                    relevant_labels = evaluate_relevance_llm(
-                        unique_labels, ms.operator_text
-                    )
-                    ms.relevance_criteria.required_classes = list(relevant_labels)
-                    _relevance_refined[0] = True  # signal writer_loop to switch to deterministic gate
-                    logging.info(
-                        "Enrichment: LLM post-filter applied on frame %d: relevant=%s",
-                        frame_idx, relevant_labels,
-                    )
-                    # Re-filter with refined classes
-                    for d in gpt_dets:
-                        decision = evaluate_relevance(d, ms.relevance_criteria)
-                        d["mission_relevant"] = decision.relevant
-                    gpt_dets = [d for d in gpt_dets if d.get("mission_relevant", True)]
-
-                # GPT threat assessment
-                if gpt_dets:
-                    # Check for cached results: passed directly or from app-level background task
-                    cached_gpt = first_frame_gpt_results
-                    if not cached_gpt and job_id:
-                        try:
-                            from jobs.storage import get_job_storage as _gjs
-                            _job = _gjs().get(job_id)
-                            if _job and _job.first_frame_gpt_results:
-                                cached_gpt = _job.first_frame_gpt_results
-                        except Exception:
-                            pass
-
-                    if cached_gpt:
-                        logging.info("Enrichment: re-using cached GPT results for frame %d", frame_idx)
-                        gpt_res = cached_gpt
-                    else:
-                        logging.info("Enrichment: running GPT estimation for frame %d...", frame_idx)
-                        frame_b64 = encode_frame_to_b64(frame_data)
-                        gpt_res = estimate_threat_gpt(
-                            detections=gpt_dets, mission_spec=ms,
-                            image_b64=frame_b64,
-                        )
-
-                    # Merge using real track_id assigned by ByteTracker
-                    for d in gpt_dets:
-                        oid = d.get('track_id')
-                        if oid and oid in gpt_res:
-                            gpt_payload = gpt_res[oid]
-                            d.update(gpt_payload)
-                            d["gpt_raw"] = gpt_payload
-                            d["assessment_frame_index"] = frame_idx
-                            d["assessment_status"] = gpt_payload.get(
-                                "assessment_status", "ASSESSED"
-                            )
-
-                    # Push GPT data back into tracker's internal STrack objects
+                gpt_res = run_enrichment(
+                    frame_idx, frame_data, gpt_dets, ms,
+                    first_frame_gpt_results=first_frame_gpt_results,
+                    job_id=job_id,
+                    relevance_refined_event=_relevance_refined,
+                )
+                if gpt_res:
                     tracker_ref.inject_metadata(gpt_dets)
                     logging.info("Enrichment: GPT results injected into tracker for frame %d", frame_idx)
 
@@ -1186,7 +1135,7 @@ def run_inference(
                         # --- RELEVANCE GATE (deterministic, fast — stays in writer) ---
                         if mission_spec:
                             if (mission_spec.parse_mode == "LLM_EXTRACTED"
-                                    and not _relevance_refined[0]):
+                                    and not _relevance_refined.is_set()):
                                 # LLM post-filter hasn't run yet — pass all through
                                 for d in dets:
                                     d["mission_relevant"] = True
@@ -1220,7 +1169,7 @@ def run_inference(
                         if enable_gpt and gpt_dets and not gpt_submitted:
                             # Tag as pending — enrichment thread will update to ASSESSED later
                             for d in gpt_dets:
-                                d["assessment_status"] = "PENDING_GPT"
+                                d["assessment_status"] = AssessmentStatus.PENDING_GPT
                             try:
                                 gpt_enrichment_queue.put(
                                     (next_idx, p_frame.copy(), gpt_dets, mission_spec),
@@ -1234,7 +1183,7 @@ def run_inference(
                         # Tag unassessed detections (INV-6)
                         for d in dets:
                             if "assessment_status" not in d:
-                                d["assessment_status"] = "UNASSESSED"
+                                d["assessment_status"] = AssessmentStatus.UNASSESSED
 
                         # --- RENDER BOXES & OVERLAYS ---
                         if dets:
@@ -2027,7 +1976,7 @@ def run_grounded_sam2_tracking(
         gpt_enrichment_queue: Queue = Queue(maxsize=4)
         gpt_data_by_track: Dict[str, Dict] = {}
         gpt_data_lock = RLock()
-        _relevance_refined = [False]
+        _relevance_refined = Event()
 
         def _gsam2_enrichment_thread_fn():
             while True:
@@ -2036,49 +1985,15 @@ def run_grounded_sam2_tracking(
                     break
                 frame_idx, frame_data, gpt_dets, ms = item
                 try:
-                    # LLM post-filter (LLM_EXTRACTED mode)
-                    if ms and ms.parse_mode == "LLM_EXTRACTED":
-                        unique_labels = list({
-                            d.get("label", "").lower()
-                            for d in gpt_dets if d.get("label")
-                        })
-                        relevant_labels = evaluate_relevance_llm(
-                            unique_labels, ms.operator_text
-                        )
-                        ms.relevance_criteria.required_classes = list(relevant_labels)
-                        _relevance_refined[0] = True
-                        logging.info(
-                            "GSAM2 enrichment: LLM post-filter frame %d: relevant=%s",
-                            frame_idx, relevant_labels,
-                        )
-                        for d in gpt_dets:
-                            decision = evaluate_relevance(d, ms.relevance_criteria)
-                            d["mission_relevant"] = decision.relevant
-                        gpt_dets = [d for d in gpt_dets if d.get("mission_relevant", True)]
-
-                    # GPT threat assessment
-                    if gpt_dets:
-                        cached_gpt = first_frame_gpt_results
-                        if not cached_gpt and job_id:
-                            try:
-                                from jobs.storage import get_job_storage as _gjs
-                                _job = _gjs().get(job_id)
-                                if _job and _job.first_frame_gpt_results:
-                                    cached_gpt = _job.first_frame_gpt_results
-                            except Exception:
-                                pass
-
-                        if cached_gpt:
-                            logging.info("GSAM2 enrichment: re-using cached GPT for frame %d", frame_idx)
-                            gpt_res = cached_gpt
-                        else:
-                            logging.info("GSAM2 enrichment: running GPT for frame %d...", frame_idx)
-                            frame_b64 = encode_frame_to_b64(frame_data)
-                            gpt_res = estimate_threat_gpt(
-                                detections=gpt_dets, mission_spec=ms,
-                                image_b64=frame_b64,
-                            )
+                    gpt_res = run_enrichment(
+                        frame_idx, frame_data, gpt_dets, ms,
+                        first_frame_gpt_results=first_frame_gpt_results,
+                        job_id=job_id,
+                        relevance_refined_event=_relevance_refined,
+                    )
 
+                    # GSAM2-specific: store results in per-track dict and persist to job storage
+                    if gpt_res:
                         for d in gpt_dets:
                             tid = d.get("track_id")
                             if tid and tid in gpt_res:
@@ -2086,7 +2001,7 @@ def run_grounded_sam2_tracking(
                                 merged["gpt_raw"] = gpt_res[tid]
                                 merged["assessment_frame_index"] = frame_idx
                                 merged["assessment_status"] = merged.get(
-                                    "assessment_status", "ASSESSED"
+                                    "assessment_status", AssessmentStatus.ASSESSED
                                 )
                                 with gpt_data_lock:
                                     gpt_data_by_track[tid] = merged
@@ -2096,6 +2011,7 @@ def run_grounded_sam2_tracking(
                         # frontend polling (/detect/status) picks them up.
                         if job_id:
                             try:
+                                from jobs.storage import get_job_storage as _gjs
                                 _st = _gjs().get(job_id)
                                 if _st and _st.first_frame_detections:
                                     for det in _st.first_frame_detections:
@@ -2184,7 +2100,7 @@ def run_grounded_sam2_tracking(
                                 # Relevance gate
                                 if mission_spec:
                                     if (mission_spec.parse_mode == "LLM_EXTRACTED"
-                                            and not _relevance_refined[0]):
+                                            and not _relevance_refined.is_set()):
                                         for d in dets:
                                             d["mission_relevant"] = True
                                             d["relevance_reason"] = "pending_llm_postfilter"
@@ -2203,7 +2119,7 @@ def run_grounded_sam2_tracking(
                                 # GPT enrichment (one-shot, first frame with detections)
                                 if gpt_dets and not gpt_submitted:
                                     for d in gpt_dets:
-                                        d["assessment_status"] = "PENDING_GPT"
+                                        d["assessment_status"] = AssessmentStatus.PENDING_GPT
                                     try:
                                         gpt_enrichment_queue.put(
                                             (
@@ -2226,9 +2142,9 @@ def run_grounded_sam2_tracking(
                                         gpt_payload = gpt_data_by_track.get(tid)
                                     if gpt_payload:
                                         det.update(gpt_payload)
-                                        det["assessment_status"] = "ASSESSED"
+                                        det["assessment_status"] = AssessmentStatus.ASSESSED
                                     elif "assessment_status" not in det:
-                                        det["assessment_status"] = "UNASSESSED"
+                                        det["assessment_status"] = AssessmentStatus.UNASSESSED
 
                                 # Build enriched display labels
                                 display_labels = []

utils/enrichment.py

@@ -0,0 +1,122 @@
+"""
+Shared enrichment workflow — single implementation of the 5-step GPT enrichment
+pipeline used by inference.py (detection + GSAM2) and app.py (first-frame).
+
+Consolidates duplicated logic from:
+  - inference.py enrichment_thread_fn
+  - inference.py _gsam2_enrichment_thread_fn
+  - app.py _enrich_first_frame_gpt
+"""
+
+import logging
+from threading import Event
+from typing import Any, Dict, List, Optional
+
+from utils.gpt_reasoning import estimate_threat_gpt, encode_frame_to_b64
+from utils.relevance import evaluate_relevance, evaluate_relevance_llm
+from utils.schemas import AssessmentStatus
+
+logger = logging.getLogger(__name__)
+
+
+def run_enrichment(
+    frame_idx: int,
+    frame_data,
+    detections: List[Dict[str, Any]],
+    mission_spec,
+    *,
+    first_frame_gpt_results: Optional[Dict] = None,
+    job_id: Optional[str] = None,
+    relevance_refined_event: Optional[Event] = None,
+) -> Optional[Dict[str, Any]]:
+    """Run the shared enrichment workflow (LLM post-filter + GPT threat assessment).
+
+    Steps:
+        1. LLM post-filter via evaluate_relevance_llm() (if LLM_EXTRACTED mode)
+        2. Signal relevance_refined_event (if provided)
+        3. Check cached GPT results (parameter or JobStorage fallback)
+        4. Call estimate_threat_gpt() if no cache
+        5. Merge results into detections by track_id
+
+    Args:
+        frame_idx: Index of the frame being enriched.
+        frame_data: OpenCV BGR frame (numpy array).
+        detections: Mutable list of detection dicts to enrich in-place.
+        mission_spec: Optional MissionSpecification.
+        first_frame_gpt_results: Pre-computed GPT results (cache hit).
+        job_id: Job identifier for JobStorage fallback cache lookup.
+        relevance_refined_event: threading.Event to signal when LLM post-filter completes.
+
+    Returns:
+        GPT results dict (object_id -> assessment), or None if all detections
+        were filtered out.
+    """
+    gpt_dets = detections
+
+    # --- Step 1: LLM post-filter (LLM_EXTRACTED mode) ---
+    if mission_spec and mission_spec.parse_mode == "LLM_EXTRACTED":
+        unique_labels = list({
+            d.get("label", "").lower()
+            for d in gpt_dets if d.get("label")
+        })
+        relevant_labels = evaluate_relevance_llm(
+            unique_labels, mission_spec.operator_text
+        )
+        mission_spec.relevance_criteria.required_classes = list(relevant_labels)
+
+        # --- Step 2: Signal writer loop ---
+        if relevance_refined_event is not None:
+            relevance_refined_event.set()
+
+        logger.info(
+            "Enrichment: LLM post-filter applied on frame %d: relevant=%s",
+            frame_idx, relevant_labels,
+        )
+        # Re-filter with refined classes
+        for d in gpt_dets:
+            decision = evaluate_relevance(d, mission_spec.relevance_criteria)
+            d["mission_relevant"] = decision.relevant
+        gpt_dets = [d for d in gpt_dets if d.get("mission_relevant", True)]
+    elif relevance_refined_event is not None:
+        # Non-LLM mode: signal immediately so writer doesn't block
+        relevance_refined_event.set()
+
+    if not gpt_dets:
+        return None
+
+    # --- Step 3: Check cached GPT results ---
+    cached_gpt = first_frame_gpt_results
+    if not cached_gpt and job_id:
+        try:
+            from jobs.storage import get_job_storage as _gjs
+            _job = _gjs().get(job_id)
+            if _job and _job.first_frame_gpt_results:
+                cached_gpt = _job.first_frame_gpt_results
+        except Exception:
+            pass
+
+    # --- Step 4: Call GPT if no cache ---
+    if cached_gpt:
+        logger.info("Enrichment: re-using cached GPT results for frame %d", frame_idx)
+        gpt_res = cached_gpt
+    else:
+        logger.info("Enrichment: running GPT estimation for frame %d...", frame_idx)
+        frame_b64 = encode_frame_to_b64(frame_data)
+        gpt_res = estimate_threat_gpt(
+            detections=gpt_dets, mission_spec=mission_spec,
+            image_b64=frame_b64,
+        )
+
+    # --- Step 5: Merge results into detections by track_id ---
+    for d in gpt_dets:
+        oid = d.get("track_id")
+        if oid and oid in gpt_res:
+            gpt_payload = gpt_res[oid]
+            d.update(gpt_payload)
+            d["gpt_raw"] = gpt_payload
+            d["assessment_frame_index"] = frame_idx
+            d["assessment_status"] = gpt_payload.get(
+                "assessment_status", AssessmentStatus.ASSESSED
+            )
+
+    return gpt_res

utils/gpt_reasoning.py

@@ -4,9 +4,8 @@ import json
 import base64
 import logging
 from typing import List, Dict, Any, Optional
-import urllib.request
-import urllib.error
-from utils.schemas import FrameThreatAnalysis
+from utils.schemas import FrameThreatAnalysis, AssessmentStatus
+from utils.openai_client import chat_completion, extract_content, get_api_key, OpenAIAPIError
 
 logger = logging.getLogger(__name__)
 
@@ -15,10 +14,16 @@ def encode_image(image_path: str) -> str:
         return base64.b64encode(image_file.read()).decode('utf-8')
 
 
-def encode_frame_to_b64(frame) -> str:
-    """Encode an OpenCV BGR frame to a base64 JPEG string in memory (no disk I/O)."""
+def encode_frame_to_b64(frame, quality=None) -> str:
+    """Encode an OpenCV BGR frame to a base64 JPEG string in memory (no disk I/O).
+
+    Args:
+        frame: OpenCV BGR numpy array.
+        quality: Optional JPEG quality (1-100). Uses OpenCV default if None.
+    """
     import cv2
-    success, buf = cv2.imencode('.jpg', frame)
+    params = [int(cv2.IMWRITE_JPEG_QUALITY), quality] if quality is not None else None
+    success, buf = cv2.imencode('.jpg', frame, params) if params else cv2.imencode('.jpg', frame)
     if not success:
         raise ValueError("Failed to encode frame to JPEG")
     return base64.b64encode(buf.tobytes()).decode('utf-8')
@@ -167,8 +172,7 @@ def estimate_threat_gpt(
     if detections is None:
         detections = []
 
-    api_key = os.environ.get("OPENAI_API_KEY")
-    if not api_key:
+    if not get_api_key():
         logger.error("OPENAI_API_KEY not set. Skipping GPT threat assessment.")
         return {}
 
@@ -200,7 +204,7 @@ def estimate_threat_gpt(
             )
             return _build_status_fallback(
                 skipped_human_ids,
-                "SKIPPED_POLICY",
+                AssessmentStatus.SKIPPED_POLICY,
                 "Human/person analysis skipped due policy constraints.",
             )
         return {}
@@ -264,25 +268,10 @@ def estimate_threat_gpt(
         "response_format": { "type": "json_object" }
     }
 
-    headers = {
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {api_key}"
-    }
-
     try:
-        req = urllib.request.Request(
-            "https://api.openai.com/v1/chat/completions",
-            data=json.dumps(payload).encode('utf-8'),
-            headers=headers,
-            method="POST"
-        )
-        with urllib.request.urlopen(req, timeout=30) as response:
-            resp_data = json.loads(response.read().decode('utf-8'))
-            
-        choice_msg = resp_data.get("choices", [{}])[0].get("message", {})
-        content = choice_msg.get("content")
+        resp_data = chat_completion(payload)
+        content, refusal = extract_content(resp_data)
         if not content:
-            refusal = choice_msg.get("refusal")
             if refusal:
                 logger.warning("GPT refused threat assessment: %s", refusal)
             else:
@@ -293,13 +282,13 @@ def estimate_threat_gpt(
                 )
             fallback = _build_status_fallback(
                 [it["obj_id"] for it in prompt_items],
-                "REFUSED",
+                AssessmentStatus.REFUSED,
                 refusal or "GPT returned empty content.",
             )
             fallback.update(
                 _build_status_fallback(
                     skipped_human_ids,
-                    "SKIPPED_POLICY",
+                    AssessmentStatus.SKIPPED_POLICY,
                     "Human/person analysis skipped due policy constraints.",
                 )
             )
@@ -320,14 +309,14 @@ def estimate_threat_gpt(
             oid = it["obj_id"]
             if oid not in objects:
                 objects[oid] = {
-                    "assessment_status": "NO_RESPONSE",
+                    "assessment_status": AssessmentStatus.NO_RESPONSE,
                     "gpt_reason": "No structured assessment returned for object.",
                 }
         for oid in skipped_human_ids:
             objects.setdefault(
                 oid,
                 {
-                    "assessment_status": "SKIPPED_POLICY",
+                    "assessment_status": AssessmentStatus.SKIPPED_POLICY,
                     "gpt_reason": "Human/person analysis skipped due policy constraints.",
                 },
             )
@@ -336,7 +325,7 @@ def estimate_threat_gpt(
         for obj_id, data in objects.items():
             if not isinstance(data, dict):
                 data = {
-                    "assessment_status": "NO_RESPONSE",
+                    "assessment_status": AssessmentStatus.NO_RESPONSE,
                     "gpt_reason": "Malformed object payload from GPT.",
                 }
                 objects[obj_id] = data
@@ -373,13 +362,13 @@ def estimate_threat_gpt(
         logger.error("GPT API call failed: %s", e, exc_info=True)
         fallback = _build_status_fallback(
             [it["obj_id"] for it in prompt_items],
-            "ERROR",
+            AssessmentStatus.ERROR,
             f"GPT API call failed: {e.__class__.__name__}",
         )
         fallback.update(
             _build_status_fallback(
                 skipped_human_ids,
-                "SKIPPED_POLICY",
+                AssessmentStatus.SKIPPED_POLICY,
                 "Human/person analysis skipped due policy constraints.",
             )
         )

utils/mission_parser.py

@@ -12,15 +12,13 @@ Internal flow:
 6. Return validated MissionSpecification or raise MissionParseError
 """
 
-import base64
 import json
 import logging
-import os
 import re
-import urllib.request
-import urllib.error
 from typing import List, Optional
 
+from utils.openai_client import chat_completion, extract_content, get_api_key, OpenAIAPIError
+
 from coco_classes import COCO_CLASSES, canonicalize_coco_name, coco_class_catalog
 from utils.schemas import MissionSpecification, RelevanceCriteria
 
@@ -209,16 +207,11 @@ def _extract_and_encode_first_frame(video_path: Optional[str]) -> Optional[str]:
     if not video_path:
         return None
     try:
-        import cv2
         from inference import extract_first_frame
+        from utils.gpt_reasoning import encode_frame_to_b64
 
         frame, _fps, _w, _h = extract_first_frame(video_path)
-        success, buf = cv2.imencode(
-            ".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 85]
-        )
-        if not success:
-            return None
-        return base64.b64encode(buf).decode("ascii")
+        return encode_frame_to_b64(frame, quality=85)
     except Exception:
         logger.warning("Failed to extract/encode first frame for vision grounding", exc_info=True)
         return None
@@ -226,8 +219,7 @@ def _extract_and_encode_first_frame(video_path: Optional[str]) -> Optional[str]:
 
 def _call_extraction_llm(raw_text: str, detector_key: str, first_frame_b64: Optional[str] = None) -> dict:
     """Call GPT-4o to extract structured mission fields from natural language."""
-    api_key = os.environ.get("OPENAI_API_KEY")
-    if not api_key:
+    if not get_api_key():
         raise MissionParseError(
             "OPENAI_API_KEY not set. Cannot parse natural language mission text. "
             "Use comma-separated class labels instead (e.g., 'person, car, boat')."
@@ -278,28 +270,15 @@ def _call_extraction_llm(raw_text: str, detector_key: str, first_frame_b64: Opti
         ],
     }
 
-    headers = {
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {api_key}",
-    }
-
     try:
-        req = urllib.request.Request(
-            "https://api.openai.com/v1/chat/completions",
-            data=json.dumps(payload).encode("utf-8"),
-            headers=headers,
-            method="POST",
-        )
-        with urllib.request.urlopen(req, timeout=timeout_s) as response:
-            resp_data = json.loads(response.read().decode("utf-8"))
-
-        content = resp_data["choices"][0]["message"].get("content")
+        resp_data = chat_completion(payload, timeout=timeout_s)
+        content, _refusal = extract_content(resp_data)
         if not content:
             raise MissionParseError("GPT returned empty content during mission parsing.")
 
         return json.loads(content)
 
-    except (urllib.error.HTTPError, urllib.error.URLError) as e:
+    except OpenAIAPIError as e:
         raise MissionParseError(f"Mission parsing API call failed: {e}")
     except json.JSONDecodeError:
         raise MissionParseError(

utils/openai_client.py

@@ -0,0 +1,80 @@
+"""
+Shared OpenAI HTTP client — single implementation of the chat-completions call.
+
+Replaces duplicated urllib boilerplate in gpt_reasoning, relevance,
+mission_parser, and threat_chat.
+"""
+
+import json
+import logging
+import os
+import urllib.request
+import urllib.error
+from typing import Dict, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+_API_URL = "https://api.openai.com/v1/chat/completions"
+
+
+class OpenAIAPIError(Exception):
+    """Raised when the OpenAI API call fails (HTTP or network error)."""
+
+    def __init__(self, message: str, status_code: Optional[int] = None):
+        self.status_code = status_code
+        super().__init__(message)
+
+
+def get_api_key() -> Optional[str]:
+    """Return the OpenAI API key from the environment, or None."""
+    return os.environ.get("OPENAI_API_KEY")
+
+
+def chat_completion(payload: Dict, *, timeout: int = 30) -> Dict:
+    """Send a chat-completion request and return the parsed JSON response.
+
+    Args:
+        payload: Full request body (model, messages, etc.).
+        timeout: HTTP timeout in seconds.
+
+    Returns:
+        Parsed response dict.
+
+    Raises:
+        OpenAIAPIError: On HTTP or network failure.
+    """
+    api_key = get_api_key()
+    if not api_key:
+        raise OpenAIAPIError("OPENAI_API_KEY not set")
+
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}",
+    }
+
+    try:
+        req = urllib.request.Request(
+            _API_URL,
+            data=json.dumps(payload).encode("utf-8"),
+            headers=headers,
+            method="POST",
+        )
+        with urllib.request.urlopen(req, timeout=timeout) as response:
+            return json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as e:
+        raise OpenAIAPIError(
+            f"HTTP {e.code}: {e.reason}", status_code=e.code
+        ) from e
+    except urllib.error.URLError as e:
+        raise OpenAIAPIError(f"URL error: {e.reason}") from e
+
+
+def extract_content(resp_data: Dict) -> Tuple[Optional[str], Optional[str]]:
+    """Safely extract content and refusal from a chat-completion response.
+
+    Returns:
+        (content, refusal) — either may be None.
+    """
+    choice = resp_data.get("choices", [{}])[0]
+    message = choice.get("message", {})
+    return message.get("content"), message.get("refusal")

utils/relevance.py

@@ -12,11 +12,10 @@ LLM-derived field. This is structural, not by convention.
 
 import json
 import logging
-import os
-import urllib.request
-import urllib.error
 from typing import Any, Dict, List, NamedTuple, Set
 
+from utils.openai_client import chat_completion, extract_content, get_api_key, OpenAIAPIError
+
 from coco_classes import canonicalize_coco_name
 from utils.schemas import RelevanceCriteria
 
@@ -91,8 +90,7 @@ def evaluate_relevance_llm(
     if not detected_labels:
         return set()
 
-    api_key = os.environ.get("OPENAI_API_KEY")
-    if not api_key:
+    if not get_api_key():
         logger.warning(
             "OPENAI_API_KEY not set — LLM relevance filter falling back to accept-all"
         )
@@ -118,22 +116,9 @@ def evaluate_relevance_llm(
         ],
     }
 
-    headers = {
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {api_key}",
-    }
-
     try:
-        req = urllib.request.Request(
-            "https://api.openai.com/v1/chat/completions",
-            data=json.dumps(payload).encode("utf-8"),
-            headers=headers,
-            method="POST",
-        )
-        with urllib.request.urlopen(req, timeout=30) as response:
-            resp_data = json.loads(response.read().decode("utf-8"))
-
-        content = resp_data["choices"][0]["message"].get("content")
+        resp_data = chat_completion(payload)
+        content, _refusal = extract_content(resp_data)
         if not content:
             logger.warning("GPT returned empty content for relevance filter — accept-all")
             return set(detected_labels)
@@ -148,7 +133,7 @@ def evaluate_relevance_llm(
         )
         return relevant_set
 
-    except (urllib.error.HTTPError, urllib.error.URLError) as e:
+    except OpenAIAPIError as e:
         logger.warning("LLM relevance API call failed: %s — accept-all fallback", e)
         return set(detected_labels)
     except (json.JSONDecodeError, KeyError, TypeError) as e:

utils/schemas.py

@@ -146,3 +146,15 @@ class MissionSpecification(BaseModel):
                     "E.g., 'term \"threat\" is not a visual class, stripped'."
     )
 
+
+class AssessmentStatus:
+    """Canonical string constants for detection assessment lifecycle."""
+    ASSESSED = "ASSESSED"
+    UNASSESSED = "UNASSESSED"
+    PENDING_GPT = "PENDING_GPT"
+    SKIPPED_POLICY = "SKIPPED_POLICY"
+    REFUSED = "REFUSED"
+    ERROR = "ERROR"
+    NO_RESPONSE = "NO_RESPONSE"
+    STALE = "STALE"
+

utils/tracker.py

@@ -3,6 +3,8 @@ import numpy as np
 from scipy.optimize import linear_sum_assignment
 import scipy.linalg
 
+from utils.schemas import AssessmentStatus
+
 
 class KalmanFilter:
     """
@@ -574,11 +576,11 @@ class ByteTracker:
             if assessment_frame is not None:
                 frames_since = self.frame_id - assessment_frame
                 if frames_since > MAX_STALE_FRAMES:
-                    d_out['assessment_status'] = 'STALE'
+                    d_out['assessment_status'] = AssessmentStatus.STALE
                     d_out['assessment_age_frames'] = frames_since
-            elif d_out.get('assessment_status') != 'ASSESSED':
+            elif d_out.get('assessment_status') != AssessmentStatus.ASSESSED:
                 # INV-6: Unassessed objects get explicit UNASSESSED status
-                d_out['assessment_status'] = 'UNASSESSED'
+                d_out['assessment_status'] = AssessmentStatus.UNASSESSED
 
             # Update history
             if 'history' not in track.gpt_data:
@@ -634,7 +636,7 @@ class ByteTracker:
                     k in meta for k in ("threat_level_score", "gpt_raw", "object_type")
                 ):
                     meta["assessment_frame_index"] = self.frame_id
-                    meta["assessment_status"] = "ASSESSED"
+                    meta["assessment_status"] = AssessmentStatus.ASSESSED
                 meta_by_tid[tid] = meta
         for track in self.tracked_stracks:
             tid_str = f"T{str(track.track_id).zfill(2)}"