From b4f1d124b9bc40e36baf3922685699352affd85a Mon Sep 17 00:00:00 2001 From: Jordan Mecom Date: Mon, 17 Aug 2026 16:24:10 -0700 Subject: [PATCH] Enforce execution-domain ACP worker isolation Signed-off-by: Jordan Mecom --- crates/buzz-acp/README.md | 19 +- crates/buzz-acp/src/acp.rs | 3 + crates/buzz-acp/src/config.rs | 40 ++- crates/buzz-acp/src/ifc.rs | 475 ++++++++++++++++++++++------- crates/buzz-acp/src/lib.rs | 107 +++++-- crates/buzz-acp/src/pool.rs | 556 +++++++++++++++++++++++++++++++--- 6 files changed, 1003 insertions(+), 197 deletions(-) diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index 4a9b54f59..6cb31a647 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -119,20 +119,29 @@ All configuration is via environment variables (or CLI flags — every env var h **Legacy env vars:** `BUZZ_ACP_PRIVATE_KEY`, `BUZZ_ACP_API_TOKEN`, and `BUZZ_ACP_TURN_TIMEOUT` (replaced by `BUZZ_ACP_IDLE_TIMEOUT`) are still accepted as fallbacks. -### Experimental information-flow audit +### Experimental information-flow modes `--information-flow audit` (or `BUZZ_ACP_INFORMATION_FLOW=audit`) enables the -audience-scoped IFC prototype. It verifies trigger events and relay-signed channel +audience-scoped IFC audit. It verifies trigger events and relay-signed channel policy, derives `D = (Audience, Context, Epoch, Capabilities)`, evaluates read, call, publish, and process-reuse rules, and keeps a conservative process-level state label. Decisions are emitted under the `buzz_acp::ifc` tracing target. +`--information-flow isolate` evaluates the same policy before pool selection and +uses the complete execution domain as the ACP worker routing key. A fresh child is +bound on first use. If its audience, context, membership epoch, or capabilities no +longer match, Buzz starts a replacement child and clears all ACP sessions and +harness-side session state before delivering the turn. Owner-private core memory +is not fetched for public or shared-conversation domains. Heartbeats use their own +owner-private domain. + The default is `off`. In that mode no IFC auditor is constructed, no extra channel policy queries run, and prompt/session behavior is unchanged. -Audit mode is observational. It does not filter prompts, block tools, split agent -processes, bind replies to a destination, or provide OS confinement. Its logs call -those gaps out explicitly; enabling it is not an enforcement claim. +Audit mode remains entirely observational. Isolate mode enforces ACP process and +core-memory separation, but it does not yet mediate tool credentials, bind replies +to a broker-controlled destination, isolate the shared workspace, or provide OS +confinement. Its logs report those remaining gaps explicitly. ### Parallel Agents & Heartbeat diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index f8373bd66..deb5180f4 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -106,6 +106,9 @@ pub enum AcpError { #[error("Protocol error: {0}")] Protocol(String), + #[error("Information-flow isolation failed: {0}")] + InformationFlow(String), + #[error("Agent reported error (code {code}): {message}")] AgentError { code: i64, message: String }, } diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index d1e77c21c..4ae1fe5cd 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -138,11 +138,12 @@ pub enum PermissionMode { Plan, } -/// Information-flow mode for the experimental audience policy. +/// Information-flow mode for audience-scoped agent execution. /// /// `Off` preserves the existing harness path without membership queries or policy -/// bookkeeping. `Audit` evaluates and logs the design-paper rules but does not alter -/// prompts, tools, session reuse, or publication. +/// bookkeeping. `Audit` evaluates and logs the design-paper rules without changing +/// runtime behavior. `Isolate` additionally binds each ACP child process and all of +/// its retained state to one execution domain. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, clap::ValueEnum)] pub enum InformationFlowMode { /// Do not construct or invoke the experimental policy evaluator. @@ -150,6 +151,8 @@ pub enum InformationFlowMode { Off, /// Evaluate and log policy without changing the existing turn. Audit, + /// Replace an ACP child before it crosses an execution-domain boundary. + Isolate, } impl std::fmt::Display for InformationFlowMode { @@ -157,10 +160,23 @@ impl std::fmt::Display for InformationFlowMode { match self { Self::Off => f.write_str("off"), Self::Audit => f.write_str("audit"), + Self::Isolate => f.write_str("isolate"), } } } +impl InformationFlowMode { + /// Whether the harness must resolve and account for execution domains. + pub(crate) fn enabled(self) -> bool { + self != Self::Off + } + + /// Whether execution-domain boundaries change ACP process reuse. + pub(crate) fn isolates_processes(self) -> bool { + self == Self::Isolate + } +} + impl PermissionMode { /// Return the wire-format string sent to the agent via /// `session/set_config_option`. @@ -466,8 +482,9 @@ pub struct CliArgs { )] pub permission_mode: PermissionMode, - /// Evaluate the audience-scoped information-flow design without enforcing it. - /// `off` is the default and leaves the existing harness behavior unchanged. + /// Apply the audience-scoped information-flow policy. `audit` only logs; + /// `isolate` also confines ACP process reuse and retained state by domain. + /// `off` is the default and leaves existing harness behavior unchanged. #[arg( long, env = "BUZZ_ACP_INFORMATION_FLOW", @@ -2244,7 +2261,7 @@ channels = "ALL" } #[test] - fn information_flow_is_default_off_and_audit_is_explicit() { + fn information_flow_modes_are_explicit_and_default_off() { let key = "0".repeat(64); let default = CliArgs::parse_from(["buzz-acp", "--private-key", &key]); assert_eq!(default.information_flow, InformationFlowMode::Off); @@ -2257,6 +2274,17 @@ channels = "ALL" "audit", ]); assert_eq!(audit.information_flow, InformationFlowMode::Audit); + + let isolate = CliArgs::parse_from([ + "buzz-acp", + "--private-key", + &key, + "--information-flow", + "isolate", + ]); + assert_eq!(isolate.information_flow, InformationFlowMode::Isolate); + assert!(isolate.information_flow.enabled()); + assert!(isolate.information_flow.isolates_processes()); } #[test] diff --git a/crates/buzz-acp/src/ifc.rs b/crates/buzz-acp/src/ifc.rs index c75f21054..b169bc338 100644 --- a/crates/buzz-acp/src/ifc.rs +++ b/crates/buzz-acp/src/ifc.rs @@ -2,8 +2,9 @@ //! //! This module is deliberately isolated from ACP transport and prompt formatting. In //! `off` mode it is not constructed at all. In `audit` mode it observes the data that -//! the existing harness admits, evaluates the proposal's rules, and emits structured -//! decisions without changing runtime behavior. +//! the existing harness admits and emits structured decisions without changing runtime +//! behavior. In `isolate` mode the harness also uses its domain key to route or replace +//! ACP children before state crosses an audience, context, epoch, or capability boundary. //! //! Comments prefixed with "Paper" refer to the matching section in the //! "Practical information-flow for Buzz agents" design paper. @@ -16,6 +17,7 @@ use nostr::{Alphabet, Event, Filter, Kind, PublicKey, SingleLetterTag}; use sha2::{Digest, Sha256}; use uuid::Uuid; +use crate::config::InformationFlowMode; use crate::queue::FlushBatch; use crate::relay::RestClient; @@ -289,6 +291,25 @@ impl CapabilitySet { #[derive(Clone, Debug, Eq, PartialEq)] struct MembershipEpoch(String); +/// Opaque routing key for one complete execution domain. +/// +/// The pool may compare and log this value, but only this module constructs it. +/// That keeps worker routing coupled to the complete paper definition rather +/// than a caller-selected subset such as channel ID or audience alone. +#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub(crate) struct DomainKey(String); + +impl DomainKey { + pub(crate) fn fingerprint(&self) -> String { + short_fingerprint(&self.0) + } + + #[cfg(test)] + pub(crate) fn for_test(value: &str) -> Self { + Self(value.to_string()) + } +} + /// Paper: `D = (Audience, Context, Epoch, Capabilities)`. #[derive(Clone, Debug, Eq, PartialEq)] struct ExecutionDomain { @@ -310,6 +331,10 @@ impl ExecutionDomain { hex::encode(hasher.finalize()) } + fn key(&self) -> DomainKey { + DomainKey(self.id()) + } + fn resource_label(&self) -> ResourceLabel { ResourceLabel { confidentiality: self.audience.clone(), @@ -574,6 +599,7 @@ enum ResolutionError { EmptyRestrictedAudience, AgentNotMember, RequesterNotMember, + NoOwner, VerificationTask, } @@ -595,19 +621,21 @@ impl fmt::Display for ResolutionError { Self::EmptyRestrictedAudience => "restricted channel has no human audience", Self::AgentNotMember => "executing agent is absent from channel membership", Self::RequesterNotMember => "trigger requester is absent from channel membership", + Self::NoOwner => "owner-private domain requires a resolved agent owner", Self::VerificationTask => "signature verification task failed", }; f.write_str(message) } } -/// Stateless policy front-end retained by `PromptContext` only in audit mode. +/// Stateless policy front-end retained by `PromptContext` in audit or isolate mode. pub(crate) struct Auditor { realm: RealmId, rest_client: RestClient, relay_self: Option, agent: Principal, owner: Option, + mode: InformationFlowMode, } impl Auditor { @@ -617,6 +645,7 @@ impl Auditor { relay_self: Option<&str>, agent: PublicKey, owner: Option, + mode: InformationFlowMode, ) -> Self { Self { realm: RealmId::from_relay_url(relay_url), @@ -624,68 +653,80 @@ impl Auditor { relay_self: relay_self.and_then(|value| PublicKey::from_hex(value).ok()), agent: Principal::from_key(agent), owner: owner.map(Principal::from_key), + mode, } } - /// Begin one audit turn. Failure to derive a trustworthy domain does not - /// block the turn in audit mode, but permanently marks this process state as - /// having unknown provenance so later publish checks cannot claim safety. - pub(crate) async fn begin_turn( + /// Resolve one signed channel batch before it is assigned to an ACP child. + /// The caller decides whether an unresolved turn is merely audited or denied. + pub(crate) async fn resolve_turn( &self, batch: &FlushBatch, turn_id: &str, - agent_index: usize, - process: &mut ProcessAuditState, + agent_index: Option, ) -> ActiveTurn { let verified = match verify_trigger_batch(batch).await { Ok(verified) => { - log_rule( + log_rule(RuleLog { + mode: self.mode, turn_id, agent_index, - None, - "event_admission", - "allow", - "all trigger IDs, signatures, and channel bindings verified", - false, - ); + domain_id: None, + rule: "event_admission", + decision: "allow", + reason: "all trigger IDs, signatures, and channel bindings verified", + enforced: self.mode.isolates_processes(), + }); verified } Err(error) => { - process.confinement.mark_unknown(); - log_rule( + log_rule(RuleLog { + mode: self.mode, turn_id, agent_index, - None, - "event_admission", - "deny", - &error.to_string(), - false, + domain_id: None, + rule: "event_admission", + decision: "deny", + reason: &error.to_string(), + enforced: self.mode.isolates_processes(), + }); + return ActiveTurn::unresolved( + turn_id, + agent_index, + self.owner.clone(), + self.mode, + TurnOrigin::Channel, ); - return ActiveTurn::unresolved(turn_id, agent_index, self.owner.clone()); } }; let domain = match self.resolve_domain(&verified).await { Ok(domain) => domain, Err(error) => { - process.confinement.mark_unknown(); - log_rule( + log_rule(RuleLog { + mode: self.mode, turn_id, agent_index, - None, - "domain_resolution", - "deny", - &error.to_string(), - false, + domain_id: None, + rule: "domain_resolution", + decision: "deny", + reason: &error.to_string(), + enforced: self.mode.isolates_processes(), + }); + return ActiveTurn::unresolved( + turn_id, + agent_index, + self.owner.clone(), + self.mode, + TurnOrigin::Channel, ); - return ActiveTurn::unresolved(turn_id, agent_index, self.owner.clone()); } }; let domain_id = domain.id(); tracing::info!( target: "buzz_acp::ifc", - ifc_mode = "audit", + ifc_mode = %self.mode, turn_id, agent_index, domain_id = %domain_id, @@ -702,28 +743,73 @@ impl Auditor { "ifc execution domain resolved" ); - let reuse = process.enter(&domain); - log_rule( - turn_id, - agent_index, - Some(&domain_id), - "reuse", - reuse.result(), - reuse.reason, - false, - ); - - let turn = ActiveTurn { + ActiveTurn { turn_id: turn_id.to_string(), agent_index, domain: Some(domain), owner: self.owner.clone(), + mode: self.mode, + origin: TurnOrigin::Channel, + } + } + + /// Heartbeats carry no external event audience. They run in an owner-only + /// domain so they cannot reuse a public or conversation-bound ACP child. + pub(crate) fn resolve_heartbeat( + &self, + turn_id: &str, + agent_index: Option, + ) -> ActiveTurn { + let Some(owner) = self.owner.clone() else { + log_rule(RuleLog { + mode: self.mode, + turn_id, + agent_index, + domain_id: None, + rule: "domain_resolution", + decision: "deny", + reason: &ResolutionError::NoOwner.to_string(), + enforced: self.mode.isolates_processes(), + }); + return ActiveTurn::unresolved( + turn_id, + agent_index, + None, + self.mode, + TurnOrigin::Heartbeat, + ); }; - turn.observe_domain_input(process, "trigger_events"); - turn.observe_domain_input(process, "channel_metadata"); - turn.log_capability_policy(); - turn.log_unmediated_coverage(); - turn + let context = DomainContext::OwnerPrivate { + realm: self.realm.clone(), + owner: owner.clone(), + }; + let domain = ExecutionDomain { + audience: ConfidentialityLabel::restricted(self.realm.clone(), BTreeSet::from([owner])), + context, + epoch: MembershipEpoch("owner-heartbeat-v1".to_string()), + capabilities: bot_capabilities(), + }; + tracing::info!( + target: "buzz_acp::ifc", + ifc_mode = %self.mode, + turn_id, + agent_index, + domain_id = %domain.id(), + realm = %self.realm.fingerprint(), + context = domain.context.kind(), + audience = "restricted", + reader_count = 1, + epoch = %short_fingerprint(&domain.epoch.0), + "ifc execution domain resolved" + ); + ActiveTurn { + turn_id: turn_id.to_string(), + agent_index, + domain: Some(domain), + owner: self.owner.clone(), + mode: self.mode, + origin: TurnOrigin::Heartbeat, + } } async fn resolve_domain( @@ -827,24 +913,121 @@ impl Auditor { } } -/// One resolved (or conservatively unresolved) invocation audit. +#[derive(Clone, Copy)] +enum TurnOrigin { + Channel, + Heartbeat, +} + +/// One resolved (or conservatively unresolved) invocation policy. pub(crate) struct ActiveTurn { turn_id: String, - agent_index: usize, + agent_index: Option, domain: Option, owner: Option, + mode: InformationFlowMode, + origin: TurnOrigin, } impl ActiveTurn { - fn unresolved(turn_id: &str, agent_index: usize, owner: Option) -> Self { + fn unresolved( + turn_id: &str, + agent_index: Option, + owner: Option, + mode: InformationFlowMode, + origin: TurnOrigin, + ) -> Self { Self { turn_id: turn_id.to_string(), agent_index, domain: None, owner, + mode, + origin, } } + pub(crate) fn domain_key(&self) -> Option { + self.domain.as_ref().map(ExecutionDomain::key) + } + + pub(crate) fn assign_agent(&mut self, agent_index: usize) { + self.agent_index = Some(agent_index); + } + + /// Commit the inputs that are about to enter the selected ACP process. + /// Resolution happens before routing; this step happens only after the + /// process is either proven reusable or replaced for this domain. + pub(crate) fn enter_process(&self, process: &mut ProcessAuditState) { + let Some(domain) = self.domain.as_ref() else { + process.confinement.mark_unknown(); + return; + }; + let domain_id = domain.id(); + let reuse = process.enter(domain); + log_rule(RuleLog { + mode: self.mode, + turn_id: &self.turn_id, + agent_index: self.agent_index, + domain_id: Some(&domain_id), + rule: "reuse", + decision: reuse.result(), + reason: reuse.reason, + enforced: self.mode.isolates_processes(), + }); + match self.origin { + TurnOrigin::Channel => { + self.observe_domain_input(process, "trigger_events"); + self.observe_domain_input(process, "channel_metadata"); + } + TurnOrigin::Heartbeat => { + self.observe_domain_input(process, "heartbeat_trigger"); + } + } + self.log_capability_policy(); + self.log_unmediated_coverage(); + } + + /// Whether owner-private material may enter this turn's domain. This is + /// checked before the broker fetches core memory in isolate mode. + pub(crate) fn permits_owner_private_input(&self) -> bool { + let (Some(domain), Some(owner)) = (self.domain.as_ref(), self.owner.as_ref()) else { + return false; + }; + let mut readers = BTreeSet::new(); + readers.insert(owner.clone()); + RuleEvaluator::read( + domain, + &ResourceLabel { + confidentiality: ConfidentialityLabel::restricted( + domain.audience.realm.clone(), + readers, + ), + context: ResourceContext::OwnerPrivate { + realm: domain.audience.realm.clone(), + owner: owner.clone(), + }, + }, + ) + .allowed + } + + pub(crate) fn log_blocked_owner_private_input(&self, source: &'static str) { + tracing::info!( + target: "buzz_acp::ifc", + ifc_mode = %self.mode, + turn_id = %self.turn_id, + agent_index = self.agent_index, + domain_id = self.domain.as_ref().map(ExecutionDomain::id), + rule = "read", + source, + decision = "deny", + reason = "owner-private input cannot flow to this execution domain", + enforced = true, + "ifc rule evaluated" + ); + } + /// Observe channel-bound material such as message history, a channel /// canvas, or huddle instructions. pub(crate) fn observe_domain_input( @@ -854,23 +1037,22 @@ impl ActiveTurn { ) { let Some(domain) = self.domain.as_ref() else { process.confinement.mark_unknown(); - log_rule( - &self.turn_id, - self.agent_index, - None, - "read", - "deny", - "execution domain is unresolved", - false, - ); + log_rule(RuleLog { + mode: self.mode, + turn_id: &self.turn_id, + agent_index: self.agent_index, + domain_id: None, + rule: "read", + decision: "deny", + reason: "execution domain is unresolved", + enforced: self.mode.isolates_processes(), + }); return; }; self.observe_resource(process, source, domain.resource_label()); } - /// Observe owner-scoped core memory. In the current harness the same NIP-AE - /// core can be injected into every channel session; audit mode makes the - /// resulting illegal flow visible without yet changing that behavior. + /// Observe owner-scoped core memory after the caller has admitted it. pub(crate) fn observe_owner_private( &self, process: &mut ProcessAuditState, @@ -878,15 +1060,17 @@ impl ActiveTurn { ) { let (Some(domain), Some(owner)) = (self.domain.as_ref(), self.owner.as_ref()) else { process.confinement.mark_unknown(); - log_rule( - &self.turn_id, - self.agent_index, - self.domain.as_ref().map(ExecutionDomain::id).as_deref(), - "read", - "deny", - "owner-private input lacks a resolved owner or domain", - false, - ); + let domain_id = self.domain.as_ref().map(ExecutionDomain::id); + log_rule(RuleLog { + mode: self.mode, + turn_id: &self.turn_id, + agent_index: self.agent_index, + domain_id: domain_id.as_deref(), + rule: "read", + decision: "deny", + reason: "owner-private input lacks a resolved owner or domain", + enforced: self.mode.isolates_processes(), + }); return; }; let mut readers = BTreeSet::new(); @@ -939,7 +1123,7 @@ impl ActiveTurn { process.confinement.mark_unknown(); tracing::info!( target: "buzz_acp::ifc", - ifc_mode = "audit", + ifc_mode = %self.mode, turn_id = %self.turn_id, agent_index = self.agent_index, domain_id = self.domain.as_ref().map(ExecutionDomain::id), @@ -963,14 +1147,16 @@ impl ActiveTurn { return; }; let decision = RuleEvaluator::read(domain, &resource); - // Audit mode does not suppress a denied input. Since the real model sees - // it, the confinement label must still include it; otherwise the later - // publish decision would incorrectly claim the process remained clean. - process.confinement.observe(&resource); + // Audit mode records denied inputs because the model still sees them. + // Isolate mode records only admitted inputs; denied owner-private input + // is stopped before the broker fetches it. + if decision.allowed || !self.mode.isolates_processes() { + process.confinement.observe(&resource); + } let domain_id = domain.id(); tracing::info!( target: "buzz_acp::ifc", - ifc_mode = "audit", + ifc_mode = %self.mode, turn_id = %self.turn_id, agent_index = self.agent_index, domain_id = %domain_id, @@ -978,7 +1164,7 @@ impl ActiveTurn { source, decision = decision.result(), reason = decision.reason, - enforced = false, + enforced = self.mode.isolates_processes(), "ifc rule evaluated" ); } @@ -999,7 +1185,7 @@ impl ActiveTurn { let decision = RuleEvaluator::call(domain, operation); tracing::info!( target: "buzz_acp::ifc", - ifc_mode = "audit", + ifc_mode = %self.mode, turn_id = %self.turn_id, agent_index = self.agent_index, domain_id = %domain_id, @@ -1020,15 +1206,16 @@ impl ActiveTurn { /// current ACP harness does not receive and bind the agent's final message. pub(crate) fn audit_reply(&self, process: &ProcessAuditState) { let Some(domain) = self.domain.as_ref() else { - log_rule( - &self.turn_id, - self.agent_index, - None, - "publish", - "deny", - "execution domain is unresolved", - false, - ); + log_rule(RuleLog { + mode: self.mode, + turn_id: &self.turn_id, + agent_index: self.agent_index, + domain_id: None, + rule: "publish", + decision: "deny", + reason: "execution domain is unresolved", + enforced: false, + }); return; }; let digest: [u8; 32] = Sha256::digest(b"audit-only-unbound-output").into(); @@ -1042,7 +1229,7 @@ impl ActiveTurn { ); tracing::info!( target: "buzz_acp::ifc", - ifc_mode = "audit", + ifc_mode = %self.mode, turn_id = %self.turn_id, agent_index = self.agent_index, domain_id = %domain.id(), @@ -1056,16 +1243,21 @@ impl ActiveTurn { } fn log_unmediated_coverage(&self) { + let gaps = if self.mode.isolates_processes() { + "ambient_workspace, credential_bearing_mcp, direct_buzz_publication, static_capability_inventory, operating_system_isolation" + } else { + "shared_agent_process, ambient_workspace, credential_bearing_mcp, direct_buzz_publication, static_capability_inventory, operating_system_isolation" + }; tracing::warn!( target: "buzz_acp::ifc", - ifc_mode = "audit", + ifc_mode = %self.mode, turn_id = %self.turn_id, agent_index = self.agent_index, domain_id = self.domain.as_ref().map(ExecutionDomain::id), rule = "confinement_coverage", decision = "not_proven", enforced = false, - gaps = "shared_agent_process, ambient_workspace, credential_bearing_mcp, direct_buzz_publication, static_capability_inventory, operating_system_isolation", + gaps, "ifc audit cannot prove confinement while these paths bypass policy" ); } @@ -1197,14 +1389,7 @@ fn effective_turn_capabilities( ) -> CapabilitySet { let conversation = CapabilitySet::from_names(["buzz.read.current", "buzz.publish.current", "memory.domain"]); - let bot = CapabilitySet::from_names([ - "buzz.read.current", - "buzz.publish.current", - "memory.domain", - "email.read", - "drive.read", - "shell.host", - ]); + let bot = bot_capabilities(); let requester_is_owner = owner.is_some_and(|owner| { !trigger.requesters.is_empty() && trigger @@ -1229,6 +1414,17 @@ fn effective_turn_capabilities( CapabilitySet::effective(&bot, &requester, &domain) } +fn bot_capabilities() -> CapabilitySet { + CapabilitySet::from_names([ + "buzz.read.current", + "buzz.publish.current", + "memory.domain", + "email.read", + "drive.read", + "shell.host", + ]) +} + fn hash_field(hasher: &mut Sha256, value: &[u8]) { hasher.update(value.len().to_be_bytes()); hasher.update(value); @@ -1239,18 +1435,31 @@ fn short_fingerprint(value: &str) -> String { hex::encode(&digest[..6]) } -fn log_rule( - turn_id: &str, - agent_index: usize, - domain_id: Option<&str>, +struct RuleLog<'a> { + mode: InformationFlowMode, + turn_id: &'a str, + agent_index: Option, + domain_id: Option<&'a str>, rule: &'static str, decision: &'static str, - reason: &str, + reason: &'a str, enforced: bool, -) { +} + +fn log_rule(entry: RuleLog<'_>) { + let RuleLog { + mode, + turn_id, + agent_index, + domain_id, + rule, + decision, + reason, + enforced, + } = entry; tracing::info!( target: "buzz_acp::ifc", - ifc_mode = "audit", + ifc_mode = %mode, turn_id, agent_index, domain_id, @@ -1469,6 +1678,46 @@ mod tests { assert!(RuleEvaluator::reuse(&first, &same).allowed); assert!(!RuleEvaluator::reuse(&first, &new_epoch).allowed); assert!(!RuleEvaluator::reuse(&first, &new_context).allowed); + assert_eq!(first.key(), same.key()); + assert_ne!(first.key(), new_epoch.key()); + assert_ne!(first.key(), new_context.key()); + } + + #[test] + fn owner_private_input_is_admitted_only_to_the_owner_domain() { + let owner = principal("alice"); + let owner_turn = ActiveTurn { + turn_id: "owner".into(), + agent_index: Some(0), + domain: Some(ExecutionDomain { + audience: label(&["alice"]), + context: DomainContext::OwnerPrivate { + realm: realm(), + owner: owner.clone(), + }, + epoch: MembershipEpoch("owner-v1".into()), + capabilities: bot_capabilities(), + }), + owner: Some(owner.clone()), + mode: InformationFlowMode::Isolate, + origin: TurnOrigin::Channel, + }; + let public_turn = ActiveTurn { + turn_id: "public".into(), + agent_index: Some(1), + domain: Some(ExecutionDomain { + audience: public_label(), + context: DomainContext::RealmPublic(realm()), + epoch: MembershipEpoch("public-v1".into()), + capabilities: CapabilitySet::default(), + }), + owner: Some(owner), + mode: InformationFlowMode::Isolate, + origin: TurnOrigin::Channel, + }; + + assert!(owner_turn.permits_owner_private_input()); + assert!(!public_turn.permits_owner_private_input()); } #[test] @@ -1826,12 +2075,12 @@ mod tests { Some(&relay.public_key().to_hex()), agent.public_key(), Some(owner.public_key()), + InformationFlowMode::Audit, ); let mut process = ProcessAuditState::default(); - let turn = auditor - .begin_turn(&batch, "test-turn", 0, &mut process) - .await; - let domain = turn.domain.expect("resolved domain"); + let turn = auditor.resolve_turn(&batch, "test-turn", Some(0)).await; + turn.enter_process(&mut process); + let domain = turn.domain.as_ref().expect("resolved domain"); assert!(matches!(domain.context, DomainContext::OwnerPrivate { .. })); assert_eq!(domain.audience.readers.explicit_count(), Some(1)); diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 57e5bd3b9..2e2cd788c 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2177,17 +2177,35 @@ async fn tokio_main() -> Result<()> { let agent_owner_pubkey = startup_owner .as_deref() .and_then(|hex| nostr::PublicKey::from_hex(hex).ok()); - let ifc_auditor = (config.information_flow == InformationFlowMode::Audit).then(|| { - tracing::warn!( - target: "buzz_acp::ifc", - "information-flow audit enabled; decisions are logged but not enforced" - ); + let ifc_auditor = config.information_flow.enabled().then(|| { + if config.information_flow == InformationFlowMode::Audit { + tracing::warn!( + target: "buzz_acp::ifc", + "information-flow audit enabled; decisions are logged but not enforced" + ); + } else { + tracing::warn!( + target: "buzz_acp::ifc", + "information-flow isolation enabled; ACP process reuse and owner-core reads are enforced, but tool and publication paths remain unmediated" + ); + } ifc::Auditor::new( &config.relay_url, relay.rest_client(), relay_self.as_deref(), config.keys.public_key(), agent_owner_pubkey, + config.information_flow, + ) + }); + let ifc_process_spec = config.information_flow.isolates_processes().then(|| { + pool::AgentProcessSpec::new( + config.agent_command.clone(), + config.agent_args.clone(), + config.persona_env_vars.clone(), + config.has_generated_codex_config, + config.model.clone(), + observer.clone(), ) }); let ctx = Arc::new(PromptContext { @@ -2218,6 +2236,8 @@ async fn tokio_main() -> Result<()> { max_turns_per_session: config.max_turns_per_session, permission_mode: config.permission_mode, ifc_auditor, + information_flow: config.information_flow, + ifc_process_spec, agent_keys: config.keys.clone(), agent_owner_pubkey, memory_enabled: config.memory_enabled, @@ -2475,7 +2495,7 @@ async fn tokio_main() -> Result<()> { // arrive when the channel is silent. if queue.has_flushable_work() { for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity).await { typing_channels.insert(channel_id, thread_tags); } @@ -2524,7 +2544,7 @@ async fn tokio_main() -> Result<()> { // next relay event arrives — which can be minutes on quiet channels. if respawn_collected { for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity).await { typing_channels.insert(channel_id, thread_tags); } @@ -3005,6 +3025,7 @@ async fn tokio_main() -> Result<()> { if pool_ready { for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + .await { typing_channels.insert(channel_id, thread_tags); } @@ -3105,6 +3126,7 @@ async fn tokio_main() -> Result<()> { tracing::debug!("heartbeat_skipped_events"); for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + .await { typing_channels.insert(channel_id, thread_tags); } @@ -3203,7 +3225,7 @@ async fn tokio_main() -> Result<()> { break; } for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity).await { typing_channels.insert(channel_id, thread_tags); } @@ -3228,7 +3250,7 @@ async fn tokio_main() -> Result<()> { break; } for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity).await { typing_channels.insert(channel_id, thread_tags); } @@ -3383,7 +3405,7 @@ async fn tokio_main() -> Result<()> { // queue drains. We still try here in case the in-flight // task has already returned. for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity).await { typing_channels.insert(channel_id, thread_tags); } @@ -3411,7 +3433,7 @@ async fn tokio_main() -> Result<()> { None, ); for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity).await { typing_channels.insert(channel_id, thread_tags); } @@ -3825,7 +3847,7 @@ fn try_native_steer( // ── dispatch_pending ────────────────────────────────────────────────────────── /// Flush queued work to available agents. -fn dispatch_pending( +async fn dispatch_pending( pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc, @@ -3843,8 +3865,20 @@ fn dispatch_pending( .last() .map(|event| queue::parse_thread_tags(&event.event)) .unwrap_or_default(); - let affinity_hit = pool.has_session_for(channel_id); - let mut agent = match pool.try_claim(Some(channel_id)) { + let turn_id = Uuid::new_v4().to_string(); + let mut prepared_ifc_turn = if ctx.information_flow.isolates_processes() { + match ctx.ifc_auditor.as_ref() { + Some(auditor) => Some(auditor.resolve_turn(&batch, &turn_id, None).await), + None => None, + } + } else { + None + }; + let domain = prepared_ifc_turn + .as_ref() + .and_then(crate::ifc::ActiveTurn::domain_key); + let affinity_hit = pool.has_affinity_for(channel_id, domain.as_ref()); + let agent = match pool.try_claim(Some(channel_id), domain.as_ref()) { Some(a) => a, None => { let pending = queue.pending_channels(); @@ -3854,6 +3888,9 @@ fn dispatch_pending( break; } }; + if let Some(turn) = prepared_ifc_turn.as_mut() { + turn.assign_agent(agent.index); + } tracing::debug!(agent = agent.index, channel = %channel_id, affinity_hit, "agent_claimed"); let recoverable_batch = match ctx.dedup_mode { @@ -3865,8 +3902,9 @@ fn dispatch_pending( let ctx_clone = Arc::clone(ctx); let agent_index = agent.index; - // Mid-turn non-cancelling steer seam: install the per-turn steer - // receiver on the read loop so the main loop's mode-gate fork + // Mid-turn non-cancelling steer seam: pass the per-turn receiver to + // the prompt task so it is installed after any domain-driven process + // replacement. The main loop's mode-gate fork // (see the `if accepted && queue.is_channel_in_flight(...)` block // in the relay event branch of the main `select!` loop) can drive // it via the matching sender stored in `TaskMeta.steer_tx`. @@ -3875,20 +3913,22 @@ fn dispatch_pending( // advertised `_session/steering` capability, and acks // `ExpectedRunIdMissing` (→ cancel+merge) when it has neither. let (tx, rx) = tokio::sync::mpsc::channel::(1); - agent.acp.install_steer_rx(rx); let steer_tx = Some(tx); // Prompt text is now built inside run_prompt_task (needs async for // context fetching). Pass None for prompt_text; batch carries the data. let (control_tx, control_rx) = tokio::sync::oneshot::channel::(); - let turn_id = Uuid::new_v4().to_string(); let task_turn_id = turn_id.clone(); let abort_handle = pool.join_set.spawn(async move { pool::run_prompt_task( agent, - Some(batch), - None, + pool::PromptDispatch { + batch: Some(batch), + prompt_text: None, + prepared_ifc_turn, + steer_rx: Some(rx), + }, ctx_clone, result_tx, Some(control_rx), @@ -4505,10 +4545,24 @@ fn dispatch_heartbeat( if *heartbeat_in_flight { return; } - let agent = match pool.try_claim(None) { + let turn_id = Uuid::new_v4().to_string(); + let mut prepared_ifc_turn = if ctx.information_flow.isolates_processes() { + ctx.ifc_auditor + .as_ref() + .map(|auditor| auditor.resolve_heartbeat(&turn_id, None)) + } else { + None + }; + let domain = prepared_ifc_turn + .as_ref() + .and_then(crate::ifc::ActiveTurn::domain_key); + let agent = match pool.try_claim(None, domain.as_ref()) { Some(a) => a, None => return, }; + if let Some(turn) = prepared_ifc_turn.as_mut() { + turn.assign_agent(agent.index); + } let prompt_text = ctx .heartbeat_prompt @@ -4517,14 +4571,17 @@ fn dispatch_heartbeat( let result_tx = pool.result_tx(); let ctx_clone = Arc::clone(ctx); let agent_index = agent.index; - let turn_id = Uuid::new_v4().to_string(); let task_turn_id = turn_id.clone(); let abort_handle = pool.join_set.spawn(async move { pool::run_prompt_task( agent, - None, - Some(prompt_text), + pool::PromptDispatch { + batch: None, + prompt_text: Some(prompt_text), + prepared_ifc_turn, + steer_rx: None, + }, ctx_clone, result_tx, None, @@ -4842,7 +4899,7 @@ async fn initialize_agent_pool( /// /// Takes owned args so it can run in a background `tokio::spawn` task without /// borrowing `Config`. All respawn/refill paths use this. -async fn spawn_and_init( +pub(crate) async fn spawn_and_init( command: &str, args: &[String], extra_env: &[(String, String)], diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 361dbf50c..46dec5f84 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -34,7 +34,7 @@ use crate::acp::{ resolve_model_switch_method, AcpClient, AcpError, EnvVar, McpServer, ModelSwitchMethod, StopReason, SystemPromptTransport, }; -use crate::config::{compose_session_title, DedupMode, PermissionMode}; +use crate::config::{compose_session_title, DedupMode, InformationFlowMode, PermissionMode}; use crate::observer; use crate::queue::{ CancelReason, ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo, @@ -132,6 +132,10 @@ pub struct SessionState { /// Process-level IFC audit state. This is intentionally not cleared when an /// ACP session is invalidated: a live process does not forget prior input. pub ifc_audit: crate::ifc::ProcessAuditState, + /// Complete execution domain to which this ACP child is bound in isolate + /// mode. It survives session invalidation and changes only when the child + /// process itself is replaced. + pub ifc_domain: Option, } impl SessionState { @@ -556,6 +560,39 @@ impl ChannelInfoResolver { } } +/// Everything needed to replace an ACP child when a slot crosses an execution +/// domain. Constructed only in isolate mode so off/audit keep the existing +/// process lifecycle and configuration footprint. +#[derive(Clone)] +pub(crate) struct AgentProcessSpec { + command: String, + args: Vec, + extra_env: Vec<(String, String)>, + has_generated_codex_config: bool, + configured_model: Option, + observer: Option, +} + +impl AgentProcessSpec { + pub(crate) fn new( + command: String, + args: Vec, + extra_env: Vec<(String, String)>, + has_generated_codex_config: bool, + configured_model: Option, + observer: Option, + ) -> Self { + Self { + command, + args, + extra_env, + has_generated_codex_config, + configured_model, + observer, + } + } +} + pub struct PromptContext { pub mcp_servers: Vec, pub initial_message: Option, @@ -590,9 +627,14 @@ pub struct PromptContext { pub max_turns_per_session: u32, /// Permission mode to apply after session creation. `Default` = skip. pub permission_mode: PermissionMode, - /// Present only for `--information-flow=audit`. `None` is the default fast - /// path and performs no membership queries or IFC bookkeeping. + /// Present for information-flow audit or isolation. `None` is the default + /// fast path and performs no membership queries or IFC bookkeeping. pub ifc_auditor: Option, + /// Information-flow behavior selected at startup. + pub information_flow: InformationFlowMode, + /// Spawn configuration used only when isolate mode must replace an ACP + /// child before binding it to another domain. + pub ifc_process_spec: Option, /// Agent identity — used to derive the NIP-AE conversation key at /// session creation for core injection. pub agent_keys: nostr::Keys, @@ -634,16 +676,26 @@ impl AgentPool { /// Try to claim an idle agent for the given channel (or heartbeat if `None`). /// - /// Pass 1: prefer an agent that already has a session for `channel_id`. - /// Pass 2: any idle agent. + /// Pass 1: prefer a compatible live session for `channel_id`. + /// Pass 2: prefer an agent bound to the resolved execution domain. + /// Pass 3: prefer an unbound agent, then fall back to any idle agent. /// /// Returns `None` if all agents are checked out. - pub fn try_claim(&mut self, channel_id: Option) -> Option { + pub fn try_claim( + &mut self, + channel_id: Option, + domain: Option<&crate::ifc::DomainKey>, + ) -> Option { // Pass 1: prefer agent with existing session for this channel. if let Some(cid) = channel_id { let idx = self.agents.iter().position(|slot| { slot.as_ref() - .map(|a| a.state.sessions.contains_key(&cid)) + .map(|agent| { + agent.state.sessions.contains_key(&cid) + && domain.is_none_or(|requested| { + agent.state.ifc_domain.as_ref() == Some(requested) + }) + }) .unwrap_or(false) }); if let Some(i) = idx { @@ -651,9 +703,32 @@ impl AgentPool { } } - // Pass 2: first idle agent. + // Pass 2: prefer a process already bound to the resolved domain. This + // lets all public channels share their public worker while keeping + // restricted conversations on their own workers. + if let Some(domain) = domain { + if let Some(index) = self.agents.iter().position(|slot| { + slot.as_ref() + .and_then(|agent| agent.state.ifc_domain.as_ref()) + == Some(domain) + }) { + return self.agents[index].take(); + } + } + + // Pass 3: a process that has never consumed domain-scoped input can be + // bound without a restart. + if let Some(index) = self.agents.iter().position(|slot| { + slot.as_ref() + .is_some_and(|agent| agent.state.ifc_domain.is_none()) + }) { + return self.agents[index].take(); + } + + // Pass 4: first idle agent. Isolate mode replaces it before use if its + // binding differs; off/audit mode preserve the historical behavior. let idx = self.agents.iter().position(|slot| slot.is_some()); - idx.map(|i| self.agents[i].take().unwrap()) + idx.and_then(|i| self.agents[i].take()) } /// Return an agent to its slot after a task completes. @@ -677,13 +752,21 @@ impl AgentPool { self.agents.iter().any(|slot| slot.is_some()) } - /// Whether any idle agent already has a session for `channel_id`. - /// Used to compute `affinity_hit` before calling `try_claim`. - pub fn has_session_for(&self, channel_id: Uuid) -> bool { - self.agents.iter().any(|slot| { - slot.as_ref() - .map(|a| a.state.sessions.contains_key(&channel_id)) - .unwrap_or(false) + /// Whether an idle process has either a live channel session or a matching + /// execution-domain binding for this turn. + pub fn has_affinity_for( + &self, + channel_id: Uuid, + domain: Option<&crate::ifc::DomainKey>, + ) -> bool { + self.agents.iter().flatten().any(|agent| { + let domain_matches = + domain.is_none_or(|requested| agent.state.ifc_domain.as_ref() == Some(requested)); + domain_matches + && (agent.state.sessions.contains_key(&channel_id) + || domain.is_some_and(|requested| { + agent.state.ifc_domain.as_ref() == Some(requested) + })) }) } @@ -1468,32 +1551,257 @@ fn send_prompt_result( }); } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum DomainBindingAction { + BindFresh, + Reuse, + Replace, +} + +fn domain_binding_action( + current: Option<&crate::ifc::DomainKey>, + requested: &crate::ifc::DomainKey, +) -> DomainBindingAction { + match current { + None => DomainBindingAction::BindFresh, + Some(current) if current == requested => DomainBindingAction::Reuse, + Some(_) => DomainBindingAction::Replace, + } +} + +/// Bind an ACP child to one complete execution domain before any turn input is +/// delivered. Crossing a domain boundary replaces the child, which clears its +/// model context, ACP sessions, process group, and harness-side session state. +/// Files and credentials outside that process group still require the separate +/// OS-confinement work described by the design. +async fn bind_agent_to_domain( + agent: &mut OwnedAgent, + requested: crate::ifc::DomainKey, + spec: Option<&AgentProcessSpec>, + turn_id: &str, +) -> Result<(), AcpError> { + match domain_binding_action(agent.state.ifc_domain.as_ref(), &requested) { + DomainBindingAction::BindFresh => { + tracing::info!( + target: "buzz_acp::ifc", + ifc_mode = "isolate", + turn_id, + agent_index = agent.index, + domain = %requested.fingerprint(), + rule = "process_binding", + decision = "allow", + action = "bind_fresh", + enforced = true, + "ifc worker routing decision" + ); + agent.state.ifc_domain = Some(requested); + Ok(()) + } + DomainBindingAction::Reuse => { + tracing::info!( + target: "buzz_acp::ifc", + ifc_mode = "isolate", + turn_id, + agent_index = agent.index, + domain = %requested.fingerprint(), + rule = "process_binding", + decision = "allow", + action = "reuse", + enforced = true, + "ifc worker routing decision" + ); + Ok(()) + } + DomainBindingAction::Replace => { + let previous = agent + .state + .ifc_domain + .as_ref() + .map(crate::ifc::DomainKey::fingerprint) + .unwrap_or_else(|| "unbound".to_string()); + tracing::warn!( + target: "buzz_acp::ifc", + ifc_mode = "isolate", + turn_id, + agent_index = agent.index, + previous_domain = %previous, + requested_domain = %requested.fingerprint(), + rule = "process_binding", + decision = "deny", + action = "replace_process", + enforced = true, + "ACP child cannot cross execution-domain boundary" + ); + let spec = spec.ok_or_else(|| { + AcpError::InformationFlow( + "isolate mode has no ACP process replacement configuration".to_string(), + ) + })?; + const REPLACEMENT_TIMEOUT: Duration = Duration::from_secs(60); + let replacement = tokio::time::timeout( + REPLACEMENT_TIMEOUT, + crate::spawn_and_init( + &spec.command, + &spec.args, + &spec.extra_env, + spec.has_generated_codex_config, + agent.index, + spec.observer.clone(), + ), + ) + .await + .map_err(|_| { + AcpError::InformationFlow(format!( + "replacement ACP child did not initialize within {REPLACEMENT_TIMEOUT:?}" + )) + })? + .map_err(|error| AcpError::InformationFlow(error.to_string()))?; + + let (new_acp, protocol_version, agent_name) = replacement; + let mut old_acp = std::mem::replace(&mut agent.acp, new_acp); + old_acp.shutdown().await; + + // Paper: "Confinement invariant." Process replacement is the only + // operation that clears retained domain state. Session rotation is + // intentionally insufficient because the old child may remember. + agent.state = SessionState::default(); + agent.state.ifc_domain = Some(requested.clone()); + agent.model_capabilities = None; + agent.desired_model = spec.configured_model.clone(); + agent.model_overridden = false; + agent.agent_name = agent_name; + agent.goose_system_prompt_supported = None; + agent.protocol_version = protocol_version; + + tracing::info!( + target: "buzz_acp::ifc", + ifc_mode = "isolate", + turn_id, + agent_index = agent.index, + domain = %requested.fingerprint(), + rule = "process_binding", + decision = "allow", + action = "replacement_bound", + enforced = true, + "fresh ACP child bound to execution domain" + ); + Ok(()) + } + } +} + +/// Values that must move together from pool dispatch into one prompt task. +/// Keeping the prepared domain beside the batch prevents the task from routing +/// on one membership snapshot and prompting with another. +pub(crate) struct PromptDispatch { + pub(crate) batch: Option, + pub(crate) prompt_text: Option, + pub(crate) prepared_ifc_turn: Option, + pub(crate) steer_rx: Option>, +} + /// Core async function spawned for each prompt. /// /// Lifecycle: -/// 1. Resolve or create a session (channel or heartbeat). -/// 2. Send `initial_message` on new channel sessions (if configured). -/// 3. Fetch conversation context if needed (thread reply or DM). -/// 4. Build the prompt text from batch + context. -/// 5. Send the actual prompt with turn timeout. +/// 1. Bind or replace the ACP child for the prepared execution domain. +/// 2. Resolve or create a session (channel or heartbeat). +/// 3. Send `initial_message` on new channel sessions (if configured). +/// 4. Fetch conversation context if needed (thread reply or DM). +/// 5. Build and send the prompt with the configured turn timeouts. /// 6. Handle all error paths, always returning the agent via `result_tx`. /// /// The agent is ALWAYS returned — even on panic the `JoinSet` detects the /// abort and the caller uses `task_map` to recover the agent index. -pub async fn run_prompt_task( +pub(crate) async fn run_prompt_task( mut agent: OwnedAgent, - batch: Option, - prompt_text: Option, + dispatch: PromptDispatch, ctx: Arc, result_tx: mpsc::UnboundedSender, control_rx: Option>, turn_id: String, ) { + let PromptDispatch { + batch, + prompt_text, + prepared_ifc_turn, + steer_rx, + } = dispatch; // Is this a channel prompt or a heartbeat? let source = match &batch { Some(b) => PromptSource::Channel(b.channel_id), None => PromptSource::Heartbeat, }; + + // Resolve the complete execution domain before the ACP child receives any + // prompt, memory, session, or tool state. Isolate-mode channel dispatch + // normally prepares this before pool selection so the domain is the routing + // key; audit mode resolves here to preserve the old non-blocking dispatcher. + let ifc_turn = match prepared_ifc_turn { + Some(turn) => Some(turn), + None => match (&ctx.ifc_auditor, batch.as_ref()) { + (Some(auditor), Some(batch)) => Some( + auditor + .resolve_turn(batch, &turn_id, Some(agent.index)) + .await, + ), + (Some(auditor), None) => Some(auditor.resolve_heartbeat(&turn_id, Some(agent.index))), + (None, _) => None, + }, + }; + + if ctx.information_flow.isolates_processes() { + let Some(turn) = ifc_turn.as_ref() else { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(AcpError::InformationFlow( + "execution domain policy is unavailable".to_string(), + )), + requeue_batch_if_queue(&ctx, batch), + ); + return; + }; + let Some(domain) = turn.domain_key() else { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(AcpError::InformationFlow( + "signed execution domain could not be resolved".to_string(), + )), + requeue_batch_if_queue(&ctx, batch), + ); + return; + }; + if let Err(error) = + bind_agent_to_domain(&mut agent, domain, ctx.ifc_process_spec.as_ref(), &turn_id).await + { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(error), + requeue_batch_if_queue(&ctx, batch), + ); + return; + } + } + + if let Some(turn) = ifc_turn.as_ref() { + turn.enter_process(&mut agent.state.ifc_audit); + } + + // Install steering only after domain binding. A replacement ACP child must + // receive this turn's receiver; installing it on the retired child would + // silently disable non-cancelling steering for the first turn in a domain. + if let Some(steer_rx) = steer_rx { + agent.acp.install_steer_rx(steer_rx); + } + let observer_channel_id = match &source { PromptSource::Channel(channel_id) => Some(*channel_id), PromptSource::Heartbeat => None, @@ -1568,20 +1876,6 @@ pub async fn run_prompt_task( .unwrap_or_default(); let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone()); - // IFC is attached at the last point where Buzz events are still typed and - // signed. A standalone ACP proxy sees only rendered prompt strings and - // cannot reconstruct trustworthy requesters, audiences, or membership - // epochs. In audit mode failures are logged and the existing turn proceeds. - let ifc_turn = if let (Some(auditor), Some(batch)) = (&ctx.ifc_auditor, batch.as_ref()) { - Some( - auditor - .begin_turn(batch, &turn_id, agent.index, &mut agent.state.ifc_audit) - .await, - ) - } else { - None - }; - // // Core memory is delivered inside the system prompt the harness already // builds (system role for protocol >= 2, the `[System]` user-message @@ -1607,7 +1901,11 @@ pub async fn run_prompt_task( // `SessionState::invalidate_channel`). // // Operator opt-out: `--no-memory` / `BUZZ_ACP_NO_MEMORY` skips the fetch. - if ctx.memory_enabled { + let owner_memory_allowed = !ctx.information_flow.isolates_processes() + || ifc_turn + .as_ref() + .is_some_and(crate::ifc::ActiveTurn::permits_owner_private_input); + if ctx.memory_enabled && owner_memory_allowed { if let (PromptSource::Channel(cid), Some(owner_pk)) = (&source, ctx.agent_owner_pubkey.as_ref()) { @@ -1644,6 +1942,14 @@ pub async fn run_prompt_task( } } } + } else if ctx.memory_enabled + && ctx.information_flow.isolates_processes() + && matches!(&source, PromptSource::Channel(_)) + && ctx.agent_owner_pubkey.is_some() + { + if let Some(turn) = ifc_turn.as_ref() { + turn.log_blocked_owner_private_input("agent_core_memory"); + } } // Canvas metadata fetch — same lifecycle as core: once per new channel session, @@ -4470,6 +4776,142 @@ mod tests { } } + #[test] + fn domain_binding_requires_replacement_only_across_domains() { + let first = crate::ifc::DomainKey::for_test("first"); + let second = crate::ifc::DomainKey::for_test("second"); + + assert_eq!( + domain_binding_action(None, &first), + DomainBindingAction::BindFresh + ); + assert_eq!( + domain_binding_action(Some(&first), &first), + DomainBindingAction::Reuse + ); + assert_eq!( + domain_binding_action(Some(&first), &second), + DomainBindingAction::Replace + ); + } + + #[tokio::test] + async fn domain_replacement_clears_every_session_state_store() { + let old_acp = AcpClient::spawn( + "bash", + &["-c".into(), "while IFS= read -r line; do :; done".into()], + &[], + false, + ) + .await + .expect("spawn old ACP child"); + let old_domain = crate::ifc::DomainKey::for_test("old-domain"); + let new_domain = crate::ifc::DomainKey::for_test("new-domain"); + let channel = Uuid::new_v4(); + let mut state = SessionState { + ifc_domain: Some(old_domain), + ..SessionState::default() + }; + state.sessions.insert(channel, "old-session".into()); + state.turn_counts.insert(channel, 7); + state.core_sections.insert(channel, "private core".into()); + state + .canvas_sections + .insert(channel, "private canvas".into()); + state + .deliveries + .insert(channel, ChannelDeliveryState::default()); + state.heartbeat_session = Some("old-heartbeat".into()); + let mut agent = OwnedAgent { + index: 0, + acp: old_acp, + state, + model_capabilities: None, + desired_model: Some("old-model".into()), + model_overridden: true, + agent_name: "old-agent".into(), + goose_system_prompt_supported: Some(true), + protocol_version: 1, + }; + let replacement_script = r#"IFS= read -r line +printf '%s\n' '{"jsonrpc":"2.0","id":0,"result":{"protocolVersion":2,"agentInfo":{"name":"replacement-agent"}}}' +while IFS= read -r line; do :; done"#; + let spec = AgentProcessSpec::new( + "bash".into(), + vec!["-c".into(), replacement_script.into()], + vec![], + false, + Some("configured-model".into()), + None, + ); + + bind_agent_to_domain(&mut agent, new_domain.clone(), Some(&spec), "test-turn") + .await + .expect("replace child"); + + assert_eq!(agent.state.ifc_domain.as_ref(), Some(&new_domain)); + assert!(agent.state.sessions.is_empty()); + assert!(agent.state.turn_counts.is_empty()); + assert!(agent.state.core_sections.is_empty()); + assert!(agent.state.canvas_sections.is_empty()); + assert!(agent.state.deliveries.is_empty()); + assert!(agent.state.heartbeat_session.is_none()); + assert_eq!(agent.desired_model.as_deref(), Some("configured-model")); + assert!(!agent.model_overridden); + assert_eq!(agent.agent_name, "replacement-agent"); + assert_eq!(agent.protocol_version, 2); + agent.acp.shutdown().await; + } + + #[tokio::test] + async fn pool_claim_prefers_the_matching_domain_over_slot_order() { + let spawn_inert = || async { + AcpClient::spawn( + "bash", + &["-c".into(), "while IFS= read -r line; do :; done".into()], + &[], + false, + ) + .await + .expect("spawn inert ACP child") + }; + let first_domain = crate::ifc::DomainKey::for_test("first-domain"); + let requested_domain = crate::ifc::DomainKey::for_test("requested-domain"); + let make_agent = |index, acp, domain| OwnedAgent { + index, + acp, + state: SessionState { + ifc_domain: Some(domain), + ..SessionState::default() + }, + model_capabilities: None, + desired_model: None, + model_overridden: false, + agent_name: "test-agent".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + }; + let channel = Uuid::new_v4(); + let mut first = make_agent(0, spawn_inert().await, first_domain); + // A stale session from an older membership epoch must not outrank the + // freshly resolved domain. + first.state.sessions.insert(channel, "stale-session".into()); + let second = make_agent(1, spawn_inert().await, requested_domain.clone()); + let mut pool = AgentPool::from_slots(vec![Some(first), Some(second)]); + + let mut claimed = pool + .try_claim(Some(channel), Some(&requested_domain)) + .expect("matching process"); + assert_eq!(claimed.index, 1); + + claimed.acp.shutdown().await; + for slot in pool.agents_mut() { + if let Some(mut agent) = slot.take() { + agent.acp.shutdown().await; + } + } + } + #[test] fn public_session_forwards_channel_origin_to_mcp() { let channel_id = Uuid::new_v4(); @@ -5724,8 +6166,12 @@ done"# for turn in 1..=3 { run_prompt_task( agent, - None, - Some(format!("heartbeat-{turn}")), + PromptDispatch { + batch: None, + prompt_text: Some(format!("heartbeat-{turn}")), + prepared_ifc_turn: None, + steer_rx: None, + }, Arc::clone(&ctx), result_tx.clone(), None, @@ -5839,8 +6285,12 @@ done"# }; run_prompt_task( agent, - Some(batch), - None, + PromptDispatch { + batch: Some(batch), + prompt_text: None, + prepared_ifc_turn: None, + steer_rx: None, + }, Arc::clone(&ctx), result_tx.clone(), None, @@ -6014,8 +6464,12 @@ done"# for (turn_id, batch) in [("merged-turn", merged_batch), ("next-turn", next_batch)] { run_prompt_task( agent, - Some(batch), - None, + PromptDispatch { + batch: Some(batch), + prompt_text: None, + prepared_ifc_turn: None, + steer_rx: None, + }, Arc::clone(&ctx), result_tx.clone(), None, @@ -6148,7 +6602,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" "live-session".into(), )); let agent = pool - .try_claim(Some(channel_id)) + .try_claim(Some(channel_id), None) .expect("claim returned agent"); let mut ctx = make_prompt_context_no_owner(); @@ -6173,8 +6627,12 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let (result_tx, mut result_rx) = mpsc::unbounded_channel(); run_prompt_task( agent, - Some(batch), - None, + PromptDispatch { + batch: Some(batch), + prompt_text: None, + prepared_ifc_turn: None, + steer_rx: None, + }, Arc::new(ctx), result_tx, None, @@ -7655,6 +8113,8 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" max_turns_per_session: 0, permission_mode: PermissionMode::Default, ifc_auditor: None, + information_flow: InformationFlowMode::Off, + ifc_process_spec: None, agent_keys: agent_keys.clone(), agent_owner_pubkey: owner_pubkey, memory_enabled: false,