From 1d8f355f0c2903f2783d36f3a8ab66e148519f4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CM4n5ter=E2=80=9D?= Date: Sat, 14 Nov 2026 22:10:13 +0800 Subject: [PATCH] fix(agent): deliver pending steering after a run ends Generated-by: OpenAI Codex --- plugins/agent/native/src/owner/queue.rs | 4 +- plugins/agent/state/src/queue.rs | 10 +- plugins/agent/state/src/queue/sql.rs | 14 +- plugins/agent/state/src/sql.rs | 12 +- plugins/agent/state/src/store/stop_tests.rs | 2 +- products/maka/cli/tests/producer_reopen.rs | 140 ++++++++++++++++---- 6 files changed, 139 insertions(+), 43 deletions(-) diff --git a/plugins/agent/native/src/owner/queue.rs b/plugins/agent/native/src/owner/queue.rs index a97173cc74..2015ed60aa 100644 --- a/plugins/agent/native/src/owner/queue.rs +++ b/plugins/agent/native/src/owner/queue.rs @@ -59,10 +59,10 @@ impl Owner { if let Err(error) = self .durable .queue() - .recover_stopped_goal_steering(self.run_owner(), UnixMillis(now() as i64)) + .recover_undelivered_steering(self.run_owner(), UnixMillis(now() as i64)) .await { - tracing::error!(%error, "Failed to recover steering from stopped Goals"); + tracing::error!(%error, "Failed to recover undelivered steering"); } let mut sources = root.capability_changes(); let mut cause = ReadyCause::CapabilitySourceChanged; diff --git a/plugins/agent/state/src/queue.rs b/plugins/agent/state/src/queue.rs index 843649ea9b..d3f1aef940 100644 --- a/plugins/agent/state/src/queue.rs +++ b/plugins/agent/state/src/queue.rs @@ -303,25 +303,25 @@ impl QueueStore { pub async fn rebuild(&self, session: String) -> Result { self.data.write(Rebuild(session)).await } - pub async fn recover_stopped_goal_steering( + pub async fn recover_undelivered_steering( &self, owner: RunOwner, at: UnixMillis, ) -> Result<(), StoreError> { self.data - .write(RecoverStoppedGoalSteering { owner, at }) + .write(RecoverUndeliveredSteering { owner, at }) .await } } -struct RecoverStoppedGoalSteering { +struct RecoverUndeliveredSteering { owner: RunOwner, at: UnixMillis, } -impl Statement for RecoverStoppedGoalSteering { +impl Statement for RecoverUndeliveredSteering { type Value = (); type Output = (); fn execute(self, c: &Connection) -> Result, StoreError> { - let sessions = sql::recover_stopped_goal_steering(c, &self.owner, self.at)?; + let sessions = sql::recover_undelivered_steering(c, &self.owner, self.at)?; Ok(Prepared { value: (), changed: !sessions.is_empty(), diff --git a/plugins/agent/state/src/queue/sql.rs b/plugins/agent/state/src/queue/sql.rs index c6b78ecd6f..30dc209e07 100644 --- a/plugins/agent/state/src/queue/sql.rs +++ b/plugins/agent/state/src/queue/sql.rs @@ -676,7 +676,7 @@ pub(crate) fn release_steering( ) } -pub(super) fn recover_stopped_goal_steering( +pub(super) fn recover_undelivered_steering( c: &Connection, owner: &RunOwner, at: UnixMillis, @@ -686,10 +686,12 @@ pub(super) fn recover_stopped_goal_steering( "SELECT DISTINCT q.session,q.target_run FROM agent_queue_items q JOIN agent_runs r ON r.run=q.target_run AND r.session=q.session AND r.owner=q.owner - JOIN agent_facts f ON f.run=r.run AND f.id=r.stop_request - JOIN contents b ON b.id=f.body_content_id - WHERE q.owner=?1 AND q.state='pending' AND q.lane='steering' - AND json_extract(b.payload,'$.fact.diagnostic')=?2", + WHERE q.owner=?1 AND q.state='pending' AND q.lane='steering' AND r.local=1 + AND (r.terminal IS NOT NULL OR EXISTS ( + SELECT 1 FROM agent_facts f JOIN contents b ON b.id=f.body_content_id + WHERE f.id=r.stop_request AND f.run=r.run + AND json_extract(b.payload,'$.fact.diagnostic')=?2 + ))", )? .query_map( params![serde_json::to_string(owner)?, GOAL_CONTROL_STOP], @@ -699,7 +701,7 @@ pub(super) fn recover_stopped_goal_steering( let mut sessions = BTreeSet::new(); for (session, run) in rows { let run = RunId::try_from( - uuid::Uuid::parse_str(&run).map_err(|_| invalid("invalid stopped steering Run"))?, + uuid::Uuid::parse_str(&run).map_err(|_| invalid("invalid steering Run"))?, ) .map_err(invalid)?; release_steering(c, &session, run, at)?; diff --git a/plugins/agent/state/src/sql.rs b/plugins/agent/state/src/sql.rs index d3366b8d9b..8715a29a69 100644 --- a/plugins/agent/state/src/sql.rs +++ b/plugins/agent/state/src/sql.rs @@ -1520,11 +1520,17 @@ pub(crate) fn append( }; view.apply(&recorded)?; record(c, &recorded)?; - if let Fact::StopRequested { diagnostic } = &recorded.fact { - crate::queue::stop_completions(c, &view.spec.session_id, run, recorded.at)?; - if diagnostic == crate::queue::GOAL_CONTROL_STOP { + match &recorded.fact { + Fact::StopRequested { diagnostic } => { + crate::queue::stop_completions(c, &view.spec.session_id, run, recorded.at)?; + if diagnostic == crate::queue::GOAL_CONTROL_STOP { + crate::queue::release_steering(c, &view.spec.session_id, run, recorded.at)?; + } + } + Fact::RunTerminated { .. } if matches!(view.spec.origin, RunOrigin::Local) => { crate::queue::release_steering(c, &view.spec.session_id, run, recorded.at)?; } + _ => {} } model_evidence::validate_partial(c, &view, &recorded)?; model_evidence::validate_source_failure(c, &view, &recorded)?; diff --git a/plugins/agent/state/src/store/stop_tests.rs b/plugins/agent/state/src/store/stop_tests.rs index df81cdd97c..02ff6b3170 100644 --- a/plugins/agent/state/src/store/stop_tests.rs +++ b/plugins/agent/state/src/store/stop_tests.rs @@ -299,7 +299,7 @@ async fn pausing_a_goal_preserves_pending_steering_as_an_independent_turn() { ); for _ in 0..2 { queue - .recover_stopped_goal_steering(spec.owner.clone(), UnixMillis(4)) + .recover_undelivered_steering(spec.owner.clone(), UnixMillis(4)) .await .unwrap(); assert_eq!(queue.read(spec.session_id.clone()).await.unwrap(), pending); diff --git a/products/maka/cli/tests/producer_reopen.rs b/products/maka/cli/tests/producer_reopen.rs index f7295ce2bb..9a14155108 100644 --- a/products/maka/cli/tests/producer_reopen.rs +++ b/products/maka/cli/tests/producer_reopen.rs @@ -68,7 +68,7 @@ use uuid::Uuid; #[derive(Clone, Copy)] enum Producer { - Goal, + Goal { pending_steering: bool }, GoalBudget, Scheduler, Automation, @@ -191,6 +191,8 @@ fn tool_text<'a>(messages: &'a [Value], id: &str) -> Result<&'a str> { .ok_or_else(|| "direct tool content must be a string".to_string()) } +const RECOVERY_STEERING: &str = "Keep this queued correction after the original Run fails."; + struct Model { producer: Producer, marker: String, @@ -284,8 +286,23 @@ impl Model { "stop", )); } - if matches!(self.producer, Producer::Goal) && iteration > 1 { - return match iteration { + if let Producer::Goal { pending_steering } = self.producer + && iteration > 1 + { + if pending_steering { + let corrections = messages + .iter() + .filter(|message| { + message["role"] == "user" + && message["content"].to_string().contains(RECOVERY_STEERING) + }) + .count(); + assert_eq!(corrections, 1, "one delivered correction in model history"); + if iteration == 2 { + return Ok(sse(json!({"content":"Correction handled."}), "stop")); + } + } + return match iteration - usize::from(pending_steering) { 2 => Ok(sse( json!({"tool_calls":[{"index":0,"id":"finish-goal","type":"function","function":{ "name":"GoalStatus","arguments":json!({"status":"achieved","note":"Continued in a new iteration."}).to_string() @@ -773,11 +790,26 @@ async fn body( let store = ConfigurationStore::for_root(Arc::new(owner)) .await .map_err(|error| error.to_string())?; - let created = store.create_connection(serde_json::from_value(json!({"expectedCatalogRevision":0,"connection":{ - "slug":"producer","name":"Producer loopback","provider":{"packageId":"maka.providers","entryId":"maka.providers","scope":"profile","name":"openai-compatible"}, + let mut draft = json!({"slug":"producer","name":"Producer loopback", + "provider":{"packageId":"maka.providers","entryId":"maka.providers","scope":"profile","name":"openai-compatible"}, "configuration":{"baseUrl":base},"enabled":true,"enabledModelIds":["gpt-4.1"], "modelOverrides":{"gpt-4.1":{"codeMode":false,"apiProtocol":"openai-chat-completions"}} - }})).unwrap()).await; + }); + if matches!( + producer, + Producer::Goal { + pending_steering: true + } + ) { + // This connection cannot interrupt a model stream to apply steering. + draft["requestBodyOverlay"] = json!({"temperature":0}); + } + let created = store + .create_connection( + serde_json::from_value(json!({"expectedCatalogRevision":0,"connection":draft})) + .unwrap(), + ) + .await; let closed = store.shutdown().await; let created = created.map_err(|error| error.to_string())?; closed.map_err(|error| error.to_string())?; @@ -813,7 +845,7 @@ async fn body( .await?; assert_eq!(saved["kind"], "saved"); let tool = match producer { - Producer::Goal | Producer::GoalBudget => "GoalStatus", + Producer::Goal { .. } | Producer::GoalBudget => "GoalStatus", Producer::Scheduler | Producer::Automation => "ScheduledTask", }; let created = mutate(client, "agent", "create", json!({"title":"Producer crash acceptance","automaticTitle":false,"labels":[],"tools":[tool], @@ -993,7 +1025,7 @@ async fn body( return Ok(()); } let (domain_id, session, message) = match producer { - Producer::Goal | Producer::GoalBudget => { + Producer::Goal { .. } | Producer::GoalBudget => { let result = mutate( client, "agent", @@ -1112,7 +1144,8 @@ async fn body( }) .await; if let Err(error) = accepted { - if matches!(producer, Producer::Goal) && error == "deadline: external physical model accept" + if matches!(producer, Producer::Goal { .. }) + && error == "deadline: external physical model accept" { eprintln!("{error}"); let snapshot = model_accept_snapshot(client, &session, &message, model).await; @@ -1132,7 +1165,7 @@ async fn body( let run = original["runId"].clone(); let user = user_message(client, &session, &message, &run).await?; let scheduled = match producer { - Producer::Goal | Producer::GoalBudget => { + Producer::Goal { .. } | Producer::GoalBudget => { assert_eq!(user["origin"]["kind"], "goal"); assert_eq!(user["origin"]["goalId"], domain_id); let goal = query(client, "goal", json!({"sessionId":session})).await?; @@ -1164,6 +1197,28 @@ async fn body( Some(detail) } }; + let correction = if matches!( + producer, + Producer::Goal { + pending_steering: true + } + ) { + let queued = mutate( + client, + "agent", + "send", + json!({"sessionId":session,"message":RECOVERY_STEERING,"delivery":"steer","expectedRunId":run}), + Attachments::new(), + ) + .await?; + assert_eq!(queued["kind"], "queued", "{queued}"); + let queue = query(client, "queue", json!({"sessionId":session})).await?; + assert_eq!(queue["entries"][0]["kind"], "steering", "{queue}"); + assert_eq!(queue["entries"][0]["messageId"], queued["messageId"]); + Some(queued["messageId"].as_str().unwrap().to_owned()) + } else { + None + }; phase("first client close before Host fault", clients[0].close()).await?; assert_eq!(model.producer_requests.load(Ordering::SeqCst), 1); assert_eq!( @@ -1233,8 +1288,14 @@ async fn body( .await?; assert_eq!(by_run["kind"], "found"); assert_eq!(by_run["execution"], recovered); - let successor = match producer { - Producer::Goal => { + let successors = match producer { + Producer::Goal { .. } => { + let mut successors = Vec::new(); + if let Some(correction) = &correction { + let corrected = work(client, &session, correction, Some("completed"), None).await?; + assert_ne!(corrected["runId"], run); + successors.push(corrected["runId"].clone()); + } phase("Goal completes a second iteration after recovery", async { loop { let goal = query(client, "goal", json!({"sessionId":session})).await?; @@ -1254,8 +1315,8 @@ async fn body( .await?; let next_message = format!("m_goal-{domain_id}-2"); let next = work(client, &session, &next_message, Some("completed"), None).await?; - assert_ne!(next["runId"], run); - Some(next["runId"].clone()) + successors.push(next["runId"].clone()); + successors } Producer::GoalBudget | Producer::Automation => unreachable!(), Producer::Scheduler => { @@ -1267,7 +1328,7 @@ async fn body( let children = children["sessions"].as_array().unwrap(); assert_eq!(children.len(), 1); assert_eq!(children[0]["id"], session); - None + vec![] } }; // Leave failed history for the existing read-only inspector, without a @@ -1287,7 +1348,7 @@ async fn body( assert_eq!( admitted_runs, std::iter::once(&run) - .chain(successor.iter()) + .chain(successors.iter()) .collect::>() ); let after: Vec<_> = after @@ -1309,14 +1370,21 @@ async fn body( one(&after, "model_attempt_admitted"), one(&before, "model_attempt_admitted") ); - let interrupted = one(&after, "model_interrupted"); - let recovery = one(&after, "attempt_recovery_recorded"); - assert_eq!(recovery["fact"]["recovery"]["kind"], "blocked"); - assert_eq!(interrupted["fact"]["attempt"], recovery["fact"]["attempt"]); - assert_eq!( - recovery["fact"]["attempt"], - one(&before, "model_attempt_prepared")["fact"]["attempt"] - ); + if correction.is_some() { + assert_eq!( + one(&after, "blocked")["fact"]["reason"]["protocol"], + "maka.native-model-input/provider-effects-or-unproved-adapter" + ); + } else { + let interrupted = one(&after, "model_interrupted"); + let recovery = one(&after, "attempt_recovery_recorded"); + assert_eq!(recovery["fact"]["recovery"]["kind"], "blocked"); + assert_eq!(interrupted["fact"]["attempt"], recovery["fact"]["attempt"]); + assert_eq!( + recovery["fact"]["attempt"], + one(&before, "model_attempt_prepared")["fact"]["attempt"] + ); + } assert_eq!(one(&after, "run_terminated")["fact"]["terminal"], "failed"); assert!(!after.iter().any(|record| matches!( record["fact"]["kind"].as_str(), @@ -1431,7 +1499,10 @@ async fn acceptance(producer: Producer, retry_workspace: bool) { return; } let (requests, producer_requests, eof) = match producer { - Producer::Goal => (3, 3, 1), + Producer::Goal { pending_steering } => { + let count = 3 + usize::from(pending_steering); + (count, count, 1) + } Producer::Scheduler => (4, 1, 1), Producer::Automation => unreachable!(), Producer::GoalBudget => (2, 2, 0), @@ -1454,7 +1525,24 @@ async fn acceptance(producer: Producer, retry_workspace: bool) { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn goal_continues_in_a_new_iteration_after_original_run_fails_on_reopen() { - acceptance(Producer::Goal, false).await; + acceptance( + Producer::Goal { + pending_steering: false, + }, + false, + ) + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn goal_continues_after_failed_run_delivers_queued_steering_as_followup() { + acceptance( + Producer::Goal { + pending_steering: true, + }, + false, + ) + .await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)]