Merge pull request #118 from FelineStateMachine/feat/decouple-adaptive-commit-awaiting

FelineStateMachine committed Jul 2372bf212

refactor(host): decouple adaptive commits from feedback cadence

commit: 72bf212524d40138ce1ebd85cfa98034feb16023
parent: 9340a142991bc687226ac231af3797a51799095a
parent: d5a0c34a28d06f570d16670efa9167f948fea1da
author: FelineStateMachine <dami.etoile@gmail.com>

3 files changed, +1107 -280wrap lines

M	crates/sigil-host/Cargo.toml	1	+
M	crates/sigil-host/src/server/adaptive.rs	1372	++++++++++++++++++++++++------
M	crates/sigil-host/src/source/in_process.rs	14	+

M crates/sigil-host/Cargo.toml +1 -0

@@ -56,3 +56,4 @@ x11rb = { version = "0.13.2", default-features = false }

5656 5757 [dev-dependencies]5858 tempfile = "3"59+tokio = { version = "1", features = ["test-util"] }

M crates/sigil-host/src/server/adaptive.rs +1092 -280

@@ -1,5 +1,6 @@
11 use super::session::{2-    AdaptiveEncoderProposal, MediaV3TelemetrySnapshot, ResolutionEncoderProposal, SessionRegistry,2+    AdaptiveEncoderProposal, FeedbackLease, MediaV3TelemetrySnapshot, ResolutionEncoderProposal,3+    SessionRegistry,34 };45 use super::*;56 @@ -260,15 +261,6 @@ impl MotionResolutionPolicy {
260261         }261262     }262263 263-    fn commit_observation_from(&mut self, candidate: &Self) {264-        self.motion_active = candidate.motion_active;265-        self.still_since = candidate.still_since;266-        self.pressure_active = candidate.pressure_active;267-        self.moderate_windows = candidate.moderate_windows;268-        self.clean_windows = candidate.clean_windows;269-        self.last_motion_sequence = candidate.last_motion_sequence;270-    }271-272264     fn downshift(273265         &mut self,274266         observed_at: Instant,@@ -825,15 +817,6 @@ impl ShadowBitrateController {
825817             trusted_host_pressure,826818         )827819     }828-829-    fn commit_observation_from(&mut self, candidate: &Self) {830-        self.next_decision_id = candidate.next_decision_id;831-        self.last_report_id = candidate.last_report_id;832-        self.last_report_received_at = candidate.last_report_received_at;833-        self.previous_telemetry = candidate.previous_telemetry;834-        self.baseline_rtt_micros = candidate.baseline_rtt_micros;835-        self.last_sequence = candidate.last_sequence;836-    }837820 }838821 839822 fn add_feedback_signal(@@ -860,6 +843,265 @@ fn queue_ratio_at_least(depth: u8, capacity: u8, numerator: u16, denominator: u1
860843             >= u16::from(capacity).saturating_mul(numerator)861844 }862845 846+#[derive(Clone, Copy, Debug)]847+struct AdaptiveCommitPlan {848+    decision: AdaptiveBitrateDecisionV1,849+    telemetry: MediaV3TelemetrySnapshot,850+    resolution_decision: Option<MotionResolutionDecision>,851+    force_keyframe: bool,852+}853+854+#[derive(Debug)]855+struct AdaptiveCommitAcknowledgement {856+    plan: AdaptiveCommitPlan,857+    generation_ended: bool,858+    recovery_acknowledged: Option<bool>,859+    resolution_applied: Option<bool>,860+    applied_bitrate_kbps: Option<u32>,861+    applied_resolution: Option<VideoDimensions>,862+}863+864+impl AdaptiveCommitAcknowledgement {865+    fn immediate(plan: AdaptiveCommitPlan) -> Self {866+        Self {867+            plan,868+            generation_ended: false,869+            recovery_acknowledged: None,870+            resolution_applied: None,871+            applied_bitrate_kbps: None,872+            applied_resolution: None,873+        }874+    }875+}876+877+/// Bounded adaptive-control actuator.878+///879+/// Feedback evaluation remains synchronous and deterministic, while this880+/// coordinator owns the slower encoder acknowledgements. At most one commit881+/// task and one latest coalesced plan exist. A queued plan carries the latest882+/// desired bitrate and resolution even when its feedback decision is a hold,883+/// so a slow or failed predecessor cannot strand an uncommitted target.884+struct AdaptiveCommitCoordinator {885+    sessions: Arc<SessionRegistry>,886+    remote: EndpointId,887+    session_id: u64,888+    actuation_enabled: bool,889+    desired_bitrate_kbps: u32,890+    requested_bitrate_kbps: u32,891+    committed_bitrate_kbps: u32,892+    desired_resolution: Option<VideoDimensions>,893+    requested_resolution: Option<VideoDimensions>,894+    committed_resolution: Option<VideoDimensions>,895+    pending: bool,896+    queued: Option<AdaptiveCommitPlan>,897+    acknowledgements: tokio::task::JoinSet<AdaptiveCommitAcknowledgement>,898+}899+900+impl AdaptiveCommitCoordinator {901+    fn new(902+        sessions: Arc<SessionRegistry>,903+        remote: EndpointId,904+        session_id: u64,905+        actuation_enabled: bool,906+        committed_bitrate_kbps: u32,907+        committed_resolution: Option<VideoDimensions>,908+    ) -> Self {909+        Self {910+            sessions,911+            remote,912+            session_id,913+            actuation_enabled,914+            desired_bitrate_kbps: committed_bitrate_kbps,915+            requested_bitrate_kbps: committed_bitrate_kbps,916+            committed_bitrate_kbps,917+            desired_resolution: committed_resolution,918+            requested_resolution: committed_resolution,919+            committed_resolution,920+            pending: false,921+            queued: None,922+            acknowledgements: tokio::task::JoinSet::new(),923+        }924+    }925+926+    fn has_pending(&self) -> bool {927+        self.pending928+    }929+930+    fn bitrate_is_applied(&self, target_kbps: u32) -> bool {931+        self.actuation_enabled932+            && !self.pending933+            && target_kbps == self.requested_bitrate_kbps934+            && target_kbps == self.committed_bitrate_kbps935+    }936+937+    fn submit(&mut self, mut plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {938+        self.desired_bitrate_kbps = plan.decision.target_kbps;939+        if let Some(resolution) = plan.resolution_decision {940+            self.desired_resolution = Some(resolution.target);941+        }942+        if self.pending {943+            if let Some(queued) = self.queued {944+                // The latest decision replaces older diagnostics, but a945+                // recovery barrier requested by any coalesced desired update946+                // must survive until that desired state is actuated.947+                plan.force_keyframe |= queued.force_keyframe;948+            }949+            self.queued = Some(plan);950+            return None;951+        }952+        if let Some(queued) = self.queued.take() {953+            // A completion can deliberately defer this slot while an unread954+            // feedback generation is waiting. The newly evaluated plan is the955+            // actual latest state, but it must inherit any recovery barrier956+            // carried by the superseded slot.957+            plan.force_keyframe |= queued.force_keyframe;958+        }959+        self.start(plan)960+    }961+962+    fn start_queued(&mut self) -> Option<AdaptiveCommitAcknowledgement> {963+        let plan = self.queued.take()?;964+        self.start(plan)965+    }966+967+    fn start(&mut self, plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {968+        if !self.actuation_enabled {969+            return Some(AdaptiveCommitAcknowledgement::immediate(plan));970+        }971+972+        let desired_resolution = plan.resolution_decision.map(|change| change.target);973+        debug_assert_eq!(plan.decision.target_kbps, self.desired_bitrate_kbps);974+        if desired_resolution.is_some() {975+            debug_assert_eq!(desired_resolution, self.desired_resolution);976+        }977+978+        // A proposal mutates EncoderControl's persistent desired watch before979+        // its acknowledgement arrives. Reconcile against both the last980+        // requested and last committed values: after a timeout, returning to981+        // the old committed target must overwrite the still-pending stale982+        // request rather than allowing it to apply later.983+        let reconciles_resolution = desired_resolution.is_some()984+            && (desired_resolution != self.requested_resolution985+                || desired_resolution != self.committed_resolution);986+        let resolution_proposal = if reconciles_resolution {987+            let target = desired_resolution.expect("resolution target was checked as present");988+            match self989+                .sessions990+                .propose_resolution_update(self.remote, self.session_id, target)991+            {992+                Ok(Some(proposal)) => {993+                    self.requested_resolution = Some(target);994+                    Some(proposal)995+                }996+                Ok(None) => {997+                    warn!(998+                        remote = %self.remote,999+                        session_id = self.session_id,1000+                        "active adaptive session lost encoder control; retaining committed resolution"1001+                    );1002+                    None1003+                }1004+                Err(error) => {1005+                    warn!(1006+                        remote = %self.remote,1007+                        session_id = self.session_id,1008+                        %error,1009+                        "adaptive resolution proposal failed; retaining committed resolution"1010+                    );1011+                    None1012+                }1013+            }1014+        } else {1015+            None1016+        };1017+1018+        let reconciles_bitrate = plan.decision.target_kbps != self.requested_bitrate_kbps1019+            || plan.decision.target_kbps != self.committed_bitrate_kbps;1020+        let force_keyframe = plan.force_keyframe && resolution_proposal.is_none();1021+        let bitrate_proposal = if reconciles_bitrate || force_keyframe {1022+            match self.sessions.propose_adaptive_encoder_update(1023+                self.remote,1024+                self.session_id,1025+                plan.decision.target_kbps,1026+                force_keyframe,1027+            ) {1028+                Ok(Some(proposal)) => {1029+                    self.requested_bitrate_kbps = plan.decision.target_kbps;1030+                    Some(proposal)1031+                }1032+                Ok(None) => {1033+                    warn!(1034+                        remote = %self.remote,1035+                        session_id = self.session_id,1036+                        "active adaptive session lost encoder control; retaining committed bitrate"1037+                    );1038+                    None1039+                }1040+                Err(error) => {1041+                    warn!(1042+                        remote = %self.remote,1043+                        session_id = self.session_id,1044+                        %error,1045+                        "adaptive encoder proposal failed; retaining committed bitrate"1046+                    );1047+                    None1048+                }1049+            }1050+        } else {1051+            None1052+        };1053+1054+        if bitrate_proposal.is_none() && resolution_proposal.is_none() {1055+            return Some(AdaptiveCommitAcknowledgement::immediate(plan));1056+        }1057+1058+        self.pending = true;1059+        let sessions = Arc::clone(&self.sessions);1060+        let remote = self.remote;1061+        let session_id = self.session_id;1062+        self.acknowledgements.spawn(async move {1063+            run_adaptive_commit(1064+                sessions,1065+                remote,1066+                session_id,1067+                plan,1068+                bitrate_proposal,1069+                resolution_proposal,1070+            )1071+            .await1072+        });1073+        None1074+    }1075+1076+    fn complete(1077+        &mut self,1078+        result: Option<std::result::Result<AdaptiveCommitAcknowledgement, tokio::task::JoinError>>,1079+    ) -> Result<AdaptiveCommitAcknowledgement> {1080+        ensure!(1081+            self.pending,1082+            "adaptive commit completed without pending work"1083+        );1084+        self.pending = false;1085+        let acknowledgement = result1086+            .context("adaptive commit task ended without a result")?1087+            .context("adaptive commit task failed")?;1088+        if let Some(applied) = acknowledgement.applied_bitrate_kbps {1089+            self.committed_bitrate_kbps = applied;1090+        }1091+        if let Some(applied) = acknowledgement.applied_resolution {1092+            self.committed_resolution = Some(applied);1093+        }1094+        Ok(acknowledgement)1095+    }1096+1097+    async fn abort_and_drain(&mut self) {1098+        self.pending = false;1099+        self.queued = None;1100+        self.acknowledgements.abort_all();1101+        while self.acknowledgements.join_next().await.is_some() {}1102+    }1103+}1104+8631105 pub(super) async fn serve_media_feedback(8641106     connection: Connection,8651107     config: &HostConfig,@@ -924,7 +1166,45 @@ pub(super) async fn serve_media_feedback(
9241166 9251167     let (feedback_sender, mut feedback_receiver) =9261168         tokio::sync::watch::channel(CumulativeFeedbackIngress::default());927-    let mut reader_task = tokio::spawn(forward_media_feedback_reports(recv, feedback_sender));1169+    let reader_task = tokio::spawn(forward_media_feedback_reports(recv, feedback_sender));1170+    let connection_closed = async {1171+        let closed = connection.closed().await;1172+        debug!(%remote, ?closed, "media feedback connection closed");1173+    };1174+    let result = run_media_feedback_loop(1175+        &mut send,1176+        connection_closed,1177+        remote,1178+        sessions,1179+        &lease,1180+        encoder_actuation_available,1181+        ceiling_kbps,1182+        &mut feedback_receiver,1183+        reader_task,1184+    )1185+    .await;1186+    let _ = send.finish();1187+    drop(lease);1188+    info!(%remote, "media feedback client released");1189+    result1190+}1191+1192+#[allow(clippy::too_many_arguments)]1193+async fn run_media_feedback_loop<W, C>(1194+    send: &mut W,1195+    connection_closed: C,1196+    remote: EndpointId,1197+    sessions: &Arc<SessionRegistry>,1198+    lease: &FeedbackLease,1199+    encoder_actuation_available: bool,1200+    ceiling_kbps: u32,1201+    feedback_receiver: &mut tokio::sync::watch::Receiver<CumulativeFeedbackIngress>,1202+    mut reader_task: tokio::task::JoinHandle<Result<()>>,1203+) -> Result<()>1204+where1205+    W: tokio::io::AsyncWrite + Unpin,1206+    C: Future<Output = ()>,1207+{9281208     let mut reader_finished = false;9291209     let mut receiver_open = true;9301210     let mut controller = ShadowBitrateController::new(ceiling_kbps, lease.telemetry.snapshot());@@ -938,267 +1218,317 @@ pub(super) async fn serve_media_feedback(
9381218     } else {9391219         None9401220     };1221+    let initial_resolution = resolution_controller1222+        .as_ref()1223+        .map(MotionResolutionPolicy::target);1224+    let mut commit_coordinator = AdaptiveCommitCoordinator::new(1225+        Arc::clone(sessions),1226+        remote,1227+        lease.session_id,1228+        encoder_actuation_available,1229+        ceiling_kbps,1230+        initial_resolution,1231+    );9411232     let mut feedback_cursor = FeedbackIngressCursor::default();942-943-    let result: Result<()> = loop {944-        let session_changed = sessions.session_changed.notified();945-        tokio::pin!(session_changed);946-        session_changed.as_mut().enable();947-        if !sessions.is_active(remote, lease.session_id) {948-            break Ok(());949-        }950-        tokio::select! {951-            biased;952-            reader_result = &mut reader_task, if !reader_finished => {953-                reader_finished = true;954-                match reader_result {955-                    Ok(Ok(())) => {956-                        debug!(%remote, "media feedback report stream closed");957-                    }958-                    Ok(Err(error)) => {959-                        break Err(error).context("reading media feedback reports");960-                    }961-                    Err(error) => {962-                        break Err(error).context("media feedback reader task failed");1233+    tokio::pin!(connection_closed);1234+1235+    // Keep cleanup on every error path: `?` below exits this async block, not1236+    // the outer function that owns the reader and actuator tasks.1237+    let result: Result<()> = async {1238+        loop {1239+            let session_changed = sessions.session_changed.notified();1240+            tokio::pin!(session_changed);1241+            session_changed.as_mut().enable();1242+            if !sessions.is_active(remote, lease.session_id) {1243+                break Ok(());1244+            }1245+            tokio::select! {1246+                biased;1247+                reader_result = &mut reader_task, if !reader_finished => {1248+                    reader_finished = true;1249+                    match reader_result {1250+                        Ok(Ok(())) => {1251+                            debug!(%remote, "media feedback report stream closed");1252+                        }1253+                        Ok(Err(error)) => {1254+                            break Err(error).context("reading media feedback reports");1255+                        }1256+                        Err(error) => {1257+                            break Err(error).context("media feedback reader task failed");1258+                        }9631259                     }9641260                 }965-            }966-            changed = feedback_receiver.changed(), if receiver_open => {967-                if changed.is_err() {968-                    receiver_open = false;969-                    if reader_finished {1261+                commit_result = commit_coordinator.acknowledgements.join_next(),1262+                    if commit_coordinator.has_pending() =>1263+                {1264+                    let acknowledgement = commit_coordinator.complete(commit_result)?;1265+                    if acknowledgement.generation_ended {9701266                         break Ok(());9711267                     }972-                    continue;973-                }974-                let ingress = *feedback_receiver.borrow_and_update();975-                let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {976-                    continue;977-                };978-                // This branch now owns the complete aggregate. Advance the979-                // ingress baseline exactly once before evaluation; reports980-                // arriving during an encoder wait accumulate beyond it.981-                feedback_cursor = next_cursor;982-                let telemetry = lease.telemetry.snapshot();983-                let observed_at = Instant::now();984-                let mut candidate = controller.clone();985-                let evaluation = candidate.evaluate(&report, telemetry, observed_at)?;986-                let mut decision = evaluation.decision;987-                let mut resolution_candidate = resolution_controller.clone();988-                let resolution_decision = resolution_candidate989-                    .as_mut()990-                    .map(|policy| policy.observe_window(&report, &evaluation, observed_at));991-                let changes_bitrate = decision.state != AdaptiveBitrateStateV1::Hold;992-                let changes_resolution = resolution_decision.is_some_and(|change| change.changed);993-                let can_actuate_bitrate = changes_bitrate && encoder_actuation_available;994-                let can_actuate_resolution = changes_resolution && encoder_actuation_available;995-                let mut recovery_acknowledged = None;996-                let bitrate_proposal = if can_actuate_bitrate {997-                    let force_keyframe =998-                        adaptive_recovery_keyframe_required(&report, &decision)999-                            && !changes_resolution;1000-                    match sessions.propose_adaptive_encoder_update(1268+                    log_adaptive_acknowledgement(10011269                         remote,10021270                         lease.session_id,1003-                        decision.target_kbps,1004-                        force_keyframe,1005-                    ) {1006-                        Ok(Some(proposal)) => Some(proposal),1007-                        Ok(None) => {1008-                            warn!(1009-                                %remote,1010-                                session_id = lease.session_id,1011-                                "active adaptive session lost encoder control; retaining committed bitrate"1012-                            );1013-                            None1014-                        }1015-                        Err(error) => {1016-                            warn!(1017-                                %remote,1018-                                session_id = lease.session_id,1019-                                %error,1020-                                "adaptive encoder proposal failed; retaining committed bitrate"1021-                            );1022-                            None1023-                        }1271+                        encoder_actuation_available,1272+                        &commit_coordinator,1273+                        &acknowledgement,1274+                    );1275+                    // If a newer report is already waiting, let its evaluation1276+                    // replace the queued actuator slot before starting more1277+                    // work. This avoids spending another full timeout on a1278+                    // plan that was stale at completion time.1279+                    if !matches!(feedback_receiver.has_changed(), Ok(true))1280+                        && let Some(immediate) = commit_coordinator.start_queued()1281+                    {1282+                        log_adaptive_acknowledgement(1283+                            remote,1284+                            lease.session_id,1285+                            encoder_actuation_available,1286+                            &commit_coordinator,1287+                            &immediate,1288+                        );10241289                     }1025-                } else {1026-                    None1027-                };1028-                let resolution_proposal = if can_actuate_resolution {1029-                    let target = resolution_decision1030-                        .expect("resolution change is required for actuation")1031-                        .target;1032-                    match sessions.propose_resolution_update(remote, lease.session_id, target) {1033-                        Ok(Some(proposal)) => Some(proposal),1034-                        Ok(None) => {1035-                            warn!(1036-                                %remote,1037-                                session_id = lease.session_id,1038-                                "active adaptive session lost encoder control; retaining committed resolution"1039-                            );1040-                            None1041-                        }1042-                        Err(error) => {1043-                            warn!(1044-                                %remote,1045-                                session_id = lease.session_id,1046-                                %error,1047-                                "adaptive resolution proposal failed; retaining committed resolution"1048-                            );1049-                            None1290+                }1291+                changed = feedback_receiver.changed(), if receiver_open => {1292+                    if changed.is_err() {1293+                        receiver_open = false;1294+                        if reader_finished {1295+                            break Ok(());10501296                         }1297+                        continue;10511298                     }1052-                } else {1053-                    None1054-                };1299+                    let ingress = *feedback_receiver.borrow_and_update();1300+                    let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {1301+                        continue;1302+                    };1303+                    // This branch now owns the complete aggregate. Advance the1304+                    // ingress baseline exactly once before evaluation; reports1305+                    // arriving during an encoder wait accumulate beyond it.1306+                    feedback_cursor = next_cursor;1307+                    let telemetry = lease.telemetry.snapshot();1308+                    let observed_at = Instant::now();1309+                    let mut candidate = controller.clone();1310+                    let evaluation = candidate.evaluate(&report, telemetry, observed_at)?;1311+                    let mut decision = evaluation.decision;1312+                    decision.applied =1313+                        commit_coordinator.bitrate_is_applied(decision.target_kbps);1314+                    let mut resolution_candidate = resolution_controller.clone();1315+                    let resolution_decision = resolution_candidate1316+                        .as_mut()1317+                        .map(|policy| policy.observe_window(&report, &evaluation, observed_at));1318+                    let force_keyframe =1319+                        adaptive_recovery_keyframe_required(&report, &decision);1320+                    let plan = AdaptiveCommitPlan {1321+                        decision,1322+                        telemetry,1323+                        resolution_decision,1324+                        force_keyframe,1325+                    };1326+1327+                    // One fixed-width decision is written for every evaluated1328+                    // report before actuator work is submitted. Encoder1329+                    // acknowledgements are correlated in bounded host1330+                    // diagnostics and never gate the Portal-facing cadence.1331+                    write_adaptive_decision(1332+                        send,1333+                        remote,1334+                        lease.session_id,1335+                        encoder_actuation_available,1336+                        &plan,1337+                    )1338+                    .await?;10551339 1056-                if let Some(proposal) = bitrate_proposal {1057-                        match commit_adaptive_encoder_proposal(1058-                            sessions,1340+                    // Evaluation state advances at feedback cadence. The1341+                    // actuator independently retains the exact requested,1342+                    // committed, and latest desired targets.1343+                    controller = candidate;1344+                    resolution_controller = resolution_candidate;1345+                    if let Some(immediate) = commit_coordinator.submit(plan) {1346+                        log_adaptive_acknowledgement(10591347                             remote,10601348                             lease.session_id,1061-                            &proposal,1062-                        )1063-                        .await1064-                        {1065-                            AdaptiveEncoderCommit::GenerationEnded => break Ok(()),1066-                            AdaptiveEncoderCommit::NotApplied(error) => {1067-                                warn!(1068-                                    %remote,1069-                                    session_id = lease.session_id,1070-                                    %error,1071-                                    "adaptive encoder application failed; retaining committed bitrate"1072-                                );1073-                            }1074-                            AdaptiveEncoderCommit::Applied => {1075-                                decision.applied = true;1076-                                controller = candidate.clone();1077-                                if proposal.force_keyframe_revision.is_some() {1078-                                    match await_adaptive_recovery_keyframe(1079-                                        sessions,1080-                                        remote,1081-                                        lease.session_id,1082-                                        &proposal,1083-                                    )1084-                                    .await1085-                                    {1086-                                        None => break Ok(()),1087-                                        Some(Ok(())) => recovery_acknowledged = Some(true),1088-                                        Some(Err(error)) => {1089-                                            recovery_acknowledged = Some(false);1090-                                            warn!(1091-                                                %remote,1092-                                                session_id = lease.session_id,1093-                                                %error,1094-                                                "adaptive bitrate applied but forced-IDR recovery was not acknowledged"1095-                                            );1096-                                        }1097-                                    }1098-                                }1099-                            }1100-                        }1349+                            encoder_actuation_available,1350+                            &commit_coordinator,1351+                            &immediate,1352+                        );1353+                    }11011354                 }1102-                if can_actuate_bitrate {1103-                    if !decision.applied {1104-                        controller.commit_observation_from(&candidate);1355+                _ = &mut session_changed => {1356+                    if !sessions.is_active(remote, lease.session_id) {1357+                        break Ok(());11051358                     }1106-                } else {1107-                    // External gst-launch and sources without EncoderControl1108-                    // remain intentionally shadow-only. Hold decisions do not1109-                    // need encoder actuation.1110-                    controller = candidate;11111359                 }1360+                _ = &mut connection_closed => {1361+                    break Ok(());1362+                }1363+            }1364+            if reader_finished1365+                && !receiver_open1366+                && !commit_coordinator.has_pending()1367+                && commit_coordinator.queued.is_none()1368+            {1369+                break Ok(());1370+            }1371+        }1372+    }1373+    .await;11121374 1113-                let mut resolution_applied = None;1114-                if let Some(proposal) = resolution_proposal {1115-                    match commit_resolution_encoder_proposal(1116-                        sessions,1117-                        remote,1118-                        lease.session_id,1119-                        &proposal,1120-                    )1121-                    .await1375+    if !reader_finished {1376+        reader_task.abort();1377+        let _ = reader_task.await;1378+    }1379+    commit_coordinator.abort_and_drain().await;1380+    result1381+}1382+1383+async fn write_adaptive_decision<W>(1384+    send: &mut W,1385+    remote: EndpointId,1386+    session_id: u64,1387+    encoder_actuation_available: bool,1388+    plan: &AdaptiveCommitPlan,1389+) -> Result<()>1390+where1391+    W: tokio::io::AsyncWrite + Unpin,1392+{1393+    let decision = plan.decision;1394+    let telemetry = plan.telemetry;1395+    let resolution_decision = plan.resolution_decision;1396+    tokio::time::timeout(1397+        FEEDBACK_WRITE_TIMEOUT,1398+        write_adaptive_bitrate_decision_v1(send, &decision),1399+    )1400+    .await1401+    .context("timed out writing adaptive bitrate decision")??;1402+    debug!(1403+        %remote,1404+        session_id,1405+        report_id = decision.report_id,1406+        decision_id = decision.decision_id,1407+        target_kbps = decision.target_kbps,1408+        ?decision.state,1409+        reasons = decision.reasons.bits(),1410+        applied = decision.applied,1411+        resolution_width = resolution_decision.map(|change| change.target.width),1412+        resolution_height = resolution_decision.map(|change| change.target.height),1413+        resolution_trigger = ?resolution_decision.map(|change| change.trigger),1414+        path_rtt_micros = telemetry.selected_path_rtt_micros,1415+        path_lost_packets = telemetry.selected_path_lost_packets,1416+        path_congestion_events = telemetry.selected_path_congestion_events,1417+        scheduler_cancellations = telemetry.scheduler_cancellations,1418+        send_failures = telemetry.send_failures,1419+        mode = if encoder_actuation_available { "active" } else { "shadow" },1420+        "adaptive bitrate decision"1421+    );1422+    Ok(())1423+}1424+1425+fn log_adaptive_acknowledgement(1426+    remote: EndpointId,1427+    session_id: u64,1428+    encoder_actuation_available: bool,1429+    coordinator: &AdaptiveCommitCoordinator,1430+    acknowledgement: &AdaptiveCommitAcknowledgement,1431+) {1432+    let decision = acknowledgement.plan.decision;1433+    let resolution_decision = acknowledgement.plan.resolution_decision;1434+    debug!(1435+        %remote,1436+        session_id,1437+        report_id = decision.report_id,1438+        decision_id = decision.decision_id,1439+        target_kbps = decision.target_kbps,1440+        bitrate_applied = acknowledgement.applied_bitrate_kbps.is_some(),1441+        applied_bitrate_kbps = ?acknowledgement.applied_bitrate_kbps,1442+        recovery_acknowledged = ?acknowledgement.recovery_acknowledged,1443+        resolution_width = resolution_decision.map(|change| change.target.width),1444+        resolution_height = resolution_decision.map(|change| change.target.height),1445+        resolution_applied = ?acknowledgement.resolution_applied,1446+        applied_resolution = ?acknowledgement.applied_resolution,1447+        desired_bitrate_kbps = coordinator.desired_bitrate_kbps,1448+        requested_bitrate_kbps = coordinator.requested_bitrate_kbps,1449+        committed_bitrate_kbps = coordinator.committed_bitrate_kbps,1450+        desired_resolution = ?coordinator.desired_resolution,1451+        requested_resolution = ?coordinator.requested_resolution,1452+        committed_resolution = ?coordinator.committed_resolution,1453+        mode = if encoder_actuation_available { "active" } else { "shadow" },1454+        "adaptive actuator acknowledgement"1455+    );1456+}1457+1458+async fn run_adaptive_commit(1459+    sessions: Arc<SessionRegistry>,1460+    remote: EndpointId,1461+    session_id: u64,1462+    plan: AdaptiveCommitPlan,1463+    bitrate_proposal: Option<AdaptiveEncoderProposal>,1464+    resolution_proposal: Option<ResolutionEncoderProposal>,1465+) -> AdaptiveCommitAcknowledgement {1466+    let mut acknowledgement = AdaptiveCommitAcknowledgement::immediate(plan);1467+    if let Some(proposal) = bitrate_proposal {1468+        match commit_adaptive_encoder_proposal(&sessions, remote, session_id, &proposal).await {1469+            AdaptiveEncoderCommit::GenerationEnded => {1470+                acknowledgement.generation_ended = true;1471+                return acknowledgement;1472+            }1473+            AdaptiveEncoderCommit::NotApplied(error) => {1474+                warn!(1475+                    %remote,1476+                    session_id,1477+                    %error,1478+                    "adaptive encoder application failed; retaining committed bitrate"1479+                );1480+            }1481+            AdaptiveEncoderCommit::Applied => {1482+                acknowledgement.plan.decision.applied = true;1483+                acknowledgement.applied_bitrate_kbps = Some(proposal.target_kbps);1484+                if proposal.force_keyframe_revision.is_some() {1485+                    match await_adaptive_recovery_keyframe(&sessions, remote, session_id, &proposal)1486+                        .await11221487                     {1123-                        AdaptiveEncoderCommit::GenerationEnded => break Ok(()),1124-                        AdaptiveEncoderCommit::NotApplied(error) => {1125-                            resolution_applied = Some(false);1488+                        None => {1489+                            acknowledgement.generation_ended = true;1490+                            return acknowledgement;1491+                        }1492+                        Some(Ok(())) => acknowledgement.recovery_acknowledged = Some(true),1493+                        Some(Err(error)) => {1494+                            acknowledgement.recovery_acknowledged = Some(false);11261495                             warn!(11271496                                 %remote,1128-                                session_id = lease.session_id,1129-                                target_width = proposal.target.width,1130-                                target_height = proposal.target.height,1497+                                session_id,11311498                                 %error,1132-                                "adaptive resolution application failed; retaining committed resolution"1499+                                "adaptive bitrate applied but forced-IDR recovery was not acknowledged"11331500                             );11341501                         }1135-                        AdaptiveEncoderCommit::Applied => {1136-                            resolution_applied = Some(true);1137-                            resolution_controller = resolution_candidate.clone();1138-                        }11391502                     }11401503                 }1141-                if can_actuate_resolution && resolution_applied != Some(true) {1142-                    if let (Some(committed), Some(candidate)) =1143-                        (resolution_controller.as_mut(), resolution_candidate.as_ref())1144-                    {1145-                        committed.commit_observation_from(candidate);1146-                    }1147-                } else if !changes_resolution {1148-                    resolution_controller = resolution_candidate;1149-                }1150-                tokio::time::timeout(1151-                    FEEDBACK_WRITE_TIMEOUT,1152-                    write_adaptive_bitrate_decision_v1(&mut send, &decision),1153-                )1154-                .await1155-                .context("timed out writing adaptive bitrate decision")??;1156-                debug!(1504+            }1505+        }1506+    }1507+1508+    if let Some(proposal) = resolution_proposal {1509+        match commit_resolution_encoder_proposal(&sessions, remote, session_id, &proposal).await {1510+            AdaptiveEncoderCommit::GenerationEnded => {1511+                acknowledgement.generation_ended = true;1512+                return acknowledgement;1513+            }1514+            AdaptiveEncoderCommit::NotApplied(error) => {1515+                acknowledgement.resolution_applied = Some(false);1516+                warn!(11571517                     %remote,1158-                    session_id = lease.session_id,1159-                    report_id = decision.report_id,1160-                    decision_id = decision.decision_id,1161-                    target_kbps = decision.target_kbps,1162-                    ?decision.state,1163-                    reasons = decision.reasons.bits(),1164-                    applied = decision.applied,1165-                    ?recovery_acknowledged,1166-                    resolution_width = resolution_decision.map(|change| change.target.width),1167-                    resolution_height = resolution_decision.map(|change| change.target.height),1168-                    resolution_trigger = ?resolution_decision.map(|change| change.trigger),1169-                    ?resolution_applied,1170-                    path_rtt_micros = telemetry.selected_path_rtt_micros,1171-                    path_lost_packets = telemetry.selected_path_lost_packets,1172-                    path_congestion_events = telemetry.selected_path_congestion_events,1173-                    scheduler_cancellations = telemetry.scheduler_cancellations,1174-                    send_failures = telemetry.send_failures,1175-                    mode = if encoder_actuation_available { "active" } else { "shadow" },1176-                    "adaptive bitrate decision"1518+                    session_id,1519+                    target_width = proposal.target.width,1520+                    target_height = proposal.target.height,1521+                    %error,1522+                    "adaptive resolution application failed; retaining committed resolution"11771523                 );11781524             }1179-            _ = &mut session_changed => {1180-                if !sessions.is_active(remote, lease.session_id) {1181-                    break Ok(());1182-                }1183-            }1184-            closed = connection.closed() => {1185-                debug!(%remote, ?closed, "media feedback connection closed");1186-                break Ok(());1525+            AdaptiveEncoderCommit::Applied => {1526+                acknowledgement.resolution_applied = Some(true);1527+                acknowledgement.applied_resolution = Some(proposal.target);11871528             }11881529         }1189-        if reader_finished && !receiver_open {1190-            break Ok(());1191-        }1192-    };1193-1194-    if !reader_finished {1195-        reader_task.abort();1196-        let _ = reader_task.await;11971530     }1198-    let _ = send.finish();1199-    drop(lease);1200-    info!(%remote, "media feedback client released");1201-    result1531+    acknowledgement12021532 }12031533 12041534 async fn forward_media_feedback_reports<R>(@@ -1439,6 +1769,25 @@ mod tests {
14391769             .unwrap()14401770     }14411771 1772+    fn evaluate_commit_plan(1773+        controller: &mut ShadowBitrateController,1774+        resolution_controller: &mut Option<MotionResolutionPolicy>,1775+        report: &MediaFeedbackReportV1,1776+        observed_at: Instant,1777+    ) -> AdaptiveCommitPlan {1778+        let telemetry = MediaV3TelemetrySnapshot::default();1779+        let evaluation = controller.evaluate(report, telemetry, observed_at).unwrap();1780+        let resolution_decision = resolution_controller1781+            .as_mut()1782+            .map(|policy| policy.observe_window(report, &evaluation, observed_at));1783+        AdaptiveCommitPlan {1784+            decision: evaluation.decision,1785+            telemetry,1786+            resolution_decision,1787+            force_keyframe: adaptive_recovery_keyframe_required(report, &evaluation.decision),1788+        }1789+    }1790+14421791     fn native_dimensions() -> VideoDimensions {14431792         VideoDimensions {14441793             width: 1_280,@@ -2155,36 +2504,6 @@ mod tests {
21552504         ));21562505     }21572506 2158-    #[test]2159-    fn failed_adaptive_application_commits_observation_but_not_target() {2160-        let mut committed =2161-            ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2162-        controller_decide(&mut committed, &clean_feedback(1), 0);2163-        let old_target = committed.target_kbps;2164-        let old_decision_id = committed.next_decision_id;2165-        let mut severe = clean_feedback(2);2166-        severe.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2167-2168-        let mut candidate = committed.clone();2169-        let decision = controller_decide(&mut candidate, &severe, 1);2170-        assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);2171-        assert!(candidate.target_kbps < old_target);2172-        committed.commit_observation_from(&candidate);2173-2174-        assert_eq!(committed.target_kbps, old_target);2175-        assert_eq!(committed.last_report_id, Some(2));2176-        assert_eq!(committed.next_decision_id, old_decision_id + 1);2177-        assert!(2178-            committed2179-                .decide(2180-                    &clean_feedback(3),2181-                    MediaV3TelemetrySnapshot::default(),2182-                    Instant::now() + Duration::from_secs(2)2183-                )2184-                .is_ok()2185-        );2186-    }2187-21882507     #[test]21892508     fn adaptive_recovery_idr_is_only_requested_for_a_bitrate_change_during_resync() {21902509         let mut report = clean_feedback(1);@@ -2306,4 +2625,497 @@ mod tests {
23062625             "a clamped aggregate must be stale rather than disguised as one fresh report"23072626         );23082627     }2628+2629+    #[tokio::test]2630+    async fn queued_adaptive_plan_keeps_latest_ids_and_recovery_barrier() {2631+        let sessions = Arc::new(SessionRegistry::default());2632+        let remote = endpoint(1);2633+        let media = sessions2634+            .claim(remote, [3; 16], InvitationGrants::VIEW)2635+            .unwrap();2636+        let harness = crate::source::EncoderControlTestHarness::new();2637+        sessions2638+            .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2639+            .unwrap();2640+        let mut coordinator = AdaptiveCommitCoordinator::new(2641+            Arc::clone(&sessions),2642+            remote,2643+            media.session_id,2644+            true,2645+            12_000,2646+            Some(native_dimensions()),2647+        );2648+        let mut controller =2649+            ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2650+        let mut resolution_controller = None;2651+        let observed_at = Instant::now();2652+2653+        let seed = evaluate_commit_plan(2654+            &mut controller,2655+            &mut resolution_controller,2656+            &clean_feedback(1),2657+            observed_at,2658+        );2659+        assert!(coordinator.submit(seed).is_some());2660+2661+        let mut first_pressure = clean_feedback(2);2662+        first_pressure.decoder_dropped_delta = 1;2663+        let first = evaluate_commit_plan(2664+            &mut controller,2665+            &mut resolution_controller,2666+            &first_pressure,2667+            observed_at + Duration::from_secs(1),2668+        );2669+        assert!(!first.force_keyframe);2670+        assert!(coordinator.submit(first).is_none());2671+2672+        let mut recovery_pressure = clean_feedback(3);2673+        recovery_pressure.decoder_dropped_delta = 1;2674+        recovery_pressure.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2675+        let recovery = evaluate_commit_plan(2676+            &mut controller,2677+            &mut resolution_controller,2678+            &recovery_pressure,2679+            observed_at + Duration::from_secs(2),2680+        );2681+        assert!(recovery.force_keyframe);2682+        assert!(coordinator.submit(recovery).is_none());2683+2684+        let latest = evaluate_commit_plan(2685+            &mut controller,2686+            &mut resolution_controller,2687+            &clean_feedback(4),2688+            observed_at + Duration::from_secs(3),2689+        );2690+        assert!(!latest.force_keyframe);2691+        assert!(coordinator.submit(latest).is_none());2692+        let retained = coordinator.queued.expect("latest plan must be retained");2693+        assert_eq!(retained.decision.decision_id, latest.decision.decision_id);2694+        assert_eq!(retained.decision.report_id, latest.decision.report_id);2695+        assert!(2696+            retained.force_keyframe,2697+            "a coalesced recovery barrier must survive a newer hold decision"2698+        );2699+2700+        coordinator.abort_and_drain().await;2701+        drop(media);2702+    }2703+2704+    #[tokio::test(start_paused = true)]2705+    async fn timed_out_actuator_requests_are_overwritten_when_policy_returns_to_committed_state() {2706+        let sessions = Arc::new(SessionRegistry::default());2707+        let remote = endpoint(1);2708+        let media = sessions2709+            .claim(remote, [8; 16], InvitationGrants::VIEW)2710+            .unwrap();2711+        let harness = crate::source::EncoderControlTestHarness::new();2712+        sessions2713+            .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2714+            .unwrap();2715+        let native = native_dimensions();2716+        let reduced = native.three_quarter().unwrap();2717+        let mut coordinator = AdaptiveCommitCoordinator::new(2718+            Arc::clone(&sessions),2719+            remote,2720+            media.session_id,2721+            true,2722+            12_000,2723+            Some(native),2724+        );2725+        let timed_out = AdaptiveCommitPlan {2726+            decision: AdaptiveBitrateDecisionV1 {2727+                decision_id: 1,2728+                report_id: 1,2729+                target_kbps: 9_000,2730+                floor_kbps: 1_000,2731+                ceiling_kbps: 12_000,2732+                state: AdaptiveBitrateStateV1::Decrease,2733+                reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,2734+                applied: false,2735+            },2736+            telemetry: MediaV3TelemetrySnapshot::default(),2737+            resolution_decision: Some(MotionResolutionDecision {2738+                tier: ResolutionTier::Reduced,2739+                target: reduced,2740+                trigger: ResolutionTrigger::Pressure,2741+                changed: true,2742+            }),2743+            force_keyframe: false,2744+        };2745+        assert!(coordinator.submit(timed_out).is_none());2746+        let stale_bitrate = harness2747+            .requested_bitrate()2748+            .expect("the first bitrate request must reach the actuator");2749+        let stale_resolution = harness2750+            .requested_resolution()2751+            .expect("the first resolution request must reach the actuator");2752+        assert_eq!(stale_bitrate.1, 9_000);2753+        assert_eq!(2754+            (stale_resolution.1, stale_resolution.2),2755+            (reduced.width, reduced.height)2756+        );2757+2758+        // Feedback evaluation recovers all the way to the previously2759+        // committed state while the old request is still awaiting readback.2760+        let recovered = AdaptiveCommitPlan {2761+            decision: AdaptiveBitrateDecisionV1 {2762+                decision_id: 2,2763+                report_id: 2,2764+                target_kbps: 12_000,2765+                floor_kbps: 1_000,2766+                ceiling_kbps: 12_000,2767+                state: AdaptiveBitrateStateV1::Increase,2768+                reasons: AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY,2769+                applied: false,2770+            },2771+            telemetry: MediaV3TelemetrySnapshot::default(),2772+            resolution_decision: Some(MotionResolutionDecision {2773+                tier: ResolutionTier::Native,2774+                target: native,2775+                trigger: ResolutionTrigger::Recovery,2776+                changed: true,2777+            }),2778+            force_keyframe: false,2779+        };2780+        assert!(coordinator.submit(recovered).is_none());2781+2782+        // The first task times out its bitrate wait and then its resolution2783+        // wait. Neither timeout changes the last acknowledged state.2784+        tokio::task::yield_now().await;2785+        tokio::time::advance(ENCODER_CONTROL_COMMIT_TIMEOUT).await;2786+        tokio::task::yield_now().await;2787+        tokio::time::advance(ENCODER_CONTROL_COMMIT_TIMEOUT).await;2788+        tokio::task::yield_now().await;2789+        let result = coordinator.acknowledgements.join_next().await;2790+        let acknowledgement = coordinator.complete(result).unwrap();2791+        assert!(!acknowledgement.plan.decision.applied);2792+        assert_eq!(acknowledgement.resolution_applied, Some(false));2793+        assert_eq!(coordinator.committed_bitrate_kbps, 12_000);2794+        assert_eq!(coordinator.committed_resolution, Some(native));2795+        assert_eq!(coordinator.requested_bitrate_kbps, 9_000);2796+        assert_eq!(coordinator.requested_resolution, Some(reduced));2797+2798+        // Starting the recovered plan must overwrite the stale persistent2799+        // actuator watch even though its targets equal the old committed2800+        // values. Otherwise the timed-out request would apply on resume.2801+        assert!(coordinator.start_queued().is_none());2802+        assert!(coordinator.has_pending());2803+        let corrected_bitrate = harness2804+            .requested_bitrate()2805+            .expect("recovered bitrate must overwrite stale actuator intent");2806+        let corrected_resolution = harness2807+            .requested_resolution()2808+            .expect("recovered resolution must overwrite stale actuator intent");2809+        assert!(corrected_bitrate.0 > stale_bitrate.0);2810+        assert!(corrected_resolution.0 > stale_resolution.0);2811+        assert_eq!(corrected_bitrate.1, 12_000);2812+        assert_eq!(2813+            (corrected_resolution.1, corrected_resolution.2),2814+            (native.width, native.height)2815+        );2816+        assert_eq!(coordinator.requested_bitrate_kbps, 12_000);2817+        assert_eq!(coordinator.requested_resolution, Some(native));2818+2819+        coordinator.abort_and_drain().await;2820+        drop(media);2821+    }2822+2823+    #[tokio::test]2824+    async fn feedback_loop_writes_every_decision_while_combined_commit_is_hung() {2825+        let sessions = Arc::new(SessionRegistry::default());2826+        let remote = endpoint(1);2827+        let nonce = [9; 16];2828+        let media = sessions2829+            .claim(remote, nonce, InvitationGrants::VIEW)2830+            .unwrap();2831+        let harness = crate::source::EncoderControlTestHarness::new();2832+        sessions2833+            .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2834+            .unwrap();2835+        let feedback = sessions.claim_feedback(remote, nonce).unwrap();2836+        let session_id = feedback.session_id;2837+2838+        let (mut report_writer, report_reader) = tokio::io::duplex(4_096);2839+        let (feedback_sender, mut feedback_receiver) =2840+            tokio::sync::watch::channel(CumulativeFeedbackIngress::default());2841+        let reader_task = tokio::spawn(forward_media_feedback_reports(2842+            report_reader,2843+            feedback_sender,2844+        ));2845+        let (mut decision_writer, mut decision_reader) = tokio::io::duplex(4_096);2846+        let loop_sessions = Arc::clone(&sessions);2847+        let loop_task = tokio::spawn(async move {2848+            run_media_feedback_loop(2849+                &mut decision_writer,2850+                std::future::pending(),2851+                remote,2852+                &loop_sessions,2853+                &feedback,2854+                true,2855+                12_000,2856+                &mut feedback_receiver,2857+                reader_task,2858+            )2859+            .await2860+        });2861+2862+        let mut observed = Vec::new();2863+        for report_id in 1..=5 {2864+            if report_id > 1 {2865+                tokio::time::sleep(Duration::from_millis(800)).await;2866+            }2867+            let mut report = clean_feedback(report_id);2868+            if report_id >= 2 {2869+                report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2870+                report.decoder_dropped_delta = 4;2871+            }2872+            sigil_protocol::write_media_feedback_report_v1(&mut report_writer, &report)2873+                .await2874+                .unwrap();2875+            let decision = tokio::time::timeout(2876+                Duration::from_secs(1),2877+                sigil_protocol::read_adaptive_bitrate_decision_v1(&mut decision_reader),2878+            )2879+            .await2880+            .expect("each evaluated report must be written before the hung commit timeout")2881+            .unwrap()2882+            .expect("feedback loop closed before writing its decision");2883+            observed.push((decision.decision_id, decision.report_id));2884+        }2885+        assert_eq!(2886+            observed,2887+            (1..=5).map(|id| (id, id)).collect::<Vec<_>>(),2888+            "slow bitrate and resolution readbacks must not coalesce Portal decisions"2889+        );2890+        assert!(2891+            harness.requested_bitrate().is_some(),2892+            "the adversarial run must actually have pending bitrate work"2893+        );2894+        assert!(2895+            harness.requested_resolution().is_some(),2896+            "the adversarial run must actually have pending resolution work"2897+        );2898+2899+        drop(media);2900+        drop(report_writer);2901+        tokio::time::timeout(Duration::from_secs(1), loop_task)2902+            .await2903+            .expect("session teardown must abort the hung commit promptly")2904+            .unwrap()2905+            .unwrap();2906+        assert!(!sessions.is_active(remote, session_id));2907+    }2908+2909+    #[tokio::test(start_paused = true)]2910+    async fn slow_adaptive_commit_does_not_pause_feedback_evaluation() {2911+        let sessions = Arc::new(SessionRegistry::default());2912+        let remote = endpoint(1);2913+        let media = sessions2914+            .claim(remote, [4; 16], InvitationGrants::VIEW)2915+            .unwrap();2916+        let harness = crate::source::EncoderControlTestHarness::new();2917+        sessions2918+            .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2919+            .unwrap();2920+        let mut coordinator = AdaptiveCommitCoordinator::new(2921+            Arc::clone(&sessions),2922+            remote,2923+            media.session_id,2924+            true,2925+            12_000,2926+            Some(native_dimensions()),2927+        );2928+        let mut controller =2929+            ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2930+        let mut resolution_controller = None;2931+        let observed_at = Instant::now();2932+2933+        let seed = evaluate_commit_plan(2934+            &mut controller,2935+            &mut resolution_controller,2936+            &clean_feedback(1),2937+            observed_at,2938+        );2939+        assert_eq!(2940+            coordinator.submit(seed).unwrap().plan.decision.decision_id,2941+            12942+        );2943+2944+        let mut severe = clean_feedback(2);2945+        severe.decoder_dropped_delta = 1;2946+        let first = evaluate_commit_plan(2947+            &mut controller,2948+            &mut resolution_controller,2949+            &severe,2950+            observed_at + Duration::from_secs(1),2951+        );2952+        assert_eq!(first.decision.state, AdaptiveBitrateStateV1::Decrease);2953+        assert!(coordinator.submit(first).is_none());2954+        assert!(coordinator.has_pending());2955+2956+        tokio::time::advance(Duration::from_secs(1)).await;2957+        let evaluation_tick = tokio::time::Instant::now();2958+        let mut newer_severe = clean_feedback(3);2959+        newer_severe.decoder_dropped_delta = 1;2960+        let queued = evaluate_commit_plan(2961+            &mut controller,2962+            &mut resolution_controller,2963+            &newer_severe,2964+            observed_at + Duration::from_secs(2),2965+        );2966+        assert_eq!(queued.decision.decision_id, 3);2967+        assert_eq!(queued.decision.state, AdaptiveBitrateStateV1::Decrease);2968+        assert!(coordinator.submit(queued).is_none());2969+        assert_eq!(tokio::time::Instant::now(), evaluation_tick);2970+        assert_eq!(coordinator.acknowledgements.len(), 1);2971+        assert_eq!(2972+            coordinator.queued.unwrap().decision.decision_id,2973+            queued.decision.decision_id2974+        );2975+2976+        // Revision one is acknowledged only after another feedback window was2977+        // evaluated. The decision retains that exact readback result.2978+        harness.status.send_modify(|status| {2979+            status.applied_bitrate_revision = Some(1);2980+            status.applied_bitrate_kbps = Some(first.decision.target_kbps);2981+        });2982+        let commit_result = coordinator.acknowledgements.join_next().await;2983+        let completed = coordinator.complete(commit_result).unwrap();2984+        assert_eq!(completed.plan.decision.decision_id, 2);2985+        assert!(completed.plan.decision.applied);2986+        assert_eq!(2987+            completed.applied_bitrate_kbps,2988+            Some(first.decision.target_kbps)2989+        );2990+2991+        assert!(coordinator.start_queued().is_none());2992+        assert!(coordinator.has_pending());2993+        assert_eq!(coordinator.acknowledgements.len(), 1);2994+2995+        let session_id = media.session_id;2996+        drop(media);2997+        assert!(!sessions.is_active(remote, session_id));2998+        let session_end_result = coordinator.acknowledgements.join_next().await;2999+        let session_end = coordinator.complete(session_end_result).unwrap();3000+        assert!(session_end.generation_ended);3001+        coordinator.abort_and_drain().await;3002+        assert!(!coordinator.has_pending());3003+        assert!(coordinator.queued.is_none());3004+        assert!(coordinator.acknowledgements.is_empty());3005+    }3006+3007+    #[tokio::test(start_paused = true)]3008+    async fn timing_out_commits_keep_uniform_hysteresis_and_bounded_work() {3009+        let sessions = Arc::new(SessionRegistry::default());3010+        let remote = endpoint(1);3011+        let media = sessions3012+            .claim(remote, [5; 16], InvitationGrants::VIEW)3013+            .unwrap();3014+        let harness = crate::source::EncoderControlTestHarness::new();3015+        sessions3016+            .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3017+            .unwrap();3018+        let mut coordinator = AdaptiveCommitCoordinator::new(3019+            Arc::clone(&sessions),3020+            remote,3021+            media.session_id,3022+            true,3023+            12_000,3024+            Some(native_dimensions()),3025+        );3026+        let mut controller =3027+            ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3028+        let mut resolution_controller =3029+            Some(MotionResolutionPolicy::new(native_dimensions()).unwrap());3030+        let observed_at = Instant::now();3031+        let virtual_started_at = tokio::time::Instant::now();3032+        let mut evaluation_ticks = Vec::new();3033+        let mut decision_ids = Vec::new();3034+        let mut states = Vec::new();3035+        let mut completed = Vec::new();3036+3037+        for report_id in 1..=9 {3038+            evaluation_ticks.push(tokio::time::Instant::now());3039+            let mut report = clean_feedback(report_id);3040+            if report_id > 1 {3041+                report.frontend_queue_depth = 2;3042+            }3043+            let plan = evaluate_commit_plan(3044+                &mut controller,3045+                &mut resolution_controller,3046+                &report,3047+                observed_at + Duration::from_secs(report_id - 1),3048+            );3049+            decision_ids.push(plan.decision.decision_id);3050+            states.push(plan.decision.state);3051+            if let Some(immediate) = coordinator.submit(plan) {3052+                completed.push(immediate);3053+            }3054+            assert!(3055+                coordinator.acknowledgements.len() <= 1,3056+                "the coordinator must own at most one commit task"3057+            );3058+            assert!(3059+                coordinator.queued.iter().count() <= 1,3060+                "the coordinator must retain only the latest plan"3061+            );3062+3063+            tokio::time::advance(Duration::from_secs(1)).await;3064+            tokio::task::yield_now().await;3065+            if let Some(result) = coordinator.acknowledgements.try_join_next() {3066+                completed.push(coordinator.complete(Some(result)).unwrap());3067+                if let Some(immediate) = coordinator.start_queued() {3068+                    completed.push(immediate);3069+                }3070+            }3071+        }3072+3073+        assert_eq!(3074+            decision_ids,3075+            (1..=9).collect::<Vec<_>>(),3076+            "coalescing must not pause or reuse decision IDs"3077+        );3078+        assert_eq!(3079+            states,3080+            [3081+                AdaptiveBitrateStateV1::Hold,3082+                AdaptiveBitrateStateV1::Hold,3083+                AdaptiveBitrateStateV1::Decrease,3084+                AdaptiveBitrateStateV1::Hold,3085+                AdaptiveBitrateStateV1::Decrease,3086+                AdaptiveBitrateStateV1::Hold,3087+                AdaptiveBitrateStateV1::Decrease,3088+                AdaptiveBitrateStateV1::Hold,3089+                AdaptiveBitrateStateV1::Decrease,3090+            ],3091+            "moderate-pressure hysteresis must advance once per feedback window"3092+        );3093+        for (index, tick) in evaluation_ticks.into_iter().enumerate() {3094+            assert_eq!(3095+                tick.duration_since(virtual_started_at),3096+                Duration::from_secs(index as u64),3097+                "feedback evaluation cadence must remain uniform during commit timeouts"3098+            );3099+        }3100+        assert!(3101+            completed.iter().any(|acknowledgement| {3102+                acknowledgement.plan.decision.decision_id == 33103+                    && !acknowledgement.plan.decision.applied3104+                    && acknowledgement.resolution_applied == Some(false)3105+            }),3106+            "the first bitrate and resolution timeout must retain exact negative acknowledgement"3107+        );3108+        assert!(coordinator.has_pending());3109+        assert_eq!(coordinator.acknowledgements.len(), 1);3110+        assert!(coordinator.queued.is_some());3111+3112+        coordinator.abort_and_drain().await;3113+        assert!(!coordinator.has_pending());3114+        assert!(coordinator.queued.is_none());3115+        assert!(coordinator.acknowledgements.is_empty());3116+3117+        let session_id = media.session_id;3118+        drop(media);3119+        assert!(!sessions.is_active(remote, session_id));3120+    }23093121 }