From 1ebc226e2c31713c2f499a5a5ace15fd97126258 Mon Sep 17 00:00:00 2001 From: daniel Date: Sun, 25 Oct 2026 00:01:55 +0100 Subject: [PATCH] fix(ahp): support server-driven hook configuration --- Cargo.lock | 4 +- Cargo.toml | 2 +- docs/user/agent-hooks-protocol.md | 18 ++++++ src/ahp.rs | 21 ++++-- src/ahp/model.rs | 8 ++- src/ahp/tests.rs | 104 +++++++++++++++++++++++++++++- src/runtime.rs | 16 +++-- 7 files changed, 153 insertions(+), 20 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 21dc3032..40134a12 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -91,9 +91,9 @@ dependencies = [ [[package]] name = "agenthooksprotocol" -version = "0.9.0" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f19905ff8831f7a0b24d336af0802881a52f778d02769375aa2fe120718328ce" +checksum = "d538f4e71bc13e4bd1c48b767188a49de5401676a94b9d7561116d4f83cd95fd" dependencies = [ "http", "jsonschema", diff --git a/Cargo.toml b/Cargo.toml index cabf37c4..fe8c1451 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,7 +36,7 @@ tokio-tungstenite = { version = "=0.30.0", default-features = false, features = a2a-protocol-client = { version = "=0.14.1", default-features = false, features = ["tls-rustls"] } a2a-protocol-server = { version = "=0.14.1", default-features = false } a2a-protocol-types = "=0.14.1" -agenthooksprotocol = { version = "=0.9.0", default-features = false, features = ["reqwest", "tokio-process"] } +agenthooksprotocol = { version = "=0.10.1", default-features = false, features = ["reqwest", "tokio-process"] } agent-client-protocol = { package = "agentkit-acp-sdk", version = "=2.2.0", features = ["unstable_session_fork", "unstable_session_inject"] } agent-client-protocol-schema = { package = "agentkit-acp-schema", version = "=1.9.1", features = ["unstable_session_notices", "unstable_session_compaction"] } agent-client-protocol-http = { package = "agentkit-acp-http", version = "=2.2.0", default-features = false, features = ["server", "client"] } diff --git a/docs/user/agent-hooks-protocol.md b/docs/user/agent-hooks-protocol.md index 7b5493bf..5d8f9857 100644 --- a/docs/user/agent-hooks-protocol.md +++ b/docs/user/agent-hooks-protocol.md @@ -49,6 +49,24 @@ hooks that you trust with those values. Hook backends run local processes or receive session content. Review both files before you run Kit in a project you do not trust. +## Use server-provided subscriptions + +Omit a backend's `subscriptions` field to let it supply subscriptions through +`hooks/configuration`. Kit resolves discovery before deciding whether to buffer +model output or disable speculative compaction. Explicit `subscriptions`, +including an empty array, bypass discovery. + +Kit trusts configured hook backends to supply upload settings, including upload +destinations, credential bindings, and forwarding of configured backend headers. +Only configure backends you trust with the selected content and credentials. +Server-provided subscriptions cannot expand the event or effect authority Kit +grants in the table below. + +If a backend's discovery request fails or its response is invalid, the SDK skips +that backend for this hook instance. Check the backend's availability and +`hooks/configuration` response, or configure explicit subscriptions when you +need client-defined routing and failure policies. + ## Events | Event | Where Kit emits it | Hook authority | diff --git a/src/ahp.rs b/src/ahp.rs index f8770005..08e8e21e 100644 --- a/src/ahp.rs +++ b/src/ahp.rs @@ -40,7 +40,6 @@ const CONFIG_FILE: &str = "hooks.json"; /// Registered hook backends with the event grants Kit implements. pub struct Hooks { sdk: Arc, - intercepts_model_response: bool, injections: injection::Injections, messages: messages::Routes, turn_events: turn::TurnQueues, @@ -49,9 +48,17 @@ pub struct Hooks { impl Hooks { /// Compaction hooks have observable effects and cannot run speculatively. - pub(crate) fn intercepts_compaction(&self) -> bool { - self.sdk.subscribes(EventType::ContextCompactBefore) - || self.sdk.subscribes(EventType::ContextCompactAfter) + pub(crate) async fn intercepts_compaction(&self) -> Result { + self.sdk + .initialize() + .await + .map_err(|error| error.to_string())?; + Ok(self.sdk.subscribes(EventType::ContextCompactBefore) + || self.sdk.subscribes(EventType::ContextCompactAfter)) + } + + fn intercepts_model_response(&self) -> bool { + self.sdk.intercepts(EventType::ModelResponseAfter) } /// Loads `~/.agents/hooks.json`, then `/.agents/hooks.json`. Project @@ -82,10 +89,8 @@ impl Hooks { fn new(registration: Registration, options: HooksOptions) -> Result { let sdk = agenthooksprotocol::Hooks::new(registration, options) .map_err(|error| error.to_string())?; - let intercepts_model_response = sdk.intercepts(EventType::ModelResponseAfter); Ok(Self { sdk: Arc::new(sdk), - intercepts_model_response, injections: injection::Injections::default(), messages: messages::Routes::default(), turn_events: turn::TurnQueues::default(), @@ -341,6 +346,10 @@ fn options() -> Result { capabilities, ); options.backend.inherit_environment = true; + // A user-configured hook is trusted to receive event bodies and choose its + // upload destinations, credential bindings, and header forwarding policy. + // Discovery does not expand Kit's host event/effect grants. + options.authorize_configuration = Some(Arc::new(|_, _| Ok(true))); Ok(options) } diff --git a/src/ahp/model.rs b/src/ahp/model.rs index a6282271..f408a927 100644 --- a/src/ahp/model.rs +++ b/src/ahp/model.rs @@ -186,6 +186,10 @@ impl ModelSession for HookedSession { let inner = self.inner.begin_turn(request, cancellation).await?; return Ok(HookedTurn::live(inner, None)); }; + // Resolve server subscriptions before deciding whether previews are safe. + hooks.sdk.initialize().await.map_err(|error| { + LoopError::InvalidState(format!("hook configuration discovery failed: {error}")) + })?; hooks .drain_turn_events(&request.session_id.to_string()) .await; @@ -411,7 +415,7 @@ impl ModelTurn for HookedTurn { if self .call .as_ref() - .is_some_and(|call| call.hooks.intercepts_model_response) + .is_some_and(|call| call.hooks.intercepts_model_response()) && matches!( event, Ok(Some(ModelTurnEvent::Delta(_) | ModelTurnEvent::ToolCall(_))) @@ -524,7 +528,7 @@ impl ModelTurn for HookedTurn { Err(error) => call.failed(error, self.latest_usage.as_ref()).await, _ => {} } - if call.hooks.intercepts_model_response + if call.hooks.intercepts_model_response() && let Ok(Some(ModelTurnEvent::Finished(result))) = event { // agentkit-loop gates tool execution on a ToolCall event, not only diff --git a/src/ahp/tests.rs b/src/ahp/tests.rs index 089ecfc9..9612dabb 100644 --- a/src/ahp/tests.rs +++ b/src/ahp/tests.rs @@ -656,14 +656,14 @@ async fn tool_after_empty_replacement_removes_output() { #[test] fn only_response_interceptors_require_buffering() { let (response_hooks, _) = hooks(vec![intercept(&["model.response.after"])]); - assert!(response_hooks.intercepts_model_response); + assert!(response_hooks.intercepts_model_response()); for subscriptions in [ vec![observe(&["model.response.after"])], vec![intercept(&["tool.before"])], vec![intercept(&["model.request.before"])], ] { let (hooks, _) = hooks(subscriptions); - assert!(!hooks.intercepts_model_response); + assert!(!hooks.intercepts_model_response()); } } @@ -1279,6 +1279,10 @@ impl ModelSession for ToolRoundModel { let mut assistant = Item::text(ItemKind::Assistant, ""); assistant.parts = vec![Part::ToolCall(call.clone())]; if self.0 { + events.push_back(ModelTurnEvent::Delta(agentkit_core::Delta::AppendText { + part_id: agentkit_core::PartId::new("preview"), + chunk: "unapproved preview".into(), + })); // A superseded provider response is retried within this model turn. events.push_back(ModelTurnEvent::ToolCall(agentkit_core::ToolCallPart::new( "discarded-call", @@ -2050,3 +2054,99 @@ async fn child_identity_is_session_local_through_the_hook_lifecycle() { } } } + +/// A real HTTP backend whose subscriptions are absent from client registration. +async fn discovered_http_hooks( + subscriptions: Vec, +) -> (Arc, tokio::task::JoinHandle<()>) { + use axum::{Json, Router, extract::State, routing::post}; + let app = Router::new().route("/hooks", post( + |State(subscriptions): State>, Json(request): Json| async move { + let result = if request["method"] == "hooks/configuration" { + json!({"protocolVersion":"draft", "configuration":{"subscriptions":subscriptions}}) + } else { + json!({"protocolVersion":"draft", "effects":[ + {"type":"modify", "target":"response", "operation":"replace", "value":[]} + ]}) + }; + Json(json!({"jsonrpc":"2.0", "id":request["id"], "result":result})) + } + )).with_state(subscriptions); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}/hooks", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let registration = serde_json::from_value(json!({ + "protocolVersion":"draft", "hooks":[{ + "id":"dev.kit.discovery", "transport":{"type":"http", "url":endpoint} + }] + })) + .unwrap(); + let mut options = options().unwrap(); + options.backend.allow_loopback_http = true; + (Arc::new(Hooks::new(registration, options).unwrap()), server) +} + +#[tokio::test] +async fn discovered_compaction_subscriptions_disable_speculative_preparation() { + for event in ["context.compact.before", "context.compact.after"] { + let (hooks, server) = discovered_http_hooks(vec![observe(&[event])]).await; + assert!(hooks.intercepts_compaction().await.unwrap()); + hooks.sdk.shutdown().await.unwrap(); + server.abort(); + } + let (hooks, server) = discovered_http_hooks(vec![observe(&["model.response.after"])]).await; + assert!(!hooks.intercepts_compaction().await.unwrap()); + hooks.sdk.shutdown().await.unwrap(); + server.abort(); +} + +#[tokio::test] +async fn discovered_response_interceptor_with_upload_config_withholds_previews_and_tool_calls() { + let mut subscription = intercept(&["model.response.after"]); + // No upload is needed for this small response, but admitting the declaration + // still requires explicit host authorization of the server's upload policy. + subscription["upload"] = json!({ + "endpoint":"https://uploads.example.test/content", "timeoutMs":1000, "maxBytes":1024 + }); + let (hooks, server) = discovered_http_hooks(vec![subscription]).await; + let adapter = HookedAdapter::new( + ToolRoundModel(true), + Some(hooks.clone()), + SessionStartInputTrigger::Startup, + ); + let mut session = adapter + .start_session(SessionConfig::new("discovered")) + .await + .unwrap(); + let mut turn = session + .begin_turn( + TurnRequest { + session_id: SessionId::new("discovered"), + turn_id: TurnId::new("turn"), + transcript: vec![], + available_tools: vec![], + cache: None, + metadata: MetadataMap::new(), + }, + None, + ) + .await + .unwrap(); + let mut finished = false; + while let Some(event) = turn.next_event(None).await.unwrap() { + match event { + ModelTurnEvent::Delta(_) | ModelTurnEvent::ToolCall(_) => { + panic!("unapproved preview escaped") + } + ModelTurnEvent::Finished(result) => { + assert!(result.output_items.is_empty()); + assert_eq!(result.finish_reason, FinishReason::Completed); + finished = true; + } + _ => {} + } + } + assert!(finished); + hooks.sdk.shutdown().await.unwrap(); + server.abort(); +} diff --git a/src/runtime.rs b/src/runtime.rs index 094b0f7c..3c3c2a43 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -771,18 +771,17 @@ impl Runtime { Ok(Arc::new(runtime)) } - fn compactor( + async fn compactor( &self, adapter: SelectableAdapter, persistence: Option, owner_session: String, ) -> Result { - let prepare = self.base_depth == 0 - && persistence.is_some() - && !self - .hooks - .as_ref() - .is_some_and(|hooks| hooks.intercepts_compaction()); + let hooks_compaction = match &self.hooks { + Some(hooks) => hooks.intercepts_compaction().await?, + None => false, + }; + let prepare = self.base_depth == 0 && persistence.is_some() && !hooks_compaction; let compactor_session = if persistence.is_some() { format!("compaction-{}", crate::session::new_id()) } else { @@ -1542,6 +1541,7 @@ impl Runtime { Some(opened.observer.clone()), session_id.clone(), ) + .await .map_err(|error| { record_runtime_failure( &session_id, @@ -1742,6 +1742,7 @@ impl Runtime { let recorded = crate::session::TransientTranscript::new(transcript.clone()); let compactor = self .compactor(self.adapter.clone(), None, session.clone()) + .await .map_err(LoopError::InvalidState)? .mirroring(recorded.clone()); let subagents = self @@ -1966,6 +1967,7 @@ impl Runtime { Some(opened.observer.clone()), session_id.clone(), ) + .await .map_err(AcpRuntimeError::Loop)?; let prompt_cache_key = crate::tools::subagent::prompt_cache_key(&session_id); let subagents = self