diff --git a/plugins/agent/facts/src/cells.rs b/plugins/agent/facts/src/cells.rs index 1a301883c7..fca57222b7 100644 --- a/plugins/agent/facts/src/cells.rs +++ b/plugins/agent/facts/src/cells.rs @@ -18,7 +18,9 @@ */ //! Durable Cell evidence; no value here represents interpreter or Host custody. -use crate::durable::{AgentStepLink, ApprovalId, ContentRef, FactId, ResourceRecovery, StepId}; +use crate::durable::{ + AgentStepLink, ApprovalId, ContentRef, FactId, ResourceRecovery, StepId, Terminal, +}; use crate::operation::{ RecordedOperationReply, RecordedOperationResult, result_reference, same_outcome, }; @@ -511,19 +513,17 @@ impl CellView { || self.children.values().any(|child| !child.known()) || self.blocks_effect_admission() } - pub fn blocks_session_retirement(&self, terminal: Option<&crate::durable::Terminal>) -> bool { - if self.owner.run().is_some() - && matches!( - terminal, - Some(crate::durable::Terminal::Cancelled | crate::durable::Terminal::Failed) - ) - { - self.has_unsettled_effects() - } else { - self.pending() + pub fn blocks_session_retirement(&self, terminal: Option<&Terminal>) -> bool { + if self.owner.run().is_none() { + return self.pending(); + } + match terminal { + Some(Terminal::Completed) => self.settlement_pending(), + Some(Terminal::Cancelled | Terminal::Failed) => self.has_unsettled_effects(), + None => self.pending(), } } - pub fn pending(&self) -> bool { + pub fn settlement_pending(&self) -> bool { matches!( self.status, CellStatus::Running @@ -533,6 +533,9 @@ impl CellView { } ) || self.children.values().any(|child| !child.known()) || self.blocks_effect_admission() + } + pub fn pending(&self) -> bool { + self.settlement_pending() || (matches!(self.owner, CellOwner::Agent { .. }) && (!self.notifications.is_empty() || (matches!(self.status, CellStatus::Interrupted { .. }) diff --git a/plugins/agent/state/src/fold.rs b/plugins/agent/state/src/fold.rs index dd5c4759a8..f26b9b47f6 100644 --- a/plugins/agent/state/src/fold.rs +++ b/plugins/agent/state/src/fold.rs @@ -2126,8 +2126,15 @@ impl RunView { }), "successful completion requires actual model cleanup acknowledgements", )?; + // FinishTurn closes model admission, so there is no step to receive Cell notifications. require( - self.cells.values().all(|cell| !cell.pending()), + self.cells.values().all(|cell| { + if self.finish_requested() { + !cell.settlement_pending() + } else { + !cell.pending() + } + }), "run has pending Cell obligations", )?; require( diff --git a/plugins/agent/state/src/fold/tests.rs b/plugins/agent/state/src/fold/tests.rs index 7d706db305..974e0cafdb 100644 --- a/plugins/agent/state/src/fold/tests.rs +++ b/plugins/agent/state/src/fold/tests.rs @@ -85,6 +85,84 @@ fn spec(backend: BackendSpec, reference: &ContentRef) -> RunSpec { } } +#[test] +fn finish_turn_settles_after_cell_cleanup_with_an_undelivered_notification() { + use maka_agent_facts::cells::{CellEvent, CellId, CellOwner, CellView}; + + let source = ContentBlob::new(vec![]).reference().clone(); + let spec = spec( + BackendSpec::Model { + binding: ModelBinding { + connection_id: "policy".into(), + connection_slug: "policy".into(), + model: "policy".into(), + }, + descriptor: source.clone(), + }, + &source, + ); + let spec_ref = ContentBlob::json(&spec).unwrap().reference().clone(); + let mut view = RunView::admitted(RunId::new(), spec).unwrap(); + apply(&mut view, Fact::RunAdmitted { spec: spec_ref }); + let link = AgentStepLink { + run: view.run, + step: StepId::new(), + attempt: ModelAttemptId::new(), + }; + let cell = CellId::for_agent(link, "background"); + let background = CellView::started(&CellEvent::Started { + cell, + owner: CellOwner::Agent { + link, + call: "background".into(), + }, + source: source.clone(), + }) + .unwrap(); + view.cells.insert(cell, background); + let complete = |view: &mut RunView| { + let fact = record( + view, + Fact::RunTerminated { + terminal: Terminal::Completed, + diagnostic: None, + }, + ); + view.apply(&fact) + }; + view.finish_requested = true; + assert!(complete(&mut view).is_err()); + + apply( + &mut view, + Fact::Cell { + event: CellEvent::Interrupted { + cell, + report: source.clone(), + }, + }, + ); + let notification = FactId::new(); + apply( + &mut view, + Fact::Cell { + event: CellEvent::FinalNotificationQueued { + cell, + notification, + content: source, + }, + }, + ); + let mut continuing = view.clone(); + continuing.finish_requested = false; + assert!(complete(&mut continuing).is_err()); + + complete(&mut view).unwrap(); + assert_eq!(view.terminal, Some(Terminal::Completed)); + assert!(view.cells[&cell].notifications.contains(¬ification)); + assert!(!view.cells[&cell].blocks_session_retirement(view.terminal.as_ref())); +} + #[test] fn unknown_local_return_keeps_risk_until_new_known_evidence_and_never_rewrites_known_result() { // Pure fold data: content loading, physical cleanup, and Native custody are not exercised.