harness-tools + app: task tool, subagent spawner, board wiring (M4)
The `task` tool delegates to a SubagentSpawner (owned by the composition root): resolves the agent, enforces the depth limit, applies permission intersection, filters tools per-agent, and runs the child session foreground or background. Foreground returns the child's final text; background registers on the job board and detaches under the parent run token. The run loop injects the board into primary-agent requests and reconciles terminal jobs each step. Integration tests cover foreground run + alias reuse continuing the same child session, depth-limit and unknown-agent rejection, and board prompt injection. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -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
|
||||
|
||||
+643
-29
@@ -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]
|
||||
|
||||
Reference in New Issue
Block a user