fix(ahp): support server-driven hook configuration

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"] }

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 |

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<agenthooksprotocol::Hooks>,
-    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<bool, String> {
+        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 `<root>/.agents/hooks.json`. Project
@@ -82,10 +89,8 @@ impl Hooks {
     fn new(registration: Registration, options: HooksOptions) -> Result<Self, String> {
         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<HooksOptions, String> {
         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)
 }
 

src/ahp/model.rs

@@ -186,6 +186,10 @@ impl<S: ModelSession> ModelSession for HookedSession<S> {
             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<T: ModelTurn> ModelTurn for HookedTurn<T> {
             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<T: ModelTurn> ModelTurn for HookedTurn<T> {
             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

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<Value>,
+) -> (Arc<Hooks>, tokio::task::JoinHandle<()>) {
+    use axum::{Json, Router, extract::State, routing::post};
+    let app = Router::new().route("/hooks", post(
+        |State(subscriptions): State<Vec<Value>>, Json(request): Json<Value>| 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();
+}

src/runtime.rs

@@ -771,18 +771,17 @@ impl Runtime {
         Ok(Arc::new(runtime))
     }
 
-    fn compactor(
+    async fn compactor(
         &self,
         adapter: SelectableAdapter,
         persistence: Option<crate::session::SessionObserver>,
         owner_session: String,
     ) -> Result<crate::compaction::AutomaticCompactor, String> {
-        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