diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md
index 8779ead8e..25c8a1bb0 100644
--- a/.claude/CLAUDE.md
+++ b/.claude/CLAUDE.md
@@ -23,6 +23,7 @@ cargo fmt && cargo clippy            # Run after changes pass tests
 - **`crates/icp-cli`**: Main CLI binary (`icp`): argument parsing, command implementations, and all terminal presentation
 - **`crates/icp-app`**: Everything about the machine the tool runs on: identities and the keyring, user settings, the global directory layout, the package cache, local networks and the launcher that runs them, telemetry, offline message signing, and the operations that act on a canister by principal
 - **`crates/icp-project`**: Everything about a project: the project model, manifest loading and consolidation, canister management, and the operations that build, install, sync and deploy
+- **`crates/icp-sync-plugin`**: The wasmtime Component Model runtime for sync plugins — one implementation of `icp-project`'s plugin-runner seam
 - **`crates/icp-events`**: Typed progress events passed from operations to the CLI's renderers
 - **`crates/icp-canister-interfaces`**: Canister interface definitions for ICP system canisters
 - **`crates/schema-gen`**: JSON schema generation for manifest validation
@@ -42,13 +43,24 @@ even though it takes a principal, and `icp-app` calls down into it.
 
 `icp-project` is meant to end up runnable inside a canister, so it must not
 reach the host directly. What it needs from the machine it asks for through a
-trait declared there and implemented in `icp-app`:
+trait declared there and implemented elsewhere — in `icp-app`, or in
+`icp-sync-plugin`, which likewise depends on `icp-project` and never the
+reverse:
 
+- `files::FileSystem` — the files a project is made of
+- `calls::CanisterCalls` — submitting a call, and reading a certified fact
 - `network::Access` — a network's endpoints, root key and friendly domains
 - `canister::wasm::Fetch` — a wasm module a manifest names by URL
 - `canister::recipe::Resolve` — a recipe's Handlebars template
+- `canister::sync::plugin::Run` — running one sync plugin
+- `canister::sync::script::ScriptRunner` — running one sync script
+- `store_id::Access` / `store_artifact::Access` — the project's `.icp` stores
 - `host::Observe` — what resolution turned up, for telemetry
 
+Host implementations of the first, the last two stores and the script runner
+ship in `icp-project` itself behind the default-on `host` feature; the rest
+have no implementation there at all.
+
 Because those are implemented across a crate boundary, their error types carry
 their cause boxed and pass it through with `#[snafu(transparent)]`, which leaves
 the wrapper out of the source chain so the cause is reported once rather than
