diff --git a/crates/imcp2-core/src/calls.rs b/crates/imcp2-core/src/calls.rs
index 796eaca..7776b24 100644
--- a/crates/imcp2-core/src/calls.rs
+++ b/crates/imcp2-core/src/calls.rs
@@ -643,7 +643,7 @@ pub(crate) fn on_deep_stack<T: Send>(f: impl FnOnce() -> T + Send) -> Option<T>
 /// — Candid lexes `type` as a keyword, never an identifier, so it cannot appear as
 /// a field name (`record { type : nat }` is a syntax error); the quoted form
 /// `record { "type" : nat }` is a string literal, which the scan already skips.
-pub(crate) fn guard_candid_text(what: &str, text: &str) -> Result<(), String> {
+pub fn guard_candid_text(what: &str, text: &str) -> Result<(), String> {
     if text.len() > MAX_CANDID_TEXT_BYTES {
         return Err(format!(
             "{what} is too large to parse ({} bytes; limit {MAX_CANDID_TEXT_BYTES})",
diff --git a/crates/imcp2-core/src/lib.rs b/crates/imcp2-core/src/lib.rs
index e7e26d6..c56e2e8 100644
--- a/crates/imcp2-core/src/lib.rs
+++ b/crates/imcp2-core/src/lib.rs
@@ -49,6 +49,12 @@ mod management;
 #[cfg(all(test, feature = "e2e"))]
 mod e2e_canister_tools;
 
+/// The bound on caller-supplied textual Candid (size, nesting depth, type
+/// declarations) that the call tools apply before parsing, for an embedder
+/// that hands Candid text from an untrusted source to a parser of its own
+/// (the local server's deploy tools, with a bundle's arguments and
+/// interfaces).
+pub use calls::guard_candid_text;
 /// The IC [`Agent`] type the components are built around, re-exported so
 /// callers construct the injected agent from the exact `ic-agent` version this
 /// crate links.
diff --git a/crates/imcp2-deploy/Cargo.toml b/crates/imcp2-deploy/Cargo.toml
index c260a6d..fe5bfc3 100644
--- a/crates/imcp2-deploy/Cargo.toml
+++ b/crates/imcp2-deploy/Cargo.toml
@@ -22,7 +22,7 @@ ic-agent = { workspace = true }
 candid = { workspace = true }
 # `rt-multi-thread`: the sync-plugin runtime runs a wasm component on a
 # blocking thread (`block_in_place`), which only a multi-thread runtime has.
-tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "fs"] }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "fs", "io-util"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tracing = { workspace = true }
diff --git a/crates/imcp2-deploy/src/calls.rs b/crates/imcp2-deploy/src/calls.rs
new file mode 100644
index 0000000..840f3d5
--- /dev/null
+++ b/crates/imcp2-deploy/src/calls.rs
@@ -0,0 +1,365 @@
+//! The project library's one seam to the network, over this connection's
+//! `ic-agent`: [`CanisterCalls`] signed as the user's management identity.
+//!
+//! Modelled on the icp CLI's own implementation (`icp_app::calls::AgentCalls`),
+//! minus proxy-canister routing: this server never names a proxy, so every call
+//! is made directly by the identity the agent carries. The certified reads are
+//! what the deploy operation decides on — whether a canister exists, what is
+//! installed on it, which subnet it sits on — so they are answered from
+//! certificates rather than from anything a canister says about itself.
+
+use async_trait::async_trait;
+use candid::Principal;
+use ic_agent::{
+    agent::{CallResponse, EffectiveId, SubnetType},
+    hash_tree::{HashTree, Label, LookupResult},
+    Agent, AgentError,
+};
+use icp_project::calls::{Authority, Call, CallError, CanisterCalls, RouteTo};
+use imcp2_core::guard_candid_text;
+
+/// The metadata section the upgrade compatibility check reads from the
+/// deployed canister and parses as Candid.
+const CANDID_SERVICE: &str = "candid:service";
+
+/// [`CanisterCalls`] over an `ic-agent`.
+pub struct AgentCalls {
+    agent: Agent,
+    /// The agent's principal, read once here. [`CanisterCalls::caller`] is
+    /// infallible while [`Agent::get_principal`] is not (an identity may
+    /// decline to name its sender), so the one place the failure can be
+    /// reported is construction; the agent's identity cannot change after it.
+    caller: Principal,
+}
+
+impl AgentCalls {
+    pub fn new(agent: Agent) -> Result<Self, String> {
+        let caller = agent.get_principal()?;
+        Ok(Self { agent, caller })
+    }
+
+    /// Turn an agent error into the shape the trait speaks in: a rejection is
+    /// a verdict and keeps its code, a call that ran out of time or never got
+    /// down the wire went unanswered, and anything else — a reply that would
+    /// not parse, a certificate that would not verify — failed for good.
+    fn wrap(canister: Principal, method: &str, err: AgentError) -> CallError {
+        match &err {
+            AgentError::CertifiedReject { reject, .. }
+            | AgentError::UncertifiedReject { reject, .. } => CallError::Rejected {
+                canister,
+                method: method.to_owned(),
+                code: reject.error_code.clone(),
+                message: reject.reject_message.clone(),
+            },
+            AgentError::TimeoutWaitingForResponse() | AgentError::TransportError(_) => {
+                CallError::unanswered(canister, method, err)
+            }
+            _ => CallError::failed(canister, method, err),
+        }
+    }
+
+    /// Whether `canister` exists, which its controllers answer — see
+    /// [`CanisterCalls::controllers`]. A check that could not be made reads as
+    /// "does not exist": every caller asks this while deciding whether to
+    /// report a read failure of its own, and an existence check that failed
+    /// too is no reason to suppress it.
+    async fn exists(&self, canister: Principal) -> bool {
+        self.controllers(canister).await.is_ok_and(|controllers| controllers.is_some())
+    }
+
+    /// Ask the target's subnet to certify a metadata section, reporting only
+    /// what the certificate proves — see [`certified_section`] for what that
+    /// is.
+    async fn certified_metadata_section(
+        &self,
+        canister: Principal,
+        path: &str,
+    ) -> Result<Option<Vec<u8>>, CallError> {
+        let (metadata_path, controllers_path) = section_paths(canister, path);
+        let cert = self
+            .agent
+            .read_state_raw(vec![metadata_path, controllers_path], canister)
+            .await
+            .map_err(|err| Self::wrap(canister, READ_STATE_METADATA, err))?;
+        certified_section(canister, path, &cert.tree)
+    }
+
+    /// A subnet-scoped update: routed to a subnet rather than to any canister
+    /// on it, which the agent only exposes through a signed submission.
+    async fn to_subnet(&self, subnet: Principal, call: &Call) -> Result<Vec<u8>, CallError> {
+        let effective = EffectiveId::Subnet(subnet);
+        let fail = |e: AgentError| Self::wrap(call.canister, &call.method, e);
+
+        let signed = self
+            .agent
+            .update(&call.canister, &call.method)
+            .with_arg(call.arg.clone())
+            .sign()
+            .map_err(fail)?;
+        let response =
+            self.agent.update_signed(effective, signed.signed_update).await.map_err(fail)?;
+        match response {
+            CallResponse::Response(bytes) => Ok(bytes),
+            CallResponse::Poll(request_id) => {
+                let status = self.agent.sign_request_status(effective, request_id).map_err(fail)?;
+                Ok(self
+                    .agent
+                    .wait_signed(&request_id, effective, status.signed_request_status)
+                    .await
+                    .map_err(fail)?
+                    .0)
+            }
+        }
+    }
+}
+
+/// The method name a certified metadata read reports itself under.
+const READ_STATE_METADATA: &str = "read_state(metadata)";
+
+/// A path into the state tree, label by label.
+type StatePath = Vec<Label<Vec<u8>>>;
+
+/// The two state-tree paths a metadata read asks for: the section, and the
+/// canister's controllers (see [`certified_section`] for why both).
+fn section_paths(canister: Principal, path: &str) -> (StatePath, StatePath) {
+    let canister_path = || vec![Label::from("canister"), Label::from_bytes(canister.as_slice())];
+    let mut metadata = canister_path();
+    metadata.extend(["metadata".into(), path.into()]);
+    let mut controllers = canister_path();
+    controllers.push("controllers".into());
+    (metadata, controllers)
+}
+
+/// What a certificate proves about section `path` of `canister`, and nothing
+/// more: `Some` for a section it holds, `None` for a canister it proves has
+/// no such section, and an error for a canister it proves absent or a path
+/// it says nothing about.
+///
+/// The section is read together with `controllers`, because a metadata path
+/// proven absent is equally what a canister that was never created looks
+/// like — `controllers` is written at creation, so its presence is what
+/// separates the two. A canister with no module installed has no sections
+/// at all, which the certificate reports as an absent path under a canister
+/// that exists, and so as `Ok(None)`.
+fn certified_section(
+    canister: Principal,
+    path: &str,
+    tree: &HashTree<Vec<u8>>,
+) -> Result<Option<Vec<u8>>, CallError> {
+    let (metadata_path, controllers_path) = section_paths(canister, path);
+    let rejected = |message: String| CallError::Rejected {
+        canister,
+        method: READ_STATE_METADATA.to_owned(),
+        code: None,
+        message,
+    };
+    let unproven =
+        |about: String| Err(rejected(format!("the certificate proves nothing about {about}")));
+    match tree.lookup_path(&metadata_path) {
+        LookupResult::Found(bytes) => Ok(Some(bytes.to_vec())),
+        LookupResult::Absent => match tree.lookup_path(&controllers_path) {
+            LookupResult::Found(_) => Ok(None),
+            LookupResult::Absent => Err(rejected(format!("canister {canister} does not exist"))),
+            _ => unproven(format!("canister {canister}")),
+        },
+        // Not proof of absence, just a certificate that says nothing about
+        // the path — reporting the section missing off this would be a
+        // guess, and a private section is exactly what it looks like.
+        _ => unproven(format!("section `{path}` of canister {canister}")),
+    }
+}
+
+#[async_trait]
+impl CanisterCalls for AgentCalls {
+    fn caller(&self) -> Principal {
+        self.caller
+    }
+
+    async fn update(&self, call: Call) -> Result<Vec<u8>, CallError> {
+        if let RouteTo::Subnet(subnet) = call.route {
+            return self.to_subnet(subnet, &call).await;
+        }
+        let mut builder =
+            self.agent.update(&call.canister, &call.method).with_arg(call.arg.clone());
+        if let RouteTo::Canister(effective) = call.route {
+            builder = builder.with_effective_canister_id(effective);
+        }
+        builder.await.map_err(|e| Self::wrap(call.canister, &call.method, e))
+    }
+
+    async fn query(&self, call: Call) -> Result<Vec<u8>, CallError> {
+        let mut builder = self.agent.query(&call.canister, &call.method).with_arg(call.arg.clone());
+        if let RouteTo::Canister(effective) = call.route {
+            builder = builder.with_effective_canister_id(effective);
+        }
+        builder.call().await.map_err(|e| Self::wrap(call.canister, &call.method, e))
+    }
+
+    /// A `candid:service` section is parsed by the library's compatibility
+    /// check, on the worker thread and without a depth guard, and the
+    /// canister it comes from is whichever the caller's mapping named: so
+    /// the interface is held to the same bound as the bundle's own before
+    /// it is handed back ([`guard_candid_text`]). The check treats a
+    /// failed read as "cannot tell" and skips, which is the right outcome
+    /// for an interface too deep to compare.
+    async fn metadata_section(
+        &self,
+        canister: Principal,
+        path: &str,
+        // With no proxy there is one authority, and both values mean it.
+        _authority: Authority,
+    ) -> Result<Option<Vec<u8>>, CallError> {
+        let section = self.certified_metadata_section(canister, path).await?;
+        if path == CANDID_SERVICE {
+            if let Some(bytes) = &section {
+                let text = String::from_utf8_lossy(bytes);
+                guard_candid_text(&format!("the `{path}` section of {canister}"), &text).map_err(
+                    |why| CallError::Rejected {
+                        canister,
+                        method: "read_state(metadata)".to_owned(),
+                        code: None,
+                        message: why,
+                    },
+                )?;
+            }
+        }
+        Ok(section)
+    }
+
+    async fn controllers(&self, canister: Principal) -> Result<Option<Vec<Principal>>, CallError> {
+        match self.agent.read_state_canister_controllers(canister).await {
+            Ok(controllers) => Ok(Some(controllers)),
+            Err(AgentError::LookupPathAbsent(_)) => Ok(None),
+            Err(err) => Err(Self::wrap(canister, "read_state(controllers)", err)),
+        }
+    }
+
+    async fn module_hash(&self, canister: Principal) -> Result<Option<Vec<u8>>, CallError> {
+        let err = match self.agent.read_state_canister_module_hash(canister).await {
+            Ok(hash) => return Ok(Some(hash)),
+            Err(err) => err,
+        };
+        // No module-hash path means nothing is installed — or that there is no
+        // canister to install into, which is not an answer this can give.
+        if matches!(err, AgentError::LookupPathAbsent(_)) && self.exists(canister).await {
+            return Ok(None);
+        }
+        Err(Self::wrap(canister, "read_state(module_hash)", err))
+    }
+
+    async fn subnet_of(&self, canister: Principal) -> Result<Principal, CallError> {
+        Ok(self
+            .agent
+            .get_subnet_by_canister(&canister)
+            .await
+            .map_err(|e| Self::wrap(canister, "subnet_of", e))?
+            .id())
+    }
+
+    async fn subnet_uses_engine_operator(&self, subnet: Principal) -> Result<bool, CallError> {
+        let info = self
+            .agent
+            .get_subnet_by_id(&subnet)
+            .await
+            .map_err(|e| Self::wrap(subnet, "subnet_type", e))?;
+        Ok(matches!(info.subnet_type(), Some(SubnetType::CloudEngine)))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use ic_agent::hash_tree::{fork, label, leaf, pruned};
+
+    const PATH: &str = "candid:service";
+
+    fn canister() -> Principal {
+        Principal::from_text("rrkah-fqaaa-aaaaa-aaaaq-cai").unwrap()
+    }
+
+    /// A state tree holding `node` under this canister's entry.
+    fn under_canister(node: HashTree<Vec<u8>>) -> HashTree<Vec<u8>> {
+        label("canister", label(Label::from_bytes(canister().as_slice()), node))
+    }
+
+    /// The `controllers` entry creation writes; its bytes are not read here.
+    fn controllers() -> HashTree<Vec<u8>> {
+        label("controllers", leaf(b"\x81\x41\x01".to_vec()))
+    }
+
+    fn rejection(result: Result<Option<Vec<u8>>, CallError>) -> String {
+        match result {
+            Err(CallError::Rejected { message, .. }) => message,
+            other => panic!("expected a rejection, got {other:?}"),
+        }
+    }
+
+    // A section the certificate holds is what the check parses.
+    #[test]
+    fn a_section_the_certificate_holds_is_returned() {
+        let tree = under_canister(fork(
+            controllers(),
+            label("metadata", label(PATH, leaf(b"service : {}".to_vec()))),
+        ));
+        assert_eq!(
+            certified_section(canister(), PATH, &tree).unwrap(),
+            Some(b"service : {}".to_vec())
+        );
+    }
+
+    // A canister that exists — its controllers are proven — with no such
+    // section: none installed yet, or a module without one. Either reads as
+    // "no section", not as "no canister".
+    #[test]
+    fn a_canister_proven_to_exist_without_the_section_has_none() {
+        let no_module = under_canister(controllers());
+        assert_eq!(certified_section(canister(), PATH, &no_module).unwrap(), None);
+        let other_section = under_canister(fork(
+            controllers(),
+            label("metadata", label("other", leaf(b"x".to_vec()))),
+        ));
+        assert_eq!(certified_section(canister(), PATH, &other_section).unwrap(), None);
+    }
+
+    // The tree proves another canister's entry and no entry for this one:
+    // the section is absent because the canister is.
+    #[test]
+    fn a_canister_the_certificate_proves_absent_does_not_exist() {
+        let other = Principal::from_text("ryjl3-tyaaa-aaaaa-aaaba-cai").unwrap();
+        let tree = label("canister", label(Label::from_bytes(other.as_slice()), controllers()));
+        let message = rejection(certified_section(canister(), PATH, &tree));
+        assert_eq!(message, format!("canister {} does not exist", canister()));
+    }
+
+    // A pruned subtree where the section would be proves neither presence
+    // nor absence — a private section looks exactly like this — so no
+    // answer is given off it.
+    #[test]
+    fn a_pruned_section_proves_nothing() {
+        let tree = under_canister(fork(controllers(), pruned([0u8; 32])));
+        let message = rejection(certified_section(canister(), PATH, &tree));
+        assert_eq!(
+            message,
+            format!(
+                "the certificate proves nothing about section `{PATH}` of canister {}",
+                canister()
+            )
+        );
+    }
+
+    // The section is proven absent but the controllers are pruned: whether
+    // the canister exists is what the certificate does not say, so neither
+    // "no section" nor "no canister" is reported.
+    #[test]
+    fn a_pruned_controllers_entry_proves_nothing_about_the_canister() {
+        let tree = under_canister(fork(
+            pruned([0u8; 32]),
+            label("metadata", label("other", leaf(b"x".to_vec()))),
+        ));
+        let message = rejection(certified_section(canister(), PATH, &tree));
+        assert_eq!(
+            message,
+            format!("the certificate proves nothing about canister {}", canister())
+        );
+    }
+}
diff --git a/crates/imcp2-deploy/src/deploy.rs b/crates/imcp2-deploy/src/deploy.rs
new file mode 100644
index 0000000..723f12a
--- /dev/null
+++ b/crates/imcp2-deploy/src/deploy.rs
@@ -0,0 +1,875 @@
+//! One deploy, start to finish: extract the bundle, load its project, seed the
+//! id store, run the project library's `deploy` operation as the user, and
+//! read the resulting ids back — reporting what happened either way.
+//!
+//! The operation is the icp CLI's own (`icp deploy`): build (here: verify the
+//! bundled wasms), create what does not exist, wire canister ids into each
+//! other's environment variables, apply settings, check Candid compatibility
+//! on upgrades, install, sync. Nothing is re-implemented; this module only
+//! supplies what the CLI's command layer would — the host, the signing
+//! identity, the parameters — and turns the progress stream into a log.
+
+use std::{
+    collections::{BTreeMap, HashMap},
+    fmt,
+    sync::Arc,
+};
+
+use camino::{Utf8Path, Utf8PathBuf};
+use candid::Principal;
+use ic_agent::{Agent, Identity};
+use icp_canister_interfaces::engine_canister::engine_canister_id;
+use icp_events::{EventKind, TaskId, TaskOutcome};
+use icp_project::{
+    calls::CanisterCalls,
+    defer::Deferred,
+    host::EnvironmentSelection,
+    operations::{
+        deploy::{deploy, resolve_targets, DeployParams, DeployReport},
+        task::Task,
+    },
+};
+
+use crate::{calls::AgentCalls, host};
+
+/// Where the bundle bytes come from.
+pub enum BundleSource {
+    /// The archive itself.
+    Bytes(Vec<u8>),
+    /// An absolute path to the archive on this machine.
+    Path(Utf8PathBuf),
+}
+
+// By hand so a request never prints megabytes of archive.
+impl fmt::Debug for BundleSource {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            BundleSource::Bytes(bytes) => write!(f, "Bytes({} bytes)", bytes.len()),
+            BundleSource::Path(path) => write!(f, "Path({path})"),
+        }
+    }
+}
+
+/// The install modes `icp deploy --mode` takes, as the library spells them.
+pub const INSTALL_MODES: [&str; 4] = ["auto", "install", "reinstall", "upgrade"];
+
+/// Everything one deploy is given.
+#[derive(Debug)]
+pub struct DeployRequest {
+    pub bundle: BundleSource,
+    /// The environment, by name, among those the bundle's manifest declares.
+    pub environment: String,
+    /// Canister ids to start from: a canister named here is upgraded in place
+    /// instead of created. Each name must be a canister of the environment.
+    pub canister_ids: BTreeMap<String, Principal>,
+    /// One of [`INSTALL_MODES`]: `auto`, `install`, `reinstall` or `upgrade`.
+    /// Checked at entry: the library resolves the mode only once it installs,
+    /// after the canisters have been created, and it panics on any other
+    /// spelling rather than failing the deploy.
+    pub mode: String,
+    /// Cycles to fund each canister this deploy creates.
+    pub cycles: u128,
+    /// Subnet to create canisters on; the registry picks one otherwise.
+    pub subnet: Option<Principal>,
+    /// Refuse, rather than create, when a canister does not exist yet.
+    pub no_create: bool,
+    /// Skip the Candid interface compatibility check on upgrades.
+    pub skip_candid_check: bool,
+}
+
+/// How a deploy ended.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Status {
+    /// Every phase succeeded.
+    Deployed,
+    /// A phase failed; `created` and `canister_ids` still hold whatever landed
+    /// before it did.
+    Failed,
+    /// The environment lists no canisters, so there was nothing to do.
+    NothingToDeploy,
+}
+
+impl Status {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Status::Deployed => "deployed",
+            Status::Failed => "failed",
+            Status::NothingToDeploy => "nothing_to_deploy",
+        }
+    }
+}
+
+/// What a deploy did. Filled in as the run progresses rather than returned at
+/// the end, so a canister created before a later phase failed is still
+/// reported — its id is what the caller needs to pick the run back up.
+#[derive(Debug)]
+pub struct DeployOutcome {
+    pub status: Status,
+    pub environment: String,
+    /// The network the environment targets.
+    pub network: String,
+    /// The canisters the deploy targeted, in manifest order.
+    pub canisters: Vec<String>,
+    /// Canisters created by this deploy, in creation order.
+    pub created: Vec<(String, Principal)>,
+    /// Every canister of the environment that has an id after the deploy —
+    /// the ids passed in, the ids created, both.
+    pub canister_ids: BTreeMap<String, Principal>,
+    /// Why the deploy failed, with its cause chain, when it did.
+    pub error: Option<String>,
+    /// The phases and per-canister tasks the operation reported, with their
+    /// outcomes, in completion order.
+    pub log: Vec<String>,
+}
+
+/// An error made before the deploy could begin: the request itself was
+/// unusable, so there is no outcome to report, only the reason.
+#[derive(Debug)]
+pub struct RequestError(pub String);
+
+impl fmt::Display for RequestError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(&self.0)
+    }
+}
+
+impl std::error::Error for RequestError {}
+
+/// Render an error with its `source()` chain, outermost first: the project
+/// library's errors say *what* failed at each layer and leave the *why* to
+/// the cause beneath, so only the whole chain reads as an explanation.
+pub fn error_chain(error: &(dyn std::error::Error + 'static)) -> String {
+    let mut text = error.to_string();
+    let mut source = error.source();
+    while let Some(cause) = source {
+        text.push_str(": ");
+        text.push_str(&cause.to_string());
+        source = cause.source();
+    }
+    text
+}
+
+/// Run one deploy as `identity`.
+///
+/// `Err` means the request never got as far as a deploy (an unreadable
+/// bundle, an environment the manifest does not declare, a canister name the
+/// id mapping has that the environment does not); `Ok` carries the outcome of
+/// one that ran, failed or not.
+pub async fn deploy_bundle(
+    identity: Arc<dyn Identity>,
+    request: DeployRequest,
+) -> Result<DeployOutcome, RequestError> {
+    if !INSTALL_MODES.contains(&request.mode.as_str()) {
+        return Err(RequestError(format!(
+            "mode must be one of {}, got '{}'",
+            INSTALL_MODES.map(|m| format!("'{m}'")).join(", "),
+            request.mode
+        )));
+    }
+    let bytes = match request.bundle {
+        BundleSource::Bytes(bytes) => bytes,
+        BundleSource::Path(path) => read_bundle_file(&path, host::MAX_BUNDLE_BYTES).await?,
+    };
+
+    // The scratch directory lives exactly as long as this function: the
+    // bundle's files, the ids the deploy records, and the build artifacts all
+    // go with it, whichever way the deploy ends.
+    let scratch = tempfile::Builder::new()
+        .prefix("imcp2-deploy-")
+        .tempdir()
+        .map_err(|e| RequestError(format!("cannot create a scratch directory: {e}")))?;
+    let dir = Utf8PathBuf::from_path_buf(scratch.path().to_path_buf()).map_err(|p| {
+        RequestError(format!("scratch directory path is not UTF-8: {}", p.display()))
+    })?;
+    let fs = {
+        let dir = dir.clone();
+        tokio::task::spawn_blocking(move || host::extract_bundle(&bytes, &dir))
+            .await
+            .map_err(|e| RequestError(format!("bundle extraction panicked: {e}")))?
+            .map_err(RequestError)?
+    };
+
+    let host = host::host_for(fs.clone());
+    // The load is where the untrusted rule holds (a script step, a module or
+    // plugin fetched from a URL or not a regular file under the root, a
+    // recipe: the project does not load), so it is the first thing a bundle outside the
+    // rule fails. It reads the manifest and, through the metered view, every
+    // file a canister names, with the filesystem's own blocking reads: like
+    // extraction it runs off the runtime's workers, and the loader keeps
+    // what it loaded for every later ask.
+    {
+        let project = host.project.clone();
+        let handle = tokio::runtime::Handle::current();
+        tokio::task::spawn_blocking(move || handle.block_on(project.load()))
+            .await
+            .map_err(|e| RequestError(format!("project load panicked: {e}")))?
+            .map_err(|e| RequestError(error_chain(&e)))?;
+    }
+    let selection = EnvironmentSelection::Named(request.environment.clone());
+    let env = host.get_environment(&selection).await.map_err(|e| RequestError(error_chain(&e)))?;
+    // Before any phase runs: a bundle that is not one — a wasm that is
+    // missing or does not hash to its checksum, a network other than the IC
+    // — is refused here, not at the phase that would reach it after
+    // canisters have been created. The check reads, hashes and gunzips every
+    // module, up to a gigabyte of them, so like extraction it runs off the
+    // runtime's workers.
+    {
+        let fs = fs.clone();
+        let network = env.network.clone();
+        let canisters: Vec<_> = env.canisters.values().cloned().collect();
+        tokio::task::spawn_blocking(move || {
+            host::check_bundle(
+                &fs,
+                &network,
+                canisters.iter().map(|(dir, canister)| (dir.as_path(), canister)),
+            )
+        })
+        .await
+        .map_err(|e| RequestError(format!("bundle check panicked: {e}")))?
+        .map_err(RequestError)?;
+    }
+
+    // Seed the id store with the mapping the caller passed: from here on the
+    // operation sees those canisters as existing, exactly as it would see the
+    // ids a previous `icp deploy` recorded in the project's `.icp` directory.
+    for (name, id) in &request.canister_ids {
+        host.set_canister_id_for_env(name, *id, &selection)
+            .await
+            .map_err(|e| RequestError(error_chain(&e)))?;
+    }
+
+    let canisters =
+        resolve_targets(&host, &selection, &[]).await.map_err(|e| RequestError(error_chain(&e)))?;
+
+    let mut outcome = DeployOutcome {
+        status: Status::NothingToDeploy,
+        environment: request.environment.clone(),
+        network: env.network.name.clone(),
+        canisters: canisters.clone(),
+        created: Vec::new(),
+        canister_ids: request.canister_ids.clone(),
+        error: None,
+        log: Vec::new(),
+    };
+    if canisters.is_empty() {
+        return Ok(outcome);
+    }
+
+    let engine_registry = engine_canister_id().map_err(RequestError)?;
+    let params = DeployParams {
+        environment: selection.clone(),
+        canisters,
+        mode: request.mode,
+        subnet: request.subnet,
+        proxy: None,
+        cycles: request.cycles,
+        engine_registry,
+        no_create: request.no_create,
+        yes: request.skip_candid_check,
+        args: None,
+    };
+
+    // The agent is made by the first phase that reaches the network, as the
+    // CLI does: a deploy that fails to load its wasms never resolves a root
+    // key. The identity is already unlocked — it is the session's — so what
+    // is deferred here is only the network resolution.
+    let network = env.network.clone();
+    let host_for_agent = host.clone();
+    let calls = Deferred::new(move || {
+        let host = host_for_agent.clone();
+        let network = network.clone();
+        let identity = identity.clone();
+        async move {
+            let access =
+                host.network.access(&network).await.map_err(|e| RequestError(error_chain(&e)))?;
+            let agent = Agent::builder()
+                .with_url(access.api_url.as_str())
+                .with_arc_identity(identity)
+                .build()
+                .map_err(|e| {
+                    RequestError(format!("cannot build an agent for {}: {e}", access.api_url))
+                })?;
+            agent.set_root_key(access.root_key);
+            let calls: Arc<dyn CanisterCalls> =
+                Arc::new(AgentCalls::new(agent).map_err(RequestError)?);
+            Ok::<_, RequestError>(calls)
+        }
+    });
+
+    // One reporter for the whole deploy; its events are turned into log lines
+    // as they arrive, so a run that fails still reports every phase before it.
+    let (reporter, receiver) = icp_events::channel::<Task>();
+    let collector = tokio::spawn(collect_log(receiver));
+
+    let mut report = DeployReport::default();
+    let result = deploy(&host, &calls, &params, &reporter, &mut report).await;
+    // The receiver ends only once every reporter handle is gone.
+    drop(reporter);
+    outcome.log = collector.await.unwrap_or_default();
+    outcome.created = report.created;
+
+    match result {
+        Ok(()) => outcome.status = Status::Deployed,
+        Err(e) => {
+            outcome.status = Status::Failed;
+            outcome.error = Some(error_chain(&e));
+        }
+    }
+
+    // Whatever the store holds now: the seeded ids plus every id created before
+    // a failure, which is what a retry needs to pass back in.
+    match host.ids_by_environment(&selection).await {
+        Ok(ids) => outcome.canister_ids = ids.into_iter().collect(),
+        Err(e) => {
+            outcome.log.push(format!("reading the canister ids back failed: {}", error_chain(&e)));
+        }
+    }
+    // A created id is reported before it is recorded, so the created list is
+    // the authority: it fills in whatever the store failed to record — or to
+    // read back — because a mapping missing a canister that exists would have
+    // the retry create it again.
+    for (name, id) in &outcome.created {
+        outcome.canister_ids.insert(name.clone(), *id);
+    }
+
+    drop(scratch);
+    Ok(outcome)
+}
+
+/// Read the bundle at `path`, which must be an absolute path to a regular
+/// file of at most `limit` bytes, and nothing else: a device or a pipe reports
+/// a length that says nothing about what a read of it produces (`/dev/zero`
+/// is 0 bytes long and endless; a FIFO blocks on open), so the size check
+/// means something for a regular file only. The path is checked before it
+/// is opened, so a pipe is never opened; the open file is checked again, so
+/// a file replaced or grown between the two is still read to the limit and
+/// no further.
+async fn read_bundle_file(path: &Utf8Path, limit: usize) -> Result<Vec<u8>, RequestError> {
+    use tokio::io::AsyncReadExt;
+    if !path.is_absolute() {
+        return Err(RequestError(format!(
+            "bundle_path must be an absolute path on this machine, got '{path}'"
+        )));
+    }
+    let cannot = |e: std::io::Error| RequestError(format!("cannot read bundle '{path}': {e}"));
+    let too_large = |len: u64| {
+        RequestError(format!("bundle '{path}' is {len} bytes, above the {limit} byte limit"))
+    };
+    let not_regular = || RequestError(format!("bundle '{path}' is not a regular file"));
+    let meta = tokio::fs::metadata(path.as_std_path()).await.map_err(cannot)?;
+    if !meta.is_file() {
+        return Err(not_regular());
+    }
+    if meta.len() > limit as u64 {
+        return Err(too_large(meta.len()));
+    }
+    let file = tokio::fs::File::open(path.as_std_path()).await.map_err(cannot)?;
+    let meta = file.metadata().await.map_err(cannot)?;
+    if !meta.is_file() {
+        return Err(not_regular());
+    }
+    if meta.len() > limit as u64 {
+        return Err(too_large(meta.len()));
+    }
+    // One byte past the limit is enough to tell that the file outgrew it.
+    let mut bytes = Vec::with_capacity(meta.len() as usize);
+    file.take(limit as u64 + 1).read_to_end(&mut bytes).await.map_err(cannot)?;
+    if bytes.len() > limit {
+        return Err(too_large(bytes.len() as u64));
+    }
+    Ok(bytes)
+}
+
+/// Turn the operation's event stream into one line per finished task:
+/// `<what>: ok` / `failed: <why>` / `skipped: <why>`, with a task nested under
+/// a phase indented beneath it. Output lines are not kept — a sync plugin's
+/// retained stderr is the exception, since the operation asked for it to
+/// survive — so the log stays a summary, not a transcript.
+async fn collect_log(
+    mut receiver: tokio::sync::mpsc::UnboundedReceiver<icp_events::Event<Task>>,
+) -> Vec<String> {
+    let mut log = Vec::new();
+    let mut started: HashMap<TaskId, (String, bool)> = HashMap::new();
+    while let Some(event) = receiver.recv().await {
+        match event.kind {
+            EventKind::TaskStarted { parent, task } => {
+                started.insert(event.task_id, (describe(&task), parent.is_some()));
+            }
+            EventKind::TaskCompleted { outcome } => {
+                let (label, nested) = started
+                    .remove(&event.task_id)
+                    .unwrap_or_else(|| (format!("task {:?}", event.task_id), false));
+                let indent = if nested { "  " } else { "" };
+                match outcome {
+                    TaskOutcome::Succeeded { retained_output } => {
+                        log.push(format!("{indent}{label}: ok"));
+                        for line in retained_output {
+                            log.push(format!("{indent}  {line}"));
+                        }
+                    }
+                    TaskOutcome::Failed { message, causes } => {
+                        let mut line = format!("{indent}{label}: failed: {message}");
+                        for cause in causes {
+                            line.push_str(": ");
+                            line.push_str(&cause);
+                        }
+                        log.push(line);
+                    }
+                    TaskOutcome::Skipped { reason } => {
+                        log.push(format!("{indent}{label}: skipped: {reason}"));
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+    log
+}
+
+/// A task's one-line description, read off its serialized form so this stays
+/// independent of the task vocabulary: a phase is its title, anything else is
+/// its kind and the canister it acts on.
+fn describe(task: &Task) -> String {
+    let value = serde_json::to_value(task).unwrap_or_default();
+    if let Some(title) = value.get("title").and_then(|v| v.as_str()) {
+        return title.trim_end_matches(':').to_string();
+    }
+    let kind = value.get("kind").and_then(|v| v.as_str()).unwrap_or("task");
+    let canister = value.get("canister").and_then(|v| v.as_str());
+    let id = value.get("canister_id").and_then(|v| v.as_str());
+    match (canister, id) {
+        (Some(c), Some(id)) => format!("{kind} {c} ({id})"),
+        (Some(c), None) => format!("{kind} {c}"),
+        _ => kind.to_string(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use ic_agent::identity::AnonymousIdentity;
+    use sha2::{Digest, Sha256};
+    use std::io::Write;
+
+    /// A bundle as `icp project bundle` writes one: the rewritten manifest
+    /// (every canister a pre-built step naming a bundled wasm by checksum) at
+    /// the archive root, the wasm under `canisters/`. The `ic` and `local`
+    /// networks and environments are implicit, as in every manifest.
+    fn bundle(sha256: &str, wasm: &[u8]) -> Vec<u8> {
+        let manifest = format!(
+            "canisters:\n  - name: backend\n    build:\n      steps:\n        - type: pre-built\n          \
+             path: canisters/backend.wasm\n          sha256: {sha256}\n"
+        );
+        archive(&[("icp.yaml", manifest.as_bytes()), ("canisters/backend.wasm", wasm)])
+    }
+
+    fn archive(entries: &[(&str, &[u8])]) -> Vec<u8> {
+        let mut archive = tar::Builder::new(flate2::write::GzEncoder::new(
+            Vec::new(),
+            flate2::Compression::fast(),
+        ));
+        for (path, data) in entries {
+            let mut header = tar::Header::new_gnu();
+            header.set_size(data.len() as u64);
+            header.set_mode(0o644);
+            header.set_cksum();
+            archive.append_data(&mut header, path, *data).unwrap();
+        }
+        let mut encoder = archive.into_inner().unwrap();
+        encoder.flush().unwrap();
+        encoder.finish().unwrap()
+    }
+
+    fn request(bytes: Vec<u8>, environment: &str) -> DeployRequest {
+        DeployRequest {
+            bundle: BundleSource::Bytes(bytes),
+            environment: environment.into(),
+            canister_ids: BTreeMap::new(),
+            mode: "auto".into(),
+            cycles: 1,
+            subnet: None,
+            // Every test here stays off the network: with nothing created,
+            // the deploy stops at the create phase, after the build phase has
+            // verified the bundled wasm.
+            no_create: true,
+            skip_candid_check: false,
+        }
+    }
+
+    fn wasm() -> (Vec<u8>, String) {
+        let wasm = wat::parse_str("(module)").unwrap();
+        let sha = hex::encode(Sha256::digest(&wasm));
+        (wasm, sha)
+    }
+
+    // The whole pipeline up to the network: extract, load the bundled
+    // manifest through the project library, resolve the implicit `ic`
+    // environment, verify the wasm against its recorded checksum (the build
+    // phase of a bundle), then stop where `--no-create` stops `icp deploy`.
+    // The outcome reports the failure, the targets, and the log of the phase
+    // that succeeded before it.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_bundle_deploys_up_to_the_create_phase_offline() {
+        let (wasm, sha) = wasm();
+        let outcome =
+            deploy_bundle(Arc::new(AnonymousIdentity), request(bundle(&sha, &wasm), "ic"))
+                .await
+                .expect("the request is valid");
+        assert_eq!(outcome.status, Status::Failed);
+        assert_eq!(outcome.network, "ic");
+        assert_eq!(outcome.canisters, ["backend"]);
+        assert!(outcome.created.is_empty());
+        assert!(outcome.canister_ids.is_empty());
+        let error = outcome.error.as_deref().unwrap();
+        assert!(error.contains("--no-create") && error.contains("backend"), "{error}");
+        assert!(outcome.log.iter().any(|l| l == "Building canisters: ok"), "{:?}", outcome.log);
+        assert!(outcome.log.iter().any(|l| l == "  build backend: ok"), "{:?}", outcome.log);
+    }
+
+    // The manifest's checksum is the bundle's integrity check: a wasm that
+    // does not hash to it is refused before any phase, as a request error
+    // that names the canister and both digests.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_wasm_that_does_not_match_its_checksum_is_refused_up_front() {
+        let (wasm, _) = wasm();
+        let wrong = "0".repeat(64);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bundle(&wrong, &wasm), "ic"))
+            .await
+            .expect_err("a checksum mismatch is refused");
+        assert!(err.0.contains("backend") && err.0.contains("checksum mismatch"), "{err}");
+        assert!(err.0.contains(&wrong), "{err}");
+    }
+
+    // Only the IC: a deploy to `local` names the icp CLI's managed network,
+    // and is refused up front with the way out; a connected testnet is not
+    // the IC either.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn environments_off_the_ic_network_are_refused() {
+        let (wasm, sha) = wasm();
+        let mut local = request(bundle(&sha, &wasm), "local");
+        local.no_create = false;
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), local)
+            .await
+            .expect_err("a managed network is refused");
+        assert!(err.0.contains("not the Internet Computer") && err.0.contains("`ic`"), "{err}");
+
+        let manifest = format!(
+            "networks:\n  - name: testnet\n    mode: connected\n    url: https://testnet.example\n    \
+             root-key: fetch\nenvironments:\n  - name: staging\n    network: testnet\n\
+             canisters:\n  - name: backend\n    build:\n      steps:\n        - type: pre-built\n          \
+             path: canisters/backend.wasm\n          sha256: {sha}\n"
+        );
+        let bytes =
+            archive(&[("icp.yaml", manifest.as_bytes()), ("canisters/backend.wasm", &wasm)]);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "staging"))
+            .await
+            .expect_err("a testnet is refused");
+        assert!(err.0.contains("testnet") && err.0.contains("not the Internet Computer"), "{err}");
+    }
+
+    // Before anything is deployed: an environment the bundle does not
+    // declare, or an id for a canister the environment does not hold, is a
+    // request error — nothing ran, so there is no outcome to report.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn unknown_environments_and_canisters_are_request_errors() {
+        let (wasm, sha) = wasm();
+        let err =
+            deploy_bundle(Arc::new(AnonymousIdentity), request(bundle(&sha, &wasm), "staging"))
+                .await
+                .expect_err("no such environment");
+        assert!(err.0.contains("staging"), "{err}");
+
+        let mut request = request(bundle(&sha, &wasm), "ic");
+        request.canister_ids.insert("frontend".into(), Principal::management_canister());
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request)
+            .await
+            .expect_err("no such canister");
+        assert!(err.0.contains("frontend"), "{err}");
+    }
+
+    // A bundle never carries a script step, and this server never runs one:
+    // the untrusted rule the project is loaded under refuses it at the load,
+    // as a request error naming the step, before any phase — so no canister
+    // is created ahead of a sync step that would then be refused.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_script_step_is_refused_before_any_phase_runs() {
+        let manifest = "canisters:\n  - name: backend\n    build:\n      steps:\n        \
+                        - type: script\n          command: touch /tmp/pwned\n";
+        let bytes = archive(&[("icp.yaml", manifest.as_bytes())]);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "ic"))
+            .await
+            .expect_err("a script step is refused");
+        assert!(
+            err.0.contains("canister 'backend' has a build step that is not a pre-built module")
+                && err.0.contains("touch /tmp/pwned"),
+            "{err}"
+        );
+    }
+
+    // A canister directory the manifest spells absolutely resolves inside the
+    // bundle for every reader that goes through the bundle filesystem — the
+    // loader finds the `canister.yaml` it looks for there — but the
+    // sync-plugin runtime takes the directory as a host path and would make
+    // `/etc` its sandbox root. Refused before any phase.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_canister_directory_outside_the_root_is_refused() {
+        let (wasm, sha) = wasm();
+        let canister = format!(
+            "name: backend\nbuild:\n  steps:\n    - type: pre-built\n      path: backend.wasm\n      \
+             sha256: {sha}\n"
+        );
+        let bytes = archive(&[
+            ("icp.yaml", b"canisters:\n  - /etc\n"),
+            ("etc/canister.yaml", canister.as_bytes()),
+            ("etc/backend.wasm", &wasm),
+        ]);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "ic"))
+            .await
+            .expect_err("a canister directory outside the root is refused");
+        // The confined filesystem finds nothing at `/etc`, so the loader does
+        // not find the canister: the spelling itself is the refusal.
+        assert!(err.0.contains("canister manifest") && err.0.contains("'/etc'"), "{err}");
+    }
+
+    // A manifest may declare a connected network that claims the mainnet root
+    // key over an endpoint of its own; an environment on it is refused like
+    // every network that is not the reserved `ic`.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_network_claiming_mainnet_elsewhere_is_refused() {
+        let (wasm, sha) = wasm();
+        let manifest = format!(
+            "networks:\n  - name: mainnet2\n    mode: connected\n    url: http://127.0.0.1:4943\n    \
+             root-key: mainnet\nenvironments:\n  - name: elsewhere\n    network: mainnet2\n\
+             canisters:\n  - name: backend\n    build:\n      steps:\n        - type: pre-built\n          \
+             path: canisters/backend.wasm\n          sha256: {sha}\n"
+        );
+        let bytes =
+            archive(&[("icp.yaml", manifest.as_bytes()), ("canisters/backend.wasm", &wasm)]);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "elsewhere"))
+            .await
+            .expect_err("a network that is not `ic` is refused");
+        assert!(err.0.contains("mainnet2") && err.0.contains("not the Internet Computer"), "{err}");
+    }
+
+    // A gzip-compressed module is held to what it expands to, since the
+    // library gunzips it in memory to read its interface: one that hashes
+    // correctly but expands past the module limit is refused before any
+    // phase.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_gzipped_wasm_that_expands_past_the_module_limit_is_refused() {
+        let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
+        encoder.write_all(&vec![0u8; host::MAX_WASM_BYTES as usize + 1]).unwrap();
+        let gz = encoder.finish().unwrap();
+        let sha = hex::encode(Sha256::digest(&gz));
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bundle(&sha, &gz), "ic"))
+            .await
+            .expect_err("an over-large gzipped module is refused");
+        assert!(err.0.contains("backend") && err.0.contains("expands past"), "{err}");
+    }
+
+    // Two canisters whose names differ only in case would share build
+    // output on a case-insensitive disk; refused before any phase.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn canister_names_that_differ_only_in_case_are_refused() {
+        let (wasm, sha) = wasm();
+        let manifest = format!(
+            "canisters:\n  - name: backend\n    build:\n      steps:\n        - type: pre-built\n          \
+             path: canisters/backend.wasm\n          sha256: {sha}\n  - name: Backend\n    build:\n      \
+             steps:\n        - type: pre-built\n          path: canisters/backend.wasm\n          \
+             sha256: {sha}\n"
+        );
+        let bytes =
+            archive(&[("icp.yaml", manifest.as_bytes()), ("canisters/backend.wasm", &wasm)]);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "ic"))
+            .await
+            .expect_err("case-colliding names are refused");
+        assert!(err.0.contains("differ only in letter case"), "{err}");
+    }
+
+    // A `bundle_path` is read only when it is a regular file: a directory
+    // here, and by the same check a device or a pipe, whose length says
+    // nothing about what reading it would produce.
+    // The library resolves the mode at install, after creation, and panics on
+    // a spelling it does not know: refused at entry instead, before anything
+    // is read.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn an_unknown_mode_is_refused_at_entry() {
+        let mut bad = request(Vec::new(), "ic");
+        bad.mode = "yolo".into();
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), bad).await.unwrap_err();
+        assert!(err.0.contains("mode must be one of"), "{}", err.0);
+        assert!(err.0.contains("'yolo'"), "{}", err.0);
+    }
+
+    // The file is read to the limit and one byte past it, never to its end:
+    // a file that grows between the size check and the read is still bounded.
+    #[tokio::test]
+    async fn a_bundle_file_is_read_to_its_limit_and_no_further() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = Utf8PathBuf::from_path_buf(dir.path().join("b.tar.gz")).unwrap();
+        std::fs::write(&path, [7u8; 32]).unwrap();
+        assert_eq!(read_bundle_file(&path, 32).await.unwrap().len(), 32);
+        let err = read_bundle_file(&path, 16).await.unwrap_err();
+        assert!(err.0.contains("above the 16 byte limit"), "{}", err.0);
+        let err = read_bundle_file(Utf8Path::new("relative/b.tar.gz"), 32).await.unwrap_err();
+        assert!(err.0.contains("absolute"), "{}", err.0);
+        let err = read_bundle_file(&path.with_file_name("missing"), 32).await.unwrap_err();
+        assert!(err.0.contains("cannot read bundle"), "{}", err.0);
+    }
+
+    // The Candid a bundle carries is parsed by the library without a depth
+    // guard, on the worker thread: an argument or an interface nested past
+    // the bound is refused before any phase, not left to overflow the stack
+    // at install or at the upgrade check.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn candid_nested_past_the_bound_is_refused_up_front() {
+        let deep = format!("({}null)", "opt ".repeat(200));
+        let (wasm, sha) = wasm();
+        let manifest = format!(
+            "canisters:\n  - name: backend\n    init_args:\n      value: '{deep}'\n    build:\n      \
+             steps:\n        - type: pre-built\n          path: canisters/backend.wasm\n          \
+             sha256: {sha}\n"
+        );
+        let bytes =
+            archive(&[("icp.yaml", manifest.as_bytes()), ("canisters/backend.wasm", &wasm)]);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "ic"))
+            .await
+            .expect_err("a deeply nested init_args value is refused");
+        assert!(err.0.contains("`init_args` is nested too deeply"), "{err}");
+
+        let interface = format!("service : {{ f : ({}null) -> () }}", "opt ".repeat(200));
+        let wasm = wat::parse_str(format!(
+            "(module (@custom \"icp:public candid:service\" \"{interface}\"))"
+        ))
+        .unwrap();
+        let sha = hex::encode(Sha256::digest(&wasm));
+        let bytes = bundle(&sha, &wasm);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "ic"))
+            .await
+            .expect_err("a deeply nested interface is refused");
+        assert!(err.0.contains("`candid:service` interface"), "{err}");
+        assert!(err.0.contains("nested too deeply"), "{err}");
+
+        // A module with an ordinary interface deploys as before.
+        let wasm = wat::parse_str(
+            "(module (@custom \"icp:public candid:service\" \"service : { f : () -> () }\"))",
+        )
+        .unwrap();
+        let sha = hex::encode(Sha256::digest(&wasm));
+        let outcome =
+            deploy_bundle(Arc::new(AnonymousIdentity), request(bundle(&sha, &wasm), "ic"))
+                .await
+                .unwrap();
+        assert!(outcome.log.iter().any(|l| l == "Building canisters: ok"), "{:?}", outcome.log);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_bundle_path_that_is_not_a_regular_file_is_refused() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).unwrap();
+        let mut local = request(Vec::new(), "ic");
+        local.bundle = BundleSource::Path(path);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), local).await.unwrap_err();
+        assert!(err.0.contains("not a regular file"), "{}", err.0);
+    }
+
+    // The manifest can name only what the archive carried: a variable file
+    // the bundle does not hold is refused at load — before any phase, since
+    // the value would otherwise be sent to the canister as a setting — and so
+    // is a wasm, whether the path is absolute or climbing: the vet asks the
+    // bundle's filesystem whether a regular file is there, and nothing
+    // outside the root is. Nothing on the machine is reached for, whatever
+    // exists at that spelling.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn paths_the_bundle_does_not_hold_are_refused() {
+        let (wasm, sha) = wasm();
+        let manifest = format!(
+            "canisters:\n  - name: backend\n    settings:\n      environment_variables:\n        \
+             SECRET:\n          path: /etc/hostname\n    build:\n      steps:\n        \
+             - type: pre-built\n          path: canisters/backend.wasm\n          sha256: {sha}\n"
+        );
+        let bytes =
+            archive(&[("icp.yaml", manifest.as_bytes()), ("canisters/backend.wasm", &wasm)]);
+        let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "ic"))
+            .await
+            .expect_err("a variable file outside the bundle is refused");
+        assert!(err.0.contains("SECRET"), "{err}");
+        assert!(err.0.contains("'/etc/hostname' is outside"), "{err}");
+
+        for path in ["/etc/hostname", "../../../../etc/hostname"] {
+            let manifest = format!(
+                "canisters:\n  - name: backend\n    build:\n      steps:\n        - type: pre-built\n          \
+                 path: {path}\n          sha256: {sha}\n"
+            );
+            let bytes = archive(&[("icp.yaml", manifest.as_bytes())]);
+            let err = deploy_bundle(Arc::new(AnonymousIdentity), request(bytes, "ic"))
+                .await
+                .expect_err("a wasm outside the bundle is refused");
+            assert!(
+                err.0.contains("etc/hostname")
+                    && err.0.contains("not a regular file in the project"),
+                "{path}: {err}"
+            );
+        }
+    }
+
+    // The log is what the agent (and the user) read to learn how far a deploy
+    // got: phases by title, canister tasks by kind and name, nested tasks
+    // indented, every outcome spelled out.
+    #[tokio::test]
+    async fn the_event_stream_becomes_a_readable_log() {
+        let (reporter, receiver) = icp_events::channel::<Task>();
+        let collector = tokio::spawn(collect_log(receiver));
+        let phase = reporter.task(Task::phase("Creating canisters:"));
+        let create = phase.reporter().task(Task::create("backend"));
+        create.finish(TaskOutcome::succeeded());
+        let install =
+            phase.reporter().task(Task::install("frontend", Principal::management_canister()));
+        install.finish(TaskOutcome::Failed {
+            message: "install_code rejected".into(),
+            causes: vec!["canister is out of cycles".into()],
+        });
+        phase.finish(TaskOutcome::failed("one canister failed"));
+        let check = reporter.task(Task::candid_check("backend", Principal::management_canister()));
+        check.finish(TaskOutcome::Skipped { reason: "not an upgrade".into() });
+        drop((reporter, phase, create, install, check));
+        let log = collector.await.unwrap();
+        assert_eq!(
+            log,
+            [
+                "  create backend: ok",
+                "  install frontend (aaaaa-aa): failed: install_code rejected: canister is out of cycles",
+                "Creating canisters: failed: one canister failed",
+                "candid_check backend (aaaaa-aa): skipped: not an upgrade",
+            ]
+        );
+    }
+
+    #[test]
+    fn error_chains_read_outermost_first() {
+        #[derive(Debug)]
+        struct Outer(Inner);
+        #[derive(Debug)]
+        struct Inner;
+        impl fmt::Display for Outer {
+            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+                f.write_str("failed to install canister 'backend'")
+            }
+        }
+        impl fmt::Display for Inner {
+            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+                f.write_str("canister rejected the call")
+            }
+        }
+        impl std::error::Error for Inner {}
+        impl std::error::Error for Outer {
+            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+                Some(&self.0)
+            }
+        }
+        assert_eq!(
+            error_chain(&Outer(Inner)),
+            "failed to install canister 'backend': canister rejected the call"
+        );
+    }
+}
diff --git a/crates/imcp2-deploy/src/host.rs b/crates/imcp2-deploy/src/host.rs
index f75f2cd..b8b3bcf 100644
--- a/crates/imcp2-deploy/src/host.rs
+++ b/crates/imcp2-deploy/src/host.rs
@@ -94,13 +94,15 @@ use icp_project::{
         Access, AccessError, CollectFriendlyDomains, Configuration, NetworkAccess, NetworkUrls,
         RootKeySource, RootKeySpec,
     },
