diff --git a/docs/user/agent-hooks-protocol.md b/docs/user/agent-hooks-protocol.md index 5d8f9857..deb6be7d 100644 --- a/docs/user/agent-hooks-protocol.md +++ b/docs/user/agent-hooks-protocol.md @@ -72,10 +72,10 @@ need client-defined routing and failure policies. | Event | Where Kit emits it | Hook authority | | --- | --- | --- | | `session.start` | A session's model connection opens | Inject context | -| `turn.start` | A logical agent-loop turn starts, with native session and turn IDs | Observe | +| `turn.start` | A logical agent-loop turn admits its input, with native session and turn IDs | Deny, replace the admitted prompt | | `turn.end` | That logical turn completes, fails, or is cancelled | Observe | | `session.end` | A prompt run finishes or an ACP session closes | Observe | -| `user.message.inbound` | A new turn admits a user prompt | Deny, replace the prompt | +| `user.message.inbound` | A user prompt arrives, including an interjection during a turn | Deny, replace the prompt | | `model.request.before` | Before each model call, with the transcript | Deny, replace messages, supply a response, inject context | | `model.response.after` | A model call finishes, with its output | Replace messages, block response | | `model.error` | A model call fails | Observe | @@ -92,10 +92,20 @@ Model message edits accept text replacements and preserve unchanged native items. Response interception withholds streaming output until hooks finish. A compaction denial fails the turn without replacing the transcript. -A denied prompt is removed, together with resources attached to it, and the -turn ends without a model call. A denied -model request fails the turn. A denied tool call returns a permission error to -the model. +`user.message.inbound` applies arrival policy before `turn.start`. The latter's +`items` contain the effective admitted input, including attachments and +notifications submitted with it, not prior-turn history. A `modify` effect with +target `prompt` and operation `replace` on `turn.start` replaces this entire +canonical item list. Retain the supplied opaque part identities to preserve +attachments; edits cannot fabricate media. Both events support user-facing +`message` effects. An interjection during an active turn emits +`user.message.inbound` without another `turn.start`. + +A denied prompt is removed together with its admitted resources and +notifications. Denial fails the current turn without a further model call, +including when the denied input is an interjection. Prior-turn history is +preserved. A denied model request fails the turn. A denied tool call returns +a permission error to the model. Kit checks edited tool input against the tool's input schema before the tool runs; an edit that does not match counts as a hook failure. An edit that the diff --git a/src/ahp.rs b/src/ahp.rs index c65fe9fe..5998a19c 100644 --- a/src/ahp.rs +++ b/src/ahp.rs @@ -30,7 +30,7 @@ use agentkit_core::{FinishReason, Item, ItemKind, Part, ToolOutput, TurnCancella use agentkit_loop::LoopError; pub(crate) use model::{HookedAdapter, HookedSession}; -pub(crate) use prompt::{PromptHook, PromptRecord}; +pub(crate) use prompt::{PromptHook, PromptHookBuilder, PromptRecord}; pub(crate) use tool::hooked_tools; pub(crate) use turn::TurnHook; @@ -281,7 +281,13 @@ fn options() -> Result { .inject_context(capability::InjectContext::Now), ), session_end: Some(capability::session_end().observe()), - turn_start: Some(capability::turn_start().observe()), + turn_start: Some( + capability::turn_start() + .message() + .observe() + .deny() + .modify_prompt(Modify::Replace), + ), turn_end: Some(capability::turn_end().observe()), user_message_inbound: Some( capability::user_message_inbound() diff --git a/src/ahp/model.rs b/src/ahp/model.rs index b4024b7e..1dbbd1dd 100644 --- a/src/ahp/model.rs +++ b/src/ahp/model.rs @@ -614,10 +614,20 @@ impl<'a> OriginalMessage<'a> { fn canonical_items<'a>( messages: impl IntoIterator, originals: &[Item], +) -> Result, LoopError> { + canonical_items_at(messages, originals, 0) +} + +/// Keep synthesized identities at their transcript offsets while limiting +/// opaque-part authority to the supplied original message scope. +pub(super) fn canonical_items_at<'a>( + messages: impl IntoIterator, + originals: &[Item], + offset: usize, ) -> Result, LoopError> { let mut by_id = HashMap::with_capacity(originals.len()); for (index, item) in originals.iter().enumerate() { - let id = super::item_id(item, index); + let id = super::item_id(item, offset + index); let original = OriginalMessage::new(item, &id); if by_id.insert(id, original).is_some() { return Err(invalid("duplicate original model message identity")); diff --git a/src/ahp/prompt.rs b/src/ahp/prompt.rs index 05b564d1..52657e31 100644 --- a/src/ahp/prompt.rs +++ b/src/ahp/prompt.rs @@ -1,25 +1,66 @@ -use std::sync::Arc; +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; use agenthooksprotocol::generated::Event; -use agenthooksprotocol::inputs::UserMessageInboundInput; -use agenthooksprotocol::{EventType, UserMessageInboundInputMessage}; +use agenthooksprotocol::inputs::{TurnStartInput, UserMessageInboundInput}; +use agenthooksprotocol::{ + EventType, TurnStartInputTrigger, TurnStartInputTurn, UserMessageInboundInputMessage, + UserMessageInboundInputTurn, +}; use agentkit_core::{ItemKind, Part}; -use agentkit_loop::{AgentEvent, LoopCtx, LoopError, LoopMutator, MutationPoint, TranscriptCursor}; +use agentkit_loop::{ + AgentBuilder, AgentEvent, LoopCtx, LoopError, LoopMutator, ModelAdapter, MutationPoint, + TranscriptCursor, TranscriptEvent, TranscriptObserver, +}; use async_trait::async_trait; use super::Hooks; use crate::session::{SessionObserver, TransientTranscript}; -const ACP_RESOURCE: &str = "acp.resource.uri"; +const ACP_MESSAGE_ID: &str = "dev.kit.acp.message_id"; -/// Runs `user.message.inbound` over the prompt a new turn just admitted. +/// Runs arrival policy before `turn.start` on the effective admitted prompt. /// -/// Removing a denied prompt leaves no pending input, so the loop finishes the -/// turn without calling the model. +/// A denied batch is removed and explicitly fails its native turn, so older +/// unfinished history cannot become pending model input. pub(crate) struct PromptHook { hooks: Arc, channel: &'static str, record: PromptRecord, + batch: PromptBatch, +} + +/// Counts the unconsumed appended input suffix, not passive transcript history. +/// The driver serializes callbacks and mutation; no guard spans hook dispatch. +#[derive(Clone, Default)] +struct PromptBatch(Arc); + +impl TranscriptObserver for PromptBatch { + fn on_transcript_event(&self, event: TranscriptEvent<'_>) { + if matches!( + event.item.kind, + ItemKind::User | ItemKind::Notification | ItemKind::Context + ) { + self.0.fetch_add(1, Ordering::Relaxed); + } else { + self.0.store(0, Ordering::Relaxed); + } + } +} + +/// Installs both halves of prompt admission so the batch boundary is exact. +/// Register before transcript-rewriting mutators, including compaction: admission +/// owns the appended batch before another policy can consume or relocate it. +pub(crate) trait PromptHookBuilder { + fn prompt_hook(self, hook: PromptHook) -> Self; +} + +impl PromptHookBuilder for AgentBuilder { + fn prompt_hook(self, hook: PromptHook) -> Self { + self.transcript_observer(hook.batch.clone()).mutator(hook) + } } /// Where the transcript a prompt edit or denial rewrites is recorded. @@ -46,113 +87,303 @@ impl PromptHook { hooks, channel, record, + batch: PromptBatch::default(), } } } -#[async_trait] -impl LoopMutator for PromptHook { - async fn mutate( +impl PromptHook { + async fn apply( &self, cursor: &mut TranscriptCursor<'_>, - ctx: LoopCtx<'_>, - ) -> Result<(), LoopError> { - if ctx.point != MutationPoint::AfterTurnEnded { - return Ok(()); - } - self.hooks - .drain_turn_events(&ctx.session_id.to_string()) - .await; - let Some(prompt) = cursor.last().filter(|item| item.kind == ItemKind::User) else { - return Ok(()); - }; - let index = cursor.len().saturating_sub(1); - let message = super::message(prompt, index); + ctx: &LoopCtx<'_>, + event_type: EventType, + index: Option, + start: usize, + trigger: TurnStartInputTrigger, + ) -> Result { + let message = index.map(|index| super::message(&cursor[index], index)); let original: String = message - .parts .iter() + .flat_map(|message| &message.parts) .filter_map(|part| match part { agenthooksprotocol::inputs::PartInput::Text(body) => editable_text(body), _ => None, }) .collect(); - let mut input = UserMessageInboundInput::new(UserMessageInboundInputMessage::new( - self.channel, - Vec::new(), - "user", - )); - input.session = Some(self.hooks.session(&ctx.session_id.to_string())); - input.turn = ctx - .turn_id - .map(|id| agenthooksprotocol::UserMessageInboundInputTurn::new(id.to_string())); - let boundary = self - .hooks - .sdk - .user_message_inbound(input.with_sources().with_message_messages(vec![message])); - let result = super::dispatch(boundary, ctx.cancellation.as_ref()).await; + let Some(turn_id) = ctx.turn_id else { + return Ok(true); + }; + let result = if event_type == EventType::UserMessageInbound { + let mut input = UserMessageInboundInput::new(UserMessageInboundInputMessage::new( + self.channel, + Vec::new(), + "user", + )); + input.session = Some(self.hooks.session(&ctx.session_id.to_string())); + input.turn = Some(UserMessageInboundInputTurn::new(turn_id.to_string())); + super::dispatch( + self.hooks.sdk.user_message_inbound( + input + .with_sources() + .with_message_messages(message.into_iter().collect()), + ), + ctx.cancellation.as_ref(), + ) + .await + } else { + let items = cursor[start..] + .iter() + .enumerate() + .map(|(offset, item)| super::message(item, start + offset)) + .collect(); + let input = TurnStartInput::new( + Vec::new(), + trigger, + TurnStartInputTurn::new(turn_id.to_string()), + ) + .with_session(self.hooks.session(&ctx.session_id.to_string())); + // Start delivery owns the session's observation gate until the SDK + // finishes. Dropping the admission wait must not drop a start that + // has not reached the transport yet, or allow its end to overtake it. + let guard = self + .hooks + .turn_delivery(&ctx.session_id.to_string()) + .lock_owned() + .await; + let sdk = Arc::clone(&self.hooks.sdk); + let cancellation = ctx.cancellation.clone(); + let operation = tokio::spawn(async move { + let _guard = guard; + let input = if cancellation + .as_ref() + .is_some_and(agentkit_core::TurnCancellation::is_cancelled) + { + input.with_sources() + } else { + input.with_sources().with_items(items) + }; + super::dispatch(sdk.turn_start(input), None).await + }); + let operation = async move { + operation + .await + .map_err(|error| LoopError::InvalidState(error.to_string()))? + }; + let operation = std::pin::pin!(operation); + if let Some(cancellation) = ctx.cancellation.as_ref() { + let cancelled = std::pin::pin!(cancellation.cancelled()); + match futures_util::future::select(cancelled, operation).await { + futures_util::future::Either::Left(_) => Err(LoopError::Cancelled), + futures_util::future::Either::Right((result, _)) => { + if cancellation.is_cancelled() { + Err(LoopError::Cancelled) + } else { + result + } + } + } + } else { + operation.await + } + }; if matches!(&result, Err(LoopError::Cancelled)) { return Err(LoopError::Cancelled); } - let edited = match self.hooks.settle( - EventType::UserMessageInbound, - result, - &ctx.session_id.to_string(), - ) { + let edited = match self + .hooks + .settle(event_type, result, &ctx.session_id.to_string()) + { Some(outcome) if outcome.outcome.is_denied() => Err(super::denial(&outcome.outcome)), + Some(outcome) + if event_type == EventType::TurnStart && !super::has_modification(&outcome) => + { + return Ok(true); + } Some(outcome) => match outcome.event { + Event::TurnStart(event) => { + // `prompt` replaces the entire admitted canonical item list, + // not just the final user's text. Reuse model-effect validation + // to retain authorized opaque parts and reject fabricated media. + if index.is_none() { + return Ok(true); + } + let mut replacement = super::model::canonical_items_at( + event.0.items.iter().map(|item| item.0.as_ref()), + &cursor[start..], + start, + )?; + // Canonical IDs belong to the hook; ACP replay identity belongs + // to the host and survives a replacement with a new ID. + if let Some(identity) = + index.and_then(|index| cursor[index].metadata.get(ACP_MESSAGE_ID)) + && let Some(user) = replacement + .iter_mut() + .rev() + .find(|item| item.kind == ItemKind::User) + { + user.metadata + .insert(ACP_MESSAGE_ID.into(), identity.clone()); + } + cursor.splice(start.., replacement); + self.persist(cursor, ctx); + return Ok(true); + } Event::UserMessageInbound(event) => Ok(event .message .messages .iter() - .map(|message| { - message - .parts - .iter() - .filter_map(|part| match part.as_ref() { - agenthooksprotocol::ContentItem::TextBodyPart(body) => { - editable_text(body) - } - _ => None, - }) - .collect::() - }) + .map(|message| message_text(&message.parts)) .collect::>() .join("\n")), - _ => return Ok(()), + _ => return Ok(true), }, None => Err("agent hook dispatch failed".into()), }; + let Some(index) = index else { + return Ok(true); + }; + let accepted = edited.is_ok(); match edited { Ok(text) => { if text == original { - return Ok(()); + return Ok(true); } - if let Some(prompt) = cursor.last_mut() { + if let Some(prompt) = cursor.get_mut(index) { prompt.parts.retain(|part| super::part_text(part).is_none()); prompt.parts.insert(0, Part::text(text)); } } Err(reason) => { - cursor.pop(); - while cursor.last().is_some_and(|item| { - item.kind == ItemKind::Context && item.metadata.contains_key(ACP_RESOURCE) - }) { - cursor.pop(); - } + cursor.truncate(start); ctx.emitter.emit(AgentEvent::Warning { message: format!("prompt blocked by agent hook: {reason}"), }); } } + self.persist(cursor, ctx); + Ok(accepted) + } + + fn persist(&self, cursor: &[agentkit_core::Item], ctx: &LoopCtx<'_>) { if let Err(error) = self.record.replace(cursor) { ctx.emitter.emit(AgentEvent::Warning { message: format!("could not persist agent hook prompt edit: {error}"), }); } + } +} + +#[async_trait] +impl LoopMutator for PromptHook { + async fn mutate( + &self, + cursor: &mut TranscriptCursor<'_>, + ctx: LoopCtx<'_>, + ) -> Result<(), LoopError> { + if !matches!( + ctx.point, + MutationPoint::AfterTurnEnded | MutationPoint::AfterToolResult + ) { + return Ok(()); + } + self.hooks + .drain_turn_events(&ctx.session_id.to_string()) + .await; + // Consume the append count before awaiting policy, including cancellation. + let pending = self.batch.0.swap(0, Ordering::Relaxed); + let start = cursor + .len() + .checked_sub(pending) + .ok_or_else(|| LoopError::InvalidState("prompt batch exceeds transcript".into()))?; + let index = cursor.len().checked_sub(1).filter(|&index| { + index >= start && matches!(cursor[index].kind, ItemKind::User | ItemKind::Notification) + }); + // Arrival and admission are separate boundaries with shared message identity. + // A2A and delegated sessions carry agent-originated input; foreground + // CLI and ACP expose user prompts. Internal notifications are external. + let trigger = if index.is_none_or(|index| cursor[index].kind != ItemKind::User) + || self.channel == "a2a" + || self.hooks.injections.is_child(&ctx.session_id.to_string()) + { + TurnStartInputTrigger::External + } else { + TurnStartInputTrigger::User + }; + // Notifications are execution input, not transport arrivals. A user + // interjection after tools is an arrival within the existing turn. + let arrival = if index.is_some_and(|index| cursor[index].kind == ItemKind::User) { + self.apply( + cursor, + &ctx, + EventType::UserMessageInbound, + index, + start, + trigger.clone(), + ) + .await + } else { + Ok(true) + }; + let new_turn = ctx.turn_id.is_some_and(|turn| { + self.hooks + .claim_turn_start(&ctx.session_id.to_string(), &turn.to_string()) + }); + let accepted = matches!(arrival, Ok(true)); + let index = match arrival { + Ok(true) => index, + Ok(false) => None, + Err(LoopError::Cancelled) => { + if new_turn { + // The native turn still ends. Deliver its start without accepting + // effects from the cancelled admission or changing the transcript. + if let Some(turn_id) = ctx.turn_id { + let input = TurnStartInput::new( + Vec::new(), + trigger, + TurnStartInputTurn::new(turn_id.to_string()), + ) + .with_session(self.hooks.session(&ctx.session_id.to_string())); + super::settle( + EventType::TurnStart, + super::dispatch(self.hooks.sdk.turn_start(input.with_sources()), None) + .await, + ); + } + } + return Err(LoopError::Cancelled); + } + Err(error) => return Err(error), + }; + if new_turn { + // Even rejected input owns a native turn. Report an empty admitted + // input rather than inventing a prompt or leaving an unmatched end. + let admitted = self + .apply(cursor, &ctx, EventType::TurnStart, index, start, trigger) + .await?; + if !admitted { + return Err(LoopError::Mutator("prompt blocked by agent hook".into())); + } + } + // Transcript shape is not a finish signal: removing this batch may expose + // a failed user prompt, notification, or tool result from an older turn. + if !accepted { + return Err(LoopError::Mutator("prompt blocked by agent hook".into())); + } Ok(()) } } +fn message_text(parts: &[Box]) -> String { + parts + .iter() + .filter_map(|part| match part.as_ref() { + agenthooksprotocol::ContentItem::TextBodyPart(body) => editable_text(body), + _ => None, + }) + .collect() +} + // Reference descriptors describe retained non-text input; do not inject their // URI/base64/filename representation into the model's text when a hook edits it. fn editable_text(body: &agenthooksprotocol::TextBodyPart) -> Option<&str> { diff --git a/src/ahp/tests.rs b/src/ahp/tests.rs index 6aed4952..4caa7e3d 100644 --- a/src/ahp/tests.rs +++ b/src/ahp/tests.rs @@ -125,7 +125,7 @@ async fn registration_headers_reach_http_hook_requests() { "X-Kit-Test": "configured-header", }, }, - "subscriptions": [intercept(&["user.message.inbound"])], + "subscriptions": [intercept(&["turn.start"])], }], })) .unwrap(), @@ -134,7 +134,7 @@ async fn registration_headers_reach_http_hook_requests() { let mut options = options().unwrap(); options.backend.allow_loopback_http = true; let hooks = Arc::new(Hooks::new(read(&path).unwrap().unwrap(), options).unwrap()); - prompt(&hooks, &Model::default(), "hello").await; + prompt(&hooks, &Model::default(), "hello").await.0.unwrap(); let headers = tokio::time::timeout(std::time::Duration::from_secs(5), received.recv()) .await .unwrap() @@ -148,6 +148,7 @@ async fn registration_headers_reach_http_hook_requests() { fn every_intercept_event_accepts_observe_subscriptions() { for event in [ "user.message.inbound", + "turn.start", "model.request.before", "tool.before", "tool.after", @@ -159,6 +160,23 @@ fn every_intercept_event_accepts_observe_subscriptions() { } } +#[test] +fn turn_start_grants_prompt_effects_without_injection_or_flow() { + let grants = options().unwrap().capabilities.to_wire(); + let start = serde_json::to_value(&grants[&EventType::TurnStart]).unwrap(); + let capabilities = &start["capabilities"]; + assert_eq!( + capabilities["effects"], + json!(["deny", "message", "modify"]) + ); + assert_eq!( + capabilities["modify"]["prompt"], + json!({"replace": true, "merge": false}) + ); + assert!(capabilities.get("inject").is_none()); + assert!(capabilities.get("flow").is_none()); +} + #[test] fn every_granted_event_accepts_wildcard_subscriptions() { hooks(vec![intercept(&["*"]), observe(&["*"])]); @@ -167,7 +185,7 @@ fn every_granted_event_accepts_wildcard_subscriptions() { #[tokio::test] async fn every_lifecycle_event_reports_harness_identity() { let (hooks, backend) = hooks(vec![intercept(&["*"]), observe(&["*"])]); - prompt(&hooks, &Model::default(), "hello").await; + prompt(&hooks, &Model::default(), "hello").await.0.unwrap(); call(&hooks, &Echo::new(), json!({"text": "hello"})).await; hooks .session_end("abc", SessionEndInputOutcome::Completed) @@ -200,7 +218,6 @@ async fn every_lifecycle_event_reports_harness_identity() { "session.end", "turn.start", "turn.end", - "user.message.inbound", "model.request.before", "model.response.after", "model.error", @@ -927,11 +944,27 @@ fn last_user_text(transcript: &[Item]) -> Option { .map(|item| item.parts.iter().filter_map(part_text).collect()) } -async fn prompt(hooks: &Arc, model: &Model, text: &str) -> (LoopStep, Vec) { +fn transient_prompt_hook(hooks: &Arc, channel: &'static str) -> PromptHook { + PromptHook::new( + hooks.clone(), + channel, + PromptRecord::Transient(crate::session::TransientTranscript::new(vec![])), + ) +} + +async fn prompt( + hooks: &Arc, + model: &Model, + text: &str, +) -> (Result, Vec) { admit(hooks, model, vec![Item::text(ItemKind::User, text)]).await } -async fn admit(hooks: &Arc, model: &Model, input: Vec) -> (LoopStep, Vec) { +async fn admit( + hooks: &Arc, + model: &Model, + input: Vec, +) -> (Result, Vec) { let root = tempfile::tempdir().unwrap(); let directory = root.path().join("sessions"); let initial = vec![Item::text(ItemKind::System, "system")]; @@ -944,7 +977,7 @@ async fn admit(hooks: &Arc, model: &Model, input: Vec) -> (LoopStep Some(hooks.clone()), SessionStartInputTrigger::Startup, )) - .mutator(PromptHook::new( + .prompt_hook(PromptHook::new( hooks.clone(), "cli", PromptRecord::Session(opened.observer.clone()), @@ -957,17 +990,585 @@ async fn admit(hooks: &Arc, model: &Model, input: Vec) -> (LoopStep .start(SessionConfig::new("abc")) .await .unwrap(); - let step = driver.next().await.unwrap(); + let step = driver.next().await; hooks.drain_turn_events("abc").await; let persisted = crate::session::load_in(root.path(), &directory, "abc").unwrap(); (step, persisted) } #[tokio::test] -async fn prompt_edits_replace_the_submitted_text() { +async fn turn_start_intercepts_initial_and_subsequent_prompts_with_channel_triggers() { + for (channel, trigger) in [("cli", "user"), ("acp", "user"), ("a2a", "external")] { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond( + "turn.start", + json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("edited")}]), + ); + let model = Model::default(); + let recorded = crate::session::TransientTranscript::new(vec![]); + let mut driver = Agent::builder() + .model(model.clone()) + .prompt_hook(PromptHook::new( + hooks.clone(), + channel, + PromptRecord::Transient(recorded.clone()), + )) + .transcript_observer(recorded.clone()) + .input(vec![Item::text(ItemKind::User, "first")]) + .build() + .unwrap() + .start(SessionConfig::new(channel)) + .await + .unwrap(); + assert!(matches!( + driver.next().await.unwrap(), + LoopStep::Finished(_) + )); + driver + .submit_input(vec![Item::text(ItemKind::User, "second")]) + .unwrap(); + assert!(matches!( + driver.next().await.unwrap(), + LoopStep::Finished(_) + )); + let starts = backend.events("turn.start"); + assert_eq!(starts.len(), 2, "{channel}"); + for (event, submitted) in starts.iter().zip(["first", "second"]) { + assert_eq!(event["trigger"], trigger); + assert_eq!(event["items"].as_array().unwrap().len(), 1); + assert_eq!(event["items"][0]["parts"][0]["text"], submitted); + assert_eq!(event["session"]["id"], channel); + assert!(event.get("message").is_none()); + } + assert_ne!(starts[0]["turn"]["id"], starts[1]["turn"]["id"]); + let requests = model.requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + for request in requests.iter() { + assert_eq!(last_user_text(request).as_deref(), Some("edited")); + } + let users: Vec<_> = recorded + .snapshot() + .into_iter() + .filter(|item| item.kind == ItemKind::User) + .collect(); + assert_eq!(users.len(), 2); + for user in users { + assert_eq!(user.parts, vec![Part::text("edited")]); + } + } +} + +#[tokio::test] +async fn delegated_acp_prompt_uses_external_turn_start_trigger() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + let mut driver = Agent::builder() + .model( + HookedAdapter::new( + Model::default(), + Some(hooks.clone()), + SessionStartInputTrigger::Startup, + ) + .with_child(true), + ) + .prompt_hook(transient_prompt_hook(&hooks, "acp")) + .input(vec![Item::text(ItemKind::User, "delegated task")]) + .build() + .unwrap() + .start(SessionConfig::new("delegated-acp")) + .await + .unwrap(); + assert!(matches!( + driver.next().await.unwrap(), + LoopStep::Finished(_) + )); + let starts = backend.events("turn.start"); + assert_eq!(starts.len(), 1); + assert_eq!(starts[0]["trigger"], "external"); + assert_eq!( + starts[0]["session"]["agent"], + json!({"id": "delegated-acp"}) + ); + assert_eq!(starts[0]["items"][0]["parts"][0]["text"], "delegated task"); +} + +#[tokio::test] +async fn turn_start_contains_attached_resources_and_multimodal_prompt() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + let mut resource = Item::text(ItemKind::Context, "attached resource"); + resource + .metadata + .insert("acp.resource.uri".into(), json!("file:///notes.txt")); + let image = Part::media( + Modality::Image, + "image/png", + DataRef::uri("file:///image.png"), + ); + let model = Model::default(); + admit( + &hooks, + &model, + vec![ + resource, + Item::new(ItemKind::User, vec![Part::text("question"), image]), + ], + ) + .await + .0 + .unwrap(); + let starts = backend.events("turn.start"); + assert_eq!(starts.len(), 1); + let items = starts[0]["items"].as_array().unwrap(); + assert_eq!(items.len(), 2); + assert_eq!(items[0]["parts"][0]["text"], "attached resource"); + assert_eq!(items[1]["parts"][0]["text"], "question"); + let descriptor: Value = + serde_json::from_str(items[1]["parts"][1]["text"].as_str().unwrap()).unwrap(); + assert_eq!(descriptor["mediaType"], "image/png"); + assert_eq!( + descriptor["data"], + serde_json::to_value(DataRef::uri("file:///image.png")).unwrap() + ); + assert_eq!( + last_user_text(&model.requests.lock().unwrap()[0]).as_deref(), + Some("question") + ); +} + +#[tokio::test] +async fn turn_start_replacement_retains_unchanged_native_attachments_by_admitted_identity() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + let image = Part::media( + Modality::Image, + "image/png", + DataRef::uri("file:///image.png"), + ); + // System history occupies index 0; attached context and user are indices 1–2. + let mut replacement = text_message("edited"); + replacement[0]["id"] = json!("item-2"); + replacement[0]["parts"][0]["id"] = json!("item-2/0"); + for part in content_parts("item-2/1", &image) { + let agenthooksprotocol::inputs::PartInput::Text(body) = part else { + panic!("expected URI descriptor"); + }; + replacement[0]["parts"] + .as_array_mut() + .unwrap() + .push(serde_json::to_value(body).unwrap()); + } + backend.respond( + "turn.start", + json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":replacement}]), + ); + let mut resource = Item::text(ItemKind::Context, "resource"); + resource + .metadata + .insert("acp.resource.uri".into(), json!("file:///notes.txt")); + let model = Model::default(); + let (_, persisted) = admit( + &hooks, + &model, + vec![ + resource, + Item::new(ItemKind::User, vec![Part::text("original"), image.clone()]), + ], + ) + .await; + let requests = model.requests.lock().unwrap(); + for items in [&persisted, &requests[0]] { + let user = items + .iter() + .find(|item| item.kind == ItemKind::User) + .unwrap(); + assert_eq!(user.parts, vec![Part::text("edited"), image.clone()]); + assert!( + items + .iter() + .all(|item| !item.metadata.contains_key("acp.resource.uri")) + ); + } +} + +#[tokio::test] +async fn turn_start_replacement_replaces_the_whole_admitted_item_list() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("replacement")} ])); + let mut resource = Item::text(ItemKind::Context, "original resource"); + resource + .metadata + .insert("acp.resource.uri".into(), json!("file:///notes.txt")); + let model = Model::default(); + let (_, persisted) = admit( + &hooks, + &model, + vec![ + resource, + Item::new( + ItemKind::User, + vec![ + Part::text("original"), + Part::media( + Modality::Image, + "image/png", + DataRef::uri("file:///image.png"), + ), + ], + ), + ], + ) + .await; + let requests = model.requests.lock().unwrap(); + for items in [&persisted, &requests[0]] { + let users: Vec<_> = items + .iter() + .filter(|item| item.kind == ItemKind::User) + .collect(); + assert_eq!(users.len(), 1); + assert_eq!(users[0].parts, vec![Part::text("replacement")]); + assert!( + items + .iter() + .all(|item| !item.metadata.contains_key("acp.resource.uri")) + ); + } +} + +#[tokio::test] +async fn turn_start_owns_prepended_notifications_without_absorbing_history() { + for deny in [false, true] { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", if deny { + json!([{"type":"deny", "reason":"blocked"}]) + } else { + json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("replacement")}]) + }); + let model = Model::default(); + let history = vec![ + Item::text(ItemKind::User, "prior user"), + Item::notification("prior notification"), + ]; + let recorded = crate::session::TransientTranscript::new(history.clone()); + let mut driver = Agent::builder() + .model(model.clone()) + .prompt_hook(PromptHook::new( + hooks.clone(), + "acp", + PromptRecord::Transient(recorded.clone()), + )) + .transcript_observer(recorded.clone()) + .transcript(history.clone()) + .input(vec![ + Item::notification("voice changed"), + Item::text(ItemKind::User, "hello"), + ]) + .build() + .unwrap() + .start(SessionConfig::new("batch")) + .await + .unwrap(); + let step = driver.next().await; + if deny { + assert!(matches!(step, Err(LoopError::Mutator(_)))); + } else { + assert!(matches!(step, Ok(LoopStep::Finished(_)))); + } + let starts = backend.events("turn.start"); + assert_eq!(starts.len(), 1); + assert_eq!(starts[0]["items"].as_array().unwrap().len(), 2); + assert_eq!(starts[0]["items"][0]["parts"][0]["text"], "voice changed"); + let persisted = recorded.snapshot(); + assert_eq!(&persisted[..history.len()], history.as_slice()); + assert!( + !persisted + .iter() + .any(|item| item.parts == vec![Part::text("voice changed")]) + ); + let requests = model.requests.lock().unwrap(); + assert_eq!(requests.len(), usize::from(!deny)); + if !deny { + assert_eq!(last_user_text(&requests[0]).as_deref(), Some("replacement")); + assert_eq!(&requests[0][..history.len()], history.as_slice()); + assert!( + !requests[0] + .iter() + .any(|item| item.parts == vec![Part::text("voice changed")]) + ); + } + } +} + +#[tokio::test] +async fn turn_start_admits_manual_compaction_before_the_command_is_consumed() { + for deny in [false, true] { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", if deny { + json!([{"type":"deny", "reason":"blocked"}]) + } else { + json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("replacement")}]) + }); + let history = vec![ + Item::text(ItemKind::User, "prior user"), + Item::text(ItemKind::Assistant, "prior answer"), + ]; + let recorded = crate::session::TransientTranscript::new(history.clone()); + let compactor = crate::compaction::automatic( + Model::default(), + agentkit_loop::TelemetryConfig::default(), + None, + "compact-admission", + "compact-admission", + None, + ) + .unwrap() + .mirroring(recorded.clone()); + let model = Model::default(); + let mut driver = Agent::builder() + .model(model.clone()) + .prompt_hook(PromptHook::new( + hooks.clone(), + "cli", + PromptRecord::Transient(recorded.clone()), + )) + .mutator(compactor) + .transcript_observer(recorded.clone()) + .transcript(history.clone()) + .input(vec![Item::text(ItemKind::User, "/compact")]) + .build() + .unwrap() + .start(SessionConfig::new("compact-admission")) + .await + .unwrap(); + let step = driver.next().await; + if deny { + assert!(matches!(step, Err(LoopError::Mutator(_)))); + } else { + assert!(matches!(step, Ok(LoopStep::Finished(_)))); + } + let starts = backend.events("turn.start"); + assert_eq!(starts.len(), 1); + assert_eq!(starts[0]["items"].as_array().unwrap().len(), 1); + assert_eq!(starts[0]["items"][0]["parts"][0]["text"], "/compact"); + assert_eq!(&recorded.snapshot()[..history.len()], history.as_slice()); + let requests = model.requests.lock().unwrap(); + assert_eq!(requests.len(), usize::from(!deny)); + if !deny { + assert_eq!(last_user_text(&requests[0]).as_deref(), Some("replacement")); + assert_eq!(&requests[0][..history.len()], history.as_slice()); + } + } +} + +#[tokio::test] +async fn turn_start_replacement_preserves_host_acp_replay_identity() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("edited")} ])); + let mut user = Item::text(ItemKind::User, "original"); + user.metadata + .insert("dev.kit.acp.message_id".into(), json!("kit-user-live")); + let model = Model::default(); + let (_, persisted) = admit(&hooks, &model, vec![user]).await; + let requests = model.requests.lock().unwrap(); + for items in [&persisted, &requests[0]] { + let user = items + .iter() + .find(|item| item.kind == ItemKind::User) + .unwrap(); + assert_eq!(user.parts, vec![Part::text("edited")]); + assert_eq!( + user.metadata["dev.kit.acp.message_id"], + json!("kit-user-live") + ); + } +} + +#[tokio::test] +async fn turn_start_messages_reach_the_session_after_inbound_and_before_model() { + let (hooks, backend) = hooks(vec![intercept(&["*"]), observe(&["turn.end"])]); + let (sender, receiver) = std::sync::mpsc::channel(); + let _subscription = hooks.subscribe_messages("abc".into(), move |text| { + sender.send(text.to_owned()).is_ok() + }); + backend.respond( + "turn.start", + json!([{"type":"message", "text":"prompt accepted"}]), + ); + let model = Model::default(); + let (_, persisted) = prompt(&hooks, &model, "hello").await; + assert_eq!(receiver.try_iter().collect::>(), ["prompt accepted"]); + assert_eq!(last_user_text(&persisted).as_deref(), Some("hello")); + assert_eq!( + last_user_text(&model.requests.lock().unwrap()[0]).as_deref(), + Some("hello") + ); + let inbound = backend.events("user.message.inbound"); + let starts = backend.events("turn.start"); + assert_eq!(inbound.len(), 1); + assert_eq!( + inbound[0]["message"]["messages"][0]["id"], + starts[0]["items"][0]["id"] + ); + let types: Vec<_> = backend + .all_events() + .iter() + .map(|event| event["type"].as_str().unwrap().to_owned()) + .collect(); + assert_eq!( + types, + [ + "session.start", + "user.message.inbound", + "turn.start", + "model.request.before", + "model.response.after", + "turn.end" + ] + ); +} + +#[tokio::test] +async fn inbound_replacement_is_the_effective_turn_start_prompt() { + let (hooks, backend) = hooks(vec![intercept(&["user.message.inbound", "turn.start"])]); + backend.respond("user.message.inbound", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("inbound edit")}])); + backend.respond("turn.start", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("turn edit")}])); + let model = Model::default(); + let (_, persisted) = prompt(&hooks, &model, "original").await; + let inbound = backend.events("user.message.inbound"); + let starts = backend.events("turn.start"); + assert_eq!(inbound.len(), 1); + assert_eq!(starts.len(), 1); + assert_eq!( + inbound[0]["message"]["messages"][0]["parts"][0]["text"], + "original" + ); + assert_eq!(starts[0]["items"][0]["parts"][0]["text"], "inbound edit"); + assert_eq!( + inbound[0]["message"]["messages"][0]["id"], + starts[0]["items"][0]["id"] + ); + assert_eq!(last_user_text(&persisted).as_deref(), Some("turn edit")); + assert_eq!( + last_user_text(&model.requests.lock().unwrap()[0]).as_deref(), + Some("turn edit") + ); +} + +#[tokio::test] +async fn inbound_only_replacement_still_reaches_model_and_persistence() { let (hooks, backend) = hooks(vec![intercept(&["user.message.inbound"])]); + backend.respond("user.message.inbound", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("inbound edit")}])); + let model = Model::default(); + let (_, persisted) = prompt(&hooks, &model, "original").await; + assert_eq!(last_user_text(&persisted).as_deref(), Some("inbound edit")); + assert_eq!( + last_user_text(&model.requests.lock().unwrap()[0]).as_deref(), + Some("inbound edit") + ); +} + +#[tokio::test] +async fn inbound_denial_emits_empty_turn_start_and_finishes_without_model() { + let (hooks, backend) = hooks(vec![ + intercept(&["user.message.inbound", "turn.start"]), + observe(&["turn.end"]), + ]); + backend.respond( + "user.message.inbound", + json!([{"type":"deny", "reason":"blocked"}]), + ); + backend.respond("turn.start", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("must not resurrect")} ])); + let model = Model::default(); + let (step, persisted) = prompt(&hooks, &model, "original").await; + assert!(matches!(step, Err(LoopError::Mutator(_)))); + assert!(model.requests.lock().unwrap().is_empty()); + assert_eq!(last_user_text(&persisted), None); + assert_eq!(backend.events("user.message.inbound").len(), 1); + let starts = backend.events("turn.start"); + let ends = backend.events("turn.end"); + assert_eq!(starts.len(), 1); + assert_eq!(starts[0]["items"], json!([])); + assert_eq!(ends.len(), 1); + assert_eq!(starts[0]["turn"], ends[0]["turn"]); +} + +#[tokio::test] +async fn notification_only_turns_emit_external_boundaries_without_user_inbound() { + let (hooks, backend) = hooks(vec![ + intercept(&["user.message.inbound", "turn.start"]), + observe(&["turn.end"]), + ]); + let model = Model::default(); + let mut driver = Agent::builder() + .model(model.clone()) + .observer(TurnHook(hooks.clone())) + .prompt_hook(transient_prompt_hook(&hooks, "acp")) + .input(vec![Item::notification("first notification")]) + .build() + .unwrap() + .start(SessionConfig::new("notifications")) + .await + .unwrap(); + for text in ["first notification", "second notification"] { + if text == "second notification" { + driver.submit_input(vec![Item::notification(text)]).unwrap(); + } + assert!(matches!( + driver.next().await.unwrap(), + LoopStep::Finished(_) + )); + hooks.drain_turn_events("notifications").await; + } + let starts = backend.events("turn.start"); + let ends = backend.events("turn.end"); + assert_eq!(starts.len(), 2); + assert_eq!(ends.len(), 2); + for ((start, end), text) in starts + .iter() + .zip(&ends) + .zip(["first notification", "second notification"]) + { + assert_eq!(start["trigger"], "external"); + assert_eq!(start["items"][0]["parts"][0]["text"], text); + assert_eq!(start["items"][0]["role"], "user"); + assert_eq!(start["turn"], end["turn"]); + } + assert_eq!(model.requests.lock().unwrap().len(), 2); + assert!(backend.events("user.message.inbound").is_empty()); +} + +#[tokio::test] +async fn notification_turn_start_replacement_reaches_model_and_persistence() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("edited notification")} ])); + let model = Model::default(); + let (_, persisted) = admit(&hooks, &model, vec![Item::notification("autonomous work")]).await; + let requests = model.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + for items in [&persisted, &requests[0]] { + assert_eq!( + last_user_text(items).as_deref(), + Some("edited notification") + ); + assert!(items.iter().all(|item| item.kind != ItemKind::Notification)); + } +} + +#[tokio::test] +async fn notification_turn_start_denial_prevents_model_call() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", json!([{"type":"deny", "reason":"blocked"}])); + let model = Model::default(); + let (step, persisted) = + admit(&hooks, &model, vec![Item::notification("autonomous work")]).await; + assert!(matches!(step, Err(LoopError::Mutator(_)))); + assert!(model.requests.lock().unwrap().is_empty()); + assert!(persisted.iter().all(|item| item.kind == ItemKind::System)); + assert_eq!(backend.events("turn.start").len(), 1); +} + +#[tokio::test] +async fn prompt_edits_replace_the_submitted_text() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); backend.respond( - "user.message.inbound", + "turn.start", json!([{"type": "modify", "target": "prompt", "operation": "replace", "value": text_message("edited")}]), ); let model = Model::default(); @@ -981,27 +1582,21 @@ async fn prompt_edits_replace_the_submitted_text() { #[tokio::test] async fn denied_prompts_finish_without_a_model_call() { - let (hooks, backend) = hooks(vec![intercept(&["user.message.inbound"])]); - backend.respond( - "user.message.inbound", - json!([{"type": "deny", "reason": "blocked"}]), - ); + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", json!([{"type": "deny", "reason": "blocked"}])); let model = Model::default(); let (step, persisted) = prompt(&hooks, &model, "original").await; - assert!(matches!(step, LoopStep::Finished(result) if result.items.is_empty())); + assert!(matches!(step, Err(LoopError::Mutator(_)))); assert!(model.requests.lock().unwrap().is_empty()); assert_eq!(last_user_text(&persisted), None); } #[tokio::test] -async fn media_only_prompts_reach_the_inbound_hook() { - let (hooks, backend) = hooks(vec![intercept(&["user.message.inbound"])]); - backend.respond( - "user.message.inbound", - json!([{"type": "deny", "reason": "blocked"}]), - ); +async fn media_only_prompts_reach_the_turn_start_hook() { + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", json!([{"type": "deny", "reason": "blocked"}])); let model = Model::default(); let image = Item::new( ItemKind::User, @@ -1014,17 +1609,14 @@ async fn media_only_prompts_reach_the_inbound_hook() { let (step, _) = admit(&hooks, &model, vec![image]).await; - assert!(matches!(step, LoopStep::Finished(result) if result.items.is_empty())); + assert!(matches!(step, Err(LoopError::Mutator(_)))); assert!(model.requests.lock().unwrap().is_empty()); } #[tokio::test] async fn denied_prompts_drop_their_attached_resources() { - let (hooks, backend) = hooks(vec![intercept(&["user.message.inbound"])]); - backend.respond( - "user.message.inbound", - json!([{"type": "deny", "reason": "blocked"}]), - ); + let (hooks, backend) = hooks(vec![intercept(&["turn.start"])]); + backend.respond("turn.start", json!([{"type": "deny", "reason": "blocked"}])); let model = Model::default(); let mut resource = Item::text(ItemKind::Context, "attached secret"); resource @@ -1077,7 +1669,7 @@ async fn session_start_and_model_response_are_observed() { let (hooks, backend) = hooks(vec![observe(&["session.start", "model.response.after"])]); let model = Model::default(); - prompt(&hooks, &model, "hello").await; + prompt(&hooks, &model, "hello").await.0.unwrap(); let started = backend.events("session.start"); assert_eq!(started.len(), 1); @@ -1125,6 +1717,7 @@ async fn turn_boundaries_keep_native_ids_across_turns_and_sessions() { let mut driver = Agent::builder() .model(Model::default()) .observer(TurnHook(hooks.clone())) + .prompt_hook(transient_prompt_hook(&hooks, "cli")) .input(vec![Item::text(ItemKind::User, "one")]) .build() .unwrap() @@ -1158,6 +1751,218 @@ async fn turn_boundaries_keep_native_ids_across_turns_and_sessions() { assert_ne!(starts[0]["turn"]["id"], starts[1]["turn"]["id"]); } +/// Holds the intercept response after transport arrival until native cancellation. +struct HoldTurnStart { + event_type: &'static str, + events: Arc, + reached: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +impl agenthooksprotocol::adapters::registered::ManagedBackend for HoldTurnStart { + fn call( + &self, + request: Value, + timeout: std::time::Duration, + ) -> agenthooksprotocol::client::LocalFuture< + '_, + Result, + > { + Box::pin(async move { + let hold = request["params"]["event"]["type"] == self.event_type; + let response = self.events.call(request, timeout).await?; + if hold { + self.reached.notify_one(); + self.release.notified().await; + } + Ok(response) + }) + } + + fn shutdown( + &self, + ) -> agenthooksprotocol::client::LocalFuture< + '_, + Result<(), agenthooksprotocol::client::HookError>, + > { + self.events.shutdown() + } +} + +#[tokio::test] +async fn notification_cancelled_before_start_dispatch_pairs_empty_start_without_edits() { + let backend = Arc::new(super::test_support::Backend::default()); + let hooks = super::test_support::hooks_with_backend( + vec![intercept(&["turn.start"]), observe(&["turn.end"])], + backend.clone(), + ); + backend.respond("turn.start", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("must not commit")}])); + let controller = agentkit_core::CancellationController::new(); + let model = Model::default(); + let recorded = crate::session::TransientTranscript::new(vec![]); + let mut driver = Agent::builder() + .model(model.clone()) + .cancellation(controller.handle()) + .observer(TurnHook(hooks.clone())) + .prompt_hook(PromptHook::new( + hooks.clone(), + "cli", + PromptRecord::Transient(recorded.clone()), + )) + .transcript_observer(recorded.clone()) + .input(vec![Item::notification("original notification")]) + .build() + .unwrap() + .start(SessionConfig::new("cancel-before-start")) + .await + .unwrap(); + // The real observation gate pauses prompt admission within the native turn, + // before the start operation is created or polled. + let gate = hooks + .turn_delivery("cancel-before-start") + .lock_owned() + .await; + let step = driver.next(); + let mut step = std::pin::pin!(step); + assert!(futures_util::poll!(step.as_mut()).is_pending()); + assert!(backend.all_events().is_empty()); + controller.interrupt(); + drop(gate); + let LoopStep::Finished(result) = step.await.unwrap() else { + panic!("expected cancelled notification turn"); + }; + assert_eq!(result.finish_reason, FinishReason::Cancelled); + assert!(recorded.snapshot().iter().any(|item| { + item.parts + .iter() + .any(|part| matches!(part, Part::Text(text) if text.text == "original notification")) + })); + assert!(model.requests.lock().unwrap().is_empty()); + hooks.drain_turn_events("cancel-before-start").await; + let events = backend.all_events(); + assert_eq!( + events + .iter() + .map(|event| event["type"].as_str().unwrap()) + .collect::>(), + ["turn.start", "turn.end"] + ); + assert_eq!(events[0]["turn"], events[1]["turn"]); + assert_eq!(events[0]["items"], json!([])); + assert_eq!(events[1]["outcome"], "cancelled"); +} + +#[tokio::test] +async fn cancelled_turn_start_preserves_prompt_and_orders_cancelled_end_before_session_end() { + cancelled_prompt_hook("turn.start").await; +} + +#[tokio::test] +async fn cancelled_user_inbound_pairs_turn_start_without_committing_effects() { + cancelled_prompt_hook("user.message.inbound").await; +} + +async fn cancelled_prompt_hook(event_type: &'static str) { + let backend = Arc::new(HoldTurnStart { + event_type, + events: Arc::default(), + reached: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + }); + backend.events.respond(event_type, json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("must not commit")}])); + backend.events.respond("turn.start", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("must not commit")}])); + let hooks = super::test_support::hooks_with_backend( + vec![ + intercept(if event_type == "turn.start" { + &["turn.start"] + } else { + &["user.message.inbound", "turn.start"] + }), + observe(&["turn.end", "session.end"]), + ], + backend.clone(), + ); + let root = tempfile::tempdir().unwrap(); + let directory = root.path().join("sessions"); + let opened = crate::session::open_in( + root.path(), + &directory, + "cancel-start", + false, + false, + vec![Item::text(ItemKind::System, "system")], + ) + .unwrap(); + let controller = agentkit_core::CancellationController::new(); + let model = Model::default(); + let mut driver = Agent::builder() + .model(model.clone()) + .cancellation(controller.handle()) + .observer(TurnHook(hooks.clone())) + .prompt_hook(PromptHook::new( + hooks.clone(), + "cli", + PromptRecord::Session(opened.observer.clone()), + )) + .transcript_observer(opened.observer) + .transcript(opened.transcript) + .input(vec![Item::text(ItemKind::User, "original")]) + .build() + .unwrap() + .start(SessionConfig::new("cancel-start")) + .await + .unwrap(); + let cancel = async { + backend.reached.notified().await; + controller.interrupt(); + backend.release.notify_one(); + }; + let (step, ()) = tokio::time::timeout(std::time::Duration::from_secs(5), async { + tokio::join!(driver.next(), cancel) + }) + .await + .expect("turn.start did not observe native cancellation"); + // Native cancellation finishes the turn with interrupted assistant output; + // it does not masquerade as hook denial or silently remove admitted input. + let LoopStep::Finished(result) = step.unwrap() else { + panic!("expected a natively cancelled turn result"); + }; + assert_eq!(result.finish_reason, FinishReason::Cancelled); + assert!( + result + .items + .iter() + .any(|item| item.kind == ItemKind::Assistant + && item.metadata.get("agentkit.interrupted") == Some(&json!(true))) + ); + assert!(model.requests.lock().unwrap().is_empty()); + let persisted = crate::session::load_in(root.path(), &directory, "cancel-start").unwrap(); + assert_eq!(last_user_text(&persisted).as_deref(), Some("original")); + hooks + .session_end("cancel-start", SessionEndInputOutcome::Completed) + .await; + let events = backend.events.all_events(); + let types: Vec<_> = events + .iter() + .map(|event| event["type"].as_str().unwrap()) + .collect(); + let expected = if event_type == "turn.start" { + vec!["turn.start", "turn.end", "session.end"] + } else { + vec![ + "user.message.inbound", + "turn.start", + "turn.end", + "session.end", + ] + }; + assert_eq!(types, expected); + let starts = backend.events.events("turn.start"); + let ends = backend.events.events("turn.end"); + assert_eq!(ends[0]["outcome"], "cancelled"); + assert_eq!(starts[0]["turn"], ends[0]["turn"]); +} + /// Holds `turn.end` at the hook transport until the test releases it. struct HoldTurnEnd { events: Arc, @@ -1207,6 +2012,7 @@ async fn cancelled_drain_still_delivers_turn_end_before_session_end() { let mut driver = Agent::builder() .model(Model::default()) .observer(TurnHook(hooks.clone())) + .prompt_hook(transient_prompt_hook(&hooks, "cli")) .input(vec![Item::text(ItemKind::User, "hello")]) .build() .unwrap() @@ -1251,6 +2057,7 @@ async fn failed_model_request_ends_the_logical_turn() { SessionStartInputTrigger::Startup, )) .observer(TurnHook(hooks.clone())) + .prompt_hook(transient_prompt_hook(&hooks, "cli")) .input(vec![Item::text(ItemKind::User, "hello")]) .build() .unwrap() @@ -1337,6 +2144,101 @@ impl ModelTurn for ToolRoundReply { } } +#[derive(Clone)] +struct RecordedToolRoundModel { + model: ToolRoundModel, + requests: Arc>>>, +} + +#[async_trait] +impl ModelAdapter for RecordedToolRoundModel { + type Session = Self; + async fn start_session(&self, _: SessionConfig) -> Result { + Ok(self.clone()) + } +} + +#[async_trait] +impl ModelSession for RecordedToolRoundModel { + type Turn = ToolRoundReply; + async fn begin_turn( + &mut self, + request: TurnRequest, + cancellation: Option, + ) -> Result { + self.requests + .lock() + .unwrap() + .push(request.transcript.clone()); + self.model.begin_turn(request, cancellation).await + } +} + +#[tokio::test] +async fn midturn_user_interjection_runs_inbound_without_another_turn_start() { + let (hooks, backend) = hooks(vec![ + intercept(&["user.message.inbound", "turn.start"]), + observe(&["turn.end"]), + ]); + let requests = Arc::new(Mutex::new(vec![])); + let mut registry = ToolRegistry::new(); + registry.register_arc(Echo::new()); + let mut driver = Agent::builder() + .model(RecordedToolRoundModel { + model: ToolRoundModel(true), + requests: requests.clone(), + }) + .observer(TurnHook(hooks.clone())) + .prompt_hook(transient_prompt_hook(&hooks, "cli")) + .add_tool_source(registry) + .input(vec![Item::text(ItemKind::User, "initial")]) + .build() + .unwrap() + .start(SessionConfig::new("interjection")) + .await + .unwrap(); + let LoopStep::Interrupt(agentkit_loop::LoopInterrupt::AfterToolResult(round)) = + driver.next().await.unwrap() + else { + panic!("expected tool-round interjection point"); + }; + backend.respond("user.message.inbound", json!([{"type":"modify", "target":"prompt", "operation":"replace", "value":text_message("edited interjection")}])); + round + .submit(&mut driver, vec![Item::text(ItemKind::User, "midturn")]) + .unwrap(); + assert!(matches!( + driver.next().await.unwrap(), + LoopStep::Finished(_) + )); + hooks.drain_turn_events("interjection").await; + let inbound = backend.events("user.message.inbound"); + let starts = backend.events("turn.start"); + let ends = backend.events("turn.end"); + assert_eq!(inbound.len(), 2); + assert_eq!(starts.len(), 1); + assert_eq!(ends.len(), 1); + assert_eq!( + inbound[0]["message"]["messages"][0]["parts"][0]["text"], + "initial" + ); + assert_eq!( + inbound[1]["message"]["messages"][0]["parts"][0]["text"], + "midturn" + ); + assert_eq!(inbound[0]["turn"], inbound[1]["turn"]); + assert_eq!(starts[0]["turn"], ends[0]["turn"]); + let requests = requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert_eq!( + last_user_text(&requests[1]).as_deref(), + Some("edited interjection") + ); + assert_eq!( + last_user_text(&driver.snapshot().transcript).as_deref(), + Some("edited interjection") + ); +} + #[tokio::test] async fn retry_and_tool_continuation_are_one_turn_and_retirement_cancels_once() { for retire in [false, true] { @@ -1354,6 +2256,7 @@ async fn retry_and_tool_continuation_are_one_turn_and_retirement_cancels_once() SessionStartInputTrigger::Startup, )) .observer(TurnHook(hooks.clone())) + .prompt_hook(transient_prompt_hook(&hooks, "cli")) .add_tool_source(registry) .input(vec![Item::text(ItemKind::User, "hello")]) .build() @@ -1683,6 +2586,7 @@ async fn turn_end_reports_usage_and_last_assistant_without_invented_continuation SessionStartInputTrigger::Startup, )) .observer(TurnHook(hooks.clone())) + .prompt_hook(transient_prompt_hook(&hooks, "cli")) .input(vec![Item::text(ItemKind::User, "hello")]) .build() .unwrap() @@ -2017,7 +2921,7 @@ async fn child_identity_is_session_local_through_the_hook_lifecycle() { .with_child(child), ) .observer(TurnHook(hooks.clone())) - .mutator(PromptHook::new( + .prompt_hook(PromptHook::new( hooks.clone(), "a2a", PromptRecord::Transient(crate::session::TransientTranscript::new(vec![])), @@ -2058,6 +2962,9 @@ async fn child_identity_is_session_local_through_the_hook_lifecycle() { .collect(); assert!(!events.is_empty(), "missing {event_type} for {id}"); for event in events { + if event_type == "turn.start" { + assert_eq!(event["trigger"], "external", "{id}"); + } if child { assert_eq!(event["session"]["agent"], json!({"id": id}), "{event_type}"); } else { @@ -2208,3 +3115,214 @@ async fn invalid_fork_session_start_preserves_dispatch_cause() { "{error}" ); } + +#[tokio::test] +async fn denied_batch_never_retries_failed_history_and_allows_the_next_prompt() { + for boundary in ["user.message.inbound", "turn.start"] { + for tail in [ItemKind::User, ItemKind::Notification, ItemKind::Tool] { + let (hooks, backend) = hooks(vec![ + intercept(&["user.message.inbound", "turn.start"]), + observe(&["turn.end"]), + ]); + backend.respond(boundary, json!([{"type":"deny", "reason":"blocked"}])); + let mut history = vec![Item::text(ItemKind::User, "previous failed prompt")]; + match tail { + ItemKind::Notification => history.push(Item::notification("previous notification")), + ItemKind::Tool => { + let call = + agentkit_core::ToolCallPart::new("old-call", "echo", json!({"text":"old"})); + history.push(Item::new(ItemKind::Assistant, vec![Part::ToolCall(call)])); + history.push(Item::new( + ItemKind::Tool, + vec![Part::ToolResult(ToolResultPart { + call_id: ToolCallId::new("old-call"), + output: ToolOutput::Text("previous result".into()), + is_error: false, + metadata: MetadataMap::new(), + })], + )); + } + _ => {} + } + let recorded = crate::session::TransientTranscript::new(history.clone()); + let model = Model::default(); + let mut driver = Agent::builder() + .model(model.clone()) + .observer(TurnHook(hooks.clone())) + .prompt_hook(PromptHook::new( + hooks.clone(), + "acp", + PromptRecord::Transient(recorded.clone()), + )) + .transcript_observer(recorded.clone()) + .transcript(history.clone()) + .input(vec![ + Item::notification("new notification"), + Item::text(ItemKind::User, "blocked new prompt"), + ]) + .build() + .unwrap() + .start(SessionConfig::new("denied-after-error")) + .await + .unwrap(); + assert!(matches!(driver.next().await, Err(LoopError::Mutator(_)))); + assert!(model.requests.lock().unwrap().is_empty()); + assert_eq!(driver.snapshot().transcript, history); + assert_eq!(recorded.snapshot(), history); + hooks.drain_turn_events("denied-after-error").await; + let starts = backend.events("turn.start"); + let ends = backend.events("turn.end"); + assert_eq!(starts.len(), 1); + assert_eq!(ends.len(), 1); + assert_eq!(starts[0]["turn"], ends[0]["turn"]); + assert_eq!(ends[0]["outcome"], "failed"); + assert_eq!( + starts[0]["items"].as_array().unwrap().len(), + if boundary == "turn.start" { 2 } else { 0 } + ); + + backend.respond(boundary, json!([])); + driver + .submit_input(vec![Item::text(ItemKind::User, "allowed next prompt")]) + .unwrap(); + assert!(matches!( + driver.next().await.unwrap(), + LoopStep::Finished(_) + )); + let requests = model.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(&requests[0][..history.len()], history.as_slice()); + assert_eq!( + last_user_text(&requests[0]).as_deref(), + Some("allowed next prompt") + ); + } + } +} + +struct DelayedBackgroundApproval { + spec: ToolSpec, + release: Arc, +} + +#[async_trait] +impl Tool for DelayedBackgroundApproval { + fn spec(&self) -> &ToolSpec { + &self.spec + } + + async fn invoke( + &self, + _: ToolRequest, + _: &mut ToolContext<'_>, + ) -> Result { + Err(ToolError::Internal( + "approval requires invoke_outcome".into(), + )) + } + + async fn invoke_outcome( + &self, + request: ToolRequest, + _: &mut ToolContext<'_>, + ) -> ToolExecutionOutcome { + self.release.acquire().await.unwrap().forget(); + ToolExecutionOutcome::Interrupted(agentkit_tools_core::ToolInterruption::ApprovalRequired( + agentkit_tools_core::ApprovalRequest { + task_id: None, + call_id: None, + id: "delayed-approval".into(), + request_kind: "test.approval".into(), + reason: agentkit_tools_core::ApprovalReason::PolicyRequiresConfirmation, + summary: "background approval".into(), + metadata: request.metadata, + }, + )) + } +} + +#[tokio::test] +async fn idle_background_approval_pairs_its_own_turn_on_resume_or_cancellation() { + use agentkit_task_manager::{AsyncTaskManager, RoutingDecision, TaskManager}; + for cancel in [false, true] { + let (hooks, backend) = hooks(vec![observe(&["turn.start", "turn.end"])]); + let release = Arc::new(tokio::sync::Semaphore::new(0)); + let tasks = AsyncTaskManager::new().routing(|_: &ToolRequest| RoutingDecision::Background); + let handle = tasks.handle(); + let mut driver = Agent::builder() + .model(ToolRoundModel(true)) + .observer(TurnHook(hooks.clone())) + .prompt_hook(transient_prompt_hook(&hooks, "cli")) + .add_tool_source(ToolRegistry::new().with(DelayedBackgroundApproval { + spec: Echo::new().spec.clone(), + release: release.clone(), + })) + .task_manager(tasks) + .input(vec![Item::text(ItemKind::User, "start background work")]) + .build() + .unwrap() + .start(SessionConfig::new("background-approval")) + .await + .unwrap(); + assert!(matches!( + driver.next().await.unwrap(), + LoopStep::Interrupt(agentkit_loop::LoopInterrupt::AwaitingInput(_)) + )); + hooks.drain_turn_events("background-approval").await; + assert_eq!(backend.events("turn.start").len(), 1); + assert_eq!(backend.events("turn.end").len(), 1); + + release.add_permits(1); + tokio::time::timeout(std::time::Duration::from_secs(5), handle.wait_for_idle()) + .await + .unwrap(); + assert!(matches!( + driver.next().await.unwrap(), + LoopStep::Interrupt(agentkit_loop::LoopInterrupt::ApprovalRequest(_)) + )); + if cancel { + assert!(matches!( + driver.cancel_pending_approvals().await.unwrap(), + Some(LoopStep::Finished(_)) + )); + } else { + driver + .resolve_approval(agentkit_tools_core::ApprovalDecision::Deny { + reason: Some("not allowed".into()), + }) + .unwrap(); + while matches!( + driver.next().await.unwrap(), + LoopStep::Interrupt(agentkit_loop::LoopInterrupt::AfterToolResult(_)) + ) {} + } + hooks.drain_turn_events("background-approval").await; + let starts = backend.events("turn.start"); + let ends = backend.events("turn.end"); + assert_eq!(starts.len(), 2); + assert_eq!(ends.len(), 2); + assert_ne!(starts[0]["turn"], starts[1]["turn"]); + assert_eq!(starts[1]["turn"], ends[1]["turn"]); + assert_eq!(starts[1]["trigger"], "external"); + if cancel { + assert_eq!(starts[1]["items"], json!([])); + } else { + // Resolving the approval contributes a new background-result + // notification, not the originating turn's user prompt. + let items = starts[1]["items"].as_array().unwrap(); + assert_eq!(items.len(), 1); + assert!( + items[0]["parts"][0]["text"] + .as_str() + .unwrap() + .contains("not allowed") + ); + } + let types: Vec<_> = backend + .all_events() + .iter() + .map(|event| event["type"].as_str().unwrap().to_owned()) + .collect(); + assert_eq!(types, ["turn.start", "turn.end", "turn.start", "turn.end"]); + } +} diff --git a/src/ahp/turn.rs b/src/ahp/turn.rs index 9c3b86af..8bd8894d 100644 --- a/src/ahp/turn.rs +++ b/src/ahp/turn.rs @@ -1,11 +1,8 @@ -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; -use agenthooksprotocol::inputs::{TurnEndInput, TurnStartInput}; -use agenthooksprotocol::{ - EventType, Integer, Session, TurnEndInputOutcome, TurnEndInputTurn, TurnStartInputTrigger, - TurnStartInputTurn, -}; +use agenthooksprotocol::inputs::TurnEndInput; +use agenthooksprotocol::{EventType, Integer, Session, TurnEndInputOutcome, TurnEndInputTurn}; use agentkit_core::FinishReason; use agentkit_loop::{AgentEvent, LoopObserver, ObservedEvent}; @@ -13,26 +10,42 @@ use super::Hooks; pub(super) type TurnQueues = Arc>>; -/// Queues native turn boundaries until the session's next async hook point. +/// Queues native turn completions until the session's next async hook point. +/// Starts run in the prompt mutator, where their effects can govern admission. #[derive(Clone)] pub(crate) struct TurnHook(pub(crate) Arc); #[derive(Default)] pub(super) struct SessionTurns { pending: VecDeque, + started: HashSet, delivery: Arc>, } impl Hooks { - /// Delivers the turn observations queued for `session`, in order. - pub(crate) async fn drain_turn_events(&self, session: &str) { - let delivery = Arc::clone( + /// Claims ownership of one start operation for a native turn. The serialized + /// driver is the only admission writer; completion draining removes the claim. + /// The operation must survive cancellation and hold the delivery gate. + pub(super) fn claim_turn_start(&self, session: &str, turn: &str) -> bool { + lock(&self.turn_events) + .entry(session.to_owned()) + .or_default() + .started + .insert(turn.to_owned()) + } + + pub(super) fn turn_delivery(&self, session: &str) -> Arc> { + Arc::clone( &lock(&self.turn_events) .entry(session.to_owned()) .or_default() .delivery, - ); - let guard = delivery.lock_owned().await; + ) + } + + /// Delivers the turn observations queued for `session`, in order. + pub(crate) async fn drain_turn_events(&self, session: &str) { + let guard = self.turn_delivery(session).lock_owned().await; if lock(&self.turn_events) .get(session) .is_none_or(|turns| turns.pending.is_empty()) @@ -46,7 +59,21 @@ impl Hooks { // A cancelled caller must not abandon an observation the SDK has begun delivering. let delivered = tokio::spawn(async move { let _guard = guard; - while let Some(event) = next(&queues, &session) { + while let Some((event, needs_start)) = next(&queues, &session) { + // Idle background approvals can be cancelled before any mutator + // runs. Pair their completion without admitting historical input. + if needs_start && let AgentEvent::TurnFinished(result) = &event { + let input = agenthooksprotocol::inputs::TurnStartInput::new( + Vec::new(), + agenthooksprotocol::TurnStartInputTrigger::External, + agenthooksprotocol::TurnStartInputTurn::new(result.turn_id.to_string()), + ) + .with_session(owner.clone()); + super::settle( + EventType::TurnStart, + super::dispatch(sdk.turn_start(input.with_sources()), None).await, + ); + } deliver(&sdk, owner.clone(), event).await; } }); @@ -61,92 +88,84 @@ impl Hooks { } fn lock(queues: &TurnQueues) -> MutexGuard<'_, HashMap> { + // Writers insert complete queues, enqueue native completion values, or pop/remove + // them. No callbacks or awaits run under this guard; poisoning cannot leave a + // partially delivered value in the queue. Start claims are complete string + // insertions/removals with no external work under the lock; delivery is + // serialized separately and cancellation cannot abandon a claimed delivery. queues.lock().unwrap_or_else(PoisonError::into_inner) } -fn next(queues: &TurnQueues, session: &str) -> Option { - lock(queues).get_mut(session)?.pending.pop_front() +fn next(queues: &TurnQueues, session: &str) -> Option<(AgentEvent, bool)> { + let mut queues = lock(queues); + let turns = queues.get_mut(session)?; + let event = turns.pending.pop_front()?; + let needs_start = matches!(&event, AgentEvent::TurnFinished(result) + if !turns.started.remove(&result.turn_id.to_string())); + Some((event, needs_start)) } async fn deliver(sdk: &agenthooksprotocol::Hooks, owner: Box, event: AgentEvent) { - match event { - AgentEvent::TurnStarted { turn_id, .. } => { - let input = TurnStartInput::new( - Vec::new(), - TurnStartInputTrigger::External, - TurnStartInputTurn::new(turn_id.to_string()), - ) - .with_session(owner); - super::settle( - EventType::TurnStart, - super::dispatch(sdk.turn_start(input), None).await, - ); - } - AgentEvent::TurnFinished(result) => { - let outcome = match result.finish_reason { - FinishReason::Cancelled => TurnEndInputOutcome::Cancelled, - FinishReason::Error | FinishReason::Blocked => TurnEndInputOutcome::Failed, - _ => TurnEndInputOutcome::Completed, - }; - // Kit grants no AHP continue effect; native tool iterations are not - // hook-requested turn continuations. - let Some(continuations) = Integer::from_i64(0) else { - return; - }; - let mut input = TurnEndInput::new( - continuations, - Vec::new(), - outcome, - TurnEndInputTurn::new(result.turn_id.to_string()), - ) - .with_session(owner); - input.last_assistant_item = result - .items - .iter() - .enumerate() - .rev() - .find(|(_, item)| item.kind == agentkit_core::ItemKind::Assistant) - .map(|(index, item)| { - let mut reference = agenthooksprotocol::TurnEndInputLastAssistantItem::new( - super::item_id(item, index), - ); - if item.id.is_none() { - reference.synthesized = agenthooksprotocol::Presence::Present(true); - } - reference - }); - if let Some(reasoning) = result - .usage - .as_ref() - .and_then(|usage| usage.tokens.as_ref()) - .and_then(|tokens| tokens.reasoning_tokens) - { - let mut extensions = agenthooksprotocol::Extensions::new(); - extensions.additional_properties.insert( - "dev.kit.usage".into(), - super::object([("reasoningTokens", reasoning.into())]), + if let AgentEvent::TurnFinished(result) = event { + let outcome = match result.finish_reason { + FinishReason::Cancelled => TurnEndInputOutcome::Cancelled, + FinishReason::Error | FinishReason::Blocked => TurnEndInputOutcome::Failed, + _ => TurnEndInputOutcome::Completed, + }; + // Kit grants no AHP continue effect; native tool iterations are not + // hook-requested turn continuations. + let Some(continuations) = Integer::from_i64(0) else { + return; + }; + let mut input = TurnEndInput::new( + continuations, + Vec::new(), + outcome, + TurnEndInputTurn::new(result.turn_id.to_string()), + ) + .with_session(owner); + input.last_assistant_item = result + .items + .iter() + .enumerate() + .rev() + .find(|(_, item)| item.kind == agentkit_core::ItemKind::Assistant) + .map(|(index, item)| { + let mut reference = agenthooksprotocol::TurnEndInputLastAssistantItem::new( + super::item_id(item, index), ); - input.extensions = Some(Box::new(extensions)); - } - input.usage = result.usage.as_ref().map(super::usage::turn).map(Box::new); - let input = input - .with_sources() - .with_items(super::messages(&result.items)); - super::settle( - EventType::TurnEnd, - super::dispatch(sdk.turn_end(input), None).await, + if item.id.is_none() { + reference.synthesized = agenthooksprotocol::Presence::Present(true); + } + reference + }); + if let Some(reasoning) = result + .usage + .as_ref() + .and_then(|usage| usage.tokens.as_ref()) + .and_then(|tokens| tokens.reasoning_tokens) + { + let mut extensions = agenthooksprotocol::Extensions::new(); + extensions.additional_properties.insert( + "dev.kit.usage".into(), + super::object([("reasoningTokens", reasoning.into())]), ); + input.extensions = Some(Box::new(extensions)); } - _ => {} + input.usage = result.usage.as_ref().map(super::usage::turn).map(Box::new); + let input = input + .with_sources() + .with_items(super::messages(&result.items)); + super::settle( + EventType::TurnEnd, + super::dispatch(sdk.turn_end(input), None).await, + ); } } impl LoopObserver for TurnHook { fn handle_event(&self, observed: ObservedEvent) { - if matches!( - observed.event, - AgentEvent::TurnStarted { .. } | AgentEvent::TurnFinished(_) - ) { + if matches!(observed.event, AgentEvent::TurnFinished(_)) { lock(&self.0.turn_events) .entry(observed.session_id.to_string()) .or_default() diff --git a/src/protocols/acp.rs b/src/protocols/acp.rs index 60e04c01..f6fc7781 100644 --- a/src/protocols/acp.rs +++ b/src/protocols/acp.rs @@ -2460,6 +2460,7 @@ async fn session_actor(actor: SessionActor) { .and_then(|meta| meta.get(model_switch::META)), )?; if decision == model_switch::Decision::Compact { + let before = driver.snapshot().transcript; let marker = model_switch::compact_marker(); let marker_id = marker.id.clone(); driver.submit_input(vec![marker]).map_err(|error| { @@ -2479,10 +2480,15 @@ async fn session_actor(actor: SessionActor) { |reason| Some(reason.clone()), ) .await - .map_err(sdk_error)?; + .map_err(|_| { + model_switch::error( + "compaction did not complete; model unchanged", + ) + })?; if !model_switch::compaction_completed( &reason, &marker_id, + &before, &driver.snapshot().transcript, cancellation.is_cancelled_since(cancellation_generation), ) { diff --git a/src/protocols/acp/model_switch.rs b/src/protocols/acp/model_switch.rs index a882a082..c000fd77 100644 --- a/src/protocols/acp/model_switch.rs +++ b/src/protocols/acp/model_switch.rs @@ -140,6 +140,7 @@ pub(super) fn compact_marker() -> Item { pub(super) fn compaction_completed( reason: &FinishReason, marker: &Option, + before: &[Item], transcript: &[Item], cancelled: bool, ) -> bool { @@ -147,6 +148,9 @@ pub(super) fn compaction_completed( && *reason == FinishReason::Completed && marker.is_some() && !transcript.iter().any(|item| &item.id == marker) + && transcript + .iter() + .any(|item| crate::compaction::is_compaction_summary(item) && !before.contains(item)) } #[cfg(test)] @@ -329,20 +333,40 @@ mod tests { } #[test] - fn model_switch_compaction_requires_exact_consumed_marker_and_success() { + fn model_switch_compaction_requires_new_persisted_summary_and_consumed_marker() { let marker = compact_marker(); - let other = compact_marker(); - assert_ne!(marker.id, other.id); - assert!(!compaction_completed( - &FinishReason::Completed, - &marker.id, - std::slice::from_ref(&marker), + let mut summary = Item::text(ItemKind::Developer, "compacted history"); + summary.metadata.insert( + crate::compaction::COMPACTION_SUMMARY_METADATA_KEY.into(), + true.into(), + ); + let check = |reason: FinishReason, before: &[Item], after: &[Item], cancelled| { + compaction_completed(&reason, &marker.id, before, after, cancelled) + }; + assert!(check( + FinishReason::Completed, + &[], + std::slice::from_ref(&summary), false )); - assert!(compaction_completed( - &FinishReason::Completed, - &marker.id, - &[other], + // Denial or replacement can remove the marker without compacting. An + // existing checkpoint must not authorize the switch either. + for after in [ + vec![], + vec![Item::text(ItemKind::User, "replacement")], + vec![summary.clone()], + ] { + assert!(!check( + FinishReason::Completed, + std::slice::from_ref(&summary), + &after, + false + )); + } + assert!(!check( + FinishReason::Completed, + &[], + &[marker.clone(), summary.clone()], false )); for reason in [ @@ -351,18 +375,19 @@ mod tests { FinishReason::MaxTokens, FinishReason::Blocked, ] { - assert!(!compaction_completed(&reason, &marker.id, &[], false)); + assert!(!check(reason, &[], std::slice::from_ref(&summary), false)); } - assert!(!compaction_completed( - &FinishReason::Completed, - &marker.id, + assert!(!check( + FinishReason::Completed, &[], + std::slice::from_ref(&summary), true )); assert!(!compaction_completed( &FinishReason::Completed, &None, &[], + &[summary], false )); } diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index f7560efc..8015f206 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -2440,8 +2440,9 @@ async fn compact_for_switch( cancellation_generation: u64, activity: &SessionActivity, ) -> Result<(), agent_client_protocol::Error> { - // Only a successful, persisted standard compaction consumes this marker. + // Require a new persisted summary as well as consumption of this marker. // Runtime telemetry cannot acknowledge a pending model switch. + let before = driver.snapshot().transcript; let marker = model_switch::compact_marker(); let marker_id = marker.id.clone(); driver @@ -2463,10 +2464,11 @@ async fn compact_for_switch( ExecutionOrigin::Prompt, ) .await - .map_err(sdk_error)?; + .map_err(|_| model_switch::error("compaction did not complete; model unchanged"))?; if !model_switch::compaction_completed( &reason, &marker_id, + &before, &driver.snapshot().transcript, handle .cancellation_handle() @@ -3505,6 +3507,8 @@ pub(crate) fn component( clippy::disallowed_macros )] mod tests { + use crate::ahp::PromptHookBuilder as _; + include!("v2/identity_tests.rs"); #[test] fn catalog_matches_providers_detects_logins_and_logouts() { @@ -8875,7 +8879,7 @@ for line in sys.stdin: #[tokio::test] async fn model_switch_compacts_with_original_model_before_selecting_and_keeps_it_on_failure_or_cancel() { - for mode in ["success", "failure", "cancel"] { + for mode in ["success", "failure", "cancel", "deny", "replace"] { let integration = AcpIntegration::default(); let sink = ResponseReplacementSink::new(RecordingSink::default()); let session_id = wire::SessionId::new(format!("switch-{mode}")); @@ -8909,12 +8913,32 @@ for line in sys.stdin: None, ) .unwrap(); + let (hooks, backend) = crate::ahp::test_support::hooks(vec![ + crate::ahp::test_support::intercept(&["turn.start"]), + ]); + if mode == "deny" { + backend.respond("turn.start", json!([{"type": "deny", "reason": "blocked"}])); + } else if mode == "replace" { + backend.respond( + "turn.start", + json!([{ + "type": "modify", "target": "prompt", "operation": "replace", + "value": crate::ahp::test_support::text_message("ordinary text"), + }]), + ); + } + let recorded = crate::session::TransientTranscript::new(vec![]); let mut driver = Agent::builder() .model(TestAdapter { outcome: TestOutcome::Content, turns: Arc::new(AtomicU64::new(0)), interrupt: None, }) + .prompt_hook(crate::ahp::PromptHook::new( + hooks, + "acp", + crate::ahp::PromptRecord::Transient(recorded), + )) .mutator(compactor) .cancellation(handle.cancellation_handle()) .transcript(vec![ @@ -8940,7 +8964,17 @@ for line in sys.stdin: ) .await; assert_eq!(selection.selection().unwrap().model, "gpt-5.4"); - assert_eq!(*seen.lock().unwrap(), ["gpt-5.4"]); + if matches!(mode, "deny" | "replace") { + assert!(seen.lock().unwrap().is_empty(), "{mode}"); + assert!( + compacted.as_ref().unwrap_err().data.as_ref() + == Some(&json!("compaction did not complete; model unchanged")), + "{mode}: {compacted:?}" + ); + assert_eq!(backend.events("turn.start").len(), 1); + } else { + assert_eq!(*seen.lock().unwrap(), ["gpt-5.4"]); + } let catalog = [crate::provider::ModelGroup { provider: ProviderKind::OpenAiSubscription, models: vec!["gpt-5.4-mini".into()], diff --git a/src/runtime.rs b/src/runtime.rs index b8dfaf38..38c1ef6f 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -37,7 +37,7 @@ use tracing::Instrument as _; use crate::{ acp_child::{AcpHarnesses, BUILTIN_HARNESS, ChildConfig}, - ahp::{HookedAdapter, HookedSession}, + ahp::{HookedAdapter, HookedSession, PromptHookBuilder as _}, provider::{ ModelSelection, ProviderKind, ReasoningEffort, SelectableAdapter, SelectableSession, }, @@ -1982,17 +1982,17 @@ impl Runtime { background_jobs.clone(), skills, )) - .task_manager(task_manager) - .mutator(compactor); + .task_manager(task_manager); if let Some(hooks) = &self.hooks { builder = builder.observer(crate::ahp::TurnHook(Arc::clone(hooks))); - builder = builder.mutator(crate::ahp::PromptHook::new( + builder = builder.prompt_hook(crate::ahp::PromptHook::new( Arc::clone(hooks), "cli", crate::ahp::PromptRecord::Session(opened.observer.clone()), )); } let agent = builder + .mutator(compactor) .transcript_observer(opened.observer) .transcript(opened.transcript) .input(vec![Item::text(ItemKind::User, prompt)]) @@ -2166,12 +2166,11 @@ impl Runtime { BackgroundJobs::default(), skills, )) - .task_manager(background_task_manager()) - .mutator(compactor); + .task_manager(background_task_manager()); let builder = match &self.hooks { Some(hooks) => builder .observer(crate::ahp::TurnHook(Arc::clone(hooks))) - .mutator(crate::ahp::PromptHook::new( + .prompt_hook(crate::ahp::PromptHook::new( Arc::clone(hooks), "a2a", crate::ahp::PromptRecord::Transient(recorded.clone()), @@ -2179,6 +2178,7 @@ impl Runtime { None => builder, }; let builder = builder + .mutator(compactor) .transcript_observer(recorded) .transcript(transcript) .input(vec![Item::text(ItemKind::User, prompt)]) @@ -2412,11 +2412,10 @@ impl Runtime { skills, mcp, )) - .task_manager(task_manager) - .mutator(compactor); + .task_manager(task_manager); if let Some(hooks) = &self.hooks { builder = builder.observer(crate::ahp::TurnHook(Arc::clone(hooks))); - builder = builder.mutator(crate::ahp::PromptHook::new( + builder = builder.prompt_hook(crate::ahp::PromptHook::new( Arc::clone(hooks), "acp", crate::ahp::PromptRecord::Session(opened.observer.clone()), @@ -2425,6 +2424,7 @@ impl Runtime { // User interaction remains foreground-only, outside backgroundable compose. builder = builder.add_tool_source(self.user_question_tools(context.ask_user)); let driver = builder + .mutator(compactor) .observer(context.integration.as_ref().clone()) .observer(crate::fatal::RetryLog) .transcript_observer(opened.observer)