diff --git a/Cargo.lock b/Cargo.lock
index ab7f8df87..2eb4681e0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3655,6 +3655,7 @@ dependencies = [
  "icp-canister-interfaces",
  "icp-events",
  "icp-project",
+ "icp-sync-plugin",
  "icrc-ledger-types",
  "indexmap",
  "indoc",
@@ -3809,13 +3810,10 @@ dependencies = [
  "handlebars",
  "hex",
  "httptest",
- "ic-agent",
  "ic-ledger-types",
  "ic-management-canister-types 0.9.0",
- "ic-utils",
  "icp-canister-interfaces",
  "icp-events",
- "icp-sync-plugin",
  "icrc-ledger-types",
  "indexmap",
  "indoc",
@@ -3855,11 +3853,8 @@ dependencies = [
  "camino-tempfile",
  "candid",
  "console 0.16.3",
- "hex",
- "ic-agent",
- "ic-management-canister-types 0.9.0",
- "icp-canister-interfaces",
  "icp-events",
+ "icp-project",
  "semver",
  "snafu",
  "tokio",
diff --git a/crates/icp-app/Cargo.toml b/crates/icp-app/Cargo.toml
index c51c890a5..1ba361e03 100644
--- a/crates/icp-app/Cargo.toml
+++ b/crates/icp-app/Cargo.toml
@@ -41,6 +41,7 @@ ic-ledger-types = { workspace = true }
 ic-management-canister-types = { workspace = true }
 ic-utils = { workspace = true }
 icp-project = { workspace = true }
+icp-sync-plugin = { workspace = true }
 icp-canister-interfaces = { workspace = true }
 icp-events = { workspace = true }
 icrc-ledger-types = { workspace = true }
diff --git a/crates/icp-app/src/calls.rs b/crates/icp-app/src/calls.rs
index 14632cc31..bd5b56708 100644
--- a/crates/icp-app/src/calls.rs
+++ b/crates/icp-app/src/calls.rs
@@ -17,7 +17,9 @@ use candid::{Encode, Nat, Principal};
 use ic_agent::{
     Agent, AgentError,
     agent::{CallResponse, EffectiveId, SubnetType},
+    hash_tree::{Label, LookupResult},
 };
+use ic_management_canister_types::{CanisterMetadataArgs, CanisterMetadataResult};
 use icp_canister_interfaces::proxy::{ProxyArgs, ProxyResult};
 use icp_project::calls::{Authority, Call, CallError, CanisterCalls, RouteTo};
 
@@ -117,6 +119,121 @@ impl AgentCalls {
             .is_ok_and(|controllers| controllers.is_some())
     }
 
+    /// Ask the target's subnet to certify a metadata section, reporting only
+    /// what the certificate proves.
+    ///
+    /// The section path is requested 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)`.
+    async fn certified_metadata_section(
+        &self,
+        canister: Principal,
+        path: &str,
+    ) -> Result<Option<Vec<u8>>, CallError> {
+        let metadata_path: Vec<Label<Vec<u8>>> = vec![
+            "canister".into(),
+            Label::from_bytes(canister.as_slice()),
+            "metadata".into(),
+            path.into(),
+        ];
+        let controllers_path: Vec<Label<Vec<u8>>> = vec![
+            "canister".into(),
+            Label::from_bytes(canister.as_slice()),
+            "controllers".into(),
+        ];
+        let method = "read_state(metadata)";
+        let cert = self
+            .agent
+            .read_state_raw(
+                vec![metadata_path.clone(), controllers_path.clone()],
+                canister,
+            )
+            .await
+            .map_err(|err| Self::wrap(canister, method, err))?;
+
+        let unproven = |about: String| {
+            Err(CallError::Rejected {
+                canister,
+                method: method.to_owned(),
+                code: None,
+                message: format!("the certificate proves nothing about {about}"),
+            })
+        };
+        match cert.tree.lookup_path(&metadata_path) {
+            LookupResult::Found(bytes) => Ok(Some(bytes.to_vec())),
+            LookupResult::Absent => match cert.tree.lookup_path(&controllers_path) {
+                LookupResult::Found(_) => Ok(None),
+                LookupResult::Absent => Err(CallError::Rejected {
+                    canister,
+                    method: method.to_owned(),
+                    code: None,
+                    message: 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}")),
+        }
+    }
+
+    /// Read a metadata section by having the proxy ask the management canister
+    /// for it, which is what reaches a section private to the proxy's control.
+    ///
+    /// `read_state` is not a canister method, so it cannot be forwarded; the
+    /// management canister's `canister_metadata` can. It does not distinguish
+    /// an absent section from one the caller may not have, so a reply claiming
+    /// absence is confirmed against a certificate before it is reported as
+    /// one.
+    async fn metadata_through_proxy(
+        &self,
+        proxy: Principal,
+        canister: Principal,
+        path: &str,
+    ) -> Result<Option<Vec<u8>>, CallError> {
+        let arg = Encode!(&CanisterMetadataArgs {
+            canister_id: canister,
+            name: path.to_owned(),
+        })
+        .map_err(|e| CallError::failed(canister, "canister_metadata", e))?;
+        let call = Call::management("canister_metadata", canister, arg);
+
+        match self.through_proxy(proxy, &call).await {
+            Ok(reply) => {
+                let (metadata,): (CanisterMetadataResult,) = candid::decode_args(&reply)
+                    .map_err(|e| CallError::failed(canister, "canister_metadata", e))?;
+                Ok(Some(metadata.value))
+            }
+            Err(err) => {
+                let claims_absent = err
+                    .message()
+                    .is_some_and(|message| rejected_as_no_such_section(message, canister, path));
+                if !claims_absent {
+                    return Err(err);
+                }
+                // The management canister says the same thing about a section
+                // that isn't there and one that is private to someone else, so
+                // its word alone cannot be reported as absence. Only a
+                // certificate proves the section absent.
+                match self.certified_metadata_section(canister, path).await? {
+                    None => Ok(None),
+                    Some(_) => Err(CallError::Rejected {
+                        canister,
+                        method: "canister_metadata".to_owned(),
+                        code: None,
+                        message: format!(
+                            "canister {canister} does not let {proxy} read section `{path}`"
+                        ),
+                    }),
+                }
+            }
+        }
+    }
+
     /// 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> {
@@ -152,6 +269,28 @@ impl AgentCalls {
     }
 }
 
+/// Whether the management canister rejected a metadata read by claiming the
+/// target has no such section, rather than because the read itself failed.
+///
+/// The claim is not proof: the same rejection covers a section private to
+/// someone other than the proxy, so the caller confirms it against a
+/// certificate. A proxied read comes back as reject text with no code
+/// attached, so recognizing the claim at all means matching the replica's
+/// wording. Both sentences name the canister and one names the section, so the
+/// match is anchored on the values this call supplied rather than on a loose
+/// phrase that text relayed from elsewhere might happen to contain. A reword
+/// upstream turns the claim into an error rather than into a wrong answer.
+fn rejected_as_no_such_section(message: &str, canister: Principal, path: &str) -> bool {
+    // A canister with no module installed has no sections at all, so it reports
+    // absence in its own words. The certificate says the same thing about it:
+    // the metadata path is absent while the canister itself is there.
+    message.contains(&format!(
+        "The canister {canister} has no Wasm module and hence no metadata is available."
+    )) || message.contains(&format!(
+        "The canister {canister} has no metadata section with the name {path}."
+    ))
+}
+
 /// Wraps a resolved agent as the caller this workspace's operations take,
 /// forwarding through `proxy` when one was asked for.
 pub fn calls(
@@ -213,23 +352,13 @@ impl CanisterCalls for AgentCalls {
         &self,
         canister: Principal,
         path: &str,
+        authority: Authority,
     ) -> Result<Option<Vec<u8>>, CallError> {
-        match self
-            .agent
-            .read_state_canister_metadata(canister, path)
-            .await
-        {
-            Ok(bytes) => Ok(Some(bytes)),
-            Err(err) => {
-                // A path the certificate will not certify looks the same as a
-                // canister that was never created, which is the error this
-                // reports rather than "no such section".
-                if self.exists(canister).await {
-                    Ok(None)
-                } else {
-                    Err(Self::wrap(canister, "read_state(metadata)", err))
-                }
+        match self.proxy {
+            Some(proxy) if authority == Authority::Mediated => {
+                self.metadata_through_proxy(proxy, canister, path).await
             }
+            _ => self.certified_metadata_section(canister, path).await,
         }
     }
 
@@ -314,4 +443,44 @@ mod tests {
         assert!(err.is_rejection());
         assert!(!err.is_transient());
     }
+
+    /// The replica's own wording for the two ways a target reports it has no
+    /// section, copied from `CanisterManagerError` in the IC repo. Both are
+    /// absence, not failure, so both must reach the plugin as `none`.
+    #[test]
+    fn management_canister_absence_rejects_are_recognized() {
+        let target = Principal::from_text("aaaaa-aa").unwrap();
+        let other = Principal::from_text("2vxsx-fae").unwrap();
+
+        let no_module = format!(
+            "Proxy call failed: The canister {target} has no Wasm module and hence no metadata is available."
+        );
+        let no_section = format!(
+            "Proxy call failed: The canister {target} has no metadata section with the name candid:service."
+        );
+        assert!(rejected_as_no_such_section(
+            &no_module,
+            target,
+            "candid:service"
+        ));
+        assert!(rejected_as_no_such_section(
+            &no_section,
+            target,
+            "candid:service"
+        ));
+
+        // A section by another name, a canister other than the one asked about,
+        // and an unrelated failure are all reads that failed.
+        assert!(!rejected_as_no_such_section(&no_section, target, "dfx"));
+        assert!(!rejected_as_no_such_section(
+            &no_module,
+            other,
+            "candid:service"
+        ));
+        assert!(!rejected_as_no_such_section(
+            &format!("Proxy call failed: Canister {target} not found."),
+            target,
+            "candid:service"
+        ));
+    }
 }
diff --git a/crates/icp-app/src/context/init.rs b/crates/icp-app/src/context/init.rs
index 3f21a2a9b..fce33938c 100644
--- a/crates/icp-app/src/context/init.rs
+++ b/crates/icp-app/src/context/init.rs
@@ -113,7 +113,10 @@ pub fn initialize(
     let builder = Arc::new(Builder::new(wasm.clone(), files.clone()));
 
     // Canister syncer
-    let syncer = Arc::new(Syncer::host(wasm.clone()));
+    let syncer = Arc::new(Syncer::host(
+        wasm.clone(),
+        Arc::new(icp_sync_plugin::Wasmtime),
+    ));
 
     // Project loader
     let pload = ProjectLoadImpl {
diff --git a/crates/icp-cli/src/commands/deploy.rs b/crates/icp-cli/src/commands/deploy.rs
index 225b3d835..ffe2c5692 100644
--- a/crates/icp-cli/src/commands/deploy.rs
+++ b/crates/icp-cli/src/commands/deploy.rs
@@ -145,7 +145,7 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow:
     // command having to await it phase by phase.
     let mut report = DeployReport::default();
     let result = rendered(ctx.debug, async |reporter| {
-        deploy(&ctx.host, &calls, &agent, &params, reporter, &mut report).await
+        deploy(&ctx.host, &calls, &params, reporter, &mut report).await
     })
     .await;
 
diff --git a/crates/icp-cli/src/commands/sync.rs b/crates/icp-cli/src/commands/sync.rs
index 673301854..20a1d8f78 100644
--- a/crates/icp-cli/src/commands/sync.rs
+++ b/crates/icp-cli/src/commands/sync.rs
@@ -61,7 +61,7 @@ pub(crate) async fn exec(ctx: &Context, args: &SyncArgs) -> Result<(), anyhow::E
     let agent = ctx
         .get_agent_for_env(&identity_selection, &environment_selection)
         .await?;
-    let calls = icp_app::calls::calls(agent.clone(), args.proxy)?;
+    let calls = icp_app::calls::calls(agent, args.proxy)?;
 
     // Prepare list of canisters with their info for syncing
     let sync_canisters = try_join_all(cnames.iter().map(|name| async {
@@ -135,7 +135,7 @@ pub(crate) async fn exec(ctx: &Context, args: &SyncArgs) -> Result<(), anyhow::E
     rendered(ctx.debug, async |reporter| {
         sync_many(
             ctx.host.syncer.clone(),
-            agent,
+            calls,
             sync_canisters,
             project_dir,
             environment_selection.name().to_owned(),
diff --git a/crates/icp-project/Cargo.toml b/crates/icp-project/Cargo.toml
index b24cc8bf5..ecc8c42ba 100644
--- a/crates/icp-project/Cargo.toml
+++ b/crates/icp-project/Cargo.toml
@@ -36,14 +36,11 @@ futures = { workspace = true }
 glob = { workspace = true }
 handlebars = { workspace = true }
 hex = { workspace = true }
-ic-agent = { workspace = true }
 ic-ledger-types = { workspace = true }
 ic-management-canister-types = { workspace = true }
-ic-utils = { workspace = true }
 icrc-ledger-types = { workspace = true }
 icp-canister-interfaces = { workspace = true }
 icp-events = { workspace = true }
-icp-sync-plugin = { workspace = true }
 indexmap = { workspace = true }
 indoc = { workspace = true }
 itertools = { workspace = true }
@@ -65,9 +62,7 @@ strum = { workspace = true }
 tar = { workspace = true }
 time = { workspace = true }
 # `io-std` is listed for `tokio::io::stdout`/`stderr`; feature unification with other
-# crates supplies it anyway, so dropping it would not fail the build. `rt-multi-thread`
-# is needed by `block_in_place` in `canister::sync::plugin` and arrives via the
-# workspace `tokio` entry.
+# crates supplies it anyway, so dropping it would not fail the build.
 tokio = { workspace = true, features = ["sync", "macros", "rt", "time", "io-util", "io-std", "process", "signal"] }
 tracing = { workspace = true }
 url = { workspace = true }
diff --git a/crates/icp-project/src/calls.rs b/crates/icp-project/src/calls.rs
index 13d7a3a7c..3399ec532 100644
--- a/crates/icp-project/src/calls.rs
+++ b/crates/icp-project/src/calls.rs
@@ -247,11 +247,17 @@ pub trait CanisterCalls: Send + Sync {
     ///
     /// `Ok(None)` means the section is certified *absent* from a canister that
     /// exists. A canister that does not exist is an error, since the two are
-    /// otherwise indistinguishable and callers use this to tell them apart.
+    /// otherwise indistinguishable and callers use this to tell them apart. A
+    /// section the reader is not allowed to have is an error too: it is not
+    /// absent, and reporting it as such would be a guess.
+    ///
+    /// `authority` decides who is doing the reading, which is what a private
+    /// section is gated on.
     async fn metadata_section(
         &self,
         canister: Principal,
         path: &str,
+        authority: Authority,
     ) -> Result<Option<Vec<u8>>, CallError>;
 
     /// A canister's controllers, or `None` when there is no such canister.
@@ -282,6 +288,52 @@ pub trait CanisterCalls: Send + Sync {
     async fn subnet_uses_engine_operator(&self, subnet: Principal) -> Result<bool, CallError>;
 }
 
+#[cfg(any(test, feature = "test-util"))]
+/// Unimplemented mock implementation of [`CanisterCalls`], for a code path
+/// under test that is not supposed to reach a canister at all.
+pub struct UnimplementedMockCalls;
+
+#[cfg(any(test, feature = "test-util"))]
+#[async_trait]
+impl CanisterCalls for UnimplementedMockCalls {
+    fn caller(&self) -> Principal {
+        Principal::anonymous()
+    }
+
+    async fn update(&self, _call: Call) -> Result<Vec<u8>, CallError> {
+        unimplemented!("UnimplementedMockCalls::update")
+    }
+
+    async fn query(&self, _call: Call) -> Result<Vec<u8>, CallError> {
+        unimplemented!("UnimplementedMockCalls::query")
+    }
+
+    async fn metadata_section(
+        &self,
+        _canister: Principal,
+        _path: &str,
+        _authority: Authority,
+    ) -> Result<Option<Vec<u8>>, CallError> {
+        unimplemented!("UnimplementedMockCalls::metadata_section")
+    }
+
+    async fn controllers(&self, _canister: Principal) -> Result<Option<Vec<Principal>>, CallError> {
+        unimplemented!("UnimplementedMockCalls::controllers")
+    }
+
+    async fn module_hash(&self, _canister: Principal) -> Result<Option<Vec<u8>>, CallError> {
+        unimplemented!("UnimplementedMockCalls::module_hash")
+    }
+
+    async fn subnet_of(&self, _canister: Principal) -> Result<Principal, CallError> {
+        unimplemented!("UnimplementedMockCalls::subnet_of")
+    }
+
+    async fn subnet_uses_engine_operator(&self, _subnet: Principal) -> Result<bool, CallError> {
+        unimplemented!("UnimplementedMockCalls::subnet_uses_engine_operator")
+    }
+}
+
 /// A typed call failed, or its arguments or reply would not encode.
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub))]
diff --git a/crates/icp-project/src/canister/sync/declared.rs b/crates/icp-project/src/canister/sync/declared.rs
new file mode 100644
index 000000000..95cf55990
--- /dev/null
+++ b/crates/icp-project/src/canister/sync/declared.rs
@@ -0,0 +1,141 @@
+//! Reducing the paths a plugin step declared to the ones that have to be acted
+//! on.
+//!
+//! A step's `dirs`/`files` entries are configuration as much as they are a
+//! grant: the same tree may legitimately be declared under several keys, or a
+//! tree and a subtree of it. Both the runtime that preopens them for a plugin
+//! and the bundler that copies them into an archive want the reduced set,
+//! while the declared list is still passed along whole.
+
+use std::collections::HashSet;
+
+/// The meaningful components of a declared relative path: the `/`-separated
+/// names, with empty and `.` components dropped, so `./data/` and `data` compare
+/// equal.
+///
+/// `\` is deliberately not a separator here. On Unix it is an ordinary character
+/// in a filename, and these comparisons decide what gets opened for a guest that
+/// will open the path exactly as written.
+fn components(path: &str) -> Vec<&str> {
+    path.split('/')
+        .filter(|part| !part.is_empty() && *part != ".")
+        .collect()
+}
+
+/// Reduce declared directories to the ones that actually have to be opened.
+///
+/// `dirs` is configuration as much as it is a sandbox grant: a plugin may
+/// legitimately be handed the same tree under several keys, or a tree and a
+/// subtree of it, and it is told about every entry that was declared. The grant
+/// behind those entries has no such multiplicity — opening a directory twice, or
+/// opening one already reachable through an ancestor, conveys no further access.
+/// Callers keep the declared list as configuration and open only what this
+/// returns; a nested declared directory is reached through the ancestor covering
+/// it.
+///
+/// Retained paths keep their written spelling and first-occurrence order.
+/// Comparison is over the written spelling rather than the resolved location,
+/// because the guest opens each entry at the spelling the manifest gave it, and
+/// is component-wise, so `data` covers `./data/inner` but not `database`.
+///
+/// A spelling prefix alone is not containment once entries may contain `..`:
+/// `..` is a prefix of `../../shared`, yet one is the canister directory's
+/// parent and the other a child of its grandparent — neither holds the other.
+/// So an entry only covers one whose remaining components descend, `..`-free.
+/// Two spellings that coincide only once resolved (`../data` and `data` from a
+/// canister in `data`'s parent) still stay separate, which merely leaves the
+/// result less reduced.
+pub fn covering_dirs<'a>(dirs: impl IntoIterator<Item = &'a str>) -> Vec<&'a str> {
+    let dirs: Vec<&str> = dirs.into_iter().collect();
+    let parts: Vec<Vec<&str>> = dirs.iter().map(|dir| components(dir)).collect();
+    dirs.iter()
+        .enumerate()
+        .filter(|(i, _)| {
+            !parts.iter().enumerate().any(|(j, other)| {
+                j != *i
+                    && parts[*i].starts_with(other)
+                    && !parts[*i][other.len()..].contains(&"..")
+                    // A strict ancestor always covers; between equals, the first written wins.
+                    && (other.len() < parts[*i].len() || j < *i)
+            })
+        })
+        .map(|(_, dir)| *dir)
+        .collect()
+}
+
+/// Reduce declared paths to the distinct ones, keeping the written spelling and
+/// first-occurrence order.
+///
+/// [`covering_dirs`] without the containment rule, for entries that name files:
+/// `./a.json` and `a.json` are one file, but a file never subsumes another the
+/// way a directory subsumes its contents.
+pub fn distinct_paths<'a>(paths: impl IntoIterator<Item = &'a str>) -> Vec<&'a str> {
+    let mut seen: HashSet<Vec<&str>> = HashSet::new();
+    paths
+        .into_iter()
+        .filter(|path| seen.insert(components(path)))
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn unrelated_dirs_are_all_kept() {
+        assert_eq!(
+            covering_dirs(["assets", "config", "data/seed"]),
+            ["assets", "config", "data/seed"],
+        );
+    }
+
+    #[test]
+    fn duplicates_collapse_to_the_first_spelling() {
+        assert_eq!(covering_dirs(["./data", "data", "data/"]), ["./data"]);
+    }
+
+    #[test]
+    fn nested_dirs_collapse_to_their_ancestor_whichever_is_written_first() {
+        assert_eq!(covering_dirs(["data", "data/inner"]), ["data"]);
+        assert_eq!(covering_dirs(["data/inner", "data"]), ["data"]);
+        // Transitive: `data` covers `data/a` covers `data/a/b`.
+        assert_eq!(covering_dirs(["data/a/b", "data/a", "data"]), ["data"]);
+    }
+
+    #[test]
+    fn a_name_prefix_is_not_an_ancestor() {
+        assert_eq!(covering_dirs(["data", "database"]), ["data", "database"]);
+    }
+
+    /// An entry reaching further out than another is not inside it, however
+    /// much of its spelling they share: `..` is the canister directory's
+    /// parent, `../../shared` a child of its grandparent. Collapsing them would
+    /// leave the second with no preopen of its own and none that contains it.
+    #[test]
+    fn an_entry_that_rises_further_is_not_covered() {
+        assert_eq!(
+            covering_dirs(["..", "../../shared"]),
+            ["..", "../../shared"],
+        );
+        assert_eq!(
+            covering_dirs(["../shared", "../shared/../assets"]),
+            ["../shared", "../shared/../assets"],
+        );
+    }
+
+    /// Entries that reach out of the canister directory still cover what
+    /// descends from them, and still collapse with a repeat of themselves.
+    #[test]
+    fn entries_outside_the_canister_dir_cover_their_own_contents() {
+        assert_eq!(covering_dirs(["../data", "../data/inner"]), ["../data"]);
+        assert_eq!(covering_dirs(["../data", "./../data"]), ["../data"]);
+    }
+
+    #[test]
+    fn distinct_paths_dedupes_without_containment() {
+        assert_eq!(
+            distinct_paths(["./a.json", "a.json", "b.json", "dir/a.json"]),
+            ["./a.json", "b.json", "dir/a.json"],
+        );
+    }
+}
diff --git a/crates/icp-project/src/canister/sync/mod.rs b/crates/icp-project/src/canister/sync/mod.rs
index 66f0e314f..f5bc22b29 100644
--- a/crates/icp-project/src/canister/sync/mod.rs
+++ b/crates/icp-project/src/canister/sync/mod.rs
@@ -3,16 +3,17 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use candid::Principal;
-use ic_agent::Agent;
 use icp_events::StepReporter;
 use snafu::prelude::*;
 
+use crate::calls::CanisterCalls;
 use crate::canister::wasm;
 use crate::manifest::canister::SyncStep;
 use crate::network::NetworkUrls;
 use crate::prelude::*;
 
-mod plugin;
+pub mod declared;
+pub mod plugin;
 pub mod script;
 
 use script::{HostScripts, ScriptInvocation, ScriptRunError, ScriptRunner};
@@ -58,28 +59,36 @@ pub trait Synchronize: Sync + Send {
         &self,
         step: &SyncStep,
         params: &Params,
-        agent: &Agent,
+        calls: &Arc<dyn CanisterCalls>,
         reporter: &StepReporter,
     ) -> Result<Vec<String>, SynchronizeError>;
 }
 
-/// Dispatches each sync step to the machinery that runs it. Plugin steps run in
-/// the wasmtime WASI sandbox, which this drives directly; script steps go through
-/// an injected [`ScriptRunner`], since spawning a subprocess is not available
-/// everywhere.
+/// Dispatches each sync step to the machinery that runs it. Neither kind can be
+/// run from here: a script step needs a subprocess and a plugin step needs a
+/// wasm component runtime, so each goes through an injected runner.
 pub struct Syncer {
     scripts: Arc<dyn ScriptRunner>,
     wasm: Arc<dyn wasm::Fetch>,
+    plugins: Arc<dyn plugin::Run>,
 }
 
 impl Syncer {
     /// A syncer that runs script steps as host subprocesses.
-    pub fn host(wasm: Arc<dyn wasm::Fetch>) -> Self {
-        Self::new(Arc::new(HostScripts), wasm)
+    pub fn host(wasm: Arc<dyn wasm::Fetch>, plugins: Arc<dyn plugin::Run>) -> Self {
+        Self::new(Arc::new(HostScripts), wasm, plugins)
     }
 
-    pub fn new(scripts: Arc<dyn ScriptRunner>, wasm: Arc<dyn wasm::Fetch>) -> Self {
-        Self { scripts, wasm }
+    pub fn new(
+        scripts: Arc<dyn ScriptRunner>,
+        wasm: Arc<dyn wasm::Fetch>,
+        plugins: Arc<dyn plugin::Run>,
+    ) -> Self {
+        Self {
+            scripts,
+            wasm,
+            plugins,
+        }
     }
 }
 
@@ -89,7 +98,7 @@ impl Synchronize for Syncer {
         &self,
         step: &SyncStep,
         params: &Params,
-        agent: &Agent,
+        calls: &Arc<dyn CanisterCalls>,
         reporter: &StepReporter,
     ) -> Result<Vec<String>, SynchronizeError> {
         match step {
@@ -100,11 +109,10 @@ impl Synchronize for Syncer {
             SyncStep::Plugin(adapter) => Ok(plugin::sync(
                 adapter,
                 params,
-                agent,
-                &params.environment,
-                params.proxy,
+                calls,
                 reporter,
                 self.wasm.as_ref(),
+                self.plugins.as_ref(),
             )
             .await?),
         }
@@ -123,7 +131,7 @@ impl Synchronize for UnimplementedMockSyncer {
         &self,
         _step: &SyncStep,
         _params: &Params,
-        _agent: &Agent,
+        _calls: &Arc<dyn CanisterCalls>,
         _reporter: &StepReporter,
     ) -> Result<Vec<String>, SynchronizeError> {
         unimplemented!("UnimplementedMockSyncer::sync")
@@ -134,10 +142,16 @@ impl Synchronize for UnimplementedMockSyncer {
 mod tests {
     use std::sync::Mutex;
 
+    use indexmap::IndexMap;
+
+    use crate::manifest::adapter::plugin::{Adapter as PluginAdapter, NamedPaths, PathOrList};
+    use crate::manifest::adapter::prebuilt::{LocalSource, SourceField};
     use crate::manifest::adapter::script::{Adapter, CommandField};
 
     use super::*;
 
+    use plugin::{Invocation, KeyedPath, RunError};
+
     /// A [`ScriptRunner`] that records what it was asked to run instead of
     /// running it, so step dispatch can be tested without spawning a shell.
     #[derive(Default)]
@@ -157,26 +171,83 @@ mod tests {
         }
     }
 
-    fn dummy_agent() -> Agent {
-        Agent::builder()
-            .with_url("http://127.0.0.1:4943")
-            .build()
-            .expect("build test agent")
+    /// A [`plugin::Run`] that records the invocation it was handed instead of
+    /// loading a wasm component, and reports stderr lines the way a plugin that
+    /// retained some would.
+    struct RecordingRun {
+        seen: Mutex<Vec<Invocation>>,
+        retained: Vec<String>,
     }
 
-    /// A script step reaches the injected runner fully resolved: the commands
-    /// from the manifest, the canister directory as cwd, and the `ICP_CLI_*`
-    /// environment assembled from the sync params. Nothing is spawned.
-    #[tokio::test]
-    async fn script_steps_are_dispatched_to_the_injected_runner() {
-        let scripts = Arc::new(RecordingScripts::default());
-        let syncer = Syncer::new(scripts.clone(), Arc::new(wasm::UnimplementedMockFetch));
+    #[async_trait]
+    impl plugin::Run for RecordingRun {
+        async fn run(&self, invocation: Invocation) -> Result<Vec<String>, RunError> {
+            self.seen.lock().unwrap().push(invocation);
+            Ok(self.retained.clone())
+        }
+    }
+
+    /// A component runtime's own error, standing in for the kind of error this
+    /// crate cannot name — the whole reason [`RunError`] carries its cause
+    /// boxed.
+    #[derive(Debug, Snafu)]
+    #[snafu(display("the component runtime gave up"))]
+    struct RuntimeGaveUp;
+
+    /// A [`plugin::Run`] that fails the way a component runtime does.
+    struct FailingRun;
+
+    #[async_trait]
+    impl plugin::Run for FailingRun {
+        async fn run(&self, _invocation: Invocation) -> Result<Vec<String>, RunError> {
+            Err(RunError::new(RuntimeGaveUp))
+        }
+    }
+
+    /// A [`wasm::Fetch`] that reports a fixed path for the module and records
+    /// what it was asked to resolve, so the step's declared source can be
+    /// checked without a wasm file on disk.
+    struct StubFetch {
+        path: PathBuf,
+        asked: Mutex<Vec<(SourceField, PathBuf, Option<String>)>>,
+    }
 
-        let cid = Principal::from_slice(&[7; 4]);
-        let params = Params {
+    impl StubFetch {
+        fn new(path: &str) -> Self {
+            Self {
+                path: path.into(),
+                asked: Mutex::default(),
+            }
+        }
+    }
+
+    #[async_trait]
+    impl wasm::Fetch for StubFetch {
+        async fn wasm(
+            &self,
+            source: &SourceField,
+            base_dir: &Path,
+            sha256: Option<&str>,
+            _reporter: &StepReporter,
+        ) -> Result<PathBuf, wasm::FetchError> {
+            self.asked.lock().unwrap().push((
+                source.clone(),
+                base_dir.to_path_buf(),
+                sha256.map(str::to_owned),
+            ));
+            Ok(self.path.clone())
+        }
+    }
+
+    fn principal(byte: u8) -> Principal {
+        Principal::from_slice(&[byte; 4])
+    }
+
+    fn params() -> Params {
+        Params {
             path: "/work/backend".into(),
             project_dir: "/work".into(),
-            cid,
+            cid: principal(7),
             name: "backend".to_owned(),
             environment: "production".to_owned(),
             network: "ic".to_owned(),
@@ -184,18 +255,54 @@ mod tests {
                 api_url: "https://icp-api.io".parse().expect("valid api url"),
                 http_gateway_url: Some("https://icp0.io".parse().expect("valid gateway url")),
             },
-            canister_ids: BTreeMap::from([(
-                "my-frontend".to_owned(),
-                Principal::from_slice(&[8; 4]),
-            )]),
+            canister_ids: BTreeMap::from([("my-frontend".to_owned(), principal(8))]),
             proxy: None,
-        };
+        }
+    }
+
+    /// A plugin step declaring a key holding one path, a key holding two, a
+    /// field, and whatever `canisters:` list the test needs. Which of the
+    /// declared paths are directories is the runner's to work out from disk, so
+    /// nothing here has to exist.
+    fn plugin_step(canisters: Option<Vec<String>>) -> SyncStep {
+        SyncStep::Plugin(Box::new(PluginAdapter {
+            source: SourceField::Local(LocalSource {
+                path: "plugins/seed.wasm".into(),
+            }),
+            sha256: Some("abc123".to_owned()),
+            dirs: None,
+            files: Some(NamedPaths::Map(IndexMap::from([
+                ("seed".to_owned(), PathOrList::One("seed-data".to_owned())),
+                (
+                    "config".to_owned(),
+                    PathOrList::Many(vec!["a.txt".to_owned(), "b.txt".to_owned()]),
+                ),
+            ]))),
+            fields: Some(BTreeMap::from([("greeting".to_owned(), "hi".to_owned())])),
+            canisters,
+        }))
+    }
+
+    /// A script step reaches the injected runner fully resolved: the commands
+    /// from the manifest, the canister directory as cwd, and the `ICP_CLI_*`
+    /// environment assembled from the sync params. Nothing is spawned.
+    #[tokio::test]
+    async fn script_steps_are_dispatched_to_the_injected_runner() {
+        let scripts = Arc::new(RecordingScripts::default());
+        let syncer = Syncer::new(
+            scripts.clone(),
+            Arc::new(wasm::UnimplementedMockFetch),
+            Arc::new(plugin::UnimplementedMockRun),
+        );
+        let calls: Arc<dyn CanisterCalls> = Arc::new(crate::calls::UnimplementedMockCalls);
+
+        let params = params();
         let step = SyncStep::Script(Adapter {
             command: CommandField::Command("./deploy.sh".to_owned()),
         });
 
         let retained = syncer
-            .sync(&step, &params, &dummy_agent(), &StepReporter::null())
+            .sync(&step, &params, &calls, &StepReporter::null())
             .await
             .expect("script step should dispatch");
         assert!(retained.is_empty());
@@ -209,12 +316,146 @@ mod tests {
             vec![
                 ("ICP_CLI_ENVIRONMENT".to_owned(), "production".to_owned()),
                 ("ICP_CLI_NETWORK".to_owned(), "ic".to_owned()),
-                ("ICP_CLI_CID".to_owned(), cid.to_text()),
-                (
-                    "ICP_CLI_CID_MY_FRONTEND".to_owned(),
-                    Principal::from_slice(&[8; 4]).to_text()
-                ),
+                ("ICP_CLI_CID".to_owned(), principal(7).to_text()),
+                ("ICP_CLI_CID_MY_FRONTEND".to_owned(), principal(8).to_text()),
             ]
         );
     }
+
+    /// A plugin step reaches the injected runner fully resolved: the wasm the
+    /// fetch seam reported, the declared paths tagged with the keys they were
+    /// written under, the canister ID table as the plugin will see it, and the
+    /// step's `canisters:` list resolved against that table. The stderr lines
+    /// the runner reports come back to the caller.
+    #[tokio::test]
+    async fn plugin_steps_are_dispatched_to_the_injected_runner() {
+        let scripts = Arc::new(RecordingScripts::default());
+        let wasm = Arc::new(StubFetch::new("/cache/seed.wasm"));
+        let plugins = Arc::new(RecordingRun {
+            seen: Mutex::default(),
+            retained: vec!["registered 3 fruits".to_owned()],
+        });
+        let syncer = Syncer::new(scripts.clone(), wasm.clone(), plugins.clone());
+        let calls: Arc<dyn CanisterCalls> = Arc::new(crate::calls::UnimplementedMockCalls);
+
+        // A canister in a subproject, so the table the plugin sees is the
+        // resolved one rather than the params' own keys.
+        let sibling = principal(3);
+        let mut params = params();
+        let cid = params.cid;
+        params.name = "services/crm:backend".to_owned();
+        params.proxy = Some(principal(9));
+        params.canister_ids = BTreeMap::from([
+            ("services/crm:backend".to_owned(), cid),
+            ("services/crm:frontend".to_owned(), sibling),
+        ]);
+
+        let step = plugin_step(Some(vec!["frontend".to_owned()]));
+        let retained = syncer
+            .sync(&step, &params, &calls, &StepReporter::null())
+            .await
+            .expect("plugin step should dispatch");
+        assert_eq!(retained, ["registered 3 fruits"]);
+        // The step went to the plugin runner alone.
+        assert!(scripts.seen.lock().unwrap().is_empty());
+
+        // The module was asked for by the step's own source, checksum and
+        // canister directory.
+        let asked = wasm.asked.lock().unwrap();
+        assert_eq!(
+            &asked[..],
+            [(
+                SourceField::Local(LocalSource {
+                    path: "plugins/seed.wasm".into()
+                }),
+                params.path.clone(),
+                Some("abc123".to_owned()),
+            )]
+        );
+
+        let seen = plugins.seen.lock().unwrap();
+        let [invocation] = &seen[..] else {
+            panic!("expected exactly one invocation, got {}", seen.len());
+        };
+        assert_eq!(invocation.wasm_path, PathBuf::from("/cache/seed.wasm"));
+        assert_eq!(invocation.base_dir, params.path);
+        assert_eq!(invocation.project_dir, params.project_dir);
+        assert!(invocation.dirs.is_empty());
+        // Every declared path arrives in written order under the key it was
+        // written beneath, a key holding a list repeating across its paths.
+        assert_eq!(
+            invocation.files,
+            [
+                KeyedPath {
+                    key: Some("seed".to_owned()),
+                    path: "seed-data".to_owned(),
+                },
+                KeyedPath {
+                    key: Some("config".to_owned()),
+                    path: "a.txt".to_owned(),
+                },
+                KeyedPath {
+                    key: Some("config".to_owned()),
+                    path: "b.txt".to_owned(),
+                },
+            ]
+        );
+        assert_eq!(
+            invocation.fields,
+            BTreeMap::from([("greeting".to_owned(), "hi".to_owned())])
+        );
+        assert_eq!(invocation.host_canister_id, cid);
+        assert_eq!(invocation.proxy, params.proxy);
+        assert_eq!(invocation.environment, params.environment);
+        assert_eq!(invocation.api_url, params.urls.api_url);
+        assert_eq!(invocation.gateway_url, params.urls.http_gateway_url);
+        assert_eq!(
+            invocation.compute_limit_secs,
+            plugin::DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS
+        );
+        // The syncing canister's own subproject names its siblings by their
+        // bare local names...
+        assert_eq!(invocation.canister_ids.get("frontend"), Some(&sibling));
+        // ...which is the table the `canisters:` list is resolved against.
+        assert_eq!(
+            invocation.callable.by_name,
+            BTreeMap::from([("frontend".to_owned(), sibling)])
+        );
+        // The plugin's calls go through the very seam the caller passed in.
+        assert!(Arc::ptr_eq(&calls, &invocation.calls));
+    }
+
+    /// The runner's own failure is what the caller sees: the seam names the
+    /// action it was attempting and passes the cause through rather than
+    /// restating it.
+    #[tokio::test]
+    async fn a_failing_plugin_runner_surfaces_its_own_cause() {
+        let syncer = Syncer::new(
+            Arc::new(RecordingScripts::default()),
+            Arc::new(StubFetch::new("/cache/seed.wasm")),
+            Arc::new(FailingRun),
+        );
+        let calls: Arc<dyn CanisterCalls> = Arc::new(crate::calls::UnimplementedMockCalls);
+
+        let err = syncer
+            .sync(&plugin_step(None), &params(), &calls, &StepReporter::null())
+            .await
+            .expect_err("a failing runner must fail the step");
+        assert!(
+            matches!(
+                err,
+                SynchronizeError::Plugin {
+                    source: plugin::PluginError::RunPlugin { .. }
+                }
+            ),
+            "unexpected error: {err}"
+        );
+        let rendered = crate::error::flatten(&err);
+        assert!(rendered.contains("failed to run plugin"), "got: {rendered}");
+        assert_eq!(
+            rendered.matches("the component runtime gave up").count(),
+            1,
+            "the runtime's own message should be reported once: {rendered}"
+        );
+    }
 }
diff --git a/crates/icp-project/src/canister/sync/plugin.rs b/crates/icp-project/src/canister/sync/plugin.rs
index 256bfd733..756194b0f 100644
--- a/crates/icp-project/src/canister/sync/plugin.rs
+++ b/crates/icp-project/src/canister/sync/plugin.rs
@@ -1,23 +1,156 @@
 use std::collections::BTreeMap;
+use std::sync::Arc;
 
-use camino::Utf8PathBuf;
+use async_trait::async_trait;
 use candid::Principal;
-use ic_agent::Agent;
 use icp_events::StepReporter;
-use icp_sync_plugin::{
-    CallableCanisters, DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS, KeyedPath, PLUGIN_COMPUTE_LIMIT_ENV,
-    PluginInvocation, RunPluginError, run_plugin,
-};
 use snafu::prelude::*;
+use url::Url;
 
 use crate::{
+    calls::CanisterCalls,
     canister::wasm,
     manifest::adapter::plugin::{Adapter, NamedPaths},
+    prelude::*,
 };
 
 use super::Params;
 
-/// Convert a manifest [`NamedPaths`] (or its absence) into the runtime's
+/// Default seconds of compute a plugin may use. This is a runaway guard, not a
+/// security boundary: it protects the machine running `icp sync` from a plugin
+/// that never terminates. Legitimately heavy plugins (e.g.
+/// brotli-compressing a large asset bundle) can exceed it, especially on
+/// slower CI runners, so it is overridable via the [`PLUGIN_COMPUTE_LIMIT_ENV`]
+/// environment variable.
+pub const DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS: u64 = 60;
+/// Environment variable that overrides [`DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS`].
+pub const PLUGIN_COMPUTE_LIMIT_ENV: &str = "ICP_CLI_PLUGIN_COMPUTE_LIMIT_SECS";
+
+/// A path a step declared, tagged with the map key it was declared under.
+///
+/// The key is `None` when the manifest wrote the setting as a plain list, and
+/// `Some(name)` when it wrote a map. It is *non-unique*: several paths share a
+/// key when a map key resolves to a list of paths. Which form a plugin accepts
+/// depends on the interface it was built against, which only the runner can
+/// know, so both forms are passed on as written.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct KeyedPath {
+    /// The map key this path was declared under, or `None` for a plain-list entry.
+    pub key: Option<String>,
+    /// Manifest-relative path, anchored at the invocation's `base_dir`.
+    pub path: String,
+}
+
+/// The canisters a sync plugin is permitted to call, beyond the canister being
+/// synced.
+///
+/// Resolved here from the step's `canisters` list against the project's
+/// canister ID table, because a name in a manifest means what the project says
+/// it means. The runner only enforces the resulting set.
+#[derive(Clone, Debug, Default)]
+pub struct CallableCanisters {
+    /// Canisters callable by name. Maps the name — as it appears in the
+    /// canister ID table — to the principal it resolves to.
+    pub by_name: BTreeMap<String, Principal>,
+}
+
+/// Everything needed to load and drive one sync plugin.
+pub struct Invocation {
+    /// The plugin's wasm component.
+    pub wasm_path: PathBuf,
+    /// Directory the declared `dirs`/`files` are anchored at (the canister dir).
+    pub base_dir: PathBuf,
+    /// The project directory: the sandbox boundary. A declared path may rise
+    /// out of `base_dir` with `..` and reach anything inside the project, but
+    /// nothing above it.
+    ///
+    /// A `base_dir` that does not lie within this directory — a dependency
+    /// project reached by an out-of-tree `path:` — is its own boundary instead,
+    /// which grants nothing above the canister directory.
+    pub project_dir: PathBuf,
+    /// The step's `dirs:` entries, in written order.
+    pub dirs: Vec<KeyedPath>,
+    /// The step's `files:` entries, in written order. Depending on the
+    /// interface the plugin implements these may name directories too.
+    pub files: Vec<KeyedPath>,
+    /// Key-value fields to pass to the plugin inline.
+    pub fields: BTreeMap<String, String>,
+    /// The canister being synced: the default target of the plugin's calls.
+    pub host_canister_id: Principal,
+    /// How the plugin's canister calls and metadata reads are made. Its
+    /// [`caller`](CanisterCalls::caller) is also surfaced to the plugin as the
+    /// identity acting on its behalf.
+    pub calls: Arc<dyn CanisterCalls>,
+    /// The proxy canister `--proxy` named, when one was. Informational: the
+    /// plugin is told which canister is acting for it, while routing calls
+    /// through it is [`calls`](Self::calls)'s business.
+    pub proxy: Option<Principal>,
+    /// Name of the environment being synced.
+    pub environment: String,
+    /// The network's API endpoint — where canister calls are submitted.
+    pub api_url: Url,
+    /// The network's HTTP gateway, when it exposes one.
+    pub gateway_url: Option<Url>,
+    /// Compute-time budget in seconds. See
+    /// [`DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS`].
+    pub compute_limit_secs: u64,
+    /// The project's canister ID table for this environment, as exposed to the
+    /// plugin. Same-project canisters appear both under their fully-qualified
+    /// key and their bare local name.
+    pub canister_ids: BTreeMap<String, Principal>,
+    /// Canisters the step declared callable, beyond the one being synced.
+    pub callable: CallableCanisters,
+    /// Reporter the plugin's live stdout/stderr is emitted on.
+    pub reporter: StepReporter,
+}
+
+/// Running a plugin failed.
+///
+/// What runs a wasm component is the implementation's business — a component
+/// runtime, a sandbox, a compute deadline — so the cause is carried whole and
+/// displayed as itself.
+#[derive(Debug, Snafu)]
+#[snafu(transparent)]
+pub struct RunError {
+    pub source: Box<dyn std::error::Error + Send + Sync + 'static>,
+}
+
+impl RunError {
+    /// Wraps an implementation's own error for the trait boundary.
+    pub fn new(source: impl std::error::Error + Send + Sync + 'static) -> Self {
+        Self {
+            source: Box::new(source),
+        }
+    }
+}
+
+/// Runs a sync plugin.
+///
+/// Everything above this line is manifest work: which paths were declared,
+/// which canisters a name resolves to, what the plugin is allowed to call.
+/// Loading a wasm component and giving it a sandbox to run in is not, so it is
+/// asked for through this. The runner reaches canisters through the
+/// invocation's [`CanisterCalls`], which is the whole reason this crate can
+/// describe a sync without being able to perform one.
+#[async_trait]
+pub trait Run: Send + Sync {
+    /// Run the plugin, returning the stderr lines it asked to have retained.
+    async fn run(&self, invocation: Invocation) -> Result<Vec<String>, RunError>;
+}
+
+#[cfg(any(test, feature = "test-util"))]
+/// Unimplemented mock implementation of [`Run`].
+pub struct UnimplementedMockRun;
+
+#[cfg(any(test, feature = "test-util"))]
+#[async_trait]
+impl Run for UnimplementedMockRun {
+    async fn run(&self, _invocation: Invocation) -> Result<Vec<String>, RunError> {
+        unimplemented!("UnimplementedMockRun::run")
+    }
+}
+
+/// Convert a manifest [`NamedPaths`] (or its absence) into the runner's
 /// key-tagged path list. A missing setting yields an empty list.
 fn keyed_paths(paths: Option<&NamedPaths>) -> Vec<KeyedPath> {
     paths
@@ -35,16 +168,13 @@ pub enum PluginError {
     #[snafu(transparent)]
     Wasm { source: wasm::FetchError },
 
-    #[snafu(display("failed to get identity principal: {err}"))]
-    GetIdentityPrincipal { err: String },
-
     #[snafu(display(
         "invalid {PLUGIN_COMPUTE_LIMIT_ENV} value '{value}': expected a positive integer number of seconds"
     ))]
     InvalidComputeLimit { value: String },
 
     #[snafu(display("failed to run plugin"))]
-    Run { source: RunPluginError },
+    RunPlugin { source: RunError },
 
     #[snafu(display(
         "sync plugin lists canister '{name}' as callable, but no canister by that name \
@@ -84,18 +214,17 @@ fn parse_compute_limit(value: &str) -> Result<u64, PluginError> {
 pub(super) async fn sync(
     adapter: &Adapter,
     params: &Params,
-    agent: &Agent,
-    environment: &str,
-    proxy: Option<Principal>,
+    calls: &Arc<dyn CanisterCalls>,
     reporter: &StepReporter,
     wasm_fetch: &dyn wasm::Fetch,
+    plugins: &dyn Run,
 ) -> Result<Vec<String>, PluginError> {
     // 0. Resolve the compute-time limit up front so a malformed
     //    ICP_CLI_PLUGIN_COMPUTE_LIMIT_SECS fails fast — before downloading the
     //    wasm or touching the network — rather than after doing that work.
     let compute_limit_secs = resolve_compute_limit_secs()?;
 
-    // 1. Determine the on-disk path for the wasm. run_plugin needs a path, not raw bytes.
+    // 1. Determine the on-disk path for the wasm. The runner needs a path, not raw bytes.
     //    - Local: sha256 is verified if present, then the original path is returned.
     //    - Remote: downloaded to cache (sha256 required, enforced at parse time) and the
     //      stable cache path is returned — no temp file needed.
@@ -108,14 +237,12 @@ pub(super) async fn sync(
         )
         .await?;
 
-    // 2. Collect inputs as manifest strings. `run_plugin` opens the declared
+    // 2. Collect inputs as manifest strings. The runner opens the declared
     //    paths itself — preopening or reading each by what is on disk, anchored
-    //    at `base_dir`, confined to `project_dir`, and subject to the runtime's
+    //    at `base_dir`, confined to `project_dir`, and subject to its
     //    path-safety checks (no escaping or symlinked paths). It also decides
     //    which of the two settings the plugin's interface accepts, so both are
     //    forwarded as written.
-    let base_dir = Utf8PathBuf::from(params.path.as_str());
-    let project_dir = Utf8PathBuf::from(params.project_dir.as_str());
     let dirs = keyed_paths(adapter.dirs.as_ref());
     let files = keyed_paths(adapter.files.as_ref());
     let fields: BTreeMap<String, String> = adapter.fields.clone().unwrap_or_default();
@@ -123,39 +250,30 @@ pub(super) async fn sync(
     // 3. Build the canister ID table exposed to the plugin, then resolve the
     //    step's `canisters` list against it.
     let canister_ids = exposed_canister_ids(params);
-    let callable = resolve_callable(adapter, &canister_ids, environment)?;
-
-    // 4. Run the plugin (blocking call — signal Tokio that this thread will block).
-    let identity_principal = agent
-        .get_principal()
-        .map_err(|err| PluginError::GetIdentityPrincipal { err })?;
-
-    let agent_clone = agent.clone();
-    let environment_owned = environment.to_owned();
-    let reporter_clone = reporter.clone();
+    let callable = resolve_callable(adapter, &canister_ids, &params.environment)?;
 
-    tokio::task::block_in_place(|| {
-        run_plugin(PluginInvocation {
+    // 4. Hand it all to the runner.
+    plugins
+        .run(Invocation {
             wasm_path,
-            base_dir,
-            project_dir,
+            base_dir: params.path.clone(),
+            project_dir: params.project_dir.clone(),
             dirs,
             files,
             fields,
             host_canister_id: params.cid,
-            agent: agent_clone,
-            proxy,
-            identity_principal,
-            environment: environment_owned,
+            calls: calls.clone(),
+            proxy: params.proxy,
+            environment: params.environment.clone(),
             api_url: params.urls.api_url.clone(),
             gateway_url: params.urls.http_gateway_url.clone(),
             compute_limit_secs,
             canister_ids,
             callable,
-            reporter: reporter_clone,
+            reporter: reporter.clone(),
         })
-    })
-    .context(RunSnafu)
+        .await
+        .context(RunPluginSnafu)
 }
 
 /// The canister ID table exposed to a sync plugin: every named canister in the
diff --git a/crates/icp-project/src/error.rs b/crates/icp-project/src/error.rs
new file mode 100644
index 000000000..d807e2eb7
--- /dev/null
+++ b/crates/icp-project/src/error.rs
@@ -0,0 +1,65 @@
+//! Reading an error's `source()` chain, for the places that have to hand an
+//! error to something that carries no chain of its own.
+
+/// The rendered `source()` chain of an error, outermost cause first. The
+/// error's own message is not included.
+pub fn causes(error: &dyn std::error::Error) -> Vec<String> {
+    let mut causes = Vec::new();
+    let mut cause = error.source();
+    while let Some(err) = cause {
+        causes.push(err.to_string());
+        cause = err.source();
+    }
+    causes
+}
+
+/// An error and its causes rendered as one `: `-separated string, for a
+/// boundary that takes a single message — a wasm guest's `result<_, string>`,
+/// say. A bare `to_string()` there drops everything the chain holds, which for
+/// the error types whose message names only the action attempted is all of the
+/// reason.
+pub fn flatten(error: &dyn std::error::Error) -> String {
+    let mut rendered = error.to_string();
+    for cause in causes(error) {
+        rendered.push_str(": ");
+        rendered.push_str(&cause);
+    }
+    rendered
+}
+
+#[cfg(test)]
+mod tests {
+    use snafu::prelude::*;
+
+    #[derive(Debug, Snafu)]
+    #[snafu(display("the innermost thing went wrong"))]
+    struct Inner;
+
+    #[derive(Debug, Snafu)]
+    #[snafu(display("the outer action failed"))]
+    struct Outer {
+        source: Inner,
+    }
+
+    fn outer() -> Outer {
+        Err::<(), _>(Inner).context(OuterSnafu).unwrap_err()
+    }
+
+    #[test]
+    fn causes_omit_the_error_itself() {
+        assert_eq!(super::causes(&outer()), ["the innermost thing went wrong"]);
+    }
+
+    #[test]
+    fn flatten_joins_the_whole_chain() {
+        assert_eq!(
+            super::flatten(&outer()),
+            "the outer action failed: the innermost thing went wrong"
+        );
+    }
+
+    #[test]
+    fn flatten_of_a_lone_error_is_its_message() {
+        assert_eq!(super::flatten(&Inner), "the innermost thing went wrong");
+    }
+}
diff --git a/crates/icp-project/src/lib.rs b/crates/icp-project/src/lib.rs
index d38f933ac..29d109d8b 100644
--- a/crates/icp-project/src/lib.rs
+++ b/crates/icp-project/src/lib.rs
@@ -28,6 +28,7 @@ use crate::{
 pub mod calls;
 pub mod canister;
 pub mod defer;
+pub mod error;
 pub mod files;
 #[cfg(feature = "host")]
 pub mod fs;
diff --git a/crates/icp-project/src/operations/bundle.rs b/crates/icp-project/src/operations/bundle.rs
index f1b6ed0da..43ed39bc8 100644
--- a/crates/icp-project/src/operations/bundle.rs
+++ b/crates/icp-project/src/operations/bundle.rs
@@ -24,13 +24,13 @@ use crate::{
 };
 use camino::Utf8Component;
 use flate2::{Compression, write::GzEncoder};
-use icp_sync_plugin::{covering_dirs, distinct_paths};
 use snafu::{OptionExt, ResultExt, Snafu};
 use tar::Builder;
 use tracing::warn;
 
 use icp_events::StepReporter;
 
+use crate::canister::sync::declared::{covering_dirs, distinct_paths};
 use crate::operations::task::Reporter;
 
 use crate::operations::build::{BuildManyError, build_many};
diff --git a/crates/icp-project/src/operations/deploy.rs b/crates/icp-project/src/operations/deploy.rs
index 885169c8f..4c23cc8bf 100644
--- a/crates/icp-project/src/operations/deploy.rs
+++ b/crates/icp-project/src/operations/deploy.rs
@@ -48,7 +48,6 @@ use crate::operations::{
 };
 use crate::project::ArgsField;
 use crate::{CanisterArgsToBytesError, ProjectLoadError};
-use ic_agent::Agent;
 
 /// Everything that can stop a deploy. Each phase's failure keeps the typed
 /// error of the operation that produced it, so a caller can still tell a
@@ -208,19 +207,16 @@ pub struct DeployReport {
 
 /// Run a full deploy, reporting progress as one task tree.
 ///
-/// `calls` and `agent` speak for the identity the caller resolved: which
-/// identity that is, and how its key was unlocked, is not this layer's
-/// business. Both are resolved at the first phase that needs the network and
-/// not before — a deploy that cannot build, or that `--no-create` refuses, has
-/// no business unlocking a key or reaching a network.
+/// `calls` speaks for the identity the caller resolved: which identity that is,
+/// and how its key was unlocked, is not this layer's business. It is resolved at
+/// the first phase that needs the network and not before — a deploy that cannot
+/// build, or that `--no-create` refuses, has no business unlocking a key or
+/// reaching a network.
 ///
 /// `report` is written as the run goes; see [`DeployReport`].
 pub async fn deploy(
     host: &Host,
     calls: &Deferred<'_, Arc<dyn CanisterCalls>>,
-    // Sync steps still run against an agent, because the wasmtime plugin
-    // runtime does. That goes when the step runners move behind their own seam.
-    agent: &Deferred<'_, Agent>,
     params: &DeployParams,
     reporter: &Reporter,
     report: &mut DeployReport,
@@ -269,7 +265,6 @@ pub async fn deploy(
     // is asked for the means — and where the identity it speaks for gets
     // unlocked.
     let calls = calls.get().await?;
-    let agent = agent.get().await?;
 
     if canisters_to_create.is_empty() {
         notice(reporter, "All canisters already exist");
@@ -411,7 +406,7 @@ pub async fn deploy(
     .await;
     finish(&phase, result)?;
 
-    sync(host, params, calls, agent, reporter).await?;
+    sync(host, params, calls, reporter).await?;
 
     Ok(())
 }
@@ -523,7 +518,6 @@ async fn sync(
     host: &Host,
     params: &DeployParams,
     calls: &Arc<dyn CanisterCalls>,
-    agent: &Agent,
     reporter: &Reporter,
 ) -> Result<(), DeployError> {
     let environment_selection = &params.environment;
@@ -608,7 +602,7 @@ async fn sync(
     let phase = reporter.task(Task::phase("Syncing canisters:"));
     let result = sync_many(
         host.syncer.clone(),
-        agent.clone(),
+        calls.clone(),
         sync_canisters,
         project_dir,
         environment_selection.name().to_owned(),
diff --git a/crates/icp-project/src/operations/misc.rs b/crates/icp-project/src/operations/misc.rs
index 2a2f92813..bb3305857 100644
--- a/crates/icp-project/src/operations/misc.rs
+++ b/crates/icp-project/src/operations/misc.rs
@@ -2,7 +2,7 @@
 
 use time::{OffsetDateTime, macros::format_description};
 
-use crate::calls::CanisterCalls;
+use crate::calls::{Authority, CanisterCalls};
 
 /// A canister's custom-section metadata as text, or `None` when it has no such
 /// section (or could not be asked).
@@ -10,12 +10,19 @@ use crate::calls::CanisterCalls;
 /// Callers use this to detect a capability — a Motoko EOP canister, a canister
 /// that publishes its Candid — so "absent" and "could not tell" lead to the
 /// same conservative answer and are not worth distinguishing.
+///
+/// The read is made under the caller's own authority: what is being asked is
+/// what the identity doing the deploying can see, not what something acting on
+/// its behalf could.
 pub async fn fetch_canister_metadata(
     calls: &dyn CanisterCalls,
     canister_id: candid::Principal,
     metadata: &str,
 ) -> Option<String> {
-    let section = calls.metadata_section(canister_id, metadata).await.ok()??;
+    let section = calls
+        .metadata_section(canister_id, metadata, Authority::Direct)
+        .await
+        .ok()??;
     Some(String::from_utf8_lossy(&section).into())
 }
 
diff --git a/crates/icp-project/src/operations/sync.rs b/crates/icp-project/src/operations/sync.rs
index 3976615f9..078c9ca28 100644
--- a/crates/icp-project/src/operations/sync.rs
+++ b/crates/icp-project/src/operations/sync.rs
@@ -1,12 +1,12 @@
 use crate::{
     Canister,
+    calls::CanisterCalls,
     canister::sync::{Params, Synchronize, SynchronizeError},
     network::NetworkUrls,
     prelude::{Path, PathBuf},
 };
 use candid::Principal;
 use futures::{StreamExt, stream::FuturesOrdered};
-use ic_agent::Agent;
 use icp_events::{StepOutcome, TaskOutcome};
 
 use crate::operations::task::{Reporter, Task, TaskReporter};
@@ -25,7 +25,7 @@ pub struct SyncOperationError {
 #[allow(clippy::too_many_arguments)]
 async fn sync_canister(
     syncer: &Arc<dyn Synchronize>,
-    agent: &Agent,
+    calls: &Arc<dyn CanisterCalls>,
     canister_path: PathBuf,
     project_dir: &Path,
     canister_id: Principal,
@@ -57,7 +57,7 @@ async fn sync_canister(
                     canister_ids: canister_ids.clone(),
                     proxy,
                 },
-                agent,
+                calls,
                 &reporter,
             )
             .await;
@@ -73,21 +73,10 @@ async fn sync_canister(
     Ok(stderr_lines)
 }
 
-/// The rendered `source()` chain of an error, outermost cause first.
-fn error_causes(error: &dyn std::error::Error) -> Vec<String> {
-    let mut causes = Vec::new();
-    let mut cause = error.source();
-    while let Some(err) = cause {
-        causes.push(err.to_string());
-        cause = err.source();
-    }
-    causes
-}
-
 /// Orchestrates syncing multiple canisters concurrently.
 pub async fn sync_many(
     syncer: Arc<dyn Synchronize>,
-    agent: Agent,
+    calls: Arc<dyn CanisterCalls>,
     canisters: Vec<(Principal, PathBuf, Canister)>,
     project_dir: PathBuf,
     environment: String,
@@ -103,7 +92,7 @@ pub async fn sync_many(
         let task = reporter.task(Task::sync(canister_info.name.clone(), cid));
 
         let fut = {
-            let agent = agent.clone();
+            let calls = calls.clone();
             let syncer = syncer.clone();
             let environment = environment.clone();
             let network = network.clone();
@@ -114,7 +103,7 @@ pub async fn sync_many(
             async move {
                 let result = sync_canister(
                     &syncer,
-                    &agent,
+                    &calls,
                     canister_path,
                     &project_dir,
                     cid,
@@ -137,7 +126,7 @@ pub async fn sync_many(
                     }),
                     Err(error) => task.finish(TaskOutcome::Failed {
                         message: error.to_string(),
-                        causes: error_causes(error),
+                        causes: crate::error::causes(error),
                     }),
                 }
 
diff --git a/crates/icp-sync-plugin/Cargo.toml b/crates/icp-sync-plugin/Cargo.toml
index edf8396d6..7cdcb33df 100644
--- a/crates/icp-sync-plugin/Cargo.toml
+++ b/crates/icp-sync-plugin/Cargo.toml
@@ -12,15 +12,14 @@ bytes.workspace = true
 camino.workspace = true
 candid.workspace = true
 console.workspace = true
-hex.workspace = true
-ic-agent.workspace = true
-ic-management-canister-types.workspace = true
-icp-canister-interfaces.workspace = true
 icp-events.workspace = true
+# Only for the seams the runtime is driven through — the plugin invocation and
+# the canister calls it makes — never for `host`-feature code, which is what
+# this crate is an implementation of.
+icp-project = { path = "../icp-project", default-features = false }
 semver.workspace = true
 snafu.workspace = true
 tokio.workspace = true
-url.workspace = true
 wasmtime.workspace = true
 wasmtime-wasi.workspace = true
 
@@ -29,6 +28,10 @@ camino.workspace = true
 
 [dev-dependencies]
 camino-tempfile.workspace = true
+# The mocks these tests use are host-side, so this one keeps the default
+# features the library half deliberately does without.
+icp-project = { path = "../icp-project", features = ["test-util"] }
+url.workspace = true
 
 [lints]
 workspace = true
diff --git a/crates/icp-sync-plugin/DESIGN.md b/crates/icp-sync-plugin/DESIGN.md
index 30dafca07..e6faf746c 100644
--- a/crates/icp-sync-plugin/DESIGN.md
+++ b/crates/icp-sync-plugin/DESIGN.md
@@ -73,29 +73,37 @@ Host-side Component Model runtime for sync plugins.
 ```
 crates/icp-sync-plugin/
   src/
-    lib.rs             — public API: run_plugin(), RunPluginError
+    lib.rs             — public API: Wasmtime
     runtime.rs         — wasmtime component setup, HostState, bindgen!, exec() call
     path.rs            — declared-path resolution and safety checks (project bound, symlinks)
   sync-plugin.wit      — current WIT interface, v0.2.0
   sync-plugin-v1.wit   — frozen WIT interface, v0.1.0
-  Cargo.toml           — wasmtime, wasmtime-wasi, ic-agent, ic-management-canister-types,
-                         candid, camino, snafu, tokio, semver
+  Cargo.toml           — wasmtime, wasmtime-wasi, icp-project, candid, camino,
+                         snafu, tokio, semver
 ```
 
-Public function:
+Public type:
 
 ```rust
-pub fn run_plugin(invocation: PluginInvocation) -> Result<Vec<String>, RunPluginError>
+pub struct Wasmtime;  // impl icp_project::canister::sync::plugin::Run
 ```
 
-`PluginInvocation` bundles the inputs: `wasm_path`, `base_dir`, `project_dir`,
+This crate is one implementation of `icp-project`'s plugin-runner seam. The
+trait, its `Invocation`, and the `KeyedPath`/`CallableCanisters` types the
+invocation is made of all live in `icp-project`, because deciding what a step
+declared and what a canister name resolves to is manifest work. What is left
+here is everything that needs a machine: a component runtime, a WASI sandbox,
+a compute deadline, and a filesystem to open the declared paths on.
+
+`Invocation` bundles the inputs: `wasm_path`, `base_dir`, `project_dir`,
 `dirs`, `files`, `fields`, `host_canister_id` (the canister being synced),
-`agent`, `proxy`, `identity_principal`, `environment`, `api_url` and
-`gateway_url` (where the network is reached — informational, since the guest has
-no sockets), `compute_limit_secs`, the exposed `canister_ids` table, the
-`callable: CallableCanisters` enforcement set, and `reporter`. The CLI resolves
-the manifest's declared `canisters:` into `CallableCanisters` before calling;
-this crate stays free of any manifest knowledge.
+`calls` (how canister calls and metadata reads are made — see below), `proxy`,
+`environment`, `api_url` and `gateway_url` (where the network is reached —
+informational, since the guest has no sockets), `compute_limit_secs`, the
+exposed `canister_ids` table, the `callable: CallableCanisters` enforcement
+set, and `reporter`. The identity surfaced to the plugin is `calls.caller()`,
+and `proxy` is informational too: which canister a call is routed through is
+`calls`'s business, not this crate's.
 
 `dirs` and `files` are the manifest's own `dirs:`/`files:` settings as
 manifest-relative paths (`KeyedPath`s carrying the map key each was declared
@@ -162,14 +170,14 @@ all, so a plugin opens `dir.path` verbatim regardless of where it points.
 
 For a v0.2.0 plugin the manifest's `files:` is the only setting, and
 `resolve_entries` records what it found at each entry — a file's contents, or
-`None` for a directory — so `run_plugin` can partition them into the interface's
+`None` for a directory — so the runtime can partition them into the interface's
 `dirs` and `files` lists while keeping each list in written order. A v0.1.0
 plugin keeps the manifest's own split instead: its `dirs:` entries must each be
 a directory (`MissingDir` otherwise), and its `files:` entries are all read.
 
 Preopens are derived from every entry that turned out to be a directory,
-whichever setting declared it, reduced by `covering_dirs` (below) so that one
-tree is opened once.
+whichever setting declared it, reduced by `icp-project`'s `covering_dirs`
+(below) so that one tree is opened once.
 
 ### `HostState` and bindgen
 
@@ -183,8 +191,7 @@ mod v1 { wasmtime::component::bindgen!({ world: "sync-plugin", path: "sync-plugi
 struct HostState {
     host_canister_id: Principal,
     callable: CallableCanisters,          // name → principal, from the manifest
-    agent: Arc<Agent>,
-    proxy: Option<Principal>,
+    calls: Arc<dyn CanisterCalls>,        // icp-project's canister-call seam
     wasi_ctx: wasmtime_wasi::WasiCtx,
     wasi_table: wasmtime_wasi::ResourceTable,
     epoch_extension: Arc<AtomicU64>,
@@ -195,43 +202,45 @@ struct HostState {
 ```
 
 `HostState` implements `WasiView` so wasmtime_wasi can access the WASI context.
-Both imports use `tokio::runtime::Handle::current().block_on(...)` because the
-caller already wraps the synchronous `run_plugin` in
-`tokio::task::block_in_place`. For a v0.2.0 plugin the target is resolved from
-the request's `call-target` by `resolve_call_target`, which enforces the
-`callable` set; for a v0.1.0 plugin the target is always `host_canister_id`.
-When a proxy is configured and the call is a non-`direct` update, it is encoded
-as `ProxyArgs` and routed through the proxy's `proxy` method; otherwise it goes
-straight to the resolved target via `ic-agent`.
-
-### Metadata reads (two routes, one answer)
-
-`canister-metadata-section` cannot reuse the call path: `read_state` is not a canister
-method, so a proxy canister has nothing to forward. The two routes are therefore
-different protocols reaching the same data, chosen by the request's `direct` flag
-exactly as `canister-call` chooses one:
-
-- **Direct** — a `read_state` signed by the sync identity, so absence is
-  *proven* by the certificate rather than asserted. It requests `controllers`
-  alongside the metadata path, since only that distinguishes a canister with no
-  such section from one that was never created.
-- **Proxied** — `ProxyArgs` aimed at the management canister's
-  `canister_metadata`, so the controller check runs against the proxy. This is
-  the same shape the CLI's own management calls take through
-  `update_or_proxy_raw`; the runtime inlines it rather than depending on the CLI.
-
-Only a certificate can make a read `none`. The management canister answers a
-section that isn't there and one private to someone else with the same
-rejection, so the proxied route treats that rejection as a claim to check rather
-than an answer, and confirms it with a certified read before reporting absence.
-A plugin then sees one answer either way: no section by that name and no module
-installed at all are `none`; a private section it may not have, a canister that
-does not exist, and any other failure are errors.
+Both imports use `tokio::runtime::Handle::current().block_on(...)`, which is
+safe because `Wasmtime::run` wraps the synchronous runtime in
+`tokio::task::block_in_place`: a wasm import cannot suspend, so it needs a
+thread it may occupy, and only this crate knows that. For a v0.2.0 plugin the
+target is resolved from the request's `call-target` by `resolve_call_target`,
+which enforces the `callable` set; for a v0.1.0 plugin the target is always
+`host_canister_id`.
+
+### Reaching canisters
+
+Both host functions go through the invocation's `CanisterCalls`. That is what
+makes this crate agnostic about *how* a canister is reached — over HTTP with an
+`ic-agent`, or from inside another canister — which the WIT interface has
+always implied and the runtime now actually reflects.
+
+The two flags a request may carry map onto the seam:
+
+- `direct` picks the `Authority` a request is made under: `Direct` for the
+  caller itself, `Mediated` for whatever is acting on its behalf (`--proxy`).
+  Encoding a proxy call, funding it with the request's `cycles`, and unwrapping
+  its reply is the implementation's business.
+- `call-type` picks `update` or `query`. A query is always issued as `Direct`,
+  whatever the request asked, because the interface documents queries as going
+  straight to the target: an intermediary accepting only updates would
+  otherwise silently turn one into an update the plugin would have paid for.
+
+The same is true of `canister-metadata-section`, which is the seam's
+`metadata_section` with the request's `direct` flag as its `Authority`. What
+that read takes — a `read_state` certificate to verify, or a
+`canister_metadata` call the proxy makes so the controller check runs against
+*it* — is behind the seam, along with the rule that only a certificate can
+make an answer `none`. A plugin sees one answer either way: no section by that
+name and no module installed at all are `none`; a private section it may not
+have, a canister that does not exist, and any other failure are errors.
 
 ### Interface versioning (parallel v0.1.0 / v0.2.0 support)
 
 A component built with wit-bindgen imports the interface it `use`s as a
-versioned instance — `icp:sync-plugin/types@0.1.0` or `@0.2.0`. `run_plugin`
+versioned instance — `icp:sync-plugin/types@0.1.0` or `@0.2.0`. The runtime
 reads that name off `Component::component_type().imports(...)` and matches the
 version with semver caret requirements (`^0.1`, `^0.2`) to pick the ABI, then
 instantiates the matching `bindgen!` world and builds the matching
@@ -260,22 +269,23 @@ call blocks the guest while the host awaits the network, both imports record the
 elapsed time (`refund_host_call_time`) and the `epoch_deadline_callback` grants
 it back via `epoch_extension` — so network latency is *not* charged against the
 limit. The
-ticker thread stops when its RAII guard drops at the end of `run_plugin`.
+ticker thread stops when its RAII guard drops at the end of the run.
 
-The deadline in seconds is the `compute_limit_secs` parameter. The CLI resolves
-it from the `ICP_CLI_PLUGIN_COMPUTE_LIMIT_SECS` environment variable, defaulting
-to `DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS` (60) when unset.
+The deadline in seconds is the invocation's `compute_limit_secs`. `icp-project`
+resolves it from the `ICP_CLI_PLUGIN_COMPUTE_LIMIT_SECS` environment variable,
+defaulting to `DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS` (60) when unset — before the
+wasm is fetched, so a malformed value fails fast.
 
 ### stdio capture
 
 `LineCapture` implements `StdoutStream`/`OutputStream`, splits guest output on
 newlines, strips ANSI codes, and emits each complete line to the `reporter` as
 an output event for the rolling step view. stderr lines are additionally
-accumulated and returned from `run_plugin` so the CLI can reprint them
+accumulated and returned from the run so the CLI can reprint them
 persistently. Each stream is capped at 1 MiB; overflow is dropped and a single
 truncation note is emitted on `finalize`.
 
-### `crates/icp/src/manifest/adapter/plugin.rs`
+### `crates/icp-project/src/manifest/adapter/plugin.rs`
 
 Deserializes the `canister.yaml` fields into:
 
@@ -297,7 +307,7 @@ back out unchanged in shape. `entries()` flattens either form to ordered
 `(key, path)` pairs: `key` is `None` for a list entry and `Some(name)` for a map
 entry, and is *non-unique* — a map key holding a list of paths yields one entry
 per path, all sharing the key. The CLI passes those to the runtime as
-`KeyedPath`s (this crate stays free of manifest types), which surface in
+`KeyedPath`s (the runtime stays free of manifest types), which surface in
 `sync-exec-input.dirs`/`files` as each entry's `key`.
 
 Both shapes stay parseable here because both remain legal *somewhere* — which
@@ -319,16 +329,18 @@ section reaches the adapter as an already-parsed `serde_yaml::Value` (see
 `Value` keeps a number a number — hence the explicit visitor. Lists, mappings,
 and empty values are rejected: there is no string to hand the plugin.
 
-### `crates/icp/src/canister/sync/plugin.rs`
-
-Resolves the wasm (local read or remote HTTP fetch into the package cache),
-verifies sha256, builds the exposed canister ID table and the `CallableCanisters`
-enforcement set (resolving `canisters:` against the project's IDs), then calls
-`icp_sync_plugin::run_plugin(...)` with a `PluginInvocation`. The runtime — not
-the CLI — opens the declared paths and enforces the path-safety checks, so the
-CLI no longer touches the plugin's input files itself; it supplies the canister
-directory and the project directory (`sync::Params::path` and `project_dir`)
-that bound them. `exposed_canister_ids`
+### `crates/icp-project/src/canister/sync/plugin.rs`
+
+Declares the runner seam (`Run`, `Invocation`, `RunError`) and the two types an
+invocation is made of (`KeyedPath`, `CallableCanisters`), then does the
+manifest half of a plugin step: resolves the wasm (local read or remote HTTP
+fetch into the package cache), verifies sha256, builds the exposed canister ID
+table and the `CallableCanisters` enforcement set (resolving `canisters:`
+against the project's IDs), and hands an `Invocation` to the injected runner.
+The runner — not this layer — opens the declared paths and enforces the
+path-safety checks, so nothing here touches the plugin's input files; it
+supplies the canister directory and the project directory
+(`sync::Params::path` and `project_dir`) that bound them. `exposed_canister_ids`
 adds a bare-local-name duplicate for every canister in the same subproject as
 the one being synced; `resolve_callable` fails the step if a name in
 `canisters:` does not resolve.
diff --git a/crates/icp-sync-plugin/src/lib.rs b/crates/icp-sync-plugin/src/lib.rs
index a6f212fca..462912aa1 100644
--- a/crates/icp-sync-plugin/src/lib.rs
+++ b/crates/icp-sync-plugin/src/lib.rs
@@ -1,8 +1,4 @@
 mod path;
 mod runtime;
 
-pub use path::{covering_dirs, distinct_paths};
-pub use runtime::{
-    CallableCanisters, DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS, KeyedPath, PLUGIN_COMPUTE_LIMIT_ENV,
-    PluginInvocation, RunPluginError, run_plugin,
-};
+pub use runtime::Wasmtime;
diff --git a/crates/icp-sync-plugin/src/path.rs b/crates/icp-sync-plugin/src/path.rs
index c8ddf4ca4..1adb6146b 100644
--- a/crates/icp-sync-plugin/src/path.rs
+++ b/crates/icp-sync-plugin/src/path.rs
@@ -1,8 +1,6 @@
 //! Path-safety helpers used by the host runtime to validate declared `dirs`/`files`
 //! entries before preopening directories or reading files inside the project.
 
-use std::collections::HashSet;
-
 use camino::{Utf8Component, Utf8Path, Utf8PathBuf};
 
 /// Why a declared `dirs`/`files` entry cannot be anchored inside the sandbox
@@ -140,137 +138,6 @@ impl Resolved<'_> {
     }
 }
 
-/// The meaningful components of a declared relative path: the `/`-separated
-/// names, with empty and `.` components dropped, so `./data/` and `data` compare
-/// equal.
-///
-/// `\` is deliberately not a separator here. On Unix it is an ordinary character
-/// in a filename, and these comparisons decide what gets opened for a guest that
-/// will open the path exactly as written.
-fn components(path: &str) -> Vec<&str> {
-    path.split('/')
-        .filter(|part| !part.is_empty() && *part != ".")
-        .collect()
-}
-
-/// Reduce declared directories to the ones that actually have to be opened.
-///
-/// `dirs` is configuration as much as it is a sandbox grant: a plugin may
-/// legitimately be handed the same tree under several keys, or a tree and a
-/// subtree of it, and it is told about every entry that was declared. The grant
-/// behind those entries has no such multiplicity — opening a directory twice, or
-/// opening one already reachable through an ancestor, conveys no further access.
-/// Callers keep the declared list as configuration and open only what this
-/// returns; a nested declared directory is reached through the ancestor covering
-/// it.
-///
-/// Retained paths keep their written spelling and first-occurrence order.
-/// Comparison is over the written spelling rather than the resolved location,
-/// because the guest opens each entry at the spelling the manifest gave it, and
-/// is component-wise, so `data` covers `./data/inner` but not `database`.
-///
-/// A spelling prefix alone is not containment once entries may contain `..`:
-/// `..` is a prefix of `../../shared`, yet one is the canister directory's
-/// parent and the other a child of its grandparent — neither holds the other.
-/// So an entry only covers one whose remaining components descend, `..`-free.
-/// Two spellings that coincide only once resolved (`../data` and `data` from a
-/// canister in `data`'s parent) still stay separate, which merely leaves the
-/// result less reduced.
-pub fn covering_dirs<'a>(dirs: impl IntoIterator<Item = &'a str>) -> Vec<&'a str> {
-    let dirs: Vec<&str> = dirs.into_iter().collect();
-    let parts: Vec<Vec<&str>> = dirs.iter().map(|dir| components(dir)).collect();
-    dirs.iter()
-        .enumerate()
-        .filter(|(i, _)| {
-            !parts.iter().enumerate().any(|(j, other)| {
-                j != *i
-                    && parts[*i].starts_with(other)
-                    && !parts[*i][other.len()..].contains(&"..")
-                    // A strict ancestor always covers; between equals, the first written wins.
-                    && (other.len() < parts[*i].len() || j < *i)
-            })
-        })
-        .map(|(_, dir)| *dir)
-        .collect()
-}
-
-/// Reduce declared paths to the distinct ones, keeping the written spelling and
-/// first-occurrence order.
-///
-/// [`covering_dirs`] without the containment rule, for entries that name files:
-/// `./a.json` and `a.json` are one file, but a file never subsumes another the
-/// way a directory subsumes its contents.
-pub fn distinct_paths<'a>(paths: impl IntoIterator<Item = &'a str>) -> Vec<&'a str> {
-    let mut seen: HashSet<Vec<&str>> = HashSet::new();
-    paths
-        .into_iter()
-        .filter(|path| seen.insert(components(path)))
-        .collect()
-}
-
-#[cfg(test)]
-mod covering_tests {
-    use super::*;
-
-    #[test]
-    fn unrelated_dirs_are_all_kept() {
-        assert_eq!(
-            covering_dirs(["assets", "config", "data/seed"]),
-            ["assets", "config", "data/seed"],
-        );
-    }
-
-    #[test]
-    fn duplicates_collapse_to_the_first_spelling() {
-        assert_eq!(covering_dirs(["./data", "data", "data/"]), ["./data"]);
-    }
-
-    #[test]
-    fn nested_dirs_collapse_to_their_ancestor_whichever_is_written_first() {
-        assert_eq!(covering_dirs(["data", "data/inner"]), ["data"]);
-        assert_eq!(covering_dirs(["data/inner", "data"]), ["data"]);
-        // Transitive: `data` covers `data/a` covers `data/a/b`.
-        assert_eq!(covering_dirs(["data/a/b", "data/a", "data"]), ["data"]);
-    }
-
-    #[test]
-    fn a_name_prefix_is_not_an_ancestor() {
-        assert_eq!(covering_dirs(["data", "database"]), ["data", "database"]);
-    }
-
-    /// An entry reaching further out than another is not inside it, however
-    /// much of its spelling they share: `..` is the canister directory's
-    /// parent, `../../shared` a child of its grandparent. Collapsing them would
-    /// leave the second with no preopen of its own and none that contains it.
-    #[test]
-    fn an_entry_that_rises_further_is_not_covered() {
-        assert_eq!(
-            covering_dirs(["..", "../../shared"]),
-            ["..", "../../shared"],
-        );
-        assert_eq!(
-            covering_dirs(["../shared", "../shared/../assets"]),
-            ["../shared", "../shared/../assets"],
-        );
-    }
-
-    /// Entries that reach out of the canister directory still cover what
-    /// descends from them, and still collapse with a repeat of themselves.
-    #[test]
-    fn entries_outside_the_canister_dir_cover_their_own_contents() {
-        assert_eq!(covering_dirs(["../data", "../data/inner"]), ["../data"]);
-        assert_eq!(covering_dirs(["../data", "./../data"]), ["../data"]);
-    }
-
-    #[test]
-    fn distinct_paths_dedupes_without_containment() {
-        assert_eq!(
-            distinct_paths(["./a.json", "a.json", "b.json", "dir/a.json"]),
-            ["./a.json", "b.json", "dir/a.json"],
-        );
-    }
-}
-
 #[cfg(test)]
 mod resolve_tests {
     use super::*;
diff --git a/crates/icp-sync-plugin/src/runtime.rs b/crates/icp-sync-plugin/src/runtime.rs
index 5f2098843..63909d1de 100644
--- a/crates/icp-sync-plugin/src/runtime.rs
+++ b/crates/icp-sync-plugin/src/runtime.rs
@@ -10,30 +10,22 @@ use std::time::{Duration, Instant};
 const MAX_PLUGIN_OUTPUT: usize = 1024 * 1024; // 1 MiB per stream
 // Maximum wasm call-stack depth (in bytes).
 const MAX_WASM_STACK: usize = 512 * 1024;
-/// Default seconds of pure wasm compute a plugin may use (host-call latency is
-/// excluded). This is a runaway guard, not a security boundary: the plugin runs
-/// locally in a read-only WASI sandbox, so the limit only protects the machine
-/// running `icp sync` from a plugin that never terminates. Legitimately heavy
-/// plugins (e.g. brotli-compressing a large asset bundle) can exceed it,
-/// especially on slower CI runners, so it is overridable via the
-/// [`PLUGIN_COMPUTE_LIMIT_ENV`] environment variable.
-pub const DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS: u64 = 60;
-/// Environment variable that overrides [`DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS`].
-pub const PLUGIN_COMPUTE_LIMIT_ENV: &str = "ICP_CLI_PLUGIN_COMPUTE_LIMIT_SECS";
 
+use async_trait::async_trait;
 use bytes::Bytes;
 use camino::{Utf8Path, Utf8PathBuf};
-use candid::{Encode, Principal};
-use ic_agent::Agent;
-use ic_agent::hash_tree::{Label, LookupResult};
-use ic_management_canister_types::{CanisterMetadataArgs, CanisterMetadataResult};
-use icp_canister_interfaces::proxy::{ProxyArgs, ProxyResult};
+use candid::Principal;
+use icp_project::calls::{Authority, Call, CanisterCalls};
+use icp_project::canister::sync::declared::covering_dirs;
+use icp_project::canister::sync::plugin::{
+    CallableCanisters, Invocation, KeyedPath, PLUGIN_COMPUTE_LIMIT_ENV, Run, RunError,
+};
+use icp_project::error;
 use semver::{Version, VersionReq};
 use snafu::prelude::*;
 // Aliased because wasmtime-wasi also has an `OutputStream` (imported below).
 use icp_events::{OutputStream as EventStream, StepReporter};
 use tokio::io::{self, AsyncWrite};
-use url::Url;
 use wasmtime::component::{Component, HasSelf, Linker};
 use wasmtime::{Config, Engine, Store};
 use wasmtime_wasi::cli::{IsTerminal, StdoutStream};
@@ -61,21 +53,6 @@ mod v1 {
 
 use v2::icp::sync_plugin::types::{CallTarget, CallType, CanisterIdEntry};
 
-/// A manifest path passed to a plugin, tagged with the map key it was declared
-/// under. Both `dirs` and `files` are lists of these.
-///
-/// The key is `None` when the manifest wrote the setting as a plain list, and
-/// `Some(name)` when it wrote a map. It is *non-unique*: several paths share a
-/// key when a map key resolves to a list of paths. Which form a plugin accepts
-/// depends on the interface it was built against — see [`PluginAbi`].
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct KeyedPath {
-    /// The map key this path was declared under, or `None` for a plain-list entry.
-    pub key: Option<String>,
-    /// Manifest-relative path, anchored at the invocation's `base_dir`.
-    pub path: String,
-}
-
 /// A declared entry the host has resolved to a location on disk, and what it
 /// found there. Held version-agnostically so it can be converted to whichever
 /// interface version's records the plugin turns out to use.
@@ -92,100 +69,6 @@ struct ResolvedEntry {
     content: Option<String>,
 }
 
-/// The canisters a sync plugin is permitted to call, beyond the canister being
-/// synced (which is always reachable via [`CallTarget::Host`]).
-///
-/// Built by the CLI from the plugin step's `canisters` list, resolved against
-/// the project's canister ID table. Keeping the resolution on the CLI side
-/// keeps this runtime crate free of any manifest knowledge.
-#[derive(Clone, Debug, Default)]
-pub struct CallableCanisters {
-    /// Canisters callable by name ([`CallTarget::Name`]). Maps the name — as it
-    /// appears in the canister ID table — to the principal it resolves to.
-    pub by_name: BTreeMap<String, Principal>,
-}
-
-/// What a certificate says about a metadata section. A section the reader may
-/// not have is neither of these: the state tree will not certify it, so it
-/// reaches the caller as an error like any other failed read.
-enum CertifiedSection {
-    Present(Vec<u8>),
-    Absent,
-}
-
-/// Ask the target's subnet to certify a metadata section, reporting only what
-/// the certificate proves.
-///
-/// The section path is requested 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 [`CertifiedSection::Absent`].
-async fn certified_metadata_section(
-    agent: &Agent,
-    target: Principal,
-    name: &str,
-) -> Result<CertifiedSection, String> {
-    let metadata_path: Vec<Label<Vec<u8>>> = vec![
-        "canister".into(),
-        Label::from_bytes(target.as_slice()),
-        "metadata".into(),
-        name.into(),
-    ];
-    let controllers_path: Vec<Label<Vec<u8>>> = vec![
-        "canister".into(),
-        Label::from_bytes(target.as_slice()),
-        "controllers".into(),
-    ];
-    let cert = agent
-        .read_state_raw(
-            vec![metadata_path.clone(), controllers_path.clone()],
-            target,
-        )
-        .await
-        .map_err(|err| format!("metadata read failed: {err}"))?;
-
-    match cert.tree.lookup_path(&metadata_path) {
-        LookupResult::Found(bytes) => Ok(CertifiedSection::Present(bytes.to_vec())),
-        LookupResult::Absent => match cert.tree.lookup_path(&controllers_path) {
-            LookupResult::Found(_) => Ok(CertifiedSection::Absent),
-            LookupResult::Absent => Err(format!("canister {target} does not exist")),
-            _ => Err(format!(
-                "metadata read failed: certificate proves nothing about canister {target}"
-            )),
-        },
-        // Not proof of absence, just a certificate that says nothing about the
-        // path — reporting the section missing off this would be a guess.
-        _ => Err(format!(
-            "metadata read failed: certificate proves nothing about section `{name}` \
-             of canister {target}"
-        )),
-    }
-}
-
-/// Whether the management canister rejected a metadata read by claiming the
-/// target has no such section, rather than because the read itself failed.
-///
-/// The claim is not proof: the same rejection covers a section private to
-/// someone other than the proxy, so the caller confirms it against a
-/// certificate. A proxied read reaches the plugin as reject text with no code
-/// attached, so recognizing the claim at all means matching the replica's
-/// wording. Both sentences name the canister and one names the section, so the
-/// match is anchored on the values this call supplied rather than on a loose
-/// phrase that text relayed from elsewhere might happen to contain. A reword
-/// upstream turns the claim into an error rather than into a wrong answer.
-fn rejected_as_no_such_section(message: &str, target: Principal, name: &str) -> bool {
-    // A canister with no module installed has no sections at all, so it reports
-    // absence in its own words. The certificate says the same thing about it:
-    // the metadata path is absent while the canister itself is there.
-    message.contains(&format!(
-        "The canister {target} has no Wasm module and hence no metadata is available."
-    )) || message.contains(&format!(
-        "The canister {target} has no metadata section with the name {name}."
-    ))
-}
-
 /// Resolve a plugin-supplied [`CallTarget`] to a concrete principal, enforcing
 /// that the plugin listed it in `canisters`. The canister being synced (`host`)
 /// is always permitted.
@@ -211,10 +94,10 @@ struct HostState {
     host_canister_id: Principal,
     /// Canisters the plugin declared in `canisters` and may also call.
     callable: CallableCanisters,
-    agent: Arc<Agent>,
-    /// Proxy canister to route update calls and metadata reads through, if
-    /// configured.
-    proxy: Option<Principal>,
+    /// How the plugin's calls and metadata reads are made. Whether they go
+    /// through a proxy canister, and what certification a read takes, is
+    /// entirely behind this.
+    calls: Arc<dyn CanisterCalls>,
     // WASI context. Preopened directories in this context are the only
     // filesystem locations the plugin can access.
     wasi_ctx: wasmtime_wasi::WasiCtx,
@@ -248,132 +131,60 @@ impl HostState {
         direct: bool,
         cycles: u64,
     ) -> Result<Vec<u8>, String> {
-        let agent = Arc::clone(&self.agent);
-        let proxy = if direct { None } else { self.proxy };
+        let calls = Arc::clone(&self.calls);
+        let query = matches!(call_type, CallType::Query);
+        let mut call = Call::new(target, method, arg_bytes).with_cycles(cycles.into());
+        // A query goes to the target itself whichever way the plugin asked,
+        // which is what the interface documents: an intermediary that only
+        // accepts updates would otherwise turn one into an update, and the
+        // plugin would have paid for it.
+        if direct || query {
+            call = call.direct();
+        }
 
-        // We are already inside tokio::task::block_in_place (see sync/plugin.rs),
+        // We are already inside tokio::task::block_in_place (see `Wasmtime`),
         // so blocking the thread here is safe.
         let start = Instant::now();
         let result = tokio::runtime::Handle::current().block_on(async move {
-            match call_type {
-                CallType::Update => {
-                    if let Some(proxy_cid) = proxy {
-                        let proxy_args = ProxyArgs {
-                            canister_id: target,
-                            method: method.clone(),
-                            args: arg_bytes,
-                            cycles: candid::Nat::from(cycles),
-                        };
-                        let encoded = Encode!(&proxy_args)
-                            .map_err(|e| format!("proxy encode failed: {e}"))?;
-                        let raw = agent
-                            .update(&proxy_cid, "proxy")
-                            .with_arg(encoded)
-                            .await
-                            .map_err(|e| format!("proxy call failed: {e}"))?;
-                        let (result,): (ProxyResult,) = candid::decode_args(&raw)
-                            .map_err(|e| format!("proxy decode failed: {e}"))?;
-                        match result {
-                            ProxyResult::Ok(ok) => Ok(ok.result),
-                            ProxyResult::Err(err) => Err(err.format_error()),
-                        }
-                    } else {
-                        agent
-                            .update(&target, &method)
-                            .with_arg(arg_bytes)
-                            .await
-                            .map_err(|e| format!("canister call failed: {e}"))
-                    }
-                }
-                CallType::Query => agent
-                    .query(&target, &method)
-                    .with_arg(arg_bytes)
-                    .call()
-                    .await
-                    .map_err(|e| format!("canister call failed: {e}")),
+            match query {
+                false => calls.update(call).await,
+                true => calls.query(call).await,
             }
+            // The guest gets one string, and a `CallError`'s own message names
+            // only the call it was; the reason it failed is down the chain.
+            .map_err(|err| error::flatten(&err))
         });
         self.refund_host_call_time(start);
         result
     }
 
     /// Read a metadata section from an already-resolved target principal.
-    /// `Ok(None)` means a certificate proved the target has no such section,
-    /// kept distinct from a failed read so a plugin can probe for an optional
-    /// section without inspecting error text. A section the reader may not have
-    /// is a failed read, not an absent one, whichever route asked.
+    /// `Ok(None)` means the target provably has no such section, kept distinct
+    /// from a failed read so a plugin can probe for an optional section
+    /// without inspecting error text.
     ///
-    /// A direct read is a certified `read_state` signed by the sync identity —
-    /// `read_state` is not a canister method, so it cannot be forwarded. A
-    /// proxied read therefore goes the other way around: the proxy calls the
-    /// management canister's `canister_metadata` on the plugin's behalf, which
-    /// checks the *proxy* against the target's controllers and so reaches
-    /// sections private to it. The management canister does not distinguish
-    /// absence from privacy, so a proxied read that comes back claiming absence
-    /// is confirmed against a certificate before it is reported as one.
+    /// `direct` picks who does the reading, which is what a private section is
+    /// gated on: the sync identity itself, or whatever is acting on its behalf.
+    /// Everything that takes — a certificate to verify, a management-canister
+    /// call to make — is behind the seam.
     fn do_canister_metadata_section(
         &mut self,
         target: Principal,
         name: String,
         direct: bool,
     ) -> Result<Option<Vec<u8>>, String> {
-        let agent = Arc::clone(&self.agent);
-        let proxy = if direct { None } else { self.proxy };
+        let calls = Arc::clone(&self.calls);
+        let authority = match direct {
+            true => Authority::Direct,
+            false => Authority::Mediated,
+        };
 
         let start = Instant::now();
         let result = tokio::runtime::Handle::current().block_on(async move {
-            let Some(proxy_cid) = proxy else {
-                return certified_metadata_section(&agent, target, &name)
-                    .await
-                    .map(|section| match section {
-                        CertifiedSection::Present(bytes) => Some(bytes),
-                        CertifiedSection::Absent => None,
-                    });
-            };
-
-            let metadata_args = Encode!(&CanisterMetadataArgs {
-                canister_id: target,
-                name: name.clone(),
-            })
-            .map_err(|e| format!("metadata encode failed: {e}"))?;
-            let proxy_args = ProxyArgs {
-                canister_id: Principal::management_canister(),
-                method: "canister_metadata".to_string(),
-                args: metadata_args,
-                cycles: candid::Nat::from(0u8),
-            };
-            let encoded = Encode!(&proxy_args).map_err(|e| format!("proxy encode failed: {e}"))?;
-            let raw = agent
-                .update(&proxy_cid, "proxy")
-                .with_arg(encoded)
+            calls
+                .metadata_section(target, &name, authority)
                 .await
-                .map_err(|e| format!("proxy call failed: {e}"))?;
-            let (result,): (ProxyResult,) =
-                candid::decode_args(&raw).map_err(|e| format!("proxy decode failed: {e}"))?;
-            match result {
-                ProxyResult::Ok(ok) => {
-                    let (metadata,): (CanisterMetadataResult,) = candid::decode_args(&ok.result)
-                        .map_err(|e| format!("metadata decode failed: {e}"))?;
-                    Ok(Some(metadata.value))
-                }
-                ProxyResult::Err(err) => {
-                    let message = err.format_error();
-                    if !rejected_as_no_such_section(&message, target, &name) {
-                        return Err(format!("metadata read failed: {message}"));
-                    }
-                    // The management canister says the same thing about a
-                    // section that isn't there and one that is private to
-                    // someone else, so its word alone cannot be reported as
-                    // absence. Only a certificate proves the section absent.
-                    match certified_metadata_section(&agent, target, &name).await? {
-                        CertifiedSection::Absent => Ok(None),
-                        CertifiedSection::Present(_) => Err(format!(
-                            "metadata read failed: canister {target} does not let the proxy \
-                             read section `{name}`"
-                        )),
-                    }
-                }
-            }
+                .map_err(|err| error::flatten(&err))
         });
         self.refund_host_call_time(start);
         result
@@ -720,65 +531,25 @@ fn resolve_entries<'a>(
     Ok(entries)
 }
 
-/// Everything [`run_plugin`] needs to load and drive one sync plugin.
-#[derive(Debug)]
-pub struct PluginInvocation {
-    /// On-disk path to the plugin's wasm component.
-    pub wasm_path: Utf8PathBuf,
-    /// Directory the declared `dirs`/`files` are anchored at (the canister dir).
-    pub base_dir: Utf8PathBuf,
-    /// The project directory: the sandbox boundary. A declared path may rise
-    /// out of `base_dir` with `..` and reach anything inside the project, but
-    /// nothing above it.
-    ///
-    /// A `base_dir` that does not lie within this directory — a dependency
-    /// project reached by an out-of-tree `path:` — is its own boundary instead,
-    /// which grants nothing above the canister directory.
-    pub project_dir: Utf8PathBuf,
-    /// The manifest's `dirs:` entries: directories to preopen read-only. Only
-    /// v0.1.0 plugins have a `dirs` list to receive them; declaring any
-    /// alongside a v0.2.0 plugin is an error.
-    pub dirs: Vec<KeyedPath>,
-    /// The manifest's `files:` entries. For a v0.1.0 plugin these are files to
-    /// read and pass inline; for a v0.2.0 plugin the list holds directories
-    /// too, and the host preopens or reads each by what is on disk.
-    pub files: Vec<KeyedPath>,
-    /// Key-value fields to pass inline. Passed to v0.2.0 plugins; ignored by
-    /// v0.1.0 plugins, whose interface has no `fields`.
-    pub fields: BTreeMap<String, String>,
-    /// The canister being synced. Reachable via `call-target::host`.
-    pub host_canister_id: Principal,
-    /// Agent used for canister calls.
-    pub agent: Agent,
-    /// Proxy canister to route update calls and metadata reads through, if
-    /// configured.
-    pub proxy: Option<Principal>,
-    /// Signing identity principal, surfaced to the plugin.
-    pub identity_principal: Principal,
-    /// Name of the environment being synced.
-    pub environment: String,
-    /// The network's API endpoint — where canister calls are submitted.
-    /// Surfaced to v0.2.0 plugins; v0.1.0 plugins have no field for it.
-    pub api_url: Url,
-    /// The network's HTTP gateway, when it exposes one. Surfaced to v0.2.0
-    /// plugins; v0.1.0 plugins have no field for it.
-    pub gateway_url: Option<Url>,
-    /// Pure-wasm compute-time budget in seconds.
-    pub compute_limit_secs: u64,
-    /// The project's canister ID table for this environment, as exposed to the
-    /// plugin. Same-project canisters appear both under their fully-qualified
-    /// key and their bare local name (see the WIT `canister-id-entry` docs).
-    pub canister_ids: BTreeMap<String, Principal>,
-    /// Canisters the plugin declared in `canisters` and may call, beyond the
-    /// canister being synced. Ignored by v0.1.0 plugins, which can only reach
-    /// the canister being synced.
-    pub callable: CallableCanisters,
-    /// Reporter the plugin's live stdout/stderr is emitted on.
-    pub reporter: StepReporter,
+/// The wasmtime Component Model runtime for sync plugins: the host half of the
+/// `icp:sync-plugin` world, which loads a plugin component, hands it a
+/// read-only WASI sandbox, and serves the imports it declares.
+#[derive(Debug, Default, Clone, Copy)]
+pub struct Wasmtime;
+
+#[async_trait]
+impl Run for Wasmtime {
+    async fn run(&self, invocation: Invocation) -> Result<Vec<String>, RunError> {
+        // The runtime is synchronous — a wasm import cannot suspend — and it
+        // blocks on the calls the plugin makes, so it needs a thread it is
+        // allowed to occupy. Only the runtime knows that; the operation
+        // driving the sync does not.
+        tokio::task::block_in_place(|| run_plugin(invocation)).map_err(RunError::new)
+    }
 }
 
-pub fn run_plugin(invocation: PluginInvocation) -> Result<Vec<String>, RunPluginError> {
-    let PluginInvocation {
+fn run_plugin(invocation: Invocation) -> Result<Vec<String>, RunPluginError> {
+    let Invocation {
         wasm_path,
         base_dir,
         project_dir,
@@ -786,9 +557,8 @@ pub fn run_plugin(invocation: PluginInvocation) -> Result<Vec<String>, RunPlugin
         files,
         fields,
         host_canister_id,
-        agent,
+        calls,
         proxy,
-        identity_principal,
         environment,
         api_url,
         gateway_url,
@@ -891,7 +661,7 @@ pub fn run_plugin(invocation: PluginInvocation) -> Result<Vec<String>, RunPlugin
         .map(|entry| (entry.path.as_str(), &entry.host_path))
         .collect();
     let mut wasi_builder = wasmtime_wasi::WasiCtxBuilder::new();
-    for dir in crate::path::covering_dirs(declared_dirs.iter().map(|entry| entry.path.as_str())) {
+    for dir in covering_dirs(declared_dirs.iter().map(|entry| entry.path.as_str())) {
         // `covering_dirs` returns a subset of the spellings it was given, so
         // every one of them is in the map.
         let host_path = host_paths[dir];
@@ -920,11 +690,11 @@ pub fn run_plugin(invocation: PluginInvocation) -> Result<Vec<String>, RunPlugin
         .stderr(stderr_capture.clone());
 
     let epoch_extension = Arc::new(AtomicU64::new(0));
+    let identity_principal = calls.caller();
     let host_state = HostState {
         host_canister_id,
         callable,
-        agent: Arc::new(agent),
-        proxy,
+        calls,
         wasi_ctx: wasi_builder.build(),
         wasi_table: wasmtime_wasi::ResourceTable::new(),
         epoch_extension: epoch_extension.clone(),
@@ -1220,14 +990,9 @@ mod tests {
     use super::*;
 
     use candid::Principal;
-    use ic_agent::Agent;
-
-    fn dummy_agent() -> Agent {
-        Agent::builder()
-            .with_url("http://127.0.0.1:4943")
-            .build()
-            .expect("build test agent")
-    }
+    use icp_project::calls::{CallError, RouteTo, UnimplementedMockCalls};
+    use icp_project::canister::sync::plugin::DEFAULT_PLUGIN_COMPUTE_LIMIT_SECS;
+    use url::Url;
 
     fn anon() -> Principal {
         Principal::anonymous()
@@ -1257,13 +1022,13 @@ mod tests {
             .collect()
     }
 
-    /// A [`PluginInvocation`] with test-friendly defaults: anonymous canister
-    /// and identity, no proxy, no declared callable canisters, the default
-    /// compute limit, a local network with no gateway of its own, and the
-    /// current directory as both the base and the project. Tests override the
-    /// few fields they care about.
-    fn invocation(wasm_path: &str, environment: &str) -> PluginInvocation {
-        PluginInvocation {
+    /// An [`Invocation`] with test-friendly defaults: an anonymous canister,
+    /// a caller that panics if the plugin actually calls one, no proxy, no
+    /// declared callable canisters, the default compute limit, a local network
+    /// with no gateway of its own, and the current directory as both the base
+    /// and the project. Tests override the few fields they care about.
+    fn invocation(wasm_path: &str, environment: &str) -> Invocation {
+        Invocation {
             wasm_path: wasm_path.into(),
             base_dir: ".".into(),
             project_dir: ".".into(),
@@ -1271,9 +1036,8 @@ mod tests {
             files: vec![],
             fields: BTreeMap::new(),
             host_canister_id: anon(),
-            agent: dummy_agent(),
+            calls: Arc::new(UnimplementedMockCalls),
             proxy: None,
-            identity_principal: anon(),
             environment: environment.to_string(),
             api_url: Url::parse("http://127.0.0.1:4943").expect("valid api url"),
             gateway_url: None,
@@ -1317,6 +1081,377 @@ mod tests {
         );
     }
 
+    // -------------------------------------------------------------------------
+    // The canister-call seam — what a plugin's request becomes on the way out
+    // -------------------------------------------------------------------------
+
+    type CallRequest = v2::icp::sync_plugin::types::CanisterCallRequest;
+    type MetadataRequest = v2::icp::sync_plugin::types::MetadataSectionRequest;
+
+    /// A [`CanisterCalls`] that records what it was asked for and answers with
+    /// canned bytes, so the request a plugin made can be read back as the call
+    /// it turned into.
+    #[derive(Default)]
+    struct RecordingCalls {
+        updates: StdMutex<Vec<Call>>,
+        queries: StdMutex<Vec<Call>>,
+        metadata: StdMutex<Vec<(Principal, String, Authority)>>,
+    }
+
+    #[async_trait]
+    impl CanisterCalls for RecordingCalls {
+        fn caller(&self) -> Principal {
+            Principal::from_slice(&[5; 4])
+        }
+
+        async fn update(&self, call: Call) -> Result<Vec<u8>, CallError> {
+            self.updates.lock().unwrap().push(call);
+            Ok(b"reply".to_vec())
+        }
+
+        async fn query(&self, call: Call) -> Result<Vec<u8>, CallError> {
+            self.queries.lock().unwrap().push(call);
+            Ok(b"reply".to_vec())
+        }
+
+        async fn metadata_section(
+            &self,
+            canister: Principal,
+            path: &str,
+            authority: Authority,
+        ) -> Result<Option<Vec<u8>>, CallError> {
+            self.metadata
+                .lock()
+                .unwrap()
+                .push((canister, path.to_owned(), authority));
+            Ok(Some(b"section".to_vec()))
+        }
+
+        async fn controllers(
+            &self,
+            _canister: Principal,
+        ) -> Result<Option<Vec<Principal>>, CallError> {
+            unimplemented!("RecordingCalls::controllers")
+        }
+
+        async fn module_hash(&self, _canister: Principal) -> Result<Option<Vec<u8>>, CallError> {
+            unimplemented!("RecordingCalls::module_hash")
+        }
+
+        async fn subnet_of(&self, _canister: Principal) -> Result<Principal, CallError> {
+            unimplemented!("RecordingCalls::subnet_of")
+        }
+
+        async fn subnet_uses_engine_operator(&self, _subnet: Principal) -> Result<bool, CallError> {
+            unimplemented!("RecordingCalls::subnet_uses_engine_operator")
+        }
+    }
+
+    /// What a call fails with underneath — an error the seam's own message does
+    /// not restate, so it only reaches the plugin if the chain is flattened.
+    #[derive(Debug, Snafu)]
+    #[snafu(display("the transport gave up"))]
+    struct TransportGaveUp;
+
+    /// A [`CanisterCalls`] whose every call fails.
+    struct FailingCalls;
+
+    #[async_trait]
+    impl CanisterCalls for FailingCalls {
+        fn caller(&self) -> Principal {
+            Principal::anonymous()
+        }
+
+        async fn update(&self, call: Call) -> Result<Vec<u8>, CallError> {
+            Err(CallError::failed(
+                call.canister,
+                call.method,
+                TransportGaveUp,
+            ))
+        }
+
+        async fn query(&self, _call: Call) -> Result<Vec<u8>, CallError> {
+            unimplemented!("FailingCalls::query")
+        }
+
+        async fn metadata_section(
+            &self,
+            _canister: Principal,
+            _path: &str,
+            _authority: Authority,
+        ) -> Result<Option<Vec<u8>>, CallError> {
+            unimplemented!("FailingCalls::metadata_section")
+        }
+
+        async fn controllers(
+            &self,
+            _canister: Principal,
+        ) -> Result<Option<Vec<Principal>>, CallError> {
+            unimplemented!("FailingCalls::controllers")
+        }
+
+        async fn module_hash(&self, _canister: Principal) -> Result<Option<Vec<u8>>, CallError> {
+            unimplemented!("FailingCalls::module_hash")
+        }
+
+        async fn subnet_of(&self, _canister: Principal) -> Result<Principal, CallError> {
+            unimplemented!("FailingCalls::subnet_of")
+        }
+
+        async fn subnet_uses_engine_operator(&self, _subnet: Principal) -> Result<bool, CallError> {
+            unimplemented!("FailingCalls::subnet_uses_engine_operator")
+        }
+    }
+
+    /// The host state a plugin's imports are served from, with an empty WASI
+    /// sandbox: these tests call the imports directly rather than through a
+    /// component, so nothing reads it.
+    fn host_state(
+        host_canister_id: Principal,
+        callable: CallableCanisters,
+        calls: Arc<dyn CanisterCalls>,
+    ) -> HostState {
+        HostState {
+            host_canister_id,
+            callable,
+            calls,
+            wasi_ctx: wasmtime_wasi::WasiCtxBuilder::new().build(),
+            wasi_table: wasmtime_wasi::ResourceTable::new(),
+            epoch_extension: Arc::new(AtomicU64::new(0)),
+        }
+    }
+
+    fn call_request(
+        target: CallTarget,
+        call_type: CallType,
+        direct: bool,
+        cycles: u64,
+    ) -> CallRequest {
+        CallRequest {
+            target,
+            method: "register".to_string(),
+            arg: b"arg".to_vec(),
+            call_type,
+            direct,
+            cycles,
+        }
+    }
+
+    /// An update request becomes an update call on the seam, routed to the
+    /// target itself and made under the caller's mediated authority — whatever
+    /// intermediary that entails is behind the seam, not decided here.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn an_update_request_becomes_a_mediated_update_call() {
+        let calls = Arc::new(RecordingCalls::default());
+        let host = Principal::from_slice(&[1; 4]);
+        let mut state = host_state(host, CallableCanisters::default(), calls.clone());
+
+        let reply = tokio::task::block_in_place(|| {
+            v2::SyncPluginImports::canister_call(
+                &mut state,
+                call_request(CallTarget::Host, CallType::Update, false, 0),
+            )
+        })
+        .expect("the call should reach the seam");
+
+        assert_eq!(reply, b"reply");
+        assert!(calls.queries.lock().unwrap().is_empty());
+        let updates = calls.updates.lock().unwrap();
+        let [call] = &updates[..] else {
+            panic!("expected exactly one update, got {}", updates.len());
+        };
+        assert_eq!(call.canister, host);
+        assert_eq!(call.method, "register");
+        assert_eq!(call.arg, b"arg");
+        assert_eq!(call.route, RouteTo::Callee);
+        assert_eq!(call.cycles, 0);
+        assert_eq!(call.authority, Authority::Mediated);
+    }
+
+    /// `direct` asks for the call to be made by the caller itself, and the
+    /// cycles the plugin attached ride along whichever way it asked.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_direct_request_carries_its_authority_and_cycles() {
+        let calls = Arc::new(RecordingCalls::default());
+        let host = Principal::from_slice(&[1; 4]);
+        let mut state = host_state(host, CallableCanisters::default(), calls.clone());
+
+        tokio::task::block_in_place(|| {
+            v2::SyncPluginImports::canister_call(
+                &mut state,
+                call_request(CallTarget::Host, CallType::Update, true, 25_000_000),
+            )
+        })
+        .expect("the call should reach the seam");
+
+        let updates = calls.updates.lock().unwrap();
+        assert_eq!(updates[0].authority, Authority::Direct);
+        assert_eq!(updates[0].cycles, 25_000_000);
+    }
+
+    /// A query is made by the caller itself however the plugin asked, which is
+    /// what the interface documents: an intermediary that only accepts updates
+    /// would otherwise turn the query into one the plugin paid for.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_query_request_is_made_directly_even_when_mediated_was_asked() {
+        let calls = Arc::new(RecordingCalls::default());
+        let host = Principal::from_slice(&[1; 4]);
+        let mut state = host_state(host, CallableCanisters::default(), calls.clone());
+
+        tokio::task::block_in_place(|| {
+            v2::SyncPluginImports::canister_call(
+                &mut state,
+                call_request(CallTarget::Host, CallType::Query, false, 0),
+            )
+        })
+        .expect("the call should reach the seam");
+
+        assert!(calls.updates.lock().unwrap().is_empty());
+        let queries = calls.queries.lock().unwrap();
+        let [call] = &queries[..] else {
+            panic!("expected exactly one query, got {}", queries.len());
+        };
+        assert_eq!(call.authority, Authority::Direct);
+    }
+
+    /// A declared target is called by the principal its name resolved to, not
+    /// by the canister being synced.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_declared_target_is_called_by_its_resolved_principal() {
+        let calls = Arc::new(RecordingCalls::default());
+        let host = Principal::from_slice(&[1; 4]);
+        let dep = Principal::from_slice(&[2; 4]);
+        let callable = CallableCanisters {
+            by_name: BTreeMap::from([("backend".to_string(), dep)]),
+        };
+        let mut state = host_state(host, callable, calls.clone());
+
+        tokio::task::block_in_place(|| {
+            v2::SyncPluginImports::canister_call(
+                &mut state,
+                call_request(
+                    CallTarget::Name("backend".into()),
+                    CallType::Update,
+                    false,
+                    0,
+                ),
+            )
+        })
+        .expect("a declared target should be callable");
+
+        assert_eq!(calls.updates.lock().unwrap()[0].canister, dep);
+    }
+
+    /// An undeclared target is refused before the seam is touched, so nothing
+    /// is submitted on the plugin's behalf.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn an_undeclared_target_never_reaches_the_seam() {
+        let calls = Arc::new(RecordingCalls::default());
+        let host = Principal::from_slice(&[1; 4]);
+        let mut state = host_state(host, CallableCanisters::default(), calls.clone());
+
+        let err = tokio::task::block_in_place(|| {
+            v2::SyncPluginImports::canister_call(
+                &mut state,
+                call_request(
+                    CallTarget::Name("frontend".into()),
+                    CallType::Update,
+                    false,
+                    0,
+                ),
+            )
+        })
+        .expect_err("an undeclared target must be rejected");
+
+        assert!(err.contains("not permitted"), "got: {err}");
+        assert!(calls.updates.lock().unwrap().is_empty());
+    }
+
+    /// `direct` on a metadata read picks who does the reading — which is what a
+    /// private section is gated on — and the section's bytes come back as they
+    /// were read.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_metadata_read_passes_the_authority_it_was_asked_for() {
+        let calls = Arc::new(RecordingCalls::default());
+        let host = Principal::from_slice(&[1; 4]);
+        let mut state = host_state(host, CallableCanisters::default(), calls.clone());
+
+        for direct in [true, false] {
+            let section = tokio::task::block_in_place(|| {
+                v2::SyncPluginImports::canister_metadata_section(
+                    &mut state,
+                    MetadataRequest {
+                        target: CallTarget::Host,
+                        name: "candid:service".to_string(),
+                        direct,
+                    },
+                )
+            })
+            .expect("the read should reach the seam");
+            assert_eq!(section.as_deref(), Some(&b"section"[..]));
+        }
+
+        assert_eq!(
+            &calls.metadata.lock().unwrap()[..],
+            [
+                (host, "candid:service".to_owned(), Authority::Direct),
+                (host, "candid:service".to_owned(), Authority::Mediated),
+            ]
+        );
+    }
+
+    /// The guest gets one string, and a failed call's own message names only
+    /// the call it was — the reason lives down its source chain, so it has to
+    /// be flattened into what the plugin is told.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_failed_call_tells_the_plugin_why() {
+        let host = Principal::from_slice(&[1; 4]);
+        let mut state = host_state(host, CallableCanisters::default(), Arc::new(FailingCalls));
+
+        let err = tokio::task::block_in_place(|| {
+            v2::SyncPluginImports::canister_call(
+                &mut state,
+                call_request(CallTarget::Host, CallType::Update, false, 0),
+            )
+        })
+        .expect_err("the call must fail");
+
+        assert!(err.contains("call to 'register'"), "got: {err}");
+        assert!(err.contains("the transport gave up"), "got: {err}");
+    }
+
+    /// The v0.1.0 interface has no target field, so its calls always reach the
+    /// canister being synced — a declared canister is unreachable from it — and
+    /// its `direct`/`cycles` are mapped the same way the current one's are.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn a_v1_request_always_targets_the_canister_being_synced() {
+        let calls = Arc::new(RecordingCalls::default());
+        let host = Principal::from_slice(&[1; 4]);
+        let callable = CallableCanisters {
+            by_name: BTreeMap::from([("backend".to_string(), Principal::from_slice(&[2; 4]))]),
+        };
+        let mut state = host_state(host, callable, calls.clone());
+
+        tokio::task::block_in_place(|| {
+            v1::SyncPluginImports::canister_call(
+                &mut state,
+                v1::icp::sync_plugin::types::CanisterCallRequest {
+                    method: "register".to_string(),
+                    arg: b"arg".to_vec(),
+                    call_type: v1::icp::sync_plugin::types::CallType::Update,
+                    direct: true,
+                    cycles: 7,
+                },
+            )
+        })
+        .expect("the call should reach the seam");
+
+        let updates = calls.updates.lock().unwrap();
+        assert_eq!(updates[0].canister, host);
+        assert_eq!(updates[0].authority, Authority::Direct);
+        assert_eq!(updates[0].cycles, 7);
+    }
+
     // -------------------------------------------------------------------------
     // Error-path tests — no fixture WASM needed
     // -------------------------------------------------------------------------
@@ -1670,46 +1805,6 @@ mod tests {
         );
     }
 
-    /// The replica's own wording for the two ways a target reports it has no
-    /// section, copied from `CanisterManagerError` in the IC repo. Both are
-    /// absence, not failure, so both must reach the plugin as `none`.
-    #[test]
-    fn management_canister_absence_rejects_are_recognized() {
-        let target = Principal::from_text("aaaaa-aa").unwrap();
-        let other = Principal::from_text("2vxsx-fae").unwrap();
-
-        let no_module = format!(
-            "Proxy call failed: The canister {target} has no Wasm module and hence no metadata is available."
-        );
-        let no_section = format!(
-            "Proxy call failed: The canister {target} has no metadata section with the name candid:service."
-        );
-        assert!(rejected_as_no_such_section(
-            &no_module,
-            target,
-            "candid:service"
-        ));
-        assert!(rejected_as_no_such_section(
-            &no_section,
-            target,
-            "candid:service"
-        ));
-
-        // A section by another name, a canister other than the one asked about,
-        // and an unrelated failure are all reads that failed.
-        assert!(!rejected_as_no_such_section(&no_section, target, "dfx"));
-        assert!(!rejected_as_no_such_section(
-            &no_module,
-            other,
-            "candid:service"
-        ));
-        assert!(!rejected_as_no_such_section(
-            &format!("Proxy call failed: Canister {target} not found."),
-            target,
-            "candid:service"
-        ));
-    }
-
     #[test]
     fn plugin_exceeding_compute_limit_is_trapped() {
         let Some(wasm_path) = option_env!("TEST_PLUGIN_WASM") else {
@@ -1720,14 +1815,9 @@ mod tests {
         let mut inv = invocation(wasm_path, "spin");
         inv.compute_limit_secs = 1;
         let err = run_plugin(inv).expect_err("spinning plugin should hit the compute limit");
-        // The trap surfaces through the CallExec source chain, so walk it and
-        // assert the message names both the limit and the override env var.
-        let mut chain = err.to_string();
-        let mut cur: &dyn std::error::Error = &err;
-        while let Some(src) = cur.source() {
-            chain = format!("{chain}: {src}");
-            cur = src;
-        }
+        // The trap surfaces through the CallExec source chain, so flatten it
+        // and assert the message names both the limit and the override env var.
+        let chain = error::flatten(&err);
         assert!(
             chain.contains("compute-time limit") && chain.contains(PLUGIN_COMPUTE_LIMIT_ENV),
             "unexpected error chain: {chain}"