diff --git a/Cargo.lock b/Cargo.lock
index 93c09c1..ce42c86 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -669,7 +669,9 @@ dependencies = [
 name = "harness-app"
 version = "0.1.0"
 dependencies = [
+ "async-trait",
  "dirs",
+ "futures",
  "harness-core",
  "harness-lsp",
  "harness-mcp",
diff --git a/crates/harness-app/Cargo.toml b/crates/harness-app/Cargo.toml
index 2eb3ea2..0fd5f60 100644
--- a/crates/harness-app/Cargo.toml
+++ b/crates/harness-app/Cargo.toml
@@ -12,12 +12,14 @@ harness-mcp = { workspace = true }
 harness-lsp = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
+async-trait = { workspace = true }
 dirs = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }
+futures = { workspace = true }
 
 [lints]
 workspace = true
diff --git a/crates/harness-app/src/lib.rs b/crates/harness-app/src/lib.rs
index 44073c2..7db5393 100644
--- a/crates/harness-app/src/lib.rs
+++ b/crates/harness-app/src/lib.rs
@@ -7,15 +7,18 @@ use std::collections::HashMap;
 use std::hash::Hasher;
 use std::path::Path;
 use std::path::PathBuf;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, OnceLock, Weak};
 use std::time::{SystemTime, UNIX_EPOCH};
 
+use async_trait::async_trait;
+use harness_core::agent::{AgentDef, AgentRegistry};
 use harness_core::config::{self, Config, ConfigError};
+use harness_core::engine::jobs::{JobBoard, JobState, LaunchSpec};
 use harness_core::engine::{run_session, RunConfig, StepContext};
 use harness_core::event::{AppEvent, EventBus, RunOutcome};
-use harness_core::permission::{PermissionReply, PermissionService};
+use harness_core::permission::{PermissionReply, PermissionService, Rule, Ruleset};
 use harness_core::store::{Store, StoreError};
-use harness_core::tool::ToolRegistry;
+use harness_core::tool::{SpawnError, SpawnOutcome, SpawnRequest, SubagentSpawner, ToolRegistry};
 use harness_core::types::{
     Message, MessageId, ModelRef, Part, PartBody, PartId, Session, SessionId,
 };