+    operations::wasm::extract_candid_service,
     prelude::{IC, IC_MAINNET_NETWORK_API_URL, IC_ROOT_KEY},
     random::HostRandom,
     store_artifact::ArtifactStore,
     store_id::AccessImpl,
     untrusted::{NoRecipes, Vetted},
-    Canister, Lazy, Network, Project, ProjectLoadImpl,
+    Canister, CanisterArgs, Lazy, Network, Project, ProjectLoadImpl,
 };
+use imcp2_core::guard_candid_text;
 use sha2::{Digest, Sha256};
 
 /// Archives above this are refused before anything is written: a bundle rides
@@ -487,8 +489,19 @@ impl ScriptRunner for NoScripts {
 /// runtime will read of them ([`check_declared`]).
 ///
 /// The work the check does is bounded by the archive, not by the manifest:
-/// a file is read, hashed and expanded once however many canisters name it
-/// ([`Verified`]), and the expansions are summed ([`MAX_WASM_BYTES_IN_ALL`]).
+/// a file is read, hashed, expanded and its interface guarded once however
+/// many canisters name it ([`Verified`]), and the expansions are summed
+/// ([`MAX_WASM_BYTES_IN_ALL`]).
+///
+/// The textual Candid the bundle carries is held to what the library can
+/// parse safely: its parser recurses without a depth guard, on the worker
+/// thread the deploy runs on, so a canister's `init_args`/`upgrade_args`
+/// value and the `candid:service` interface in its module are checked for
+/// size and nesting first ([`guard_candid_text`], the bound the call tools
+/// apply to caller-supplied Candid). The library parses the arguments at
+/// install and the interface at the upgrade compatibility check; a value it
+/// cannot parse fails there, but one that would overflow the stack aborts
+/// the process, so the bound comes first.
 ///
 /// Where each canister lives is the loader's business: it finds a canister
 /// directory through the confined filesystem, so one outside the bundle root
@@ -553,6 +566,19 @@ fn check_bundle_within<'a>(
             ))));
         };
         checksum_matches(&local.path, expected, &actual).map_err(at)?;
