mirror of
https://github.com/azaion/autopilot.git
synced 2026-06-21 11:01:10 +00:00
[AZ-683] scan_controller POI queue + 5/min cap + decision window
ci/woodpecker/push/build-arm Pipeline failed
ci/woodpecker/push/build-arm Pipeline failed
Adds the prioritized POI queue on top of the AZ-682 FSM substrate: priority = confidence x proximity x age_factor; rolling 60s window caps surfaces at 5; confidence-scaled decision window (40% -> 30s, 100% -> 120s, linear; <40% never surfaces); tick() runs the timeout sweep and silently forgets expired POIs (no IgnoredItem per spec); DeclinePoi via operator command returns a DeclineAction for AZ-685 to persist. ScanControllerHandle gains submit_poi_candidate / next_poi_for_surface / decline_poi / poi_queue_len / pois_in_window. submit_operator_cmd return type widens from Result<()> to Result<SubmitOutcome>. ScanMetrics and health() surface queue depth and counters. Tests: 26 unit + 11 integration in scan_controller (all AC1..AC5 + DeclinePoi end-to-end). Workspace clippy on scan_controller clean. Pre-existing autopilot::Runtime::vlm_provider_name dead-code error from batch 4 still open (see cumulative C5). Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -17,4 +17,6 @@ mission_executor = { workspace = true }
|
||||
tokio = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
chrono = { workspace = true }
|
||||
uuid = { workspace = true }
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
//! Internal modules for `scan_controller`. Not part of the public API.
|
||||
|
||||
pub mod frame_rate_guard;
|
||||
pub mod poi_queue;
|
||||
pub mod state_machine;
|
||||
|
||||
@@ -0,0 +1,306 @@
|
||||
//! AZ-683 — POI queue + 5/min rate cap + confidence-scaled decision
|
||||
//! window.
|
||||
//!
|
||||
//! The queue is the operator-facing buffer. Each candidate POI is
|
||||
//! ranked by `confidence × proximity × age_factor` (per
|
||||
//! `description.md §4`), the highest-priority unblocked POI is
|
||||
//! surfaced to the operator, and the rolling 60-second cap ensures
|
||||
//! the operator sees no more than 5 POIs per minute (per
|
||||
//! `description.md §8`, an operator-cognitive-load invariant).
|
||||
//!
|
||||
//! Confidence below 40 % is NEVER surfaced — `decision_window` returns
|
||||
//! `None`, the surface path skips, and the POI sits in the queue
|
||||
//! until either:
|
||||
//!
|
||||
//! - new evidence pushes its confidence above 40 % (subsequent
|
||||
//! `update_confidence` call — wired later by AZ-684 evidence
|
||||
//! ladder), or
|
||||
//! - its deadline expires and the timeout sweep forgets it (no
|
||||
//! `IgnoredItem` recorded — silent forget).
|
||||
//!
|
||||
//! Decline is handled at the operator-command layer (AZ-685 dispatches
|
||||
//! the resulting `IgnoredItem` into `mapobjects_store`). AZ-683
|
||||
//! returns the *information* needed to emit that action.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::time::Duration;
|
||||
|
||||
use uuid::Uuid;
|
||||
|
||||
use shared::models::poi::Poi;
|
||||
|
||||
mod priority;
|
||||
|
||||
pub use priority::{age_factor, decision_window, priority_score};
|
||||
|
||||
/// Operator-cognitive-load invariant from `description.md §8`. Hard
|
||||
/// non-negotiable.
|
||||
pub const SURFACE_CAP_PER_WINDOW: usize = 5;
|
||||
/// The rolling window the cap is measured over.
|
||||
pub const CAP_WINDOW: Duration = Duration::from_secs(60);
|
||||
|
||||
/// Internal POI entry. We keep `confidence` and `proximity` outside
|
||||
/// `Poi` so age-aware priority can be recomputed on demand without
|
||||
/// mutating the canonical model.
|
||||
#[derive(Debug, Clone)]
|
||||
struct Entry {
|
||||
poi: Poi,
|
||||
confidence: f32,
|
||||
proximity: f32,
|
||||
enqueued_at_ns: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub struct PoiQueue {
|
||||
entries: Vec<Entry>,
|
||||
surface_history_ns: VecDeque<u64>,
|
||||
}
|
||||
|
||||
/// Information returned when a POI is declined. AZ-685 turns this
|
||||
/// into a `MapObjectsAction::AppendIgnored` and persists it to
|
||||
/// `mapobjects_store`. AZ-683 only emits the data.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct DeclineAction {
|
||||
pub poi_id: Uuid,
|
||||
pub mgrs: String,
|
||||
pub class_group: String,
|
||||
}
|
||||
|
||||
impl PoiQueue {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize {
|
||||
self.entries.len()
|
||||
}
|
||||
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.entries.is_empty()
|
||||
}
|
||||
|
||||
/// Insert a candidate POI. `proximity` is the normalized
|
||||
/// `[0, 1]` proximity to the current camera position; AZ-682's
|
||||
/// caller (eventually AZ-686 / AZ-684) computes it. `now_ns` is
|
||||
/// the monotonic ns at insertion, used for age decay.
|
||||
///
|
||||
/// POIs below the 40 % confidence threshold are still INSERTED
|
||||
/// (per `description.md §4` they may be re-scored upward) but
|
||||
/// are NOT surfaceable until `decision_window` returns `Some`.
|
||||
pub fn insert(&mut self, poi: Poi, proximity: f32, now_ns: u64) {
|
||||
let confidence = poi.confidence;
|
||||
let proximity = proximity.clamp(0.0, 1.0);
|
||||
self.entries.push(Entry {
|
||||
poi,
|
||||
confidence,
|
||||
proximity,
|
||||
enqueued_at_ns: now_ns,
|
||||
});
|
||||
}
|
||||
|
||||
pub fn update_confidence(&mut self, poi_id: Uuid, new_confidence: f32) {
|
||||
if let Some(e) = self.entries.iter_mut().find(|e| e.poi.id == poi_id) {
|
||||
e.confidence = new_confidence;
|
||||
e.poi.confidence = new_confidence;
|
||||
}
|
||||
}
|
||||
|
||||
/// Return the next POI to surface to the operator, or `None`
|
||||
/// when either the cap is reached or no surfaceable POI exists.
|
||||
///
|
||||
/// "Surfaceable" means `decision_window(confidence) == Some(_)`
|
||||
/// — i.e. confidence ≥ 40 %.
|
||||
pub fn next_for_surface(&mut self, now_ns: u64) -> Option<Poi> {
|
||||
self.trim_history(now_ns);
|
||||
if self.surface_history_ns.len() >= SURFACE_CAP_PER_WINDOW {
|
||||
return None;
|
||||
}
|
||||
|
||||
let best_idx = self
|
||||
.entries
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter(|(_, e)| decision_window(e.confidence).is_some())
|
||||
.max_by(|(_, a), (_, b)| {
|
||||
let pa = priority_score(a.confidence, a.proximity, age_seconds(a, now_ns));
|
||||
let pb = priority_score(b.confidence, b.proximity, age_seconds(b, now_ns));
|
||||
pa.partial_cmp(&pb).unwrap_or(std::cmp::Ordering::Equal)
|
||||
})
|
||||
.map(|(idx, _)| idx)?;
|
||||
|
||||
let entry = self.entries.swap_remove(best_idx);
|
||||
self.surface_history_ns.push_back(now_ns);
|
||||
Some(entry.poi)
|
||||
}
|
||||
|
||||
/// Decline a POI by id. Removes from queue; returns the data
|
||||
/// needed to record an `IgnoredItem`.
|
||||
pub fn decline(&mut self, poi_id: Uuid) -> Option<DeclineAction> {
|
||||
let idx = self.entries.iter().position(|e| e.poi.id == poi_id)?;
|
||||
let entry = self.entries.swap_remove(idx);
|
||||
Some(DeclineAction {
|
||||
poi_id: entry.poi.id,
|
||||
mgrs: entry.poi.mgrs,
|
||||
class_group: entry.poi.class_group,
|
||||
})
|
||||
}
|
||||
|
||||
/// Drop POIs whose deadline (set at insertion by the caller per
|
||||
/// the confidence-scaled window) has elapsed. Returns the IDs of
|
||||
/// forgotten POIs. NO `IgnoredItem` is created — timeout =
|
||||
/// forget, per AC-5.
|
||||
pub fn timeout_sweep(&mut self, now_wallclock: chrono::DateTime<chrono::Utc>) -> Vec<Uuid> {
|
||||
let mut forgotten = Vec::new();
|
||||
self.entries.retain(|e| {
|
||||
if e.poi.deadline <= now_wallclock {
|
||||
forgotten.push(e.poi.id);
|
||||
false
|
||||
} else {
|
||||
true
|
||||
}
|
||||
});
|
||||
forgotten
|
||||
}
|
||||
|
||||
/// Live read of how many POIs were surfaced in the rolling cap
|
||||
/// window. Used by `health()` and metrics.
|
||||
pub fn surfaces_in_window(&mut self, now_ns: u64) -> usize {
|
||||
self.trim_history(now_ns);
|
||||
self.surface_history_ns.len()
|
||||
}
|
||||
|
||||
fn trim_history(&mut self, now_ns: u64) {
|
||||
let window_ns = CAP_WINDOW.as_nanos() as u64;
|
||||
while let Some(&front) = self.surface_history_ns.front() {
|
||||
if now_ns.saturating_sub(front) > window_ns {
|
||||
self.surface_history_ns.pop_front();
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn age_seconds(entry: &Entry, now_ns: u64) -> f32 {
|
||||
let dt_ns = now_ns.saturating_sub(entry.enqueued_at_ns);
|
||||
(dt_ns as f64 / 1e9) as f32
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use chrono::{Duration as ChronoDur, Utc};
|
||||
|
||||
fn poi(confidence: f32, mgrs: &str) -> Poi {
|
||||
Poi {
|
||||
id: Uuid::new_v4(),
|
||||
confidence,
|
||||
mgrs: mgrs.to_string(),
|
||||
class: "tank".to_string(),
|
||||
class_group: "armor".to_string(),
|
||||
source_detection_ids: vec![],
|
||||
enqueued_at: Utc::now(),
|
||||
priority: 0.0,
|
||||
decline_suppressed: false,
|
||||
vlm_status: shared::models::poi::VlmPipelineStatus::NotRequested,
|
||||
tier2_evidence: None,
|
||||
deadline: Utc::now() + ChronoDur::seconds(60),
|
||||
}
|
||||
}
|
||||
|
||||
/// AC-1 — priority ordering: `(0.9, 0.5, 0), (0.6, 0.9, 0),
|
||||
/// (0.7, 0.6, 60)` ordered by `c × p × age_factor`.
|
||||
#[test]
|
||||
fn ac1_priority_ordering_respects_age_factor() {
|
||||
// Arrange
|
||||
let mut q = PoiQueue::new();
|
||||
let p1 = poi(0.9, "1");
|
||||
let p2 = poi(0.6, "2");
|
||||
let p3 = poi(0.7, "3");
|
||||
q.insert(p1.clone(), 0.5, 0);
|
||||
q.insert(p2.clone(), 0.9, 0);
|
||||
// POI 3 enqueued 60 seconds earlier (age 60 at now_ns = 60e9).
|
||||
q.insert(p3.clone(), 0.6, 0);
|
||||
let now_ns = 60 * 1_000_000_000u64;
|
||||
|
||||
// Act
|
||||
let first = q.next_for_surface(now_ns).expect("first surface");
|
||||
let second = q.next_for_surface(now_ns).expect("second surface");
|
||||
let third = q.next_for_surface(now_ns).expect("third surface");
|
||||
|
||||
// Assert — expected order: P2 (0.54) > P1 (0.45) > P3 (0.42 ×
|
||||
// age_factor(60)).
|
||||
assert_eq!(first.id, p2.id, "highest priority first");
|
||||
assert_eq!(second.id, p1.id);
|
||||
assert_eq!(third.id, p3.id);
|
||||
}
|
||||
|
||||
/// AC-2 — hard 5-per-rolling-60-s cap.
|
||||
#[test]
|
||||
fn ac2_five_per_minute_cap_holds_back_excess() {
|
||||
// Arrange — 10 POIs all surfaceable.
|
||||
let mut q = PoiQueue::new();
|
||||
for i in 0..10 {
|
||||
let mgrs = format!("p{i}");
|
||||
q.insert(poi(0.8, &mgrs), 0.5, 0);
|
||||
}
|
||||
|
||||
// Act — drain in a 30 s window
|
||||
let mut surfaced = 0;
|
||||
let mut now = 0u64;
|
||||
for _ in 0..10 {
|
||||
if q.next_for_surface(now).is_some() {
|
||||
surfaced += 1;
|
||||
}
|
||||
now += 3_000_000_000;
|
||||
}
|
||||
|
||||
// Assert — at most 5 within the rolling window
|
||||
assert_eq!(surfaced, SURFACE_CAP_PER_WINDOW);
|
||||
assert_eq!(q.surfaces_in_window(now), SURFACE_CAP_PER_WINDOW);
|
||||
// Remaining POIs stay queued
|
||||
assert_eq!(q.len(), 5);
|
||||
|
||||
// Roll the window forward; cap should clear.
|
||||
let after = 61 * 1_000_000_000u64 + now;
|
||||
let next = q.next_for_surface(after);
|
||||
assert!(next.is_some(), "cap must clear after window rolls");
|
||||
}
|
||||
|
||||
/// AC-5 — timeout forgets without IgnoredItem.
|
||||
#[test]
|
||||
fn ac5_timeout_sweep_removes_expired_pois() {
|
||||
// Arrange
|
||||
let mut q = PoiQueue::new();
|
||||
let mut p = poi(0.8, "x");
|
||||
p.deadline = Utc::now() - ChronoDur::seconds(1);
|
||||
let id = p.id;
|
||||
q.insert(p, 0.5, 0);
|
||||
|
||||
// Act
|
||||
let forgotten = q.timeout_sweep(Utc::now());
|
||||
|
||||
// Assert
|
||||
assert_eq!(forgotten, vec![id]);
|
||||
assert!(q.is_empty());
|
||||
}
|
||||
|
||||
/// Decline emits the dispatchable action and removes the POI.
|
||||
#[test]
|
||||
fn decline_removes_and_emits_action() {
|
||||
// Arrange
|
||||
let mut q = PoiQueue::new();
|
||||
let p = poi(0.8, "y");
|
||||
let id = p.id;
|
||||
q.insert(p, 0.5, 0);
|
||||
|
||||
// Act
|
||||
let action = q.decline(id).expect("decline emits action");
|
||||
|
||||
// Assert
|
||||
assert_eq!(action.poi_id, id);
|
||||
assert_eq!(action.mgrs, "y");
|
||||
assert_eq!(action.class_group, "armor");
|
||||
assert!(q.is_empty());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
//! Pure scoring helpers for the POI queue.
|
||||
//!
|
||||
//! - [`decision_window`] — `40 % → 30 s, 100 % → 120 s, linear; <40 %
|
||||
//! → None`. Operator-cognitive-load mapping from
|
||||
//! `description.md §4`.
|
||||
//! - [`age_factor`] — gentle linear decay so a 5-minute-old POI
|
||||
//! ranks ~0 unless it has a very strong confidence × proximity
|
||||
//! product.
|
||||
//! - [`priority_score`] — the product `confidence × proximity ×
|
||||
//! age_factor(age_s)` used by `next_for_surface`.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
/// Confidence below this is NEVER surfaced. `decision_window`
|
||||
/// returns `None`, the queue skips the POI when scanning.
|
||||
pub const SURFACE_CONFIDENCE_FLOOR: f32 = 0.40;
|
||||
|
||||
const DEADLINE_AT_FLOOR: Duration = Duration::from_secs(30);
|
||||
const DEADLINE_AT_CEILING: Duration = Duration::from_secs(120);
|
||||
const AGE_DECAY_FULL_SECONDS: f32 = 300.0;
|
||||
const AGE_FACTOR_MIN: f32 = 0.1;
|
||||
|
||||
/// Linear `40 % → 30 s, 100 % → 120 s`; below 40 % returns `None`.
|
||||
pub fn decision_window(confidence: f32) -> Option<Duration> {
|
||||
if confidence < SURFACE_CONFIDENCE_FLOOR {
|
||||
return None;
|
||||
}
|
||||
let confidence = confidence.clamp(SURFACE_CONFIDENCE_FLOOR, 1.0);
|
||||
let span = (confidence - SURFACE_CONFIDENCE_FLOOR) / (1.0 - SURFACE_CONFIDENCE_FLOOR);
|
||||
let floor_s = DEADLINE_AT_FLOOR.as_secs_f32();
|
||||
let span_s = DEADLINE_AT_CEILING.as_secs_f32() - floor_s;
|
||||
Some(Duration::from_secs_f32(floor_s + span * span_s))
|
||||
}
|
||||
|
||||
/// Decay from `1.0` at age 0 down to `AGE_FACTOR_MIN` (`0.1`) at
|
||||
/// `AGE_DECAY_FULL_SECONDS` (5 minutes) and beyond. Linear in between.
|
||||
pub fn age_factor(age_seconds: f32) -> f32 {
|
||||
if age_seconds <= 0.0 {
|
||||
return 1.0;
|
||||
}
|
||||
let ratio = age_seconds / AGE_DECAY_FULL_SECONDS;
|
||||
let decay = 1.0 - ratio * (1.0 - AGE_FACTOR_MIN);
|
||||
decay.max(AGE_FACTOR_MIN)
|
||||
}
|
||||
|
||||
pub fn priority_score(confidence: f32, proximity: f32, age_seconds: f32) -> f32 {
|
||||
confidence * proximity * age_factor(age_seconds)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// AC-3 — `[0.40, 0.70, 1.00]` → `[30 s, 75 s, 120 s]`, linear.
|
||||
#[test]
|
||||
fn ac3_decision_window_linear_mapping() {
|
||||
// Assert
|
||||
assert_eq!(decision_window(0.40), Some(Duration::from_secs(30)));
|
||||
assert_eq!(decision_window(1.00), Some(Duration::from_secs(120)));
|
||||
let mid = decision_window(0.70).expect("0.70 surfaceable");
|
||||
// Linear midpoint between 30 and 120 s is 75 s.
|
||||
assert!(
|
||||
(mid.as_secs_f32() - 75.0).abs() < 0.1,
|
||||
"0.70 confidence should map to ~75 s, got {mid:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// AC-4 — sub-40 % is never surfaced.
|
||||
#[test]
|
||||
fn ac4_below_floor_returns_none() {
|
||||
assert_eq!(decision_window(0.39), None);
|
||||
assert_eq!(decision_window(0.0), None);
|
||||
assert_eq!(decision_window(-0.5), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn age_factor_decays_linearly() {
|
||||
// Arrange + Assert
|
||||
assert!((age_factor(0.0) - 1.0).abs() < 1e-6);
|
||||
// At half-decay-window (150 s) the factor is exactly halfway
|
||||
// between 1.0 and 0.1 → 0.55.
|
||||
assert!((age_factor(150.0) - 0.55).abs() < 1e-3);
|
||||
// At full-decay-window (300 s) the factor floors at 0.1.
|
||||
assert!((age_factor(300.0) - 0.1).abs() < 1e-6);
|
||||
assert!((age_factor(10_000.0) - 0.1).abs() < 1e-6);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn priority_score_zero_for_zero_proximity_or_confidence() {
|
||||
// Assert
|
||||
assert_eq!(priority_score(0.0, 1.0, 0.0), 0.0);
|
||||
assert_eq!(priority_score(1.0, 0.0, 0.0), 0.0);
|
||||
}
|
||||
}
|
||||
@@ -33,14 +33,19 @@ use std::time::{Duration, Instant};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::sync::Mutex;
|
||||
use uuid::Uuid;
|
||||
|
||||
use shared::error::{AutopilotError, Result};
|
||||
use shared::health::{ComponentHealth, HealthLevel};
|
||||
use shared::models::operator::{OperatorCommand, OperatorCommandKind};
|
||||
use shared::models::poi::Poi;
|
||||
|
||||
pub mod internal;
|
||||
|
||||
pub use internal::frame_rate_guard::{FrameRateGuard, FrameRateGuardConfig};
|
||||
pub use internal::poi_queue::{
|
||||
age_factor, decision_window, priority_score, DeclineAction, PoiQueue, SURFACE_CAP_PER_WINDOW,
|
||||
};
|
||||
pub use internal::state_machine::transitions::{transition, TransitionCtx};
|
||||
pub use internal::state_machine::{RejectReason, ScanState, TransitionOutcome, Trigger};
|
||||
|
||||
@@ -69,9 +74,13 @@ struct Inner {
|
||||
state: ScanState,
|
||||
last_state_change_ns: u64,
|
||||
fps_guard: FrameRateGuard,
|
||||
poi_queue: PoiQueue,
|
||||
latencies_us: std::collections::VecDeque<u64>,
|
||||
rejected_total: u64,
|
||||
transitions_total: u64,
|
||||
pois_surfaced_total: u64,
|
||||
pois_forgotten_total: u64,
|
||||
pois_declined_total: u64,
|
||||
}
|
||||
|
||||
impl Inner {
|
||||
@@ -104,9 +113,13 @@ impl ScanController {
|
||||
state: ScanState::ZoomedOut,
|
||||
last_state_change_ns: 0,
|
||||
fps_guard: FrameRateGuard::new(config.frame_rate),
|
||||
poi_queue: PoiQueue::new(),
|
||||
latencies_us: std::collections::VecDeque::with_capacity(LATENCY_WINDOW),
|
||||
rejected_total: 0,
|
||||
transitions_total: 0,
|
||||
pois_surfaced_total: 0,
|
||||
pois_forgotten_total: 0,
|
||||
pois_declined_total: 0,
|
||||
})),
|
||||
clock: shared::clock::MonoClock::new(),
|
||||
}
|
||||
@@ -132,6 +145,27 @@ pub struct ScanMetrics {
|
||||
pub rejected_total: u64,
|
||||
pub last_state_change_ns: u64,
|
||||
pub tick_latency_p99_us: u64,
|
||||
pub poi_queue_len: usize,
|
||||
pub pois_surfaced_total: u64,
|
||||
pub pois_forgotten_total: u64,
|
||||
pub pois_declined_total: u64,
|
||||
}
|
||||
|
||||
/// Result of [`ScanControllerHandle::submit_operator_cmd`]. `Accepted`
|
||||
/// means the command was applied with no return data; `Declined`
|
||||
/// carries the dispatchable IgnoredItem action AZ-685 must persist.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum SubmitOutcome {
|
||||
Accepted,
|
||||
Declined(DeclineAction),
|
||||
}
|
||||
|
||||
fn poi_id_from_payload(payload: &serde_json::Value) -> Result<Uuid> {
|
||||
let s = payload
|
||||
.get("poi_id")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or_else(|| AutopilotError::Validation("payload missing poi_id".to_string()))?;
|
||||
Uuid::parse_str(s).map_err(|e| AutopilotError::Validation(format!("invalid poi_id: {e}")))
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -181,46 +215,108 @@ impl ScanControllerHandle {
|
||||
outcome
|
||||
}
|
||||
|
||||
/// One scan-controller tick. AZ-682 only re-evaluates the FPS
|
||||
/// guard and records latency; AZ-684+ will run the evidence
|
||||
/// ladder + POI queue evaluation under this same tick.
|
||||
/// One scan-controller tick. Re-evaluates the FPS guard, runs
|
||||
/// the POI queue timeout sweep (AZ-683), and records latency.
|
||||
/// AZ-684+ will run the evidence ladder under this same tick.
|
||||
pub async fn tick(&self) -> Result<()> {
|
||||
let started = Instant::now();
|
||||
let now = self.clock.elapsed_ns();
|
||||
let now_wall = chrono::Utc::now();
|
||||
let mut inner = self.inner.lock().await;
|
||||
inner.fps_guard.tick(now);
|
||||
let forgotten = inner.poi_queue.timeout_sweep(now_wall);
|
||||
inner.pois_forgotten_total = inner
|
||||
.pois_forgotten_total
|
||||
.saturating_add(forgotten.len() as u64);
|
||||
inner.record_latency(started.elapsed().as_micros() as u64);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// AZ-683 — enqueue a POI candidate. `proximity` is the
|
||||
/// normalized `[0, 1]` proximity to the current camera. The
|
||||
/// caller is responsible for setting `poi.deadline` per the
|
||||
/// confidence-scaled window (use `decision_window(confidence)`
|
||||
/// to compute it).
|
||||
pub async fn submit_poi_candidate(&self, poi: Poi, proximity: f32) {
|
||||
let now = self.clock.elapsed_ns();
|
||||
let mut inner = self.inner.lock().await;
|
||||
inner.poi_queue.insert(poi, proximity, now);
|
||||
}
|
||||
|
||||
/// AZ-683 — pull the next surfaceable POI subject to the rolling
|
||||
/// 5/min cap. `None` if the cap is hit or no POI clears the
|
||||
/// 40 % confidence floor.
|
||||
pub async fn next_poi_for_surface(&self) -> Option<Poi> {
|
||||
let now = self.clock.elapsed_ns();
|
||||
let mut inner = self.inner.lock().await;
|
||||
let p = inner.poi_queue.next_for_surface(now);
|
||||
if p.is_some() {
|
||||
inner.pois_surfaced_total = inner.pois_surfaced_total.saturating_add(1);
|
||||
}
|
||||
p
|
||||
}
|
||||
|
||||
/// AZ-683 — decline a POI. Returns the dispatchable
|
||||
/// `IgnoredItem` data; the caller (AZ-685 mapobjects dispatch)
|
||||
/// is responsible for persisting it.
|
||||
pub async fn decline_poi(&self, poi_id: Uuid) -> Option<DeclineAction> {
|
||||
let mut inner = self.inner.lock().await;
|
||||
let action = inner.poi_queue.decline(poi_id);
|
||||
if action.is_some() {
|
||||
inner.pois_declined_total = inner.pois_declined_total.saturating_add(1);
|
||||
}
|
||||
action
|
||||
}
|
||||
|
||||
pub async fn poi_queue_len(&self) -> usize {
|
||||
self.inner.lock().await.poi_queue.len()
|
||||
}
|
||||
|
||||
pub async fn pois_in_window(&self) -> usize {
|
||||
let now = self.clock.elapsed_ns();
|
||||
self.inner.lock().await.poi_queue.surfaces_in_window(now)
|
||||
}
|
||||
|
||||
/// Translate an operator command into a trigger and apply it.
|
||||
///
|
||||
/// **AZ-682 mapping** (partial — POI queue lookups belong to
|
||||
/// AZ-683; until then, ConfirmPoi alone has no roi to bind so
|
||||
/// it returns `Validation`).
|
||||
/// AZ-682 / AZ-683 mapping (subset complete):
|
||||
///
|
||||
/// - `MissionAbort` → `Trigger::OperatorAbort`
|
||||
/// - `MissionAbort` → `Trigger::OperatorAbort` (AZ-682).
|
||||
/// - `ReleaseTargetFollow` → `Trigger::OperatorReleaseFollow`
|
||||
/// - `StartTargetFollow` (payload-bound) → not yet supported,
|
||||
/// returns `NotImplemented(AZ-683)` since the target_id has to
|
||||
/// be resolved via the POI queue.
|
||||
/// - `ConfirmPoi` / `DeclinePoi` / `AcknowledgeBitDegraded` /
|
||||
/// `SafetyOverride` → `NotImplemented(AZ-683/AZ-684)`.
|
||||
pub async fn submit_operator_cmd(&self, command: OperatorCommand) -> Result<()> {
|
||||
/// (AZ-682).
|
||||
/// - `DeclinePoi { poi_id }` → queue decline; returns the
|
||||
/// resulting `DeclineAction` in [`SubmitOutcome::Declined`]
|
||||
/// for the caller (AZ-685 mapobjects dispatch) to persist
|
||||
/// (AZ-683).
|
||||
/// - `ConfirmPoi` / `StartTargetFollow` → still
|
||||
/// `NotImplemented(AZ-684)` since ROI / target_id resolution
|
||||
/// needs the evidence ladder.
|
||||
/// - `AcknowledgeBitDegraded` / `SafetyOverride` →
|
||||
/// `NotImplemented(AZ-684)`.
|
||||
pub async fn submit_operator_cmd(&self, command: OperatorCommand) -> Result<SubmitOutcome> {
|
||||
match command.kind {
|
||||
OperatorCommandKind::MissionAbort => {
|
||||
self.submit_trigger(Trigger::OperatorAbort).await;
|
||||
Ok(())
|
||||
Ok(SubmitOutcome::Accepted)
|
||||
}
|
||||
OperatorCommandKind::ReleaseTargetFollow => {
|
||||
self.submit_trigger(Trigger::OperatorReleaseFollow).await;
|
||||
Ok(())
|
||||
Ok(SubmitOutcome::Accepted)
|
||||
}
|
||||
OperatorCommandKind::DeclinePoi => {
|
||||
let poi_id = poi_id_from_payload(&command.payload)?;
|
||||
match self.decline_poi(poi_id).await {
|
||||
Some(action) => Ok(SubmitOutcome::Declined(action)),
|
||||
None => Err(AutopilotError::Validation(format!(
|
||||
"DeclinePoi: unknown poi_id {poi_id}"
|
||||
))),
|
||||
}
|
||||
}
|
||||
OperatorCommandKind::ConfirmPoi | OperatorCommandKind::StartTargetFollow => {
|
||||
Err(AutopilotError::NotImplemented(
|
||||
"scan_controller::submit_operator_cmd (AZ-684 evidence ladder)",
|
||||
))
|
||||
}
|
||||
OperatorCommandKind::ConfirmPoi
|
||||
| OperatorCommandKind::DeclinePoi
|
||||
| OperatorCommandKind::StartTargetFollow => Err(AutopilotError::NotImplemented(
|
||||
"scan_controller::submit_operator_cmd (AZ-683 POI queue wiring)",
|
||||
)),
|
||||
OperatorCommandKind::AcknowledgeBitDegraded => Err(AutopilotError::NotImplemented(
|
||||
"scan_controller::submit_operator_cmd (AZ-684 evidence ladder)",
|
||||
)),
|
||||
@@ -268,6 +364,10 @@ impl ScanControllerHandle {
|
||||
rejected_total: inner.rejected_total,
|
||||
last_state_change_ns: inner.last_state_change_ns,
|
||||
tick_latency_p99_us: inner.p99_us(),
|
||||
poi_queue_len: inner.poi_queue.len(),
|
||||
pois_surfaced_total: inner.pois_surfaced_total,
|
||||
pois_forgotten_total: inner.pois_forgotten_total,
|
||||
pois_declined_total: inner.pois_declined_total,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -276,10 +376,14 @@ impl ScanControllerHandle {
|
||||
let fps_active = inner.fps_guard.is_floor_active();
|
||||
let p99 = inner.p99_us();
|
||||
let state = inner.state;
|
||||
let queue_len = inner.poi_queue.len();
|
||||
drop(inner);
|
||||
|
||||
let mut h = ComponentHealth::green(NAME);
|
||||
let mut details: Vec<String> = vec![format!("state={}", state.discriminant())];
|
||||
let mut details: Vec<String> = vec![
|
||||
format!("state={}", state.discriminant()),
|
||||
format!("poi_queue={queue_len}"),
|
||||
];
|
||||
|
||||
if fps_active {
|
||||
h.level = HealthLevel::Yellow;
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
//! AZ-683 integration tests — POI queue + rate cap + decision-window
|
||||
//! mapping exercised through the public `ScanControllerHandle` API.
|
||||
|
||||
use chrono::{Duration as ChronoDur, Utc};
|
||||
use serde_json::json;
|
||||
use uuid::Uuid;
|
||||
|
||||
use scan_controller::{decision_window, ScanController, SubmitOutcome, SURFACE_CAP_PER_WINDOW};
|
||||
use shared::models::operator::{OperatorCommand, OperatorCommandKind};
|
||||
use shared::models::poi::{Poi, VlmPipelineStatus};
|
||||
|
||||
fn poi(confidence: f32, mgrs: &str) -> Poi {
|
||||
Poi {
|
||||
id: Uuid::new_v4(),
|
||||
confidence,
|
||||
mgrs: mgrs.to_string(),
|
||||
class: "tank".to_string(),
|
||||
class_group: "armor".to_string(),
|
||||
source_detection_ids: vec![],
|
||||
enqueued_at: Utc::now(),
|
||||
priority: 0.0,
|
||||
decline_suppressed: false,
|
||||
vlm_status: VlmPipelineStatus::NotRequested,
|
||||
tier2_evidence: None,
|
||||
deadline: Utc::now() + ChronoDur::seconds(60),
|
||||
}
|
||||
}
|
||||
|
||||
/// AC-1 — priority ordering through the public API.
|
||||
#[tokio::test]
|
||||
async fn ac1_priority_ordering_via_handle() {
|
||||
// Arrange
|
||||
let h = ScanController::new().handle();
|
||||
let p1 = poi(0.9, "1");
|
||||
let p2 = poi(0.6, "2");
|
||||
h.submit_poi_candidate(p1.clone(), 0.5).await;
|
||||
h.submit_poi_candidate(p2.clone(), 0.9).await;
|
||||
|
||||
// Act
|
||||
let first = h.next_poi_for_surface().await.expect("first surface");
|
||||
|
||||
// Assert — p2 (0.6 × 0.9 = 0.54) outranks p1 (0.9 × 0.5 = 0.45).
|
||||
assert_eq!(first.id, p2.id);
|
||||
}
|
||||
|
||||
/// AC-2 — the 5/min cap holds back excess POIs.
|
||||
#[tokio::test]
|
||||
async fn ac2_five_per_minute_cap_via_handle() {
|
||||
// Arrange
|
||||
let h = ScanController::new().handle();
|
||||
for i in 0..10 {
|
||||
h.submit_poi_candidate(poi(0.8, &format!("m{i}")), 0.5)
|
||||
.await;
|
||||
}
|
||||
|
||||
// Act
|
||||
let mut surfaced = 0;
|
||||
for _ in 0..10 {
|
||||
if h.next_poi_for_surface().await.is_some() {
|
||||
surfaced += 1;
|
||||
}
|
||||
}
|
||||
|
||||
// Assert
|
||||
assert_eq!(surfaced, SURFACE_CAP_PER_WINDOW);
|
||||
assert_eq!(h.pois_in_window().await, SURFACE_CAP_PER_WINDOW);
|
||||
assert_eq!(h.poi_queue_len().await, 5);
|
||||
}
|
||||
|
||||
/// AC-3 — decision window linear mapping is exported via the
|
||||
/// `decision_window` re-export. (The pure logic is tested in unit;
|
||||
/// this is the smoke test that the public function is wired up.)
|
||||
#[tokio::test]
|
||||
async fn ac3_decision_window_public_mapping() {
|
||||
// Assert
|
||||
assert_eq!(
|
||||
decision_window(0.40).unwrap().as_secs(),
|
||||
30,
|
||||
"floor maps to 30 s"
|
||||
);
|
||||
assert_eq!(
|
||||
decision_window(1.00).unwrap().as_secs(),
|
||||
120,
|
||||
"ceiling maps to 120 s"
|
||||
);
|
||||
assert!(decision_window(0.39).is_none(), "sub-floor returns None");
|
||||
}
|
||||
|
||||
/// AC-4 — POIs below 40 % confidence enqueue but never surface.
|
||||
#[tokio::test]
|
||||
async fn ac4_below_floor_never_surfaces() {
|
||||
// Arrange
|
||||
let h = ScanController::new().handle();
|
||||
h.submit_poi_candidate(poi(0.39, "low"), 0.9).await;
|
||||
h.submit_poi_candidate(poi(0.20, "lower"), 0.9).await;
|
||||
|
||||
// Act
|
||||
let surfaced = h.next_poi_for_surface().await;
|
||||
|
||||
// Assert
|
||||
assert!(surfaced.is_none(), "sub-40% POIs must not surface");
|
||||
assert_eq!(h.poi_queue_len().await, 2, "POIs remain in queue");
|
||||
}
|
||||
|
||||
/// AC-5 — timeout sweep forgets expired POIs without emitting any
|
||||
/// IgnoredItem.
|
||||
#[tokio::test]
|
||||
async fn ac5_tick_sweep_forgets_expired_pois() {
|
||||
// Arrange — POI with an already-expired deadline.
|
||||
let h = ScanController::new().handle();
|
||||
let mut p = poi(0.8, "expired");
|
||||
p.deadline = Utc::now() - ChronoDur::seconds(1);
|
||||
h.submit_poi_candidate(p, 0.5).await;
|
||||
assert_eq!(h.poi_queue_len().await, 1);
|
||||
|
||||
// Act
|
||||
h.tick().await.expect("tick");
|
||||
|
||||
// Assert
|
||||
assert_eq!(h.poi_queue_len().await, 0);
|
||||
let metrics = h.metrics().await;
|
||||
assert_eq!(metrics.pois_forgotten_total, 1);
|
||||
assert_eq!(metrics.pois_declined_total, 0, "no IgnoredItem on timeout");
|
||||
}
|
||||
|
||||
/// DeclinePoi via operator command returns a `SubmitOutcome::Declined`
|
||||
/// carrying the IgnoredItem payload AZ-685 will persist.
|
||||
#[tokio::test]
|
||||
async fn decline_poi_via_operator_command_emits_action() {
|
||||
// Arrange
|
||||
let h = ScanController::new().handle();
|
||||
let p = poi(0.8, "decline-me");
|
||||
let id = p.id;
|
||||
h.submit_poi_candidate(p, 0.5).await;
|
||||
|
||||
let cmd = OperatorCommand {
|
||||
command_id: Uuid::new_v4(),
|
||||
session_token: "s".to_string(),
|
||||
sequence_number: 1,
|
||||
issued_at_wallclock: Utc::now(),
|
||||
kind: OperatorCommandKind::DeclinePoi,
|
||||
payload: json!({ "poi_id": id.to_string() }),
|
||||
signature: vec![],
|
||||
};
|
||||
|
||||
// Act
|
||||
let outcome = h.submit_operator_cmd(cmd).await.expect("decline accepted");
|
||||
|
||||
// Assert
|
||||
match outcome {
|
||||
SubmitOutcome::Declined(action) => {
|
||||
assert_eq!(action.poi_id, id);
|
||||
assert_eq!(action.mgrs, "decline-me");
|
||||
assert_eq!(action.class_group, "armor");
|
||||
}
|
||||
SubmitOutcome::Accepted => panic!("decline must return Declined action"),
|
||||
}
|
||||
assert_eq!(h.poi_queue_len().await, 0);
|
||||
}
|
||||
Reference in New Issue
Block a user