fix(host): reconcile adaptive control cadence

Dami committed Jul 23d5a0c34

Write every evaluated feedback decision before bounded actuator work, retain correlated commit diagnostics, and reconcile desired/requested/committed encoder state after timeouts. Add adversarial wire-cadence and stale-request recovery coverage.

commit: d5a0c34a28d06f570d16670efa9167f948fea1da
parent: f409f48e271ab9fd85ebc4988d9ada2ddfd67ee4
author: Dami <dami.etoile@gmail.com>

2 files changed, +494 -123wrap lines

M	crates/sigil-host/src/server/adaptive.rs	603	++++++++++++++++++++++++------
M	crates/sigil-host/src/source/in_process.rs	14	+

M crates/sigil-host/src/server/adaptive.rs +480 -123

@@ -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 @@ -885,7 +886,11 @@ struct AdaptiveCommitCoordinator {
885886     remote: EndpointId,886887     session_id: u64,887888     actuation_enabled: bool,889+    desired_bitrate_kbps: u32,890+    requested_bitrate_kbps: u32,888891     committed_bitrate_kbps: u32,892+    desired_resolution: Option<VideoDimensions>,893+    requested_resolution: Option<VideoDimensions>,889894     committed_resolution: Option<VideoDimensions>,890895     pending: bool,891896     queued: Option<AdaptiveCommitPlan>,@@ -906,7 +911,11 @@ impl AdaptiveCommitCoordinator {
906911             remote,907912             session_id,908913             actuation_enabled,914+            desired_bitrate_kbps: committed_bitrate_kbps,915+            requested_bitrate_kbps: committed_bitrate_kbps,909916             committed_bitrate_kbps,917+            desired_resolution: committed_resolution,918+            requested_resolution: committed_resolution,910919             committed_resolution,911920             pending: false,912921             queued: None,@@ -918,7 +927,18 @@ impl AdaptiveCommitCoordinator {
918927         self.pending919928     }920929 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+921937     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+        }922942         if self.pending {923943             if let Some(queued) = self.queued {924944                 // The latest decision replaces older diagnostics, but a@@ -929,6 +949,13 @@ impl AdaptiveCommitCoordinator {
929949             self.queued = Some(plan);930950             return None;931951         }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+        }932959         self.start(plan)933960     }934961 @@ -943,21 +970,34 @@ impl AdaptiveCommitCoordinator {
943970         }944971 945972         let desired_resolution = plan.resolution_decision.map(|change| change.target);946-        let changes_resolution =947-            desired_resolution.is_some() && desired_resolution != self.committed_resolution;948-        let bitrate_proposal = if plan.decision.target_kbps != self.committed_bitrate_kbps {949-            match self.sessions.propose_adaptive_encoder_update(950-                self.remote,951-                self.session_id,952-                plan.decision.target_kbps,953-                plan.force_keyframe && !changes_resolution,954-            ) {955-                Ok(Some(proposal)) => Some(proposal),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+                }956996                 Ok(None) => {957997                     warn!(958998                         remote = %self.remote,959999                         session_id = self.session_id,960-                        "active adaptive session lost encoder control; retaining committed bitrate"1000+                        "active adaptive session lost encoder control; retaining committed resolution"9611001                     );9621002                     None9631003                 }@@ -966,7 +1006,7 @@ impl AdaptiveCommitCoordinator {
9661006                         remote = %self.remote,9671007                         session_id = self.session_id,9681008                         %error,969-                        "adaptive encoder proposal failed; retaining committed bitrate"1009+                        "adaptive resolution proposal failed; retaining committed resolution"9701010                     );9711011                     None9721012                 }@@ -975,18 +1015,25 @@ impl AdaptiveCommitCoordinator {
9751015             None9761016         };9771017 978-        let resolution_proposal = if changes_resolution {979-            let target = desired_resolution.expect("resolution target was checked as present");980-            match self981-                .sessions982-                .propose_resolution_update(self.remote, self.session_id, target)983-            {984-                Ok(Some(proposal)) => Some(proposal),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+                }9851032                 Ok(None) => {9861033                     warn!(9871034                         remote = %self.remote,9881035                         session_id = self.session_id,989-                        "active adaptive session lost encoder control; retaining committed resolution"1036+                        "active adaptive session lost encoder control; retaining committed bitrate"9901037                     );9911038                     None9921039                 }@@ -995,7 +1042,7 @@ impl AdaptiveCommitCoordinator {
9951042                         remote = %self.remote,9961043                         session_id = self.session_id,9971044                         %error,998-                        "adaptive resolution proposal failed; retaining committed resolution"1045+                        "adaptive encoder proposal failed; retaining committed bitrate"9991046                     );10001047                     None10011048                 }@@ -1119,7 +1166,45 @@ pub(super) async fn serve_media_feedback(
11191166 11201167     let (feedback_sender, mut feedback_receiver) =11211168         tokio::sync::watch::channel(CumulativeFeedbackIngress::default());1122-    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+{11231208     let mut reader_finished = false;11241209     let mut receiver_open = true;11251210     let mut controller = ShadowBitrateController::new(ceiling_kbps, lease.telemetry.snapshot());@@ -1145,133 +1230,169 @@ pub(super) async fn serve_media_feedback(
11451230         initial_resolution,11461231     );11471232     let mut feedback_cursor = FeedbackIngressCursor::default();1148-    let mut ready_acknowledgement: Option<AdaptiveCommitAcknowledgement> = None;1149-1150-    let result: Result<()> = loop {1151-        if let Some(acknowledgement) = ready_acknowledgement.take() {1152-            if acknowledgement.generation_ended {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) {11531243                 break Ok(());11541244             }1155-            write_adaptive_acknowledgement(1156-                &mut send,1157-                remote,1158-                lease.session_id,1159-                encoder_actuation_available,1160-                &acknowledgement,1161-            )1162-            .await?;1163-            ready_acknowledgement = commit_coordinator.start_queued();1164-            continue;1165-        }1166-1167-        let session_changed = sessions.session_changed.notified();1168-        tokio::pin!(session_changed);1169-        session_changed.as_mut().enable();1170-        if !sessions.is_active(remote, lease.session_id) {1171-            break Ok(());1172-        }1173-        tokio::select! {1174-            biased;1175-            reader_result = &mut reader_task, if !reader_finished => {1176-                reader_finished = true;1177-                match reader_result {1178-                    Ok(Ok(())) => {1179-                        debug!(%remote, "media feedback report stream closed");1180-                    }1181-                    Ok(Err(error)) => {1182-                        break Err(error).context("reading media feedback reports");1183-                    }1184-                    Err(error) => {1185-                        break Err(error).context("media feedback reader task failed");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+                        }11861259                     }11871260                 }1188-            }1189-            commit_result = commit_coordinator.acknowledgements.join_next(),1190-                if commit_coordinator.has_pending() =>1191-            {1192-                let acknowledgement = commit_coordinator.complete(commit_result)?;1193-                ready_acknowledgement = Some(acknowledgement);1194-            }1195-            changed = feedback_receiver.changed(), if receiver_open => {1196-                if changed.is_err() {1197-                    receiver_open = false;1198-                    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 {11991266                         break Ok(());12001267                     }1201-                    continue;1268+                    log_adaptive_acknowledgement(1269+                        remote,1270+                        lease.session_id,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+                        );1289+                    }12021290                 }1203-                let ingress = *feedback_receiver.borrow_and_update();1204-                let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {1205-                    continue;1206-                };1207-                // This branch now owns the complete aggregate. Advance the1208-                // ingress baseline exactly once before evaluation; reports1209-                // arriving during an encoder wait accumulate beyond it.1210-                feedback_cursor = next_cursor;1211-                let telemetry = lease.telemetry.snapshot();1212-                let observed_at = Instant::now();1213-                let mut candidate = controller.clone();1214-                let evaluation = candidate.evaluate(&report, telemetry, observed_at)?;1215-                let decision = evaluation.decision;1216-                let mut resolution_candidate = resolution_controller.clone();1217-                let resolution_decision = resolution_candidate1218-                    .as_mut()1219-                    .map(|policy| policy.observe_window(&report, &evaluation, observed_at));1220-                let force_keyframe =1221-                    adaptive_recovery_keyframe_required(&report, &decision);1222-                // Evaluation state advances at feedback cadence. The actuator1223-                // independently retains the exact committed target and retries1224-                // the latest desired state after a slow or failed predecessor.1225-                controller = candidate;1226-                resolution_controller = resolution_candidate;1227-                ready_acknowledgement =1228-                    commit_coordinator.submit(AdaptiveCommitPlan {1291+                changed = feedback_receiver.changed(), if receiver_open => {1292+                    if changed.is_err() {1293+                        receiver_open = false;1294+                        if reader_finished {1295+                            break Ok(());1296+                        }1297+                        continue;1298+                    }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 {12291321                         decision,12301322                         telemetry,12311323                         resolution_decision,12321324                         force_keyframe,1233-                    });1234-            }1235-            _ = &mut session_changed => {1236-                if !sessions.is_active(remote, lease.session_id) {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?;1339+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(1347+                            remote,1348+                            lease.session_id,1349+                            encoder_actuation_available,1350+                            &commit_coordinator,1351+                            &immediate,1352+                        );1353+                    }1354+                }1355+                _ = &mut session_changed => {1356+                    if !sessions.is_active(remote, lease.session_id) {1357+                        break Ok(());1358+                    }1359+                }1360+                _ = &mut connection_closed => {12371361                     break Ok(());12381362                 }12391363             }1240-            closed = connection.closed() => {1241-                debug!(%remote, ?closed, "media feedback connection closed");1364+            if reader_finished1365+                && !receiver_open1366+                && !commit_coordinator.has_pending()1367+                && commit_coordinator.queued.is_none()1368+            {12421369                 break Ok(());12431370             }12441371         }1245-        if reader_finished1246-            && !receiver_open1247-            && !commit_coordinator.has_pending()1248-            && commit_coordinator.queued.is_none()1249-        {1250-            break Ok(());1251-        }1252-    };1372+    }1373+    .await;12531374 12541375     if !reader_finished {12551376         reader_task.abort();12561377         let _ = reader_task.await;12571378     }12581379     commit_coordinator.abort_and_drain().await;1259-    let _ = send.finish();1260-    drop(lease);1261-    info!(%remote, "media feedback client released");12621380     result12631381 }12641382 1265-async fn write_adaptive_acknowledgement(1266-    send: &mut SendStream,1383+async fn write_adaptive_decision<W>(1384+    send: &mut W,12671385     remote: EndpointId,12681386     session_id: u64,12691387     encoder_actuation_available: bool,1270-    acknowledgement: &AdaptiveCommitAcknowledgement,1271-) -> Result<()> {1272-    let decision = acknowledgement.plan.decision;1273-    let telemetry = acknowledgement.plan.telemetry;1274-    let resolution_decision = acknowledgement.plan.resolution_decision;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;12751396     tokio::time::timeout(12761397         FEEDBACK_WRITE_TIMEOUT,12771398         write_adaptive_bitrate_decision_v1(send, &decision),@@ -1287,11 +1408,9 @@ async fn write_adaptive_acknowledgement(
12871408         ?decision.state,12881409         reasons = decision.reasons.bits(),12891410         applied = decision.applied,1290-        recovery_acknowledged = ?acknowledgement.recovery_acknowledged,12911411         resolution_width = resolution_decision.map(|change| change.target.width),12921412         resolution_height = resolution_decision.map(|change| change.target.height),12931413         resolution_trigger = ?resolution_decision.map(|change| change.trigger),1294-        resolution_applied = ?acknowledgement.resolution_applied,12951414         path_rtt_micros = telemetry.selected_path_rtt_micros,12961415         path_lost_packets = telemetry.selected_path_lost_packets,12971416         path_congestion_events = telemetry.selected_path_congestion_events,@@ -1303,6 +1422,39 @@ async fn write_adaptive_acknowledgement(
13031422     Ok(())13041423 }13051424 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+13061458 async fn run_adaptive_commit(13071459     sessions: Arc<SessionRegistry>,13081460     remote: EndpointId,@@ -2549,6 +2701,211 @@ mod tests {
25492701         drop(media);25502702     }25512703 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+25522909     #[tokio::test(start_paused = true)]25532910     async fn slow_adaptive_commit_does_not_pause_feedback_evaluation() {25542911         let sessions = Arc::new(SessionRegistry::default());