+        for (field, args) in
+            [("init_args", &canister.init_args), ("upgrade_args", &canister.upgrade_args)]
+        {
+            // The format type is not nameable from here (the library exposes
+            // it through this field alone); its default is `candid`, the one
+            // format that is parsed rather than decoded.
+            if let Some(CanisterArgs::Text { content, format }) = args {
+                if *format == Default::default() {
+                    guard_candid_text(&format!("`{field}`"), content)
+                        .map_err(|e| at(Unsupported(e)))?;
+                }
+            }
+        }
 
         for step in &canister.sync.steps {
             let SyncStep::Plugin(plugin) = step else { continue };
@@ -576,9 +602,9 @@ fn check_bundle_within<'a>(
 
 /// What [`check_bundle`] has read so far, by where each file lies on disk:
 /// a module or plugin that several canisters name — or one canister names
-/// under several spellings — is read, hashed and expanded once, so the work
-/// the check does is bounded by what the archive holds, not by what the
-/// manifest repeats.
+/// under several spellings — is read, hashed, expanded and its interface
+/// guarded once, so the work the check does is bounded by what the archive
+/// holds, not by what the manifest repeats.
 struct Verified {
     /// Each file's sha256, in hex, by its on-disk path.
     hashes: BTreeMap<Utf8PathBuf, String>,
@@ -589,7 +615,9 @@ struct Verified {
 impl Verified {
     /// The sha256 of the file at `path` under `dir`, read and hashed on first
     /// sight — when a gzip-compressed one is also held to [`MAX_WASM_BYTES`]
-    /// expanded, and its expansion counted against the sum.
+    /// expanded, its expansion counted against the sum, and the
+    /// `candid:service` interface a module carries held to what the library
+    /// can parse.
     fn sha256(
         &mut self,
         fs: &BundleFs,
@@ -610,6 +638,11 @@ impl Verified {
             )));
         }
         self.wasm_left -= expanded;
+        // Bounded above, so the extractor's own gunzip is bounded too.
+        if let Some(interface) = extract_candid_service(&bytes) {
+            guard_candid_text(&format!("the `candid:service` interface of '{path}'"), &interface)
+                .map_err(Unsupported)?;
+        }
         let hash = hex::encode(Sha256::digest(&bytes));
         self.hashes.insert(on_disk, hash.clone());
         Ok(hash)
diff --git a/crates/imcp2-deploy/src/lib.rs b/crates/imcp2-deploy/src/lib.rs
index 0eb4390..162a0cb 100644
--- a/crates/imcp2-deploy/src/lib.rs
+++ b/crates/imcp2-deploy/src/lib.rs
@@ -11,9 +11,16 @@
 //! bundle` writes, treated as untrusted input. The rule it is held to is the
 //! library's own, `icp_project::untrusted`, the one `icp --untrusted`
 //! applies (dfinity/icp-cli#873, on #868 for what `icp project bundle`
-//! writes; issue #865): this crate pins icp-cli's main at their merge,
-//! composes the rule as the CLI does, and adds the bounds on the archive
-//! itself and the checks on the bundle's shape. The deploy pipeline over it
-//! and the MCP tools that expose it follow in their own changes.
+//! writes; issue #865): this crate pins icp-cli's main at
+//! their merge, composes the rule as the CLI does, and adds the bounds on the
+//! archive itself and the checks on the bundle's shape. [`deploy_bundle`]
+//! runs the library's deploy operation over it as a given identity: create
+//! the canisters the bundle names that do not exist yet, wire their ids,
+//! install or upgrade, run the sync steps. The MCP tools that expose it
+//! follow in their own change.
 
+mod calls;
+mod deploy;
 pub mod host;
+
+pub use deploy::{deploy_bundle, BundleSource, DeployOutcome, DeployRequest, RequestError, Status};