feat: Add mission-driven object relevance abstractions

Introduces MissionSpecification extraction, object relevance gating,
domain-aware GPT prompts, assessment provenance/staleness tracking,
and UNASSESSED status distinct from score 0. Mission text is now
parsed into structured intent before reaching the detector, and only
mission-relevant objects are sent to GPT for threat assessment.

Key changes:
- MissionSpecification + RelevanceCriteria schemas (utils/schemas.py)
- Mission text parser with fast-path and LLM extraction (utils/mission_parser.py)
- Deterministic relevance gate between detection and GPT (utils/relevance.py)
- Domain-aware GPT system prompts with mission context injection
- Temporal validity tracking (ASSESSED/UNASSESSED/STALE) in tracker
- LEGACY mode when no mission text provided (GPT auto-disabled)
- Frontend: deterministic sort, UNASSESSED/STALE badges, range qualifiers

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app.py

@@ -57,6 +57,7 @@ from jobs.storage import (
 )
 from utils.gpt_reasoning import estimate_threat_gpt
 from utils.threat_chat import chat_about_threats
+from utils.mission_parser import parse_mission_text, MissionParseError
 
 logging.basicConfig(level=logging.INFO)
 
@@ -266,14 +267,24 @@ async def detect_endpoint(
     fd, output_path = tempfile.mkstemp(prefix="output_", suffix=".mp4", dir="/tmp")
     os.close(fd)
 
-    # Parse queries
-    query_list = [q.strip() for q in queries.split(",") if q.strip()]
+    # Parse queries with mission awareness
+    detector_name = "drone_yolo" if mode == "drone_detection" else detector
+    mission_spec = None
+
+    if queries.strip():
+        try:
+            mission_spec = parse_mission_text(queries.strip(), detector_name)
+            query_list = mission_spec.object_classes
+        except MissionParseError as e:
+            raise HTTPException(status_code=422, detail=str(e))
+    else:
+        query_list = _default_queries_for_mode(mode)
+
     if mode == "drone_detection" and not query_list:
         query_list = ["drone"]
 
     # Run inference
     try:
-        detector_name = "drone_yolo" if mode == "drone_detection" else detector
         
         # Determine depth estimator
         active_depth = "depth" if enable_depth else None
@@ -348,9 +359,36 @@ async def detect_async_endpoint(
     finally:
         await video.close()
 
-    query_list = [q.strip() for q in queries.split(",") if q.strip()]
-    if not query_list:
+    # --- Mission-Driven Query Parsing ---
+    mission_spec = None
+    mission_mode = "LEGACY"
+
+    detector_name = detector
+    if mode == "drone_detection":
+        detector_name = "drone_yolo"
+
+    if queries.strip():
+        try:
+            mission_spec = parse_mission_text(queries.strip(), detector_name)
+            query_list = mission_spec.object_classes
+            mission_mode = "MISSION"
+            logging.info(
+                "Mission parsed: mode=%s classes=%s domain=%s(%s)",
+                mission_mode, query_list, mission_spec.domain, mission_spec.domain_source,
+            )
+        except MissionParseError as e:
+            raise HTTPException(
+                status_code=422,
+                detail=str(e),
+            )
+    else:
+        # LEGACY mode: no mission context, use defaults, disable GPT
         query_list = _default_queries_for_mode(mode)
+        enable_gpt = False
+        mission_mode = "LEGACY"
+        logging.info(
+            "LEGACY mode: no mission text, defaults=%s, GPT disabled", query_list
+        )
 
     available_depth_estimators = set(list_depth_estimators())
     if depth_estimator not in available_depth_estimators:
@@ -362,11 +400,7 @@ async def detect_async_endpoint(
             ),
         )
 
-    detector_name = detector
-    if mode == "drone_detection":
-        detector_name = "drone_yolo"
-        
-    # Determine actve depth estimator (Legacy)
+    # Determine active depth estimator (Legacy)
     active_depth = depth_estimator if enable_depth else None
 
     try:
@@ -380,6 +414,7 @@ async def detect_async_endpoint(
             depth_scale=depth_scale,
             enable_depth_estimator=enable_depth,
             enable_gpt=enable_gpt,
+            mission_spec=mission_spec,
         )
         cv2.imwrite(str(first_frame_path), processed_frame)
         
@@ -417,17 +452,13 @@ async def detect_async_endpoint(
         depth_output_path=str(depth_output_path),
         first_frame_depth_path=str(first_frame_depth_path),
         enable_gpt=enable_gpt,
+        mission_spec=mission_spec,
+        mission_mode=mission_mode,
     )
     get_job_storage().create(job)
     asyncio.create_task(process_video_async(job_id))
 
-    return {
-        "job_id": job_id,
-        "first_frame_url": f"/detect/first-frame/{job_id}",
-        "first_frame_depth_url": f"/detect/first-frame-depth/{job_id}",
-        "status_url": f"/detect/status/{job_id}",
-        "video_url": f"/detect/video/{job_id}",
-        "depth_video_url": f"/detect/depth-video/{job_id}",
+    response_data = {
         "job_id": job_id,
         "first_frame_url": f"/detect/first-frame/{job_id}",
         "first_frame_depth_url": f"/detect/first-frame-depth/{job_id}",
@@ -437,8 +468,23 @@ async def detect_async_endpoint(
         "stream_url": f"/detect/stream/{job_id}",
         "status": job.status.value,
         "first_frame_detections": detections,
+        "mission_mode": mission_mode,
     }
 
+    if mission_spec:
+        response_data["mission_spec"] = {
+            "object_classes": mission_spec.object_classes,
+            "mission_intent": mission_spec.mission_intent,
+            "domain": mission_spec.domain,
+            "domain_source": mission_spec.domain_source,
+            "parse_confidence": mission_spec.parse_confidence,
+            "parse_warnings": mission_spec.parse_warnings,
+            "context_phrases": mission_spec.context_phrases,
+            "stripped_modifiers": mission_spec.stripped_modifiers,
+        }
+
+    return response_data
+
 
 @app.get("/detect/status/{job_id}")
 async def detect_status(job_id: str):
@@ -692,34 +738,46 @@ async def reason_track(
 @app.post("/chat/threat")
 async def chat_threat_endpoint(
     question: str = Form(...),
-    detections: str = Form(...)  # JSON string of current detections
+    detections: str = Form(...),  # JSON string of current detections
+    mission_context: str = Form(""),  # Optional JSON string of mission spec
 ):
     """
     Chat about detected threats using GPT.
-    
+
     Args:
         question: User's question about the current threat situation.
         detections: JSON string of detection list with threat analysis data.
-        
+        mission_context: Optional JSON string of mission specification.
+
     Returns:
         GPT response about the threats.
     """
     import json as json_module
-    
+
     if not question.strip():
         raise HTTPException(status_code=400, detail="Question cannot be empty.")
-    
+
     try:
         detection_list = json_module.loads(detections)
     except json_module.JSONDecodeError:
         raise HTTPException(status_code=400, detail="Invalid detections JSON.")
-    
+
     if not isinstance(detection_list, list):
         raise HTTPException(status_code=400, detail="Detections must be a list.")
-    
+
+    # Parse optional mission context
+    mission_spec_dict = None
+    if mission_context.strip():
+        try:
+            mission_spec_dict = json_module.loads(mission_context)
+        except json_module.JSONDecodeError:
+            pass  # Non-critical, proceed without mission context
+
     # Run chat in thread to avoid blocking
     try:
-        response = await asyncio.to_thread(chat_about_threats, question, detection_list)
+        response = await asyncio.to_thread(
+            chat_about_threats, question, detection_list, mission_spec_dict
+        )
         return {"response": response}
     except Exception as e:
         logging.exception("Threat chat failed")

frontend/js/core/tracker.js

@@ -193,13 +193,32 @@ APP.core.tracker.syncWithBackend = async function (frameIdx) {
                 score: d.score,
                 angle_deg: d.angle_deg,
                 gpt_distance_m: d.gpt_distance_m,
-                angle_deg: d.angle_deg,
-                gpt_distance_m: d.gpt_distance_m,
+                gpt_direction: d.gpt_direction,
+                gpt_description: d.gpt_description,
                 speed_kph: d.speed_kph,
                 depth_est_m: d.depth_est_m,
                 depth_rel: d.depth_rel,
                 depth_valid: d.depth_valid,
-
+                // Threat intelligence
+                threat_level_score: d.threat_level_score || 0,
+                threat_classification: d.threat_classification || "Unknown",
+                weapon_readiness: d.weapon_readiness || "Unknown",
+                // Mission relevance and assessment status
+                mission_relevant: d.mission_relevant ?? null,
+                relevance_reason: d.relevance_reason || null,
+                assessment_status: d.assessment_status || "UNASSESSED",
+                assessment_frame_index: d.assessment_frame_index ?? null,
+                // GPT raw data for feature table
+                gpt_raw: d.gpt_raw || null,
+                features: d.gpt_raw ? {
+                    "Vessel Class": d.gpt_raw.specific_class || d.gpt_raw.vessel_category || "Unknown",
+                    "Threat Lvl": d.gpt_raw.threat_level_score + "/10",
+                    "Status": d.gpt_raw.threat_classification || "?",
+                    "Weapons": (d.gpt_raw.visible_weapons || []).join(", ") || "None Visible",
+                    "Readiness": d.gpt_raw.weapon_readiness || "Unknown",
+                    "Motion": d.gpt_raw.motion_status || "Unknown",
+                    "Range": d.gpt_raw.range_estimation_nm ? "~" + d.gpt_raw.range_estimation_nm + " NM (est.)" : "Unknown",
+                } : {},
                 // Keep UI state fields
                 lastSeen: Date.now(),
                 state: "TRACK"

frontend/js/main.js

@@ -505,13 +505,17 @@ document.addEventListener("DOMContentLoaded", () => {
                 ? { x: d.bbox[0], y: d.bbox[1], w: d.bbox[2] - d.bbox[0], h: d.bbox[3] - d.bbox[1] }
                 : { x: 0, y: 0, w: 10, h: 10 };
 
+            // Range display: qualify GPT-estimated distances (INV-10)
+            const rangeDisplay = d.gpt_raw && d.gpt_raw.range_estimation_nm
+                ? "~" + d.gpt_raw.range_estimation_nm + " NM (est.)"
+                : "Unknown";
+
             return {
                 id,
                 label: d.label || d.class,
                 score: d.score || 0.5,
                 bbox,
                 aim: { ...ap },
-                aim: { ...ap },
                 features: d.gpt_raw ? {
                     "Vessel Class": d.gpt_raw.specific_class || d.gpt_raw.vessel_category || "Unknown",
                     "Threat Lvl": d.gpt_raw.threat_level_score + "/10",
@@ -522,7 +526,7 @@ document.addEventListener("DOMContentLoaded", () => {
                     "Sensors": (d.gpt_raw.sensor_profile || []).join(", ") || "None",
                     "Flags/ID": (d.gpt_raw.identity_markers || []).join(", ") || (d.gpt_raw.flag_state || "Unknown"),
                     "Activity": d.gpt_raw.deck_activity || "None",
-                    "Range": (d.gpt_raw.range_estimation_nm ? d.gpt_raw.range_estimation_nm + " NM" : "Unknown"),
+                    "Range": rangeDisplay,
                     "Wake": d.gpt_raw.wake_description || "None"
                 } : {},
                 baseRange_m: null,
@@ -531,17 +535,22 @@ document.addEventListener("DOMContentLoaded", () => {
                 reqP_kW: 40,
                 maxP_kW: 0,
                 pkill: 0,
-                // New depth fields
+                // Depth fields
                 depth_est_m: (d.depth_est_m !== undefined && d.depth_est_m !== null) ? d.depth_est_m : null,
                 depth_rel: (d.depth_rel !== undefined && d.depth_rel !== null) ? d.depth_rel : null,
                 depth_valid: d.depth_valid ?? false,
                 gpt_distance_m: d.gpt_distance_m || null,
                 gpt_direction: d.gpt_direction || null,
                 gpt_description: d.gpt_description || null,
-                // New Threat Intelligence
+                // Threat Intelligence
                 threat_level_score: d.threat_level_score || 0,
                 threat_classification: d.threat_classification || "Unknown",
-                weapon_readiness: d.weapon_readiness || "Unknown"
+                weapon_readiness: d.weapon_readiness || "Unknown",
+                // Mission relevance and assessment status
+                mission_relevant: d.mission_relevant ?? null,
+                relevance_reason: d.relevance_reason || null,
+                assessment_status: d.assessment_status || "UNASSESSED",
+                assessment_frame_index: d.assessment_frame_index ?? null,
             };
         });
 

frontend/js/ui/cards.js

@@ -9,7 +9,14 @@ APP.ui.cards.renderFrameTrackList = function () {
     if (!frameTrackList) return;
     frameTrackList.innerHTML = "";
 
-    const dets = state.detections || [];
+    // Filter: only show mission-relevant detections (or all in LEGACY mode)
+    const dets = (state.detections || []).filter(d => {
+        // LEGACY mode: mission_relevant is null -> show all
+        if (d.mission_relevant === null || d.mission_relevant === undefined) return true;
+        // MISSION mode: only show relevant
+        return d.mission_relevant === true;
+    });
+
     if (trackCount) trackCount.textContent = dets.length;
 
     if (dets.length === 0) {
@@ -17,7 +24,18 @@ APP.ui.cards.renderFrameTrackList = function () {
         return;
     }
 
-    const sorted = [...dets].sort((a, b) => (b.threat_level_score || 0) - (a.threat_level_score || 0));
+    // Deterministic sort: ASSESSED first (by threat score), then UNASSESSED, then STALE
+    // Within each group, sort by threat_level_score descending, then by confidence
+    const statusOrder = { "ASSESSED": 0, "UNASSESSED": 1, "STALE": 2 };
+    const sorted = [...dets].sort((a, b) => {
+        const statusA = statusOrder[a.assessment_status] ?? 1;
+        const statusB = statusOrder[b.assessment_status] ?? 1;
+        if (statusA !== statusB) return statusA - statusB;
+        const scoreA = a.threat_level_score || 0;
+        const scoreB = b.threat_level_score || 0;
+        if (scoreB !== scoreA) return scoreB - scoreA;
+        return (b.score || 0) - (a.score || 0);
+    });
 
     sorted.forEach((det, i) => {
         const id = det.id || `T${String(i + 1).padStart(2, '0')}`;
@@ -28,7 +46,7 @@ APP.ui.cards.renderFrameTrackList = function () {
         if (det.depth_valid && det.depth_est_m != null) {
             rangeStr = `${Math.round(det.depth_est_m)}m (Depth)`;
         } else if (det.gpt_distance_m) {
-            rangeStr = `${det.gpt_distance_m}m (GPT)`;
+            rangeStr = `~${det.gpt_distance_m}m (est.)`;
         } else if (det.baseRange_m) {
             rangeStr = `${Math.round(det.baseRange_m)}m (Area)`;
         }
@@ -51,11 +69,22 @@ APP.ui.cards.renderFrameTrackList = function () {
             ? `<div class="track-card-body"><span class="gpt-text">${det.gpt_description}</span></div>`
             : "";
 
+        // Assessment status badge (INV-6: UNASSESSED distinct from score 0)
+        let statusBadge = "";
+        const assessStatus = det.assessment_status || "UNASSESSED";
+        if (assessStatus === "UNASSESSED") {
+            statusBadge = '<span class="badgemini" style="background:#6c757d; color:white">UNASSESSED</span>';
+        } else if (assessStatus === "STALE") {
+            statusBadge = '<span class="badgemini" style="background:#ffc107; color:#333">STALE</span>';
+        } else if (det.threat_level_score > 0) {
+            statusBadge = `<span class="badgemini" style="background:${det.threat_level_score >= 8 ? '#ff4d4d' : '#ff9f43'}; color:white">T-${det.threat_level_score}</span>`;
+        }
+
         card.innerHTML = `
             <div class="track-card-header">
                 <span>${id} · ${det.label}</span>
                 <div style="display:flex; gap:4px">
-                    ${det.threat_level_score > 0 ? `<span class="badgemini" style="background:${det.threat_level_score >= 8 ? '#ff4d4d' : '#ff9f43'}; color:white">T-${det.threat_level_score}</span>` : ''}
+                    ${statusBadge}
                     <span class="badgemini">${(det.score * 100).toFixed(0)}%</span>
                 </div>
             </div>

inference.py

@@ -23,8 +23,10 @@ from models.depth_estimators.model_loader import load_depth_estimator, load_dept
 from models.depth_estimators.base import DepthEstimator
 from utils.video import extract_frames, write_video, VideoReader, VideoWriter, AsyncVideoReader
 from utils.gpt_reasoning import estimate_threat_gpt
+from utils.relevance import evaluate_relevance
 from jobs.storage import set_track_data
 import tempfile
+import json as json_module
 
 
 class AsyncVideoReader:
@@ -715,6 +717,7 @@ def process_first_frame(
     depth_scale: Optional[float] = None,
     enable_depth_estimator: bool = False,
     enable_gpt: bool = True,  # ENABLED BY DEFAULT
+    mission_spec=None,  # Optional[MissionSpecification]
 ) -> Tuple[np.ndarray, List[Dict[str, Any]], Optional[np.ndarray]]:
     frame, _, _, _ = extract_first_frame(video_path)
     if mode == "segmentation":
@@ -722,34 +725,61 @@ def process_first_frame(
             frame, text_queries=queries, segmenter_name=segmenter_name
         )
         return processed, [], None
-    
+
     processed, detections = infer_frame(
         frame, queries, detector_name=detector_name
     )
 
+    # --- RELEVANCE GATE (between detection and GPT) ---
+    if mission_spec:
+        relevant_dets = []
+        for det in detections:
+            decision = evaluate_relevance(det, mission_spec.relevance_criteria)
+            det["mission_relevant"] = decision.relevant
+            det["relevance_reason"] = decision.reason
+            if decision.relevant:
+                relevant_dets.append(det)
+            else:
+                logging.info(
+                    json_module.dumps({
+                        "event": "relevance_decision",
+                        "label": det.get("label"),
+                        "relevant": False,
+                        "reason": decision.reason,
+                        "required_classes": mission_spec.relevance_criteria.required_classes,
+                        "frame": 0,
+                    })
+                )
+        gpt_input_dets = relevant_dets
+    else:
+        # LEGACY mode: all detections pass, tagged as unresolved
+        for det in detections:
+            det["mission_relevant"] = None
+        gpt_input_dets = detections
+
     # 1. Synchronous Depth Estimation (HF Backend)
     depth_map = None
     # If a specific depth estimator is requested OR if generic "enable" flag is on
     should_run_depth = (depth_estimator_name is not None) or enable_depth_estimator
-    
+
     if should_run_depth and detections:
         try:
             # Resolve name: if none given, default to "depth"
             d_name = depth_estimator_name if depth_estimator_name else "depth"
             scale = depth_scale if depth_scale is not None else 1.0
-            
+
             logging.info(f"Running synchronous depth estimation with {d_name} (scale={scale})...")
             estimator = load_depth_estimator(d_name)
-            
+
             # Run prediction
             with _get_model_lock("depth", estimator.name):
                  result = estimator.predict(frame)
-            
+
             depth_map = result.depth_map
-            
+
             # Compute per-detection depth metrics
             detections = compute_depth_per_detection(depth_map, detections, scale)
-            
+
         except Exception as e:
             logging.exception(f"First frame depth failed: {e}")
             # Mark all detections as depth_valid=False just in case
@@ -759,40 +789,41 @@ def process_first_frame(
                 det["depth_valid"] = False
 
     # 2. GPT-based Distance/Direction Estimation (Explicitly enabled)
-    if enable_gpt:
-        # We need to save the frame temporarily to pass to GPT (or refactor gpt_reasoning to take buffer)
-        # For now, write to temp file
+    # Only assess mission-relevant detections
+    if enable_gpt and gpt_input_dets:
         try:
             with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp_img:
                 cv2.imwrite(tmp_img.name, frame)
-                gpt_results = estimate_threat_gpt(tmp_img.name, detections)
-                logging.info(f"GPT Output for First Frame:\n{gpt_results}") # Expose to HF logs
-                os.remove(tmp_img.name) # Clean up immediatey
+                gpt_results = estimate_threat_gpt(
+                    tmp_img.name, gpt_input_dets, mission_spec=mission_spec
+                )
+                logging.info(f"GPT Output for First Frame:\n{gpt_results}")
+                os.remove(tmp_img.name)
 
             # Merge GPT results into detections
-            # GPT returns { "T01": { "distance_m": ..., "direction": ... } }
-            # Detections are list of dicts. We assume T01 maps to index 0, T02 to index 1...
-            for i, det in enumerate(detections):
-                 # Index-based IDs are intentional here: no tracker runs for first-frame
-                 # preview, so GPT, inference merge, and frontend all use the same
-                 # index-based scheme (T01=index 0, T02=index 1, ...), keeping it
-                 # self-consistent. The video pipeline uses real ByteTracker IDs instead.
+            for i, det in enumerate(gpt_input_dets):
                  obj_id = f"T{str(i+1).zfill(2)}"
                  if obj_id in gpt_results:
                      info = gpt_results[obj_id]
                      det["gpt_distance_m"] = info.get("distance_m")
                      det["gpt_direction"] = info.get("direction")
                      det["gpt_description"] = info.get("description")
-                     # Threat Intelligence
                      det["threat_level_score"] = info.get("threat_level_score")
                      det["threat_classification"] = info.get("threat_classification")
                      det["weapon_readiness"] = info.get("weapon_readiness")
-                     # Full Metadata for Feature Table
                      det["gpt_raw"] = info
-                     
+                     # Provenance: tag assessment frame
+                     det["assessment_frame_index"] = 0
+                     det["assessment_status"] = "ASSESSED"
+
         except Exception as e:
             logging.error(f"GPT Threat estimation failed: {e}")
 
+    # Tag unassessed detections (INV-6: distinct from score 0)
+    for det in detections:
+        if "assessment_status" not in det:
+            det["assessment_status"] = "UNASSESSED"
+
     return processed, detections, depth_map
 
 
@@ -807,6 +838,7 @@ def run_inference(
     depth_scale: float = 1.0,
     enable_gpt: bool = True,
     stream_queue: Optional[Queue] = None,
+    mission_spec=None,  # Optional[MissionSpecification]
 ) -> Tuple[str, List[List[Dict[str, Any]]]]:
     
     # 1. Setup Video Reader
@@ -1115,27 +1147,59 @@ def run_inference(
                         dets = tracker.update(dets)
                         speed_est.estimate(dets)
 
+                        # --- RELEVANCE GATE ---
+                        if mission_spec:
+                            for d in dets:
+                                decision = evaluate_relevance(d, mission_spec.relevance_criteria)
+                                d["mission_relevant"] = decision.relevant
+                                d["relevance_reason"] = decision.reason
+                                if not decision.relevant:
+                                    logging.info(
+                                        json_module.dumps({
+                                            "event": "relevance_decision",
+                                            "track_id": d.get("track_id"),
+                                            "label": d.get("label"),
+                                            "relevant": False,
+                                            "reason": decision.reason,
+                                            "required_classes": mission_spec.relevance_criteria.required_classes,
+                                            "frame": next_idx,
+                                        })
+                                    )
+                            gpt_dets = [d for d in dets if d.get("mission_relevant", True)]
+                        else:
+                            for d in dets:
+                                d["mission_relevant"] = None
+                            gpt_dets = dets
+
                         # --- GPT ESTIMATION (Frame 0 Only) ---
-                        if next_idx == 0 and enable_gpt and dets:
+                        if next_idx == 0 and enable_gpt and gpt_dets:
                             try:
                                 logging.info("Running GPT estimation for video start (Frame 0)...")
                                 with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
                                     cv2.imwrite(tmp.name, p_frame)
-                                    gpt_res = estimate_threat_gpt(tmp.name, dets)
+                                    gpt_res = estimate_threat_gpt(
+                                        tmp.name, gpt_dets, mission_spec=mission_spec
+                                    )
                                     os.remove(tmp.name)
 
                                     # Merge using real track_id assigned by ByteTracker
-                                    for d in dets:
+                                    for d in gpt_dets:
                                         oid = d.get('track_id')
                                         if oid and oid in gpt_res:
                                             d.update(gpt_res[oid])
+                                            d["assessment_frame_index"] = 0
+                                            d["assessment_status"] = "ASSESSED"
 
                                 # Push GPT data back into tracker's internal STrack objects
-                                # so it persists across subsequent frames via _sync_data
-                                tracker.inject_metadata(dets)
+                                tracker.inject_metadata(gpt_dets)
 
                             except Exception as e:
                                 logging.error("GPT failed for Frame 0: %s", e)
+
+                        # Tag unassessed detections (INV-6)
+                        for d in dets:
+                            if "assessment_status" not in d:
+                                d["assessment_status"] = "UNASSESSED"
                         
                         # --- RENDER BOXES & OVERLAYS ---
                         # We need to convert list of dicts back to boxes array for draw_boxes

jobs/background.py

@@ -52,6 +52,7 @@ async def process_video_async(job_id: str) -> None:
                 job.depth_scale,
                 job.enable_gpt,
                 stream_queue,
+                job.mission_spec,  # Forward mission spec for relevance gating
             )
             detection_path, detections_list = result_pkg
 

jobs/models.py

@@ -34,3 +34,6 @@ class JobInfo:
     partial_success: bool = False  # True if one component failed but job completed
     depth_error: Optional[str] = None  # Error message if depth failed
     enable_gpt: bool = True  # Whether to use GPT for distance estimation
+    # Mission specification (None = LEGACY mode)
+    mission_spec: Optional[Any] = None  # utils.schemas.MissionSpecification
+    mission_mode: str = "LEGACY"  # "MISSION" or "LEGACY"

utils/gpt_reasoning.py

@@ -13,19 +13,112 @@ def encode_image(image_path: str) -> str:
     with open(image_path, "rb") as image_file:
         return base64.b64encode(image_file.read()).decode('utf-8')
 
+def _build_domain_system_prompt(domain: str, mission_spec=None) -> str:
+    """Select domain-appropriate system prompt based on MissionSpecification."""
+
+    # Mission context block (injected regardless of domain)
+    mission_context = ""
+    if mission_spec:
+        mission_context = (
+            "\n\nMISSION CONTEXT:\n"
+            f"- Operator Intent: {mission_spec.mission_intent}\n"
+            f"- Domain: {mission_spec.domain}\n"
+            f"- Target Classes: {', '.join(mission_spec.object_classes)}\n"
+        )
+        if mission_spec.context_phrases:
+            mission_context += f"- Situational Context: {'; '.join(mission_spec.context_phrases)}\n"
+        if mission_spec.stripped_modifiers:
+            mission_context += f"- Operator Modifiers (stripped): {', '.join(mission_spec.stripped_modifiers)}\n"
+        mission_context += (
+            "\nUse the mission context to inform your analysis. "
+            "Focus assessment on the target classes and domain specified."
+        )
+
+    if domain == "NAVAL":
+        return (
+            "You are an elite Naval Intelligence Officer and Threat Analyst. "
+            "Your task is to analyze optical surveillance imagery and provide a detailed tactical assessment for every detected object. "
+            "You must output a STRICT JSON object that matches the following schema for every object ID provided:\n\n"
+            "RESPONSE SCHEMA (JSON):\n"
+            "{\n"
+            "  \"objects\": {\n"
+            "    \"T01\": {\n"
+            "      \"vessel_category\": \"Warship\" | \"Commercial\" | \"Fishing\" | \"Small Boat\" | \"Aircraft\" | \"Unknown\",\n"
+            "      \"specific_class\": \"string (e.g., Arleigh Burke, Skiff)\",\n"
+            "      \"identity_markers\": [\"string (hull numbers, flags)\"],\n"
+            "      \"flag_state\": \"string (Country)\",\n"
+            "      \"visible_weapons\": [\"string\"],\n"
+            "      \"weapon_readiness\": \"Stowed/PEACE\" | \"Trained/Aiming\" | \"Firing/HOSTILE\",\n"
+            "      \"sensor_profile\": [\"string (radars)\"],\n"
+            "      \"motion_status\": \"Dead in Water\" | \"Underway Slow\" | \"Underway Fast\" | \"Flank Speed\",\n"
+            "      \"wake_description\": \"string\",\n"
+            "      \"aspect\": \"Bow-on\" | \"Stern-on\" | \"Broadside\",\n"
+            "      \"range_estimation_nm\": float (Nautical Miles),\n"
+            "      \"bearing_clock\": \"string (e.g. 12 o'clock)\",\n"
+            "      \"deck_activity\": \"string\",\n"
+            "      \"special_features\": [\"string (anomalies)\"],\n"
+            "      \"threat_level_score\": int (1-10),\n"
+            "      \"threat_classification\": \"Friendly\" | \"Neutral\" | \"Suspect\" | \"Hostile\",\n"
+            "      \"tactical_intent\": \"string (e.g., Transit, Attack)\"\n"
+            "    }\n"
+            "  }\n"
+            "}\n\n"
+            "ASSUMPTIONS:\n"
+            "- Unknown small boats approaching larger vessels are HIGH threat (Suspect/Hostile).\n"
+            "- Visible trained weapons are IMMINENT threat (Score 9-10).\n"
+            "- Ignore artifacts, focus on the objects."
+            + mission_context
+        )
+
+    # Generic / non-naval domains use a simplified schema
+    return (
+        f"You are a surveillance analyst specializing in the {domain} domain. "
+        "Your task is to analyze optical surveillance imagery and provide a tactical assessment for every detected object. "
+        "You must output a STRICT JSON object that matches the following schema for every object ID provided:\n\n"
+        "RESPONSE SCHEMA (JSON):\n"
+        "{\n"
+        "  \"objects\": {\n"
+        "    \"T01\": {\n"
+        "      \"vessel_category\": \"string (object category)\",\n"
+        "      \"specific_class\": \"string (specific type if identifiable)\",\n"
+        "      \"identity_markers\": [\"string (visible identifiers)\"],\n"
+        "      \"flag_state\": \"string (origin if identifiable)\",\n"
+        "      \"visible_weapons\": [\"string\"],\n"
+        "      \"weapon_readiness\": \"Stowed/PEACE\" | \"Trained/Aiming\" | \"Firing/HOSTILE\" | \"Unknown\",\n"
+        "      \"sensor_profile\": [\"string\"],\n"
+        "      \"motion_status\": \"Stationary\" | \"Moving Slow\" | \"Moving Fast\" | \"Unknown\",\n"
+        "      \"wake_description\": \"string\",\n"
+        "      \"aspect\": \"string (orientation relative to camera)\",\n"
+        "      \"range_estimation_nm\": float,\n"
+        "      \"bearing_clock\": \"string (e.g. 12 o'clock)\",\n"
+        "      \"deck_activity\": \"string\",\n"
+        "      \"special_features\": [\"string (anomalies)\"],\n"
+        "      \"threat_level_score\": int (1-10),\n"
+        "      \"threat_classification\": \"Friendly\" | \"Neutral\" | \"Suspect\" | \"Hostile\",\n"
+        "      \"tactical_intent\": \"string\"\n"
+        "    }\n"
+        "  }\n"
+        "}\n\n"
+        "Assess each object based on its visual signatures and the operational context."
+        + mission_context
+    )
+
+
 def estimate_threat_gpt(
-    image_path: str, 
-    detections: List[Dict[str, Any]]
+    image_path: str,
+    detections: List[Dict[str, Any]],
+    mission_spec=None,  # Optional[MissionSpecification]
 ) -> Dict[str, Any]:
     """
-    Perform Naval Threat Assessment on detected objects using GPT-4o.
-    
+    Perform Threat Assessment on detected objects using GPT-4o.
+
     Args:
         image_path: Path to the image file.
         detections: List of detection dicts (bbox, label, etc.).
-        
+        mission_spec: Optional MissionSpecification for domain-aware assessment.
+
     Returns:
-        Dict mapping object ID (e.g., T01) to NavalThreatAssessment dict.
+        Dict mapping object ID (e.g., T01) to threat assessment dict.
     """
     api_key = os.environ.get("OPENAI_API_KEY")
     if not api_key:
@@ -35,14 +128,13 @@ def estimate_threat_gpt(
     # 1. Prepare detections summary for prompt
     det_summary = []
     for i, det in enumerate(detections):
-        # UI uses T01, T02... logic usually matches index + 1
         obj_id = det.get("track_id") or det.get("id") or f"T{str(i+1).zfill(2)}"
         bbox = det.get("bbox", [])
         label = det.get("label", "object")
         det_summary.append(f"- ID: {obj_id}, Classification Hint: {label}, BBox: {bbox}")
 
     det_text = "\n".join(det_summary)
-    
+
     if not det_text:
         return {}
 
@@ -53,45 +145,20 @@ def estimate_threat_gpt(
         logger.error(f"Failed to encode image for GPT: {e}")
         return {}
 
-    # 3. Construct Prompt (Naval Focused)
-    system_prompt = (
-        "You are an elite Naval Intelligence Officer and Threat Analyst. "
-        "Your task is to analyze optical surveillance imagery and provide a detailed tactical assessment for every detected object. "
-        "You must output a STRICT JSON object that matches the following schema for every object ID provided:\n\n"
-        "RESPONSE SCHEMA (JSON):\n"
-        "{\n"
-        "  \"objects\": {\n"
-        "    \"T01\": {\n"
-        "      \"vessel_category\": \"Warship\" | \"Commercial\" | \"Fishing\" | \"Small Boat\" | \"Aircraft\" | \"Unknown\",\n"
-        "      \"specific_class\": \"string (e.g., Arleigh Burke, Skiff)\",\n"
-        "      \"identity_markers\": [\"string (hull numbers, flags)\"],\n"
-        "      \"flag_state\": \"string (Country)\",\n"
-        "      \"visible_weapons\": [\"string\"],\n"
-        "      \"weapon_readiness\": \"Stowed/PEACE\" | \"Trained/Aiming\" | \"Firing/HOSTILE\",\n"
-        "      \"sensor_profile\": [\"string (radars)\"],\n"
-        "      \"motion_status\": \"Dead in Water\" | \"Underway Slow\" | \"Underway Fast\" | \"Flank Speed\",\n"
-        "      \"wake_description\": \"string\",\n"
-        "      \"aspect\": \"Bow-on\" | \"Stern-on\" | \"Broadside\",\n"
-        "      \"range_estimation_nm\": float (Nautical Miles),\n"
-        "      \"bearing_clock\": \"string (e.g. 12 o'clock)\",\n"
-        "      \"deck_activity\": \"string\",\n"
-        "      \"special_features\": [\"string (anomalies)\"],\n"
-        "      \"threat_level_score\": int (1-10),\n"
-        "      \"threat_classification\": \"Friendly\" | \"Neutral\" | \"Suspect\" | \"Hostile\",\n"
-        "      \"tactical_intent\": \"string (e.g., Transit, Attack)\"\n"
-        "    }\n"
-        "  }\n"
-        "}\n\n"
-        "ASSUMPTIONS:\n"
-        "- Unknown small boats approaching larger vessels are HIGH threat (Suspect/Hostile).\n"
-        "- Visible trained weapons are IMMINENT threat (Score 9-10).\n"
-        "- Ignore artifacts, focus on the objects."
-    )
+    # 3. Domain-aware prompt selection (INV-7)
+    domain = "NAVAL"  # default for backward compatibility
+    if mission_spec:
+        domain = mission_spec.domain
+        if mission_spec.domain_source == "INFERRED":
+            logger.info("GPT assessment using inferred domain=%s (domain_inferred=True)", domain)
+
+    system_prompt = _build_domain_system_prompt(domain, mission_spec)
 
+    domain_label = domain.lower() if domain != "NAVAL" else "naval"
     user_prompt = (
-        f"Analyze this naval surveillance image. The following objects have been detected:\n"
+        f"Analyze this {domain_label} surveillance image. The following objects have been detected:\n"
         f"{det_text}\n\n"
-        "Provide a detailed Naval Threat Assessment for each object based on its visual signatures."
+        f"Provide a detailed Threat Assessment for each object based on its visual signatures."
     )
 
     # 4. Call API

utils/mission_parser.py

@@ -0,0 +1,381 @@
+"""
+Mission text parser — converts raw operator text into a validated MissionSpecification.
+
+Single public function: parse_mission_text(raw_text, detector_key) -> MissionSpecification
+
+Internal flow:
+1. Fast-path regex check -> skip LLM if comma-separated labels
+2. LLM extraction call (GPT-4o, temperature 0.0)
+3. Deterministic validation pipeline
+4. COCO vocabulary mapping for COCO-only detectors
+5. Build RelevanceCriteria deterministically from mapped classes
+6. Return validated MissionSpecification or raise MissionParseError
+"""
+
+import json
+import logging
+import os
+import re
+import urllib.request
+import urllib.error
+from typing import List, Optional
+
+from coco_classes import COCO_CLASSES, canonicalize_coco_name, coco_class_catalog
+from utils.schemas import MissionSpecification, RelevanceCriteria
+
+logger = logging.getLogger(__name__)
+
+# Detectors that only support COCO class vocabulary
+_COCO_ONLY_DETECTORS = frozenset({"hf_yolov8", "detr_resnet50"})
+
+
+class MissionParseError(ValueError):
+    """Raised when mission text cannot be parsed into a valid MissionSpecification."""
+    def __init__(self, message: str, warnings: Optional[List[str]] = None):
+        self.warnings = warnings or []
+        super().__init__(message)
+
+
+def _is_comma_separated_labels(text: str) -> bool:
+    """Fast-path: detect simple comma-separated class labels (no LLM needed)."""
+    # Match: word tokens separated by commas, each token <= 3 words
+    pattern = r"^[\w\s]+(,\s*[\w\s]+)*$"
+    if not re.match(pattern, text.strip()):
+        return False
+    tokens = [t.strip() for t in text.split(",") if t.strip()]
+    return all(len(t.split()) <= 3 for t in tokens)
+
+
+def _is_coco_only(detector_key: str) -> bool:
+    return detector_key in _COCO_ONLY_DETECTORS
+
+
+def _map_coco_classes(
+    object_classes: List[str], detector_key: str
+) -> tuple[List[str], List[str], List[str]]:
+    """Map object classes to COCO vocabulary for COCO-only detectors.
+
+    Returns:
+        (mapped_classes, unmappable_classes, warnings)
+    """
+    if not _is_coco_only(detector_key):
+        return object_classes, [], []
+
+    mapped = []
+    unmappable = []
+    warnings = []
+    seen = set()
+
+    for cls in object_classes:
+        canonical = canonicalize_coco_name(cls)
+        if canonical is not None:
+            if canonical not in seen:
+                mapped.append(canonical)
+                seen.add(canonical)
+            if canonical.lower() != cls.lower():
+                warnings.append(
+                    f"'{cls}' mapped to COCO class '{canonical}'."
+                )
+        else:
+            unmappable.append(cls)
+            warnings.append(
+                f"'{cls}' is not in COCO vocabulary. Will not be detected by {detector_key}."
+            )
+
+    return mapped, unmappable, warnings
+
+
+def _build_fast_path_spec(
+    raw_text: str, object_classes: List[str], detector_key: str
+) -> MissionSpecification:
+    """Build MissionSpecification for simple comma-separated input (no LLM call)."""
+    mapped, unmappable, warnings = _map_coco_classes(object_classes, detector_key)
+
+    if _is_coco_only(detector_key) and not mapped:
+        raise MissionParseError(
+            f"None of the requested objects ({', '.join(object_classes)}) match the "
+            f"{detector_key} vocabulary. This detector supports: "
+            f"{coco_class_catalog()}. "
+            f"Use an open-vocabulary detector (Grounding DINO) or adjust your mission.",
+            warnings=warnings,
+        )
+
+    final_classes = mapped if _is_coco_only(detector_key) else object_classes
+
+    return MissionSpecification(
+        object_classes=final_classes,
+        mission_intent="DETECT",
+        domain="GENERIC",
+        domain_source="INFERRED",
+        relevance_criteria=RelevanceCriteria(
+            required_classes=final_classes,
+            min_confidence=0.0,
+        ),
+        context_phrases=[],
+        stripped_modifiers=[],
+        operator_text=raw_text,
+        parse_confidence="HIGH",
+        parse_warnings=warnings,
+    )
+
+
+# --- LLM Extraction ---
+
+_SYSTEM_PROMPT = (
+    "You are a mission text parser for an object detection system. Your ONLY job is to extract "
+    "structured fields from operator mission text. You do NOT assess threats. You do NOT reason "
+    "about tactics. You extract and classify.\n\n"
+    "OUTPUT SCHEMA (strict JSON):\n"
+    "{\n"
+    '  "object_classes": ["string"],\n'
+    '  "mission_intent": "ENUM",\n'
+    '  "domain": "ENUM",\n'
+    '  "context_phrases": ["string"],\n'
+    '  "stripped_modifiers": ["string"],\n'
+    '  "parse_confidence": "ENUM",\n'
+    '  "parse_warnings": ["string"]\n'
+    "}\n\n"
+    "EXTRACTION RULES:\n\n"
+    "1. OBJECT_CLASSES — What to extract:\n"
+    "   - Extract nouns and noun phrases that refer to PHYSICAL, VISUALLY DETECTABLE objects.\n"
+    "   - Keep visual descriptors that narrow the category: 'small boat', 'military vehicle', 'cargo ship'.\n"
+    "   - Use singular form: 'vessels' -> 'vessel', 'people' -> 'person'.\n"
+    "   - If the input is already comma-separated class labels (e.g., 'person, car, boat'),\n"
+    "     use them directly without modification.\n\n"
+    "2. OBJECT_CLASSES — What to strip:\n"
+    "   - Remove threat/intent adjectives: 'hostile', 'suspicious', 'friendly', 'dangerous', 'enemy'.\n"
+    "     -> Move these to stripped_modifiers.\n"
+    "   - Remove action verbs: 'approaching', 'fleeing', 'attacking'.\n"
+    "     -> Move the full phrase to context_phrases.\n"
+    "   - Remove spatial/temporal phrases: 'from the east', 'near the harbor', 'at night'.\n"
+    "     -> Move to context_phrases.\n"
+    "   - Do NOT extract abstract concepts: 'threat', 'danger', 'hazard', 'risk' are not objects.\n\n"
+    "3. MISSION_INTENT — Infer from verbs:\n"
+    "   - 'detect', 'find', 'locate', 'spot', 'search for' -> DETECT\n"
+    "   - 'classify', 'identify', 'determine type of' -> CLASSIFY\n"
+    "   - 'track', 'follow', 'monitor movement of' -> TRACK\n"
+    "   - 'assess threat', 'evaluate danger', 'threat assessment' -> ASSESS_THREAT\n"
+    "   - 'monitor', 'watch', 'observe', 'surveil' -> MONITOR\n"
+    "   - If no verb present (bare class list), default to DETECT.\n\n"
+    "4. DOMAIN — Infer from contextual clues:\n"
+    "   - Maritime vocabulary (vessel, ship, boat, harbor, naval, maritime, wake, sea) -> NAVAL\n"
+    "   - Ground vocabulary (vehicle, convoy, checkpoint, road, building, infantry) -> GROUND\n"
+    "   - Aerial vocabulary (aircraft, drone, UAV, airspace, altitude, flight) -> AERIAL\n"
+    "   - Urban vocabulary (pedestrian, intersection, storefront, crowd, building) -> URBAN\n"
+    "   - If no domain clues present -> GENERIC\n\n"
+    "5. PARSE_CONFIDENCE:\n"
+    "   - HIGH: Clear object classes extracted, domain identifiable.\n"
+    "   - MEDIUM: Some ambiguity but reasonable extraction possible. Include warnings.\n"
+    "   - LOW: Cannot extract meaningful object classes. Input is too abstract,\n"
+    "     contradictory, or contains no visual object references.\n"
+    "     Examples of LOW: 'keep us safe', 'do your job', 'analyze everything'.\n\n"
+    "FORBIDDEN:\n"
+    "- Do NOT infer object classes not implied by the text. If the text says 'boats',\n"
+    "  do not add 'person' or 'vehicle' unless mentioned.\n"
+    "- Do NOT add threat scores, engagement rules, or tactical recommendations.\n"
+    "- Do NOT interpret what 'threat' or 'danger' means in terms of specific objects.\n"
+    "  If the operator writes 'detect threats', set parse_confidence to LOW and warn:\n"
+    "  \"'threats' is not a visual object class. Specify what objects to detect.\""
+)
+
+
+def _call_extraction_llm(raw_text: str, detector_key: str) -> dict:
+    """Call GPT-4o to extract structured mission fields from natural language."""
+    api_key = os.environ.get("OPENAI_API_KEY")
+    if not 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')."
+        )
+
+    detector_type = "COCO_ONLY" if _is_coco_only(detector_key) else "OPEN_VOCAB"
+
+    user_prompt = (
+        f'OPERATOR MISSION TEXT:\n"{raw_text}"\n\n'
+        f"DETECTOR TYPE: {detector_type}\n\n"
+        "Extract the structured mission specification from the above text."
+    )
+
+    payload = {
+        "model": "gpt-4o",
+        "temperature": 0.0,
+        "max_tokens": 500,
+        "response_format": {"type": "json_object"},
+        "messages": [
+            {"role": "system", "content": _SYSTEM_PROMPT},
+            {"role": "user", "content": user_prompt},
+        ],
+    }
+
+    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")
+        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:
+        raise MissionParseError(f"Mission parsing API call failed: {e}")
+    except json.JSONDecodeError:
+        raise MissionParseError(
+            "GPT returned invalid JSON. Please rephrase your mission."
+        )
+
+
+def _validate_and_build(
+    llm_output: dict, raw_text: str, detector_key: str
+) -> MissionSpecification:
+    """Deterministic validation pipeline (Section 7.3 decision tree)."""
+
+    # Step 2: Extract fields with defaults
+    object_classes = llm_output.get("object_classes", [])
+    mission_intent = llm_output.get("mission_intent", "DETECT")
+    domain = llm_output.get("domain", "GENERIC")
+    context_phrases = llm_output.get("context_phrases", [])
+    stripped_modifiers = llm_output.get("stripped_modifiers", [])
+    parse_confidence = llm_output.get("parse_confidence", "LOW")
+    parse_warnings = llm_output.get("parse_warnings", [])
+
+    # Validate enum values
+    valid_intents = {"DETECT", "CLASSIFY", "TRACK", "ASSESS_THREAT", "MONITOR"}
+    if mission_intent not in valid_intents:
+        mission_intent = "DETECT"
+        parse_warnings.append(f"Invalid mission_intent '{llm_output.get('mission_intent')}', defaulted to DETECT.")
+
+    valid_domains = {"NAVAL", "GROUND", "AERIAL", "URBAN", "GENERIC"}
+    if domain not in valid_domains:
+        domain = "GENERIC"
+        parse_warnings.append(f"Invalid domain '{llm_output.get('domain')}', defaulted to GENERIC.")
+
+    valid_confidence = {"HIGH", "MEDIUM", "LOW"}
+    if parse_confidence not in valid_confidence:
+        parse_confidence = "LOW"
+
+    # Step 3: Parse confidence check
+    if parse_confidence == "LOW":
+        warnings_str = "; ".join(parse_warnings) if parse_warnings else "No details"
+        raise MissionParseError(
+            f"Could not extract object classes from mission text. "
+            f"Warnings: {warnings_str}. "
+            f"Please specify concrete objects to detect (e.g., 'vessel, small boat').",
+            warnings=parse_warnings,
+        )
+
+    # Validate object_classes is non-empty
+    if not object_classes:
+        raise MissionParseError(
+            "Mission text produced no detectable object classes. "
+            "Please specify concrete objects to detect.",
+            warnings=parse_warnings,
+        )
+
+    # Filter out empty strings
+    object_classes = [c.strip() for c in object_classes if c and c.strip()]
+    if not object_classes:
+        raise MissionParseError(
+            "All extracted object classes were empty after cleanup.",
+            warnings=parse_warnings,
+        )
+
+    # Step 4: COCO vocabulary mapping
+    mapped, unmappable, coco_warnings = _map_coco_classes(object_classes, detector_key)
+    parse_warnings.extend(coco_warnings)
+
+    if _is_coco_only(detector_key):
+        if not mapped:
+            raise MissionParseError(
+                f"None of the requested objects ({', '.join(object_classes)}) match the "
+                f"{detector_key} vocabulary. "
+                f"This detector supports: {coco_class_catalog()}. "
+                f"Use an open-vocabulary detector (Grounding DINO) or adjust your mission.",
+                warnings=parse_warnings,
+            )
+        final_classes = mapped
+    else:
+        final_classes = object_classes
+
+    # Step 5: Build RelevanceCriteria deterministically
+    relevance_criteria = RelevanceCriteria(
+        required_classes=final_classes,
+        min_confidence=0.0,
+    )
+
+    # Step 6: Construct MissionSpecification
+    return MissionSpecification(
+        object_classes=final_classes,
+        mission_intent=mission_intent,
+        domain=domain,
+        domain_source="INFERRED",
+        relevance_criteria=relevance_criteria,
+        # INVARIANT INV-13: context_phrases are forwarded to LLM reasoning layers
+        # (GPT threat assessment, threat chat) as situational context ONLY.
+        # They must NEVER be used in evaluate_relevance(), prioritization,
+        # or any deterministic filtering/sorting logic.
+        context_phrases=context_phrases,
+        stripped_modifiers=stripped_modifiers,
+        operator_text=raw_text,
+        parse_confidence=parse_confidence,
+        parse_warnings=parse_warnings,
+    )
+
+
+def parse_mission_text(
+    raw_text: str,
+    detector_key: str,
+) -> MissionSpecification:
+    """Parse raw mission text into a validated MissionSpecification.
+
+    Args:
+        raw_text: Verbatim mission text from the operator.
+        detector_key: Detector model key (determines COCO vocabulary constraints).
+
+    Returns:
+        Validated MissionSpecification.
+
+    Raises:
+        MissionParseError: If mission text cannot produce a valid specification.
+    """
+    if not raw_text or not raw_text.strip():
+        raise MissionParseError(
+            "Mission text is empty. Specify objects to detect or use the default queries."
+        )
+
+    raw_text = raw_text.strip()
+
+    # Fast-path: simple comma-separated labels -> skip LLM
+    if _is_comma_separated_labels(raw_text):
+        object_classes = [t.strip() for t in raw_text.split(",") if t.strip()]
+        logger.info(
+            "Mission fast-path: comma-separated labels %s", object_classes
+        )
+        return _build_fast_path_spec(raw_text, object_classes, detector_key)
+
+    # LLM path: natural language mission text
+    logger.info("Mission LLM-path: extracting from natural language")
+    llm_output = _call_extraction_llm(raw_text, detector_key)
+    logger.info("Mission LLM extraction result: %s", llm_output)
+
+    mission_spec = _validate_and_build(llm_output, raw_text, detector_key)
+    logger.info(
+        "Mission parsed: classes=%s intent=%s domain=%s(%s) confidence=%s",
+        mission_spec.object_classes,
+        mission_spec.mission_intent,
+        mission_spec.domain,
+        mission_spec.domain_source,
+        mission_spec.parse_confidence,
+    )
+    return mission_spec

utils/relevance.py

@@ -0,0 +1,71 @@
+"""
+Object relevance evaluation — deterministic gate between detection and GPT assessment.
+
+Single public function: evaluate_relevance(detection, criteria) -> RelevanceDecision
+
+INVARIANT INV-13 enforcement: This function accepts RelevanceCriteria, NOT
+MissionSpecification. It cannot see context_phrases, stripped_modifiers, or any
+LLM-derived field. This is structural, not by convention.
+"""
+
+import logging
+from typing import Any, Dict, NamedTuple
+
+from coco_classes import canonicalize_coco_name
+from utils.schemas import RelevanceCriteria
+
+logger = logging.getLogger(__name__)
+
+
+class RelevanceDecision(NamedTuple):
+    relevant: bool
+    reason: str  # "ok" | "label_not_in_required_classes" | "below_confidence"
+
+
+def evaluate_relevance(
+    detection: Dict[str, Any],
+    criteria: RelevanceCriteria,
+) -> RelevanceDecision:
+    """Evaluate whether a detection is relevant to the mission.
+
+    Pure deterministic predicate — no LLM involvement.
+
+    Args:
+        detection: Detection dict with at least 'label' and 'score' keys.
+        criteria: RelevanceCriteria with required_classes and min_confidence.
+
+    Returns:
+        RelevanceDecision(relevant=bool, reason=str).
+    """
+    label = (detection.get("label") or "").lower().strip()
+    confidence = detection.get("score", 0.0)
+
+    if not label:
+        return RelevanceDecision(False, "label_not_in_required_classes")
+
+    # Build lowercase set of required classes for comparison
+    required_lower = {c.lower() for c in criteria.required_classes}
+
+    # Direct match
+    if label in required_lower:
+        if confidence < criteria.min_confidence:
+            return RelevanceDecision(False, "below_confidence")
+        return RelevanceDecision(True, "ok")
+
+    # Synonym match via COCO canonicalization
+    canonical = canonicalize_coco_name(label)
+    if canonical and canonical.lower() in required_lower:
+        if confidence < criteria.min_confidence:
+            return RelevanceDecision(False, "below_confidence")
+        return RelevanceDecision(True, "ok")
+
+    # Check if any required class canonicalizes to the same COCO class as the label
+    if canonical:
+        for req in criteria.required_classes:
+            req_canonical = canonicalize_coco_name(req)
+            if req_canonical and req_canonical.lower() == canonical.lower():
+                if confidence < criteria.min_confidence:
+                    return RelevanceDecision(False, "below_confidence")
+                return RelevanceDecision(True, "ok")
+
+    return RelevanceDecision(False, "label_not_in_required_classes")

utils/schemas.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 from typing import List, Optional, Literal
 
 class NavalThreatAssessment(BaseModel):
@@ -40,3 +40,107 @@ class NavalThreatAssessment(BaseModel):
 
 class FrameThreatAnalysis(BaseModel):
     objects: dict[str, NavalThreatAssessment] = Field(..., description="Map of Object ID (e.g., 'T01') to its assessment.")
+
+
+# --- Mission-Driven Abstractions ---
+
+
+class RelevanceCriteria(BaseModel):
+    """Deterministic boolean predicate for filtering detections against a mission.
+
+    This is the ONLY input to evaluate_relevance(). It intentionally excludes
+    context_phrases, stripped_modifiers, and all LLM-derived context so that
+    relevance filtering remains purely deterministic (INV-13).
+    """
+    required_classes: List[str] = Field(
+        ..., min_length=1,
+        description="Object categories that satisfy the mission. "
+                    "Detections whose label is not in this list are excluded."
+    )
+    min_confidence: float = Field(
+        default=0.0, ge=0.0, le=1.0,
+        description="Minimum detector confidence to consider a detection relevant."
+    )
+
+
+class MissionSpecification(BaseModel):
+    """Structured representation of operator intent.
+
+    Created once from raw mission text at the API boundary (app.py).
+    Forwarded to: detector (object_classes), GPT (full spec), chat (full spec),
+    relevance gate (relevance_criteria only — INV-13).
+
+    INVARIANT INV-13: context_phrases are forwarded to LLM reasoning layers
+    (GPT threat assessment, threat chat) as situational context ONLY.
+    They must NEVER be used in evaluate_relevance(), prioritization,
+    or any deterministic filtering/sorting logic.
+    """
+
+    # --- Extracted by LLM or fast-path ---
+    object_classes: List[str] = Field(
+        ..., min_length=1,
+        description="Concrete, visually detectable object categories to detect. "
+                    "These become detector queries. Must be nouns, not adjectives or verbs."
+    )
+    mission_intent: Literal[
+        "DETECT", "CLASSIFY", "TRACK", "ASSESS_THREAT", "MONITOR"
+    ] = Field(
+        ...,
+        description="Operator purpose. DETECT=find objects, CLASSIFY=identify type, "
+                    "TRACK=follow over time, ASSESS_THREAT=evaluate danger, MONITOR=passive watch."
+    )
+    domain: Literal[
+        "NAVAL", "GROUND", "AERIAL", "URBAN", "GENERIC"
+    ] = Field(
+        ...,
+        description="Operational domain. Selects the GPT assessment schema and system prompt."
+    )
+    domain_source: Literal["INFERRED", "OPERATOR_SET"] = Field(
+        default="INFERRED",
+        description="Whether domain was LLM-inferred or explicitly set by operator."
+    )
+
+    # --- Deterministic (derived from object_classes) ---
+    relevance_criteria: RelevanceCriteria = Field(
+        ...,
+        description="Boolean predicate for filtering detections. "
+                    "Built deterministically from object_classes after extraction."
+    )
+
+    # --- Context preservation ---
+    context_phrases: List[str] = Field(
+        default_factory=list,
+        description="Non-class contextual phrases from mission text. "
+                    "E.g., 'approaching from the east', 'near the harbor'. "
+                    "Forwarded to GPT as situational context, NOT used for detection."
+    )
+    stripped_modifiers: List[str] = Field(
+        default_factory=list,
+        description="Adjectives/modifiers removed during extraction. "
+                    "E.g., 'hostile', 'suspicious', 'friendly'. Logged for audit."
+    )
+    operator_text: str = Field(
+        ...,
+        description="Original unmodified mission text from the operator. Preserved for audit."
+    )
+
+    # --- LLM self-assessment ---
+    parse_confidence: Literal["HIGH", "MEDIUM", "LOW"] = Field(
+        ...,
+        description="Confidence in the extraction. "
+                    "LOW = could not reliably extract classes -> triggers rejection."
+    )
+    parse_warnings: List[str] = Field(
+        default_factory=list,
+        description="Specific issues encountered during extraction. "
+                    "E.g., 'term \"threat\" is not a visual class, stripped'."
+    )
+
+    @model_validator(mode="after")
+    def reject_generic_threat_assessment(self):
+        if self.domain == "GENERIC" and self.mission_intent == "ASSESS_THREAT":
+            raise ValueError(
+                "Cannot assess threats without a specific domain. "
+                "Set domain to NAVAL, GROUND, AERIAL, or URBAN."
+            )
+        return self

utils/threat_chat.py

@@ -10,20 +10,25 @@ from typing import List, Dict, Any
 logger = logging.getLogger(__name__)
 
 
-def chat_about_threats(question: str, detections: List[Dict[str, Any]]) -> str:
+def chat_about_threats(
+    question: str,
+    detections: List[Dict[str, Any]],
+    mission_spec_dict: Dict[str, Any] = None,
+) -> str:
     """
     Answer user questions about detected threats using GPT.
-    
+
     Args:
         question: User's question about the current threat situation.
         detections: List of detection dicts with gpt_raw threat analysis.
-        
+        mission_spec_dict: Optional dict of mission specification fields.
+
     Returns:
         GPT's response as a string.
     """
     import urllib.request
     import urllib.error
-    
+
     api_key = os.environ.get("OPENAI_API_KEY")
     if not api_key:
         logger.warning("OPENAI_API_KEY not set. Cannot process threat chat.")
@@ -34,12 +39,41 @@ def chat_about_threats(question: str, detections: List[Dict[str, Any]]) -> str:
 
     # Build threat context from detections
     threat_context = _build_threat_context(detections)
-    
+
+    # Domain-aware role selection
+    domain = "NAVAL"
+    role_label = "Naval Tactical Intelligence Officer"
+    if mission_spec_dict:
+        domain = mission_spec_dict.get("domain", "NAVAL")
+        if domain == "GROUND":
+            role_label = "Ground Surveillance Intelligence Officer"
+        elif domain == "AERIAL":
+            role_label = "Air Surveillance Intelligence Officer"
+        elif domain == "URBAN":
+            role_label = "Urban Surveillance Intelligence Officer"
+        elif domain == "GENERIC":
+            role_label = "Tactical Intelligence Officer"
+
+    # Build mission context block (INV-8: mission context forwarded to LLM calls)
+    mission_block = ""
+    if mission_spec_dict:
+        mission_block = "\nMISSION CONTEXT:\n"
+        if mission_spec_dict.get("mission_intent"):
+            mission_block += f"- Intent: {mission_spec_dict['mission_intent']}\n"
+        if mission_spec_dict.get("domain"):
+            mission_block += f"- Domain: {mission_spec_dict['domain']}\n"
+        if mission_spec_dict.get("object_classes"):
+            mission_block += f"- Target Classes: {', '.join(mission_spec_dict['object_classes'])}\n"
+        if mission_spec_dict.get("context_phrases"):
+            mission_block += f"- Situation: {'; '.join(mission_spec_dict['context_phrases'])}\n"
+        mission_block += "\n"
+
     system_prompt = (
-        "You are a Naval Tactical Intelligence Officer providing real-time threat analysis support. "
+        f"You are a {role_label} providing real-time threat analysis support. "
         "You have access to the current threat assessment data from optical surveillance. "
         "Answer questions concisely and tactically. Use military terminology where appropriate. "
         "If asked about engagement recommendations, always note that final decisions rest with the commanding officer.\n\n"
+        f"{mission_block}"
         "CURRENT THREAT PICTURE:\n"
         f"{threat_context}\n\n"
         "Respond to the operator's question based on this threat data."

utils/tracker.py

@@ -195,6 +195,9 @@ class KalmanFilter:
         return ret
 
 
+# Default staleness threshold: GPT metadata older than this many frames is flagged STALE
+MAX_STALE_FRAMES = 300
+
 GPT_SYNC_KEYS = frozenset({
     # Legacy fields
     "gpt_distance_m", "gpt_direction", "gpt_description", "gpt_raw",
@@ -207,6 +210,10 @@ GPT_SYNC_KEYS = frozenset({
     "special_features", "tactical_intent",
     # Computed fields
     "distance_m", "direction", "description",
+    # Provenance and temporal validity
+    "assessment_frame_index", "assessment_status",
+    # Mission relevance
+    "mission_relevant", "relevance_reason",
 })
 
 
@@ -506,25 +513,28 @@ class ByteTracker:
         
         results = []
         for track in output_stracks:
-            # Reconstruct dictionary
-            # Get latest bbox from Kalman State for smoothness, or original? 
-            # Usually we use the detection box if matched, or predicted if lost (but logic above separates them).
-            # If matched, we have updated KF.
-            
             d_out = track.original_data.copy() if hasattr(track, 'original_data') else {}
-            
-            # Update bbox to tracked bbox? Or keep raw?
-            # Keeping raw is safer for simple visualizer, but tracked bbox is smoother.
-            # Let's use tracked bbox (tlbr).
+
             tracked_bbox = track.tlbr
             d_out['bbox'] = [float(x) for x in tracked_bbox]
             d_out['track_id'] = f"T{str(track.track_id).zfill(2)}"
-            
+
             # Restore GPT data if track has it and current detection didn't
             for k, v in track.gpt_data.items():
                 if k not in d_out:
                     d_out[k] = v
-                    
+
+            # --- Temporal validity check (INV-5, INV-11) ---
+            assessment_frame = d_out.get('assessment_frame_index')
+            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_age_frames'] = frames_since
+            elif d_out.get('assessment_status') != 'ASSESSED':
+                # INV-6: Unassessed objects get explicit UNASSESSED status
+                d_out['assessment_status'] = 'UNASSESSED'
+
             # Update history
             if 'history' not in track.gpt_data:
                 track.gpt_data['history'] = []
@@ -532,9 +542,9 @@ class ByteTracker:
             if len(track.gpt_data['history']) > 30:
                 track.gpt_data['history'].pop(0)
             d_out['history'] = track.gpt_data['history']
-            
+
             results.append(d_out)
-            
+
         return results
 
     def _sync_data(self, track, det_source):
@@ -553,6 +563,8 @@ class ByteTracker:
         Needed because GPT results are added to detection dicts *after* tracker.update()
         returns, so the tracker's internal state doesn't have GPT data unless we
         explicitly push it back in.
+
+        Records assessment_frame_index for temporal validity tracking (INV-5).
         """
         meta_by_tid = {}
         for d in tracked_dets:
@@ -561,6 +573,12 @@ class ByteTracker:
                 continue
             meta = {k: d[k] for k in GPT_SYNC_KEYS if k in d}
             if meta:
+                # Ensure assessment_frame_index is recorded
+                if "assessment_frame_index" not in meta and any(
+                    k in meta for k in ("threat_level_score", "gpt_raw", "vessel_category")
+                ):
+                    meta["assessment_frame_index"] = self.frame_id
+                    meta["assessment_status"] = "ASSESSED"
                 meta_by_tid[tid] = meta
         for track in self.tracked_stracks:
             tid_str = f"T{str(track.track_id).zfill(2)}"