@@ -84,9 +87,16 @@ struct EngineInner {
     tools: ToolRegistry,
     providers: ProviderRegistry,
     catalog: ModelCatalog,
+    agents: AgentRegistry,
     cwd: PathBuf,
     data_dir: PathBuf,
     runs: Mutex<HashMap<SessionId, RunHandle>>,
+    /// One job board per session (scoped to that session as a parent). Lazily created and
+    /// shared between a session's run loop (board injection) and the spawner (registration).
+    boards: Mutex<HashMap<SessionId, Arc<JobBoard>>>,
+    /// Self-reference so the spawner can hand an `Arc<EngineInner>` to child contexts (for
+    /// nested delegation) and to detached background tasks. Set once in `new`.
+    me: OnceLock<Weak<EngineInner>>,
 }
 
 struct RunHandle {
@@ -111,11 +121,6 @@ impl EngineHandle {
 
     fn new(cwd: PathBuf, store: Store) -> Result<Self, AppError> {
         let config = config::load(&cwd)?;
-        let bus = EventBus::new();
-        let permissions = Arc::new(PermissionService::new(bus.clone()));
-
-        let mut tools = ToolRegistry::new();
-        harness_tools::register_builtins(&mut tools);
 
         let mut providers = ProviderRegistry::new();
         if let Some(key) = config
@@ -141,6 +146,32 @@ impl EngineHandle {
             providers.register(Arc::new(provider));
         }
 
+        Self::build(cwd, store, config, providers)
+    }
+
+    /// Shared construction over an explicit provider set — the seam tests use to inject a mock.
+    fn build(
+        cwd: PathBuf,
+        store: Store,
+        config: Config,
+        providers: ProviderRegistry,
+    ) -> Result<Self, AppError> {
+        let bus = EventBus::new();
+        let permissions = Arc::new(PermissionService::new(bus.clone()));
+
+        let mut tools = ToolRegistry::new();
+        harness_tools::register_builtins(&mut tools);
+        harness_tools::register_task_tool(&mut tools);
+
+        // Layered agent registry: bundled markdown → global dir → project dir → config patches.
+        let global_agent_dir = dirs::config_dir().map(|d| d.join("ai-harness").join("agent"));
+        let project_agent_dir = cwd.join(".harness").join("agent");
+        let agents = AgentRegistry::load(
+            &config.agents,
+            global_agent_dir.as_deref(),
+            Some(project_agent_dir.as_path()),
+        );
+
         let data_dir = dirs::data_dir()
             .unwrap_or_else(std::env::temp_dir)
             .join("ai-harness")
@@ -150,20 +181,23 @@ impl EngineHandle {
         // `init` warms the cache in the background for the next launch.
         let catalog = ModelCatalog::load_cached_or_baked(&ModelCatalog::default_cache_path());
 
-        Ok(Self {
-            inner: Arc::new(EngineInner {
-                config,
-                store,
-                bus,
-                permissions,
-                tools,
-                providers,
-                catalog,
-                cwd,
-                data_dir,
-                runs: Mutex::new(HashMap::new()),
-            }),
-        })
+        let inner = Arc::new(EngineInner {
+            config,
+            store,
+            bus,
+            permissions,
+            tools,
+            providers,
+            catalog,
+            agents,
+            cwd,
+            data_dir,
+            runs: Mutex::new(HashMap::new()),
+            boards: Mutex::new(HashMap::new()),
+            me: OnceLock::new(),
+        });
+        let _ = inner.me.set(Arc::downgrade(&inner));
+        Ok(Self { inner })
     }
 
     /// Fire-and-forget refresh of the models.dev cache so the next launch has current pricing.
@@ -251,27 +285,43 @@ impl EngineHandle {
             .bus
             .publish(AppEvent::PartUpdated { part: user_part });
 
+        // Resolve the session's agent from the registry (falls back to a generic assistant).
+        let session_agent = self
+            .inner
+            .store
+            .session(session_id.clone())
+            .await?
+            .map(|s| s.agent)
+            .unwrap_or_else(|| "orchestrator".to_string());
+        let agent = self.inner.agent_def(&session_agent);
+        let inject_job_board = self.inner.config.orchestration.job_board && agent.mode.is_primary();
+        let board = self.inner.board_for(&session_id).await?;
+
         let ctx = StepContext {
             store: self.inner.store.clone(),
             bus: self.inner.bus.clone(),
-            tools: self.inner.tools.clone(),
+            tools: self.inner.filtered_tools(&agent),
             permissions: self.inner.permissions.clone(),
-            static_rules: self.inner.config.permissions.clone(),
+            static_rules: self.inner.effective_static_rules(&agent, 0),
             extra_rules: Arc::new(Mutex::new(Vec::new())),
+            parent_rules: Vec::new(), // root session: no intersection
             session_id: session_id.clone(),
             cwd: self.inner.cwd.clone(),
             data_dir: self.inner.data_dir.join(session_id.to_string()),
             cancel: CancellationToken::new(),
             now,
+            spawner: Some(self.inner.clone()),
+            job_board: Some(board),
         };
         let run_config = RunConfig {
-            agent_name: "orchestrator".to_string(),
-            agent_prompt: DEFAULT_AGENT_PROMPT.to_string(),
+            agent_name: agent.name.clone(),
+            agent_prompt: self.inner.agent_prompt(&agent),
             model,
-            temperature: None,
-            max_steps: DEFAULT_MAX_STEPS,
+            temperature: agent.temperature,
+            max_steps: agent.max_steps.unwrap_or(DEFAULT_MAX_STEPS),
             instructions: self.inner.config.instructions.clone(),
             cost: Some(self.inner.catalog.cost(provider_id, model_id)),
+            inject_job_board,
         };
 
         // Reserve the run slot *before* spawning. If we inserted after spawning, a run that
@@ -356,6 +406,388 @@ impl EngineHandle {
     }
 }
 
+/// Deny rules injected as low-priority defaults for any session at depth > 0: a subagent may
+/// not itself delegate (`task`) or manage todos unless config/agent rules re-enable it.
+fn depth_guard_rules() -> Ruleset {
+    ["task", "todo", "todo*"]
+        .into_iter()
+        .map(|perm| Rule {
+            permission: perm.to_string(),
+            pattern: "*".to_string(),
+            action: harness_core::permission::Action::Deny,
+        })
+        .collect()
+}
+
+impl EngineInner {
+    fn arc(&self) -> Arc<EngineInner> {
+        self.me
+            .get()
+            .and_then(Weak::upgrade)
+            .expect("EngineInner self-reference set in new()")
+    }
+
+    /// Resolves an agent definition by name, falling back to a generic assistant so an
+    /// unknown/misconfigured agent still runs rather than failing the session.
+    fn agent_def(&self, name: &str) -> AgentDef {
+        self.agents.get(name).cloned().unwrap_or_else(|| AgentDef {
+            name: name.to_string(),
+            description: String::new(),
+            mode: harness_core::agent::AgentMode::Primary,
+            model: None,
+            temperature: None,
+            prompt: String::new(),
+            permissions: Vec::new(),
+            tools: std::collections::HashMap::new(),
+            max_steps: None,
+            source: harness_core::agent::AgentSource::Config,
+        })
+    }
+
+    /// A tool registry filtered to what `agent` enables. Exact tool names override a `*`
+    /// wildcard; absent = enabled. Honors the `tools:` frontmatter map (docs/04-multiagent.md).
+    fn filtered_tools(&self, agent: &AgentDef) -> ToolRegistry {
+        if agent.tools.is_empty() {
+            return self.tools.clone();
+        }
+        let mut reg = ToolRegistry::new();
+        for tool in self.tools.all() {
+            let name = tool.name();
+            let enabled = agent
+                .tools
+                .get(name)
+                .or_else(|| agent.tools.get("*"))
+                .copied()
+                .unwrap_or(true);
+            if enabled {
+                reg.register(tool);
+            }
+        }
+        reg
+    }
+
+    fn agent_prompt(&self, agent: &AgentDef) -> String {
+        if agent.prompt.trim().is_empty() {
+            DEFAULT_AGENT_PROMPT.to_string()
+        } else {
+            agent.prompt.clone()
+        }
+    }
+
+    /// The ruleset a session evaluates its own tool calls against: low-priority depth guards
+    /// (depth > 0), then global config permissions, then the agent's own rules (highest).
+    fn effective_static_rules(&self, agent: &AgentDef, depth: u8) -> Ruleset {
+        let mut rules = Ruleset::new();
+        if depth > 0 {
+            rules.extend(depth_guard_rules());
+        }
+        rules.extend(self.config.permissions.clone());
+        rules.extend(agent.permissions.clone());
+        rules
+    }
+
+    /// The parent-effective ruleset a *child* intersects against: global config, the parent
+    /// agent's rules, then the parent session's `Always` grants. Frozen at spawn time.
+    fn parent_effective_rules(&self, session: &Session, agent: &AgentDef) -> Ruleset {
+        let mut rules = self.config.permissions.clone();
+        rules.extend(agent.permissions.clone());
+        rules.extend(session.extra_rules.clone());
+        rules
+    }
+
+    /// Gets (or lazily loads) the job board scoped to `session_id` as a parent.
+    async fn board_for(&self, session_id: &SessionId) -> Result<Arc<JobBoard>, StoreError> {
+        if let Some(board) = self.boards.lock().unwrap().get(session_id).cloned() {
+            return Ok(board);
+        }
+        let board = Arc::new(
+            JobBoard::load(
+                self.store.clone(),
+                self.bus.clone(),
+                session_id,
+                self.config.orchestration.max_reusable_per_agent,
+            )
+            .await?,
+        );
+        // Double-check under lock in case a concurrent caller inserted first.
+        let mut boards = self.boards.lock().unwrap();
+        Ok(boards.entry(session_id.clone()).or_insert(board).clone())
+    }
+
+    /// Appends `text` as a user message + text part to `session_id` (mirrors the root prompt
+    /// path so a child session's first turn has an instruction to act on).
+    async fn append_user_message(
+        &self,
+        session_id: &SessionId,
+        text: String,
+        now: i64,
+    ) -> Result<(), StoreError> {
+        let message = Message::new_user(session_id.clone(), now);
+        self.store.upsert_message(message.clone()).await?;
+        self.bus.publish(AppEvent::MessageCreated {
+            message: message.clone(),
+        });
+        let part = Part {
+            id: PartId::new(),
+            message_id: message.id.clone(),
+            session_id: session_id.clone(),
+            idx: 0,
+            body: PartBody::Text {
+                text,
+                synthetic: false,
+            },
+        };
+        self.store.upsert_part(part.clone()).await?;
+        self.bus.publish(AppEvent::PartUpdated { part });
+        Ok(())
+    }
+
+    /// Concatenates the `Text` parts of a session's last message — a child run's final answer.
+    async fn session_final_text(&self, session_id: &SessionId) -> String {
+        let Ok(messages) = self.store.messages(session_id.clone()).await else {
+            return String::new();
+        };
+        let Some(last) = messages.last() else {
+            return String::new();
+        };
+        let Ok(parts) = self.store.parts(last.id.clone()).await else {
+            return String::new();
+        };
+        parts
+            .iter()
+            .filter_map(|p| match &p.body {
+                PartBody::Text { text, .. } => Some(text.as_str()),
+                _ => None,
+            })
+            .collect::<Vec<_>>()
+            .join("")
+    }
+
+    /// Builds a child session's step context (permission intersection + its own board).
+    #[allow(clippy::too_many_arguments)]
+    fn child_ctx(
+        &self,
+        session_id: SessionId,
+        tools: ToolRegistry,
+        static_rules: Ruleset,
+        parent_rules: Ruleset,
+        extra_rules: Ruleset,
+        board: Arc<JobBoard>,
+        cancel: CancellationToken,
+        now: i64,
+    ) -> StepContext {
+        StepContext {
+            store: self.store.clone(),
+            bus: self.bus.clone(),
+            tools,
+            permissions: self.permissions.clone(),
+            static_rules,
+            extra_rules: Arc::new(Mutex::new(extra_rules)),
+            parent_rules,
+            session_id: session_id.clone(),
+            cwd: self.cwd.clone(),
+            data_dir: self.data_dir.join(session_id.to_string()),
+            cancel,
+            now,
+            spawner: Some(self.arc()),
+            job_board: Some(board),
+        }
+    }
+}
+
+#[async_trait]
+impl SubagentSpawner for EngineInner {
+    async fn spawn(&self, req: SpawnRequest) -> Result<SpawnOutcome, SpawnError> {
+        let other = |e: StoreError| SpawnError::Other(e.to_string());
+
+        // 1. Load the parent and enforce the depth limit.
+        let parent = self
+            .store
+            .session(req.parent_session_id.clone())
+            .await
+            .map_err(other)?
+            .ok_or_else(|| SpawnError::Other("parent session not found".into()))?;
+        let child_depth = parent.depth + 1;
+        if child_depth as u32 > self.config.orchestration.depth_limit {
+            return Err(SpawnError::DepthExceeded);
+        }
+
+        // 2. Resolve the agent — must be usable as a subagent.
+        let agent = self
+            .agents
+            .get(&req.agent)
+            .filter(|a| a.mode.is_subagent())
+            .cloned()
+            .ok_or_else(|| SpawnError::InvalidAgent(req.agent.clone()))?;
+
+        // Parent-effective rules for permission intersection (frozen snapshot).
+        let parent_agent = self.agent_def(&parent.agent);
+        let parent_rules = self.parent_effective_rules(&parent, &parent_agent);
+
+        let now = now_ms();
+        let board = self
+            .board_for(&req.parent_session_id)
+            .await
+            .map_err(other)?;
+
+        // 3. Reuse a completed child session by alias/id, or create a fresh one.
+        let (child_session, task_id, reused) = if let Some(alias) = &req.reuse_task_id {
+            let job = board
+                .resolve_reusable(&req.parent_session_id, alias)
+                .ok_or_else(|| SpawnError::ReuseNotFound(alias.clone()))?;
+            let child = self
+                .store
+                .session(job.child_session.clone())
+                .await
+                .map_err(other)?
+                .ok_or_else(|| SpawnError::ReuseNotFound(alias.clone()))?;
+            let _ = board.touch(&job.task_id, now).await;
+            (child, job.task_id, true)
+        } else {
+            let mut child = Session::new_child(&parent, req.agent.clone(), now);
+            if let Some(model) = &agent.model {
+                child.model = model.clone();
+            }
+            self.store
+                .upsert_session(child.clone())
+                .await
+                .map_err(other)?;
+            self.bus.publish(AppEvent::SessionCreated {
+                session: child.clone(),
+            });
+            let task_id = child.id.to_string();
+            (child, task_id, false)
+        };
+        let child_id = child_session.id.clone();
+
+        // 4. Append the instruction as the child's next user turn.
+        self.append_user_message(&child_id, req.prompt.clone(), now)
+            .await
+            .map_err(other)?;
+
+        // 5. Resolve the provider for the child model.
+        let provider = self
+            .providers
+            .get(&child_session.model.provider_id)
+            .ok_or_else(|| {
+                SpawnError::Other(format!(
+                    "no provider registered for {:?}",
+                    child_session.model.provider_id
+                ))
+            })?;
+
+        let child_static = self.effective_static_rules(&agent, child_depth);
+        let child_tools = self.filtered_tools(&agent);
+        let child_board = self.board_for(&child_id).await.map_err(other)?;
+        let run_config = RunConfig {
+            agent_name: agent.name.clone(),
+            agent_prompt: self.agent_prompt(&agent),
+            model: child_session.model.clone(),
+            temperature: agent.temperature,
+            max_steps: agent.max_steps.unwrap_or(DEFAULT_MAX_STEPS),
+            instructions: self.config.instructions.clone(),
+            cost: Some(self.catalog.cost(
+                &child_session.model.provider_id,
+                &child_session.model.model_id,
+            )),
+            inject_job_board: agent.mode.is_primary(),
+        };
+
+        // Register the launch on the parent board (also makes foreground results reusable).
+        if !reused {
+            let _ = board
+                .register_launch(
+                    LaunchSpec {
+                        task_id: task_id.clone(),
+                        parent_session: req.parent_session_id.clone(),
+                        child_session: child_id.clone(),
+                        agent: req.agent.clone(),
+                        description: req.description.clone(),
+                        objective: Some(req.description.clone()),
+                    },
+                    now,
+                )
+                .await;
+        }
+
+        if req.background {
+            // Child of the parent session's *run* token so the job dies with the session, not
+            // when the (already-returned) tool call completes.
+            let cancel = {
+                let runs = self.runs.lock().unwrap();
+                runs.get(&req.parent_session_id)
+                    .map(|r| r.cancel.child_token())
+            }
+            .unwrap_or_default();
+
+            let ctx = self.child_ctx(
+                child_id.clone(),
+                child_tools,
+                child_static,
+                parent_rules,
+                child_session.extra_rules.clone(),
+                child_board,
+                cancel,
+                now,
+            );
+            let this = self.arc();
+            let board_bg = board.clone();
+            let task_id_bg = task_id.clone();
+            let child_id_bg = child_id.clone();
+            tokio::spawn(async move {
+                let outcome = run_session(provider, ctx, &run_config, now_ms).await;
+                let (state, summary) = match &outcome {
+                    RunOutcome::Stopped => (
+                        JobState::Completed,
+                        Some(this.session_final_text(&child_id_bg).await),
+                    ),
+                    RunOutcome::Aborted => (JobState::Cancelled, None),
+                    RunOutcome::Errored { message } => (JobState::Error, Some(message.clone())),
+                };
+                let _ = board_bg.finish(&task_id_bg, state, summary, now_ms()).await;
+            });
+
+            let alias = board
+                .snapshot()
+                .into_iter()
+                .find(|j| j.task_id == task_id)
+                .map(|j| j.alias);
+            Ok(SpawnOutcome {
+                child_session_id: child_id,
+                background: true,
+                alias,
+                final_text: None,
+            })
+        } else {
+            // Foreground: run to completion under the tool call's token and return the text.
+            let ctx = self.child_ctx(
+                child_id.clone(),
+                child_tools,
+                child_static,
+                parent_rules,
+                child_session.extra_rules.clone(),
+                child_board,
+                req.cancel.clone(),
+                now,
+            );
+            let outcome = run_session(provider, ctx, &run_config, now_ms).await;
+            let final_text = self.session_final_text(&child_id).await;
+            let (state, summary) = match &outcome {
+                RunOutcome::Stopped => (JobState::Completed, Some(final_text.clone())),
+                RunOutcome::Aborted => (JobState::Cancelled, None),
+                RunOutcome::Errored { message } => (JobState::Error, Some(message.clone())),
+            };
+            let _ = board.finish(&task_id, state, summary, now_ms()).await;
+            Ok(SpawnOutcome {
+                child_session_id: child_id,
+                background: false,
+                alias: None,
+                final_text: Some(final_text),
+            })
+        }
+    }
+}
+
 pub struct App {
     engine: EngineHandle,
     // Keeps the auto-approve task alive for the lifetime of the App.
@@ -448,13 +880,195 @@ fn spawn_auto_approve_task(engine: &EngineHandle) -> JoinHandle<()> {
 mod tests {
     use super::*;
 
+    use async_trait::async_trait;
+    use harness_core::llm::{
+        FinishReason, LlmEvent, LlmEventStream, LlmRequest, Provider, ProviderError,
+    };
+    use harness_core::types::{ModelInfo, TokenUsage};
+
+    /// Emits a single final-text turn for every stream call — enough to drive a child
+    /// subagent session to completion deterministically.
+    struct MockProvider;
+
+    #[async_trait]
+    impl Provider for MockProvider {
+        fn id(&self) -> &str {
+            "mock"
+        }
+        async fn list_models(&self) -> Result<Vec<ModelInfo>, ProviderError> {
+            Ok(vec![])
+        }
+        async fn stream(
+            &self,
+            _req: LlmRequest,
+            _cancel: CancellationToken,
+        ) -> Result<LlmEventStream, ProviderError> {
+            let events = vec![
+                Ok(LlmEvent::TextStart { id: "t".into() }),
+                Ok(LlmEvent::TextDelta {
+                    id: "t".into(),
+                    text: "explored the code".into(),
+                }),
+                Ok(LlmEvent::TextEnd { id: "t".into() }),
+                Ok(LlmEvent::Finish {
+                    reason: FinishReason::Stop,
+                    usage: TokenUsage::default(),
+                }),
+            ];
+            Ok(Box::pin(futures::stream::iter(events)))
+        }
+    }
+
+    fn mock_engine(cwd: PathBuf) -> EngineHandle {
+        let mut providers = ProviderRegistry::new();
+        providers.register(Arc::new(MockProvider));
+        EngineHandle::build(
+            cwd,
+            Store::open_in_memory().unwrap(),
+            Config::default(),
+            providers,
+        )
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn foreground_task_runs_a_subagent_and_alias_reuse_continues_the_same_session() {
+        let dir = tempfile::tempdir().unwrap();
+        let engine = mock_engine(dir.path().to_path_buf());
+        let root = engine
+            .new_session("orchestrator", "mock/mock-model")
+            .await
+            .unwrap();
+
+        // Foreground spawn of the explorer subagent.
+        let out = engine
+            .inner
+            .spawn(SpawnRequest {
+                parent_session_id: root.clone(),
+                parent_message_id: MessageId::new(),
+                agent: "explorer".into(),
+                description: "map the auth flow".into(),
+                prompt: "find the login handler".into(),
+                reuse_task_id: None,
+                background: false,
+                cancel: CancellationToken::new(),
+            })
+            .await
+            .unwrap();
+        assert!(!out.background);
+        assert_eq!(out.final_text.as_deref(), Some("explored the code"));
+        let child_id = out.child_session_id.clone();
+
+        // The completed job is on the parent board, reusable by alias.
+        let board = engine.inner.board_for(&root).await.unwrap();
+        let job = board
+            .snapshot()
+            .into_iter()
+            .find(|j| j.child_session == child_id)
+            .expect("job registered");
+        assert_eq!(job.alias, "exp-1");
+        assert_eq!(job.state, JobState::Completed);
+
+        let child_messages_before = engine
+            .inner
+            .store
+            .messages(child_id.clone())
+            .await
+            .unwrap()
+            .len();
+
+        // Reuse by alias → same child session, with the new prompt appended.
+        let reused = engine
+            .inner
+            .spawn(SpawnRequest {
+                parent_session_id: root.clone(),
+                parent_message_id: MessageId::new(),
+                agent: "explorer".into(),
+                description: "follow up".into(),
+                prompt: "now find the logout handler".into(),
+                reuse_task_id: Some("exp-1".into()),
+                background: false,
+                cancel: CancellationToken::new(),
+            })
+            .await
+            .unwrap();
+        assert_eq!(
+            reused.child_session_id, child_id,
+            "reuse must continue the same child session"
+        );
+
+        let child_messages_after = engine.inner.store.messages(child_id).await.unwrap().len();
+        assert!(
+            child_messages_after > child_messages_before,
+            "reused session should have the follow-up turn appended ({child_messages_before} -> {child_messages_after})"
+        );
+    }
+
+    #[tokio::test]
+    async fn depth_limit_is_enforced() {
+        let dir = tempfile::tempdir().unwrap();
+        let engine = mock_engine(dir.path().to_path_buf());
+        // Build a session already at the depth limit so any child would exceed it.
+        let now = now_ms();
+        let mut deep = Session::new_root("explorer", ModelRef::new("mock", "mock-model"), now);
+        deep.depth = engine.inner.config.orchestration.depth_limit as u8;
+        engine
+            .inner
+            .store
+            .upsert_session(deep.clone())
+            .await
+            .unwrap();
+
+        let err = engine
+            .inner
+            .spawn(SpawnRequest {
+                parent_session_id: deep.id.clone(),
+                parent_message_id: MessageId::new(),
+                agent: "explorer".into(),
+                description: "too deep".into(),
+                prompt: "go deeper".into(),
+                reuse_task_id: None,
+                background: false,
+                cancel: CancellationToken::new(),
+            })
+            .await
+            .unwrap_err();
+        assert!(matches!(err, SpawnError::DepthExceeded));
+    }
+
+    #[tokio::test]
+    async fn unknown_subagent_is_rejected() {
+        let dir = tempfile::tempdir().unwrap();
+        let engine = mock_engine(dir.path().to_path_buf());
+        let root = engine
+            .new_session("orchestrator", "mock/mock-model")
+            .await
+            .unwrap();
+        let err = engine
+            .inner
+            .spawn(SpawnRequest {
+                parent_session_id: root,
+                parent_message_id: MessageId::new(),
+                agent: "nonexistent".into(),
+                description: "x".into(),
+                prompt: "y".into(),
+                reuse_task_id: None,
+                background: false,
+                cancel: CancellationToken::new(),
+            })
+            .await
+            .unwrap_err();
+        assert!(matches!(err, SpawnError::InvalidAgent(_)));
+    }
+
     #[tokio::test]
     async fn init_loads_defaults_with_no_providers_configured() {
         let dir = tempfile::tempdir().unwrap();
         std::env::remove_var("ANTHROPIC_API_KEY");
         let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
         assert!(app.engine.inner.providers.get("anthropic").is_none());
-        assert_eq!(app.engine.inner.tools.all().len(), 6);
+        // 6 built-ins + the multiagent `task` tool.
+        assert_eq!(app.engine.inner.tools.all().len(), 7);
     }
 
     #[tokio::test]
diff --git a/crates/harness-core/src/agent/mod.rs b/crates/harness-core/src/agent/mod.rs
index 9037b2f..c8f7196 100644
--- a/crates/harness-core/src/agent/mod.rs
+++ b/crates/harness-core/src/agent/mod.rs
@@ -18,10 +18,16 @@ use crate::types::ModelRef;
 const SUBAGENTS_MARKER: &str = "{{SUBAGENTS}}";
 
 const BUNDLED: &[(&str, &str)] = &[
-    ("orchestrator", include_str!("../../assets/agents/orchestrator.md")),
+    (
+        "orchestrator",
+        include_str!("../../assets/agents/orchestrator.md"),
+    ),
     ("explorer", include_str!("../../assets/agents/explorer.md")),
     ("oracle", include_str!("../../assets/agents/oracle.md")),
-    ("librarian", include_str!("../../assets/agents/librarian.md")),
+    (
+        "librarian",
+        include_str!("../../assets/agents/librarian.md"),
+    ),
     ("fixer", include_str!("../../assets/agents/fixer.md")),
     ("designer", include_str!("../../assets/agents/designer.md")),
 ];
@@ -107,7 +113,10 @@ fn parse_model_ref(s: &str) -> Option<ModelRef> {
 /// Splits a markdown agent file into (frontmatter, body). A file without a leading `---`
 /// fence is treated as an all-body prompt with empty frontmatter.
 fn split_frontmatter(content: &str) -> (&str, &str) {
-    let rest = match content.strip_prefix("---\n").or_else(|| content.strip_prefix("---\r\n")) {
+    let rest = match content
+        .strip_prefix("---\n")
+        .or_else(|| content.strip_prefix("---\r\n"))
+    {
         Some(r) => r,
         None => return ("", content),
     };
@@ -149,7 +158,11 @@ fn parse_agent(
     Ok(Some(AgentDef {
         name: name.to_string(),
         description: fm.description.unwrap_or_default(),
-        mode: fm.mode.as_deref().and_then(AgentMode::parse).unwrap_or_default(),
+        mode: fm
+            .mode
+            .as_deref()
+            .and_then(AgentMode::parse)
+            .unwrap_or_default(),
         model: fm.model.as_deref().and_then(parse_model_ref),
         temperature: fm.temperature,
         prompt: body.trim_end().to_string(),
@@ -207,7 +220,10 @@ impl AgentRegistry {
     /// Just the bundled agents — the default when no overrides are configured (tests, headless).
     pub fn bundled() -> Self {
         let mut reg = Self::default();
-        reg.load_markdown_layer(BUNDLED.iter().map(|(n, c)| (n.to_string(), *c)), AgentSource::Bundled);
+        reg.load_markdown_layer(
+            BUNDLED.iter().map(|(n, c)| (n.to_string(), *c)),
+            AgentSource::Bundled,
+        );
         reg.generate_routing();
         reg
     }
@@ -338,7 +354,14 @@ mod tests {
     #[test]
     fn bundled_loads_all_six_agents() {
         let reg = AgentRegistry::bundled();
-        for name in ["orchestrator", "explorer", "oracle", "librarian", "fixer", "designer"] {
+        for name in [
+            "orchestrator",
+            "explorer",
+            "oracle",
+            "librarian",
+            "fixer",
+            "designer",
+        ] {
             assert!(reg.get(name).is_some(), "missing {name}");
         }
         assert_eq!(reg.len(), 6);
@@ -356,10 +379,15 @@ mod tests {
             \x20 - { permission: \"edit\", pattern: \"*\", action: deny }\n\
             ---\n\
             You are a test agent.\n";
-        let def = parse_agent("tester", AgentSource::Bundled, md).unwrap().unwrap();
+        let def = parse_agent("tester", AgentSource::Bundled, md)
+            .unwrap()
+            .unwrap();
         assert_eq!(def.description, "test agent");
         assert_eq!(def.mode, AgentMode::Subagent);
-        assert_eq!(def.model, Some(ModelRef::new("anthropic", "claude-haiku-4-5")));
+        assert_eq!(
+            def.model,
+            Some(ModelRef::new("anthropic", "claude-haiku-4-5"))
+        );
         assert_eq!(def.temperature, Some(0.1));
         assert_eq!(def.tools.get("write"), Some(&false));
         assert_eq!(def.tools.get("bash"), Some(&true));
@@ -379,9 +407,11 @@ mod tests {
 
     #[test]
     fn disable_true_removes_agent() {
-        assert!(parse_agent("x", AgentSource::Config, "---\ndisable: true\n---\nbody")
-            .unwrap()
-            .is_none());
+        assert!(
+            parse_agent("x", AgentSource::Config, "---\ndisable: true\n---\nbody")
+                .unwrap()
+                .is_none()
+        );
     }
 
     #[test]
diff --git a/crates/harness-core/src/engine/jobs.rs b/crates/harness-core/src/engine/jobs.rs
index f2efa67..1f49a02 100644
--- a/crates/harness-core/src/engine/jobs.rs
+++ b/crates/harness-core/src/engine/jobs.rs
@@ -413,15 +413,24 @@ mod tests {
     async fn alias_increments_per_agent() {
         let (_store, board, parent) = board(2).await;
         let a1 = board
-            .register_launch(spec("t1", parent.clone(), SessionId::new(), "explorer", None), 1)
+            .register_launch(
+                spec("t1", parent.clone(), SessionId::new(), "explorer", None),
+                1,
+            )
             .await
             .unwrap();
         let a2 = board
-            .register_launch(spec("t2", parent.clone(), SessionId::new(), "explorer", None), 2)
+            .register_launch(
+                spec("t2", parent.clone(), SessionId::new(), "explorer", None),
+                2,
+            )
             .await
             .unwrap();
         let f1 = board
-            .register_launch(spec("t3", parent.clone(), SessionId::new(), "fixer", None), 3)
+            .register_launch(
+                spec("t3", parent.clone(), SessionId::new(), "fixer", None),
+                3,
+            )
             .await
             .unwrap();
         assert_eq!(a1, "exp-1");
@@ -492,7 +501,13 @@ mod tests {
         for (i, ts) in [(1, 10), (2, 20), (3, 30)] {
             board
                 .register_launch(
-                    spec(&format!("t{i}"), parent.clone(), SessionId::new(), "explorer", None),
+                    spec(
+                        &format!("t{i}"),
+                        parent.clone(),
+                        SessionId::new(),
+                        "explorer",
+                        None,
+                    ),
                     ts,
                 )
                 .await
@@ -543,7 +558,10 @@ mod tests {
                 .await
                 .unwrap();
             board
-                .register_launch(spec("t1", parent.clone(), SessionId::new(), "explorer", None), 1)
+                .register_launch(
+                    spec("t1", parent.clone(), SessionId::new(), "explorer", None),
+                    1,
+                )
                 .await
                 .unwrap();
         }
diff --git a/crates/harness-core/src/engine/loop.rs b/crates/harness-core/src/engine/loop.rs
index 2b8423a..200358a 100644
--- a/crates/harness-core/src/engine/loop.rs
+++ b/crates/harness-core/src/engine/loop.rs
@@ -22,6 +22,8 @@ pub struct RunConfig {
     pub instructions: Vec<String>,
     /// Pricing for `model`, from models.dev metadata. `None` leaves cost at 0.
     pub cost: Option<crate::types::ModelCost>,
+    /// Whether to append the background job board to requests (primary/delegating agents).
+    pub inject_job_board: bool,
 }
 
 /// Adds a step's usage/cost onto the persisted session and republishes it. Cost accounting is
@@ -198,6 +200,22 @@ pub async fn run_session(
             wire_messages.extend(convert_message(message, &parts));
         }
 
+        // Append the (synthetic, non-persisted) job board to the last user message so the
+        // orchestrator sees running/reusable subtasks. docs/04-multiagent.md.
+        if run_config.inject_job_board {
+            if let Some(board) = &ctx.job_board {
+                if let Some(block) = board.format_for_prompt() {
+                    if let Some(last_user) = wire_messages
+                        .iter_mut()
+                        .rev()
+                        .find(|m| m.role == WireRole::User)
+                    {
+                        last_user.content.push(WireContent::Text { text: block });
+                    }
+                }
+            }
+        }
+
         let system_blocks = system::assemble(
             system::env_header(&ctx.cwd),
             &run_config.agent_prompt,
@@ -260,6 +278,15 @@ pub async fn run_session(
             }
             Ok(outcome) => {
                 accumulate_session_usage(&ctx, &outcome.usage, outcome.cost, now_fn()).await;
+                // A completed step means the orchestrator has now seen any terminal jobs
+                // that were on the board this turn; mark them reconciled.
+                if run_config.inject_job_board {
+                    if let Some(board) = &ctx.job_board {
+                        if let Err(e) = board.reconcile_terminal(now_fn()).await {
+                            tracing::warn!(error = %e, "failed to reconcile job board");
+                        }
+                    }
+                }
                 match outcome.result {
                     StepResult::Continue => continue,
                     StepResult::Stop => return RunOutcome::Stopped,
@@ -376,11 +403,14 @@ mod tests {
             permissions,
             static_rules: Vec::new(),
             extra_rules: Arc::new(std::sync::Mutex::new(Vec::new())),
+            parent_rules: Vec::new(),
             session_id,
             cwd: cwd.clone(),
             data_dir: cwd.join("tool-output"),
             cancel: CancellationToken::new(),
             now: 1,
+            spawner: None,
+            job_board: None,
         }
     }
 
@@ -462,6 +492,7 @@ mod tests {
                 output: 15.0,
                 ..Default::default()
             }),
+            inject_job_board: false,
         };
 
         let outcome = run_session(Arc::new(provider), ctx, &run_config, || 2).await;
@@ -569,6 +600,7 @@ mod tests {
             max_steps: 10,
             instructions: Vec::new(),
             cost: None,
+            inject_job_board: false,
         };
 
         let outcome = run_session(Arc::new(provider), ctx, &run_config, || 2).await;
@@ -578,4 +610,128 @@ mod tests {
         let messages = store.messages(session_id).await.unwrap();
         assert_eq!(messages.len(), 1);
     }
+
+    /// Records the last request it was asked to stream so tests can assert on prompt content.
+    struct CapturingProvider {
+        last: StdMutex<Option<LlmRequest>>,
+    }
+
+    #[async_trait]
+    impl Provider for CapturingProvider {
+        fn id(&self) -> &str {
+            "mock"
+        }
+        async fn list_models(&self) -> Result<Vec<ModelInfo>, ProviderError> {
+            Ok(vec![])
+        }
+        async fn stream(
+            &self,
+            req: LlmRequest,
+            _cancel: CancellationToken,
+        ) -> Result<LlmEventStream, ProviderError> {
+            *self.last.lock().unwrap() = Some(req);
+            let events = vec![
+                Ok(LlmEvent::TextStart { id: "t".into() }),
+                Ok(LlmEvent::TextDelta {
+                    id: "t".into(),
+                    text: "ok".into(),
+                }),
+                Ok(LlmEvent::TextEnd { id: "t".into() }),
+                Ok(LlmEvent::Finish {
+                    reason: FinishReason::Stop,
+                    usage: usage(1, 1),
+                }),
+            ];
+            Ok(Box::pin(futures::stream::iter(events)))
+        }
+    }
+
+    #[tokio::test]
+    async fn job_board_is_injected_into_the_last_user_message() {
+        use crate::engine::jobs::{JobBoard, LaunchSpec};
+
+        let store = Store::open_in_memory().unwrap();
+        let bus = EventBus::new();
+        let model = ModelRef::new("mock", "mock-model");
+        let session = Session::new_root("orchestrator", model.clone(), 1);
+        let session_id = session.id.clone();
+        store.upsert_session(session).await.unwrap();
+
+        let user_message = Message::new_user(session_id.clone(), 1);
+        store.upsert_message(user_message.clone()).await.unwrap();
+        store
+            .upsert_part(Part {
+                id: crate::types::PartId::new(),
+                message_id: user_message.id.clone(),
+                session_id: session_id.clone(),
+                idx: 0,
+                body: PartBody::Text {
+                    text: "carry on".into(),
+                    synthetic: false,
+                },
+            })
+            .await
+            .unwrap();
+
+        // A board with one running job for this session.
+        let board = std::sync::Arc::new(
+            JobBoard::load(store.clone(), bus.clone(), &session_id, 2)
+                .await
+                .unwrap(),
+        );
+        board
+            .register_launch(
+                LaunchSpec {
+                    task_id: "t1".into(),
+                    parent_session: session_id.clone(),
+                    child_session: SessionId::new(),
+                    agent: "explorer".into(),
+                    description: "map auth".into(),
+                    objective: Some("map the auth flow".into()),
+                },
+                1,
+            )
+            .await
+            .unwrap();
+
+        let cwd = tempfile::tempdir().unwrap();
+        let mut ctx = make_ctx(store, bus, session_id, cwd.path().to_path_buf()).await;
+        ctx.job_board = Some(board);
+        let run_config = RunConfig {
+            agent_name: "orchestrator".into(),
+            agent_prompt: "You orchestrate.".into(),
+            model,
+            temperature: None,
+            max_steps: 1,
+            instructions: Vec::new(),
+            cost: None,
+            inject_job_board: true,
+        };
+
+        let provider = std::sync::Arc::new(CapturingProvider {
+            last: StdMutex::new(None),
+        });
+        let outcome = run_session(provider.clone(), ctx, &run_config, || 2).await;
+        assert!(matches!(outcome, RunOutcome::Stopped));
+
+        let req = provider.last.lock().unwrap().clone().expect("a request");
+        let last_user = req
+            .messages
+            .iter()
+            .rev()
+            .find(|m| m.role == WireRole::User)
+            .expect("a user message");
+        let text: String = last_user
+            .content
+            .iter()
+            .filter_map(|c| match c {
+                WireContent::Text { text } => Some(text.as_str()),
+                _ => None,
+            })
+            .collect::<Vec<_>>()
+            .join("\n");
+        assert!(text.contains("Background Job Board"), "got: {text}");
+        assert!(text.contains("exp-1"), "got: {text}");
+        assert!(text.contains("map the auth flow"), "got: {text}");
+    }
 }
diff --git a/crates/harness-core/src/engine/processor.rs b/crates/harness-core/src/engine/processor.rs
index bdc2839..89befed 100644
--- a/crates/harness-core/src/engine/processor.rs
+++ b/crates/harness-core/src/engine/processor.rs
@@ -10,10 +10,13 @@ use crate::event::{AppEvent, EventBus};
 use crate::llm::{FinishReason, LlmEvent, LlmEventStream, ProviderError};
 use crate::permission::{PermissionService, Ruleset};
 use crate::store::Store;
-use crate::tool::{MetadataSink, PermissionHandle, Tool, ToolCtx, ToolError, ToolRegistry};
+use crate::tool::{
+    MetadataSink, PermissionHandle, SubagentSpawner, Tool, ToolCtx, ToolError, ToolRegistry,
+};
 use crate::types::{Message, MessageId, Part, PartBody, PartId, SessionId, TokenUsage, ToolState};
 
 use super::doomloop::DoomLoopGuard;
+use super::jobs::JobBoard;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum StepResult {
@@ -47,6 +50,9 @@ pub struct StepContext {
     pub permissions: Arc<PermissionService>,
     pub static_rules: Ruleset,
     pub extra_rules: Arc<Mutex<Ruleset>>,
+    /// Parent-effective ruleset for a subagent session; empty for a root session. Enables
+    /// permission intersection on this session's tool calls.
+    pub parent_rules: Ruleset,
     pub session_id: SessionId,
     pub cwd: PathBuf,
     /// Session's `tool-output` spill directory (see `tool::truncate`).
@@ -54,6 +60,11 @@ pub struct StepContext {
     pub cancel: CancellationToken,
     /// Wall-clock for `created_at` stamps — passed in so tests stay deterministic.
     pub now: i64,
+    /// Lets the `task` tool spawn subagents. `None` disables delegation (headless/tests).
+    pub spawner: Option<Arc<dyn SubagentSpawner>>,
+    /// This session's background job board (as a parent). Injected into requests when the
+    /// running agent can delegate; `None` disables the board.
+    pub job_board: Option<Arc<JobBoard>>,
 }
 
 struct FlushTracker {
@@ -483,7 +494,8 @@ impl<'a> Run<'a> {
             self.ctx.static_rules.clone(),
             self.ctx.extra_rules.clone(),
             call_cancel.clone(),
-        );
+        )
+        .with_parent_rules(self.ctx.parent_rules.clone());
         let tool_ctx = ToolCtx {
             session_id: self.ctx.session_id.clone(),
             message_id: self.message_id(),
@@ -493,6 +505,7 @@ impl<'a> Run<'a> {
             cancel: call_cancel.clone(),
             ask,
             metadata: metadata_sink,
+            spawner: self.ctx.spawner.clone(),
         };
 
         let result = tokio::select! {
diff --git a/crates/harness-core/src/permission/service.rs b/crates/harness-core/src/permission/service.rs
index f4dc957..2cb48fa 100644
--- a/crates/harness-core/src/permission/service.rs
+++ b/crates/harness-core/src/permission/service.rs
@@ -8,7 +8,7 @@ use ulid::Ulid;
 use crate::event::{AppEvent, EventBus, PermissionRequest};
 use crate::types::SessionId;
 
-use super::rule::{evaluate, Action, Rule, Ruleset};
+use super::rule::{evaluate, evaluate_intersected, Action, Rule, Ruleset};
 
 pub struct AskInput {
     pub permission: String,
@@ -73,6 +73,24 @@ impl PermissionService {
         }
     }
 
+    /// Like [`ask`](Self::ask), but for a subagent: the verdict is the more restrictive of
+    /// the `parent_stack` (rules inherited from the spawning chain) and `child_stack` (the
+    /// subagent's own rules). Used so a child can never widen what its parent forbids.
+    pub async fn ask_intersected(
+        &self,
+        session_id: &SessionId,
+        parent_stack: &[&Ruleset],
+        child_stack: &[&Ruleset],
+        input: AskInput,
+        cancel: &CancellationToken,
+    ) -> Result<AskDecision, AskError> {
+        match evaluate_intersected(parent_stack, child_stack, &input.permission, &input.pattern) {
+            Action::Allow => Ok(AskDecision::Allowed),
+            Action::Deny => Err(AskError::Denied),
+            Action::Ask => self.ask_user(session_id, input, cancel).await,
+        }
+    }
+
     /// Bypasses ruleset evaluation entirely — used by the doom-loop guard, which must ask
     /// regardless of any `Allow` rule.
     pub async fn force_ask(
diff --git a/crates/harness-core/src/store/api.rs b/crates/harness-core/src/store/api.rs
index a1694c1..db6abf8 100644
--- a/crates/harness-core/src/store/api.rs
+++ b/crates/harness-core/src/store/api.rs
@@ -104,10 +104,7 @@ impl Store {
         self.call(|reply| StoreCmd::DeleteJob(task_id, reply)).await
     }
 
-    pub async fn jobs_for_parent(
-        &self,
-        parent: SessionId,
-    ) -> Result<Vec<JobRecord>, StoreError> {
+    pub async fn jobs_for_parent(&self, parent: SessionId) -> Result<Vec<JobRecord>, StoreError> {
         self.call(|reply| StoreCmd::JobsForParent(parent, reply))
             .await
     }
diff --git a/crates/harness-core/src/tool/mod.rs b/crates/harness-core/src/tool/mod.rs
index 92a9433..685624e 100644
--- a/crates/harness-core/src/tool/mod.rs
+++ b/crates/harness-core/src/tool/mod.rs
@@ -10,6 +10,50 @@ use tokio_util::sync::CancellationToken;
 use crate::permission::{AskDecision, AskError, AskInput, PermissionService, Ruleset};
 use crate::types::{MessageId, SessionId};
 
+/// A request from the `task` tool to run a subagent. The spawner (owned by the composition
+/// root) resolves the agent, enforces the depth limit, applies permission intersection, and
+/// runs the child session foreground or background. See `docs/04-multiagent.md`.
+pub struct SpawnRequest {
+    pub parent_session_id: SessionId,
+    pub parent_message_id: MessageId,
+    pub agent: String,
+    pub description: String,
+    pub prompt: String,
+    /// Alias or task id of a completed job to reuse (continue its child session).
+    pub reuse_task_id: Option<String>,
+    pub background: bool,
+    /// The tool call's cancellation token — used for foreground child runs. Background runs
+    /// are childed from the parent session's run token by the spawner instead.
+    pub cancel: CancellationToken,
+}
+
+#[derive(Debug)]
+pub struct SpawnOutcome {
+    pub child_session_id: SessionId,
+    pub background: bool,
+    /// Board alias assigned to a background launch.
+    pub alias: Option<String>,
+    /// Final assistant text of a foreground run.
+    pub final_text: Option<String>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum SpawnError {
+    #[error("unknown subagent {0:?}, or it is not usable as a subagent")]
+    InvalidAgent(String),
+    #[error("subagent depth limit reached — do this work yourself instead of delegating further")]
+    DepthExceeded,
+    #[error("cannot reuse {0:?}: no completed job with that alias for this session")]
+    ReuseNotFound(String),
+    #[error("{0}")]
+    Other(String),
+}
+
+#[async_trait]
+pub trait SubagentSpawner: Send + Sync {
+    async fn spawn(&self, req: SpawnRequest) -> Result<SpawnOutcome, SpawnError>;
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum ToolError {
     #[error("permission denied")]
@@ -60,6 +104,9 @@ pub struct PermissionHandle {
     session_id: SessionId,
     static_rules: Ruleset,
     extra_rules: Arc<Mutex<Ruleset>>,
+    /// Parent-effective ruleset for a subagent session; empty for a root session. When
+    /// non-empty, verdicts are intersected so a child can only ever be *more* restricted.
+    parent_rules: Ruleset,
     cancel: CancellationToken,
 }
 
@@ -76,10 +123,17 @@ impl PermissionHandle {
             session_id,
             static_rules,
             extra_rules,
+            parent_rules: Vec::new(),
             cancel,
         }
     }
 
+    /// Sets the parent-effective ruleset so this handle intersects verdicts (subagent runs).
+    pub fn with_parent_rules(mut self, parent_rules: Ruleset) -> Self {
+        self.parent_rules = parent_rules;
+        self
+    }
+
     pub async fn ask(
         &self,
         permission: impl Into<String>,
@@ -88,21 +142,29 @@ impl PermissionHandle {
         metadata: serde_json::Value,
     ) -> Result<(), ToolError> {
         let extra_snapshot = self.extra_rules.lock().unwrap().clone();
-        let stack: [&Ruleset; 2] = [&self.static_rules, &extra_snapshot];
-        let decision = self
-            .service
-            .ask(
-                &self.session_id,
-                &stack,
-                AskInput {
-                    permission: permission.into(),
-                    pattern: pattern.into(),
-                    always_pattern: always_pattern.into(),
-                    metadata,
-                },
-                &self.cancel,
-            )
-            .await?;
+        let child_stack: [&Ruleset; 2] = [&self.static_rules, &extra_snapshot];
+        let input = AskInput {
+            permission: permission.into(),
+            pattern: pattern.into(),
+            always_pattern: always_pattern.into(),
+            metadata,
+        };
+        let decision = if self.parent_rules.is_empty() {
+            self.service
+                .ask(&self.session_id, &child_stack, input, &self.cancel)
+                .await?
+        } else {
+            let parent_stack: [&Ruleset; 1] = [&self.parent_rules];
+            self.service
+                .ask_intersected(
+                    &self.session_id,
+                    &parent_stack,
+                    &child_stack,
+                    input,
+                    &self.cancel,
+                )
+                .await?
+        };
         if let AskDecision::AllowedAlways(rule) = decision {
             self.extra_rules.lock().unwrap().push(rule);
         }
@@ -120,6 +182,9 @@ pub struct ToolCtx {
     pub cancel: CancellationToken,
     pub ask: PermissionHandle,
     pub metadata: MetadataSink,
+    /// Present when the engine can spawn subagents (the `task` tool's capability). `None`
+    /// in headless/test contexts with no orchestration wired in.
+    pub spawner: Option<Arc<dyn SubagentSpawner>>,
 }
 
 #[derive(Debug)]
diff --git a/crates/harness-tools/src/bash.rs b/crates/harness-tools/src/bash.rs
index 257c39d..0c3f831 100644
--- a/crates/harness-tools/src/bash.rs
+++ b/crates/harness-tools/src/bash.rs
@@ -149,6 +149,7 @@ mod tests {
                 CancellationToken::new(),
             ),
             metadata,
+            spawner: None,
         }
     }
 
diff --git a/crates/harness-tools/src/edit/mod.rs b/crates/harness-tools/src/edit/mod.rs
index 5e53fb8..6cd140a 100644
--- a/crates/harness-tools/src/edit/mod.rs
+++ b/crates/harness-tools/src/edit/mod.rs
@@ -232,6 +232,7 @@ mod tests {
                 CancellationToken::new(),
             ),
             metadata,
+            spawner: None,
         }
     }
 
diff --git a/crates/harness-tools/src/glob.rs b/crates/harness-tools/src/glob.rs
index 2e6bdcb..44b59db 100644
--- a/crates/harness-tools/src/glob.rs
+++ b/crates/harness-tools/src/glob.rs
@@ -130,6 +130,7 @@ mod tests {
                 CancellationToken::new(),
             ),
             metadata,
+            spawner: None,
         }
     }
 
diff --git a/crates/harness-tools/src/grep.rs b/crates/harness-tools/src/grep.rs
index f9556ae..012f33c 100644
--- a/crates/harness-tools/src/grep.rs
+++ b/crates/harness-tools/src/grep.rs
@@ -159,6 +159,7 @@ mod tests {
                 CancellationToken::new(),
             ),
             metadata,
+            spawner: None,
         }
     }
 
diff --git a/crates/harness-tools/src/lib.rs b/crates/harness-tools/src/lib.rs
index 211e25b..175d96a 100644
--- a/crates/harness-tools/src/lib.rs
+++ b/crates/harness-tools/src/lib.rs
@@ -4,6 +4,7 @@ mod glob;
 mod grep;
 mod paths;
 mod read;
+mod task;
 mod write;
 
 pub use bash::BashTool;
@@ -11,6 +12,7 @@ pub use edit::EditTool;
 pub use glob::GlobTool;
 pub use grep::GrepTool;
 pub use read::ReadTool;
+pub use task::TaskTool;
 pub use write::WriteTool;
 
 use std::sync::Arc;
@@ -26,3 +28,10 @@ pub fn register_builtins(registry: &mut ToolRegistry) {
     registry.register(Arc::new(GlobTool));
     registry.register(Arc::new(GrepTool));
 }
+
+/// Registers the multiagent `task` tool (M4). Kept separate from [`register_builtins`] so
+/// non-orchestrating contexts can omit delegation; the tool no-ops with an error if the
+/// session has no spawner wired in.
+pub fn register_task_tool(registry: &mut ToolRegistry) {
+    registry.register(Arc::new(TaskTool));
+}
diff --git a/crates/harness-tools/src/read.rs b/crates/harness-tools/src/read.rs
index 5e20a6c..4ba2aa3 100644
--- a/crates/harness-tools/src/read.rs
+++ b/crates/harness-tools/src/read.rs
@@ -139,6 +139,7 @@ mod tests {
                 CancellationToken::new(),
             ),
             metadata,
+            spawner: None,
         }
     }
 
diff --git a/crates/harness-tools/src/task.rs b/crates/harness-tools/src/task.rs
new file mode 100644
index 0000000..e934cc0
--- /dev/null
+++ b/crates/harness-tools/src/task.rs
@@ -0,0 +1,140 @@
+//! The `task` tool: delegate work to a specialist subagent, foreground or background.
+//!
+//! This tool is deliberately thin — it validates input, gates on a `task/<agent>` permission,
+//! and hands off to the engine's `SubagentSpawner` (owned by the composition root), which
+//! resolves the agent, enforces the depth limit, applies permission intersection, and runs
+//! the child session. See `docs/04-multiagent.md`.
+
+use async_trait::async_trait;
+use serde::Deserialize;
+use serde_json::json;
+
+use harness_core::tool::{
+    invalid_input, SpawnError, SpawnRequest, Tool, ToolCtx, ToolError, ToolOutput,
+};
+
+#[derive(Debug, Deserialize)]
+struct TaskInput {
+    /// Short human-facing label for the subtask (shown on the job board).
+    description: String,
+    /// The full instruction handed to the subagent.
+    prompt: String,
+    /// Which specialist to run (must be a subagent-capable agent).
+    subagent_type: String,
+    /// Alias or task id of a completed job to continue instead of starting fresh.
+    #[serde(default)]
+    task_id: Option<String>,
+    /// Run in the background and return immediately (tracked on the job board).
+    #[serde(default)]
+    background: bool,
+}
+
+pub struct TaskTool;
+
+#[async_trait]
+impl Tool for TaskTool {
+    fn name(&self) -> &str {
+        "task"
+    }
+
+    fn description(&self) -> &str {
+        "Delegate a self-contained unit of work to a specialist subagent. Set `background: \
+         true` to launch it without blocking (track progress on the job board); reuse a \
+         completed subagent by passing its `task_id`/alias to continue the same session."
+    }
+
+    fn parameters(&self) -> serde_json::Value {
+        json!({
+            "type": "object",
+            "properties": {
+                "description": { "type": "string", "description": "Short label for the subtask." },
+                "prompt": { "type": "string", "description": "Full instruction for the subagent." },
+                "subagent_type": { "type": "string", "description": "Specialist to run." },
+                "task_id": { "type": "string", "description": "Alias/id of a completed job to reuse." },
+                "background": { "type": "boolean", "description": "Launch without blocking." }
+            },
+            "required": ["description", "prompt", "subagent_type"]
+        })
+    }
+
+    async fn execute(
+        &self,
+        input: serde_json::Value,
+        ctx: ToolCtx,
+    ) -> Result<ToolOutput, ToolError> {
+        let args: TaskInput = serde_json::from_value(input).map_err(|e| invalid_input(self, e))?;
+
+        let Some(spawner) = ctx.spawner.clone() else {
+            return Err(ToolError::Other(
+                "subagent delegation is not available in this session".into(),
+            ));
+        };
+
+        // Gate on task/<agent>. `Always` grants blanket delegation to this specialist.
+        ctx.ask
+            .ask(
+                "task",
+                &args.subagent_type,
+                &args.subagent_type,
+                json!({
+                    "agent": args.subagent_type,
+                    "description": args.description,
+                    "background": args.background,
+                }),
+            )
+            .await?;
+
+        let req = SpawnRequest {
+            parent_session_id: ctx.session_id.clone(),
+            parent_message_id: ctx.message_id.clone(),
+            agent: args.subagent_type.clone(),
+            description: args.description.clone(),
+            prompt: args.prompt,
+            reuse_task_id: args.task_id,
+            background: args.background,
+            cancel: ctx.cancel.clone(),
+        };
+
+        let outcome = spawner.spawn(req).await.map_err(map_spawn_error)?;
+
+        if outcome.background {
+            let alias = outcome.alias.unwrap_or_default();
+            Ok(ToolOutput {
+                title: format!("launched {} ({alias})", args.subagent_type),
+                output: format!(
+                    "Launched background task {alias} ({}). Check the job board; do not poll — \
+                     wait for completion.",
+                    outcome.child_session_id
+                ),
+                metadata: json!({
+                    "child_session": outcome.child_session_id,
+                    "agent": args.subagent_type,
+                    "alias": alias,
+                    "background": true,
+                }),
+            })
+        } else {
+            let text = outcome.final_text.unwrap_or_default();
+            Ok(ToolOutput {
+                title: format!("{} — {}", args.subagent_type, args.description),
+                output: text,
+                metadata: json!({
+                    "child_session": outcome.child_session_id,
+                    "agent": args.subagent_type,
+                    "background": false,
+                }),
+            })
+        }
+    }
+}
+
+fn map_spawn_error(err: SpawnError) -> ToolError {
+    match err {
+        // Depth/agent problems are the model's to fix — surface as tool errors it can read
+        // and act on, not hard failures.
+        SpawnError::DepthExceeded => ToolError::Other(err.to_string()),
+        SpawnError::InvalidAgent(_) => ToolError::Invalid(err.to_string()),
+        SpawnError::ReuseNotFound(_) => ToolError::Invalid(err.to_string()),
+        SpawnError::Other(msg) => ToolError::Other(msg),
+    }
+}
diff --git a/crates/harness-tools/src/write.rs b/crates/harness-tools/src/write.rs
index a3268db..7011030 100644
--- a/crates/harness-tools/src/write.rs
+++ b/crates/harness-tools/src/write.rs
@@ -114,6 +114,7 @@ mod tests {
                 CancellationToken::new(),
             ),
             metadata,
+            spawner: None,
         }
     }