Enforce execution-domain ACP worker isolation

Signed-off-by: Jordan Mecom <jm@squareup.com>

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
 

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 },
 }

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]

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<PublicKey>,
     agent: Principal,
     owner: Option<Principal>,
+    mode: InformationFlowMode,
 }
 
 impl Auditor {
@@ -617,6 +645,7 @@ impl Auditor {
         relay_self: Option<&str>,
         agent: PublicKey,
         owner: Option<PublicKey>,
+        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<usize>,
     ) -> 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<usize>,
+    ) -> 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<usize>,
     domain: Option<ExecutionDomain>,
     owner: Option<Principal>,
+    mode: InformationFlowMode,
+    origin: TurnOrigin,
 }
 
 impl ActiveTurn {
-    fn unresolved(turn_id: &str, agent_index: usize, owner: Option<Principal>) -> Self {
+    fn unresolved(
+        turn_id: &str,
+        agent_index: Option<usize>,
+        owner: Option<Principal>,
+        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<DomainKey> {
+        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<usize>,
+    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));

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<PromptContext>,
@@ -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::<pool::SteerRequest>(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::<ControlSignal>();
-        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)],

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<crate::ifc::DomainKey>,
 }
 
 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<String>,
+    extra_env: Vec<(String, String)>,
+    has_generated_codex_config: bool,
+    configured_model: Option<String>,
+    observer: Option<observer::ObserverHandle>,
+}
+
+impl AgentProcessSpec {
+    pub(crate) fn new(
+        command: String,
+        args: Vec<String>,
+        extra_env: Vec<(String, String)>,
+        has_generated_codex_config: bool,
+        configured_model: Option<String>,
+        observer: Option<observer::ObserverHandle>,
+    ) -> Self {
+        Self {
+            command,
+            args,
+            extra_env,
+            has_generated_codex_config,
+            configured_model,
+            observer,
+        }
+    }
+}
+
 pub struct PromptContext {
     pub mcp_servers: Vec<McpServer>,
     pub initial_message: Option<String>,
@@ -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<crate::ifc::Auditor>,
+    /// 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<AgentProcessSpec>,
     /// 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<Uuid>) -> Option<OwnedAgent> {
+    pub fn try_claim(
+        &mut self,
+        channel_id: Option<Uuid>,
+        domain: Option<&crate::ifc::DomainKey>,
+    ) -> Option<OwnedAgent> {
         // 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<FlushBatch>,
+    pub(crate) prompt_text: Option<String>,
+    pub(crate) prepared_ifc_turn: Option<crate::ifc::ActiveTurn>,
+    pub(crate) steer_rx: Option<tokio::sync::mpsc::Receiver<SteerRequest>>,
+}
+
 /// 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<FlushBatch>,
-    prompt_text: Option<String>,
+    dispatch: PromptDispatch,
     ctx: Arc<PromptContext>,
     result_tx: mpsc::UnboundedSender<PromptResult>,
     control_rx: Option<tokio::sync::oneshot::Receiver<ControlSignal>>,
     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,