feat(engine): Phase 2 — execution loop, capability system, thread runtime

Add the core execution engine to ironclaw_engine crate:

- CapabilityRegistry: register/get/list capabilities and actions
- LeaseManager: async lease lifecycle (grant, check, consume, revoke, expire)
- PolicyEngine: deterministic effect-level allow/deny/approve
- ThreadTree: parent-child relationship tracking
- ThreadSignal/ThreadOutcome: inter-thread messaging via mpsc
- ThreadManager: spawn threads as tokio tasks, stop, inject messages, join
- ExecutionLoop: core loop replacing run_agentic_loop() with signals,
  context building, LLM calls, action execution, and event recording
- Structured executor (Tier 0): lease lookup → policy check → effect execution
- Tool intent nudge detection
- MemoryStore + RetrievalEngine stubs for Phase 4
- Full 8-phase architecture plan in docs/plans/
- CLAUDE.md spec for the engine crate

74 tests passing, zero clippy warnings.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
This commit is contained in:
2026-03-21 00:16:41 -07:00
co-authored by Claude Opus 4.6
parent 8be19a4128
commit bf7dfb8c49
19 changed files with 2497 additions and 10 deletions
+1
View File
@@ -191,6 +191,7 @@ When modifying a module with a spec, read the spec first. Code follows spec; spe
| `src/setup/` | `src/setup/README.md` |
| `src/tools/` | `src/tools/README.md` |
| `src/workspace/` | `src/workspace/README.md` |
| `crates/ironclaw_engine/` | `crates/ironclaw_engine/CLAUDE.md` |
| `tests/e2e/` | `tests/e2e/CLAUDE.md` |
## Job State Machine
+139
View File
@@ -0,0 +1,139 @@
# IronClaw Engine Crate
Unified thread-capability-CodeAct execution model. Replaces ~10 separate abstractions (Session, Job, Routine, Channel, Tool, Skill, Hook, Observer, Extension, LoopDelegate) with 5 primitives.
## Full Architecture Plan
See `docs/plans/2026-03-20-engine-v2-architecture.md` for the 8-phase roadmap.
## Five Primitives
| Primitive | Purpose | Replaces |
|-----------|---------|----------|
| **Thread** | Unit of work with lifecycle, parent-child tree, capability leases | Session + Job + Routine + Sub-agent |
| **Step** | Unit of execution (one LLM call + its action executions) | Agentic loop iteration + tool calls |
| **Capability** | Unit of effect (actions + knowledge + policies) | Tool + Skill + Hook + Extension |
| **MemoryDoc** | Unit of durable knowledge (summaries, lessons, playbooks) | Workspace memory blobs |
| **Project** | Unit of context (scopes memory, threads, missions) | Flat workspace namespace |
## Build & Test
```bash
cargo check -p ironclaw_engine
cargo clippy -p ironclaw_engine --all-targets -- -D warnings
cargo test -p ironclaw_engine
```
## Module Map
```
src/
├── lib.rs # Public API, re-exports
├── types/ # Core data structures (no async, no I/O)
│ ├── thread.rs # Thread, ThreadId, ThreadState (state machine), ThreadType, ThreadConfig
│ ├── step.rs # Step, StepId, LlmResponse, ActionCall, ActionResult, TokenUsage
│ ├── capability.rs # Capability, ActionDef, EffectType, CapabilityLease, PolicyRule
│ ├── memory.rs # MemoryDoc, DocId, DocType (Summary/Lesson/Playbook/Issue/Spec/Note)
│ ├── project.rs # Project, ProjectId
│ ├── event.rs # ThreadEvent, EventKind (16 variants for event sourcing)
│ ├── message.rs # ThreadMessage, MessageRole
│ ├── provenance.rs # Provenance enum (User/System/ToolOutput/LlmGenerated/etc.)
│ └── error.rs # EngineError, ThreadError, StepError, CapabilityError
├── traits/ # External dependency abstractions (host implements these)
│ ├── llm.rs # LlmBackend trait
│ ├── store.rs # Store trait (18 CRUD methods)
│ └── effect.rs # EffectExecutor trait
├── capability/ # Capability management
│ ├── registry.rs # CapabilityRegistry — register/get/list capabilities
│ ├── lease.rs # LeaseManager — grant/check/consume/revoke/expire leases
│ └── policy.rs # PolicyEngine — deterministic effect-level allow/deny/approve
├── runtime/ # Thread lifecycle management
│ ├── manager.rs # ThreadManager — spawn, stop, inject messages, join threads
│ ├── tree.rs # ThreadTree — parent-child relationships
│ └── messaging.rs # ThreadSignal, ThreadOutcome, signal channels
├── executor/ # Step execution
│ ├── loop_engine.rs # ExecutionLoop — core loop replacing run_agentic_loop()
│ ├── structured.rs # Tier 0: structured tool call execution
│ ├── context.rs # Context builder (messages + actions from leases)
│ └── intent.rs # Tool intent nudge detection
├── memory/ # Memory document system
│ ├── store.rs # MemoryStore — project-scoped doc CRUD
│ └── retrieval.rs # RetrievalEngine — context building (stub, Phase 4)
└── reflection/ # Post-thread reflection (stub, Phase 4)
└── mod.rs
```
## Thread State Machine
```
Created → Running → Waiting → Running (resume)
→ Suspended → Running (resume)
→ Completed → Reflecting → Done
→ Failed
```
Validated by `ThreadState::can_transition_to()`. Terminal states: `Done`, `Failed`.
## External Trait Boundaries
The engine defines three traits that the host crate implements:
| Trait | Purpose | Host wraps |
|-------|---------|------------|
| `LlmBackend` | `complete(messages, actions, config) -> LlmOutput` | `LlmProvider` |
| `Store` | Thread/Step/Event/Project/Doc/Lease CRUD | `Database` (PostgreSQL + libSQL) |
| `EffectExecutor` | `execute_action(name, params, lease, ctx) -> ActionResult` | `ToolRegistry` + `SafetyLayer` |
## Execution Loop
`ExecutionLoop::run()` mirrors `run_agentic_loop()`:
1. Check signals (Stop, InjectMessage) via `mpsc::Receiver`
2. Build context (messages + available actions from active leases)
3. Call LLM via `LlmBackend::complete()`
4. If text: check tool intent nudge, return if final response
5. If action calls: for each call, find lease → check policy → consume use → execute via `EffectExecutor` → record result
6. Record Step, emit ThreadEvents
7. Repeat until: text response, stop signal, max iterations, or approval needed
## Capability Leases
Threads don't have static permissions. They receive **leases** — scoped, time-limited, use-limited grants:
```rust
CapabilityLease {
thread_id, capability_name, granted_actions,
expires_at: Option<DateTime>, // time-limited
max_uses: Option<u32>, // use-limited
revoked: bool,
}
```
The `PolicyEngine` evaluates actions against leases deterministically: `Deny > RequireApproval > Allow`.
## Effect Types
Every action declares its side effects. The policy engine uses these for allow/deny:
```
ReadLocal, ReadExternal, WriteLocal, WriteExternal,
CredentialedNetwork, Compute, Financial
```
## Key Design Decisions
1. **No dependency on main `ironclaw` crate** — clean separation, testable in isolation
2. **No safety logic** — sanitization/leak detection is applied at the adapter boundary (`EffectExecutor` impl)
3. **Event sourcing from day one** — every thread records a complete event log via `ThreadEvent`
4. **Tier 0 only (MVP)** — structured tool calls. CodeAct (Tier 1-3) added in Phase 3
5. **Engine owns its message type**`ThreadMessage` is simpler than `ChatMessage`; bridge adapters handle conversion
## Code Style
Follows the main crate's conventions from `/CLAUDE.md`:
- No `.unwrap()` or `.expect()` in production code (tests are fine)
- `thiserror` for error types
- Map errors with context
- Prefer strong types over strings (newtypes for IDs)
- All I/O is async with tokio
- `Arc<T>` for shared state, `RwLock` for concurrent access
@@ -0,0 +1,225 @@
//! Lease manager — grants, validates, and expires capability leases.
use std::collections::HashMap;
use chrono::Utc;
use tokio::sync::RwLock;
use crate::types::capability::{CapabilityLease, LeaseId};
use crate::types::error::EngineError;
use crate::types::thread::ThreadId;
/// Manages the lifecycle of capability leases.
///
/// Leases are the mechanism by which threads gain access to capabilities.
/// They are scoped (time-limited, use-limited, action-restricted) to bound
/// the blast radius of any single thread.
pub struct LeaseManager {
active: RwLock<HashMap<LeaseId, CapabilityLease>>,
}
impl LeaseManager {
pub fn new() -> Self {
Self {
active: RwLock::new(HashMap::new()),
}
}
/// Grant a new lease to a thread.
pub async fn grant(
&self,
thread_id: ThreadId,
capability_name: impl Into<String>,
granted_actions: Vec<String>,
duration: Option<chrono::Duration>,
max_uses: Option<u32>,
) -> CapabilityLease {
let now = Utc::now();
let lease = CapabilityLease {
id: LeaseId::new(),
thread_id,
capability_name: capability_name.into(),
granted_actions,
granted_at: now,
expires_at: duration.map(|d| now + d),
max_uses,
uses_remaining: max_uses,
revoked: false,
};
self.active.write().await.insert(lease.id, lease.clone());
lease
}
/// Check whether a lease is still valid. Returns the lease if valid.
pub async fn check(&self, lease_id: LeaseId) -> Result<CapabilityLease, EngineError> {
let leases = self.active.read().await;
let lease = leases.get(&lease_id).ok_or_else(|| EngineError::LeaseExpired {
capability_name: format!("lease {lease_id:?} not found"),
})?;
if !lease.is_valid() {
return Err(EngineError::LeaseExpired {
capability_name: lease.capability_name.clone(),
});
}
Ok(lease.clone())
}
/// Consume one use of a lease. Returns error if the lease is invalid or exhausted.
pub async fn consume_use(&self, lease_id: LeaseId) -> Result<(), EngineError> {
let mut leases = self.active.write().await;
let lease = leases.get_mut(&lease_id).ok_or_else(|| EngineError::LeaseExpired {
capability_name: format!("lease {lease_id:?} not found"),
})?;
if !lease.is_valid() {
return Err(EngineError::LeaseExpired {
capability_name: lease.capability_name.clone(),
});
}
if !lease.consume_use() {
return Err(EngineError::LeaseExpired {
capability_name: lease.capability_name.clone(),
});
}
Ok(())
}
/// Revoke a lease by ID.
pub async fn revoke(&self, lease_id: LeaseId, _reason: &str) {
let mut leases = self.active.write().await;
if let Some(lease) = leases.get_mut(&lease_id) {
lease.revoked = true;
}
}
/// Remove all expired or revoked leases from the active set.
pub async fn expire_stale(&self) -> usize {
let mut leases = self.active.write().await;
let before = leases.len();
leases.retain(|_, lease| lease.is_valid());
before - leases.len()
}
/// Get all active (valid) leases for a thread.
pub async fn active_for_thread(&self, thread_id: ThreadId) -> Vec<CapabilityLease> {
let leases = self.active.read().await;
leases
.values()
.filter(|l| l.thread_id == thread_id && l.is_valid())
.cloned()
.collect()
}
/// Find the lease that grants a specific action to a thread.
pub async fn find_lease_for_action(
&self,
thread_id: ThreadId,
action_name: &str,
) -> Option<CapabilityLease> {
let leases = self.active.read().await;
leases
.values()
.find(|l| l.thread_id == thread_id && l.is_valid() && l.covers_action(action_name))
.cloned()
}
}
impl Default for LeaseManager {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::thread::ThreadId;
#[tokio::test]
async fn grant_and_check() {
let mgr = LeaseManager::new();
let tid = ThreadId::new();
let lease = mgr.grant(tid, "github", vec![], None, None).await;
assert!(mgr.check(lease.id).await.is_ok());
}
#[tokio::test]
async fn check_nonexistent_fails() {
let mgr = LeaseManager::new();
assert!(mgr.check(LeaseId::new()).await.is_err());
}
#[tokio::test]
async fn consume_use_works() {
let mgr = LeaseManager::new();
let tid = ThreadId::new();
let lease = mgr.grant(tid, "github", vec![], None, Some(2)).await;
assert!(mgr.consume_use(lease.id).await.is_ok());
assert!(mgr.consume_use(lease.id).await.is_ok());
assert!(mgr.consume_use(lease.id).await.is_err());
}
#[tokio::test]
async fn revoke_invalidates() {
let mgr = LeaseManager::new();
let tid = ThreadId::new();
let lease = mgr.grant(tid, "github", vec![], None, None).await;
mgr.revoke(lease.id, "test").await;
assert!(mgr.check(lease.id).await.is_err());
}
#[tokio::test]
async fn expire_stale_removes_revoked() {
let mgr = LeaseManager::new();
let tid = ThreadId::new();
let lease = mgr.grant(tid, "github", vec![], None, None).await;
mgr.revoke(lease.id, "done").await;
let removed = mgr.expire_stale().await;
assert_eq!(removed, 1);
assert!(mgr.active_for_thread(tid).await.is_empty());
}
#[tokio::test]
async fn active_for_thread_filters_correctly() {
let mgr = LeaseManager::new();
let t1 = ThreadId::new();
let t2 = ThreadId::new();
mgr.grant(t1, "github", vec![], None, None).await;
mgr.grant(t1, "memory", vec![], None, None).await;
mgr.grant(t2, "slack", vec![], None, None).await;
assert_eq!(mgr.active_for_thread(t1).await.len(), 2);
assert_eq!(mgr.active_for_thread(t2).await.len(), 1);
}
#[tokio::test]
async fn find_lease_for_action_respects_grants() {
let mgr = LeaseManager::new();
let tid = ThreadId::new();
mgr.grant(
tid,
"github",
vec!["create_issue".into(), "list_prs".into()],
None,
None,
)
.await;
assert!(mgr.find_lease_for_action(tid, "create_issue").await.is_some());
assert!(mgr.find_lease_for_action(tid, "delete_repo").await.is_none());
}
#[tokio::test]
async fn expired_lease_not_active() {
let mgr = LeaseManager::new();
let tid = ThreadId::new();
let lease = mgr
.grant(
tid,
"github",
vec![],
Some(chrono::Duration::seconds(-10)),
None,
)
.await;
assert!(mgr.check(lease.id).await.is_err());
assert!(mgr.active_for_thread(tid).await.is_empty());
}
}
+11 -2
View File
@@ -1,4 +1,13 @@
//! Capability management.
//!
//! Registry, lease management, and deterministic policy engine.
//! Implemented in Phase 2.
//! - [`CapabilityRegistry`] — stores known capabilities and their actions
//! - [`LeaseManager`] — grants, validates, and expires capability leases
//! - [`PolicyEngine`] — deterministic effect-level allow/deny/approve
pub mod lease;
pub mod policy;
pub mod registry;
pub use lease::LeaseManager;
pub use policy::{PolicyDecision, PolicyEngine};
pub use registry::CapabilityRegistry;
@@ -0,0 +1,283 @@
//! Deterministic policy engine.
//!
//! Evaluates whether an action is allowed, denied, or requires approval
//! based on effect types, capability policies, and thread leases.
//! No LLM calls — purely deterministic.
use crate::types::capability::{
ActionDef, CapabilityLease, EffectType, PolicyCondition, PolicyEffect, PolicyRule,
};
/// The result of a policy evaluation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PolicyDecision {
Allow,
Deny { reason: String },
RequireApproval { reason: String },
}
/// Deterministic policy engine.
///
/// Evaluation precedence: Deny > RequireApproval > Allow.
/// Checks are evaluated in order: global policies, then capability policies,
/// then action-level `requires_approval`, then effect-type checks against
/// the lease's allowed effects.
pub struct PolicyEngine {
global_policies: Vec<PolicyRule>,
/// Effect types that are always denied unless explicitly overridden.
denied_effects: Vec<EffectType>,
}
impl PolicyEngine {
pub fn new() -> Self {
Self {
global_policies: Vec::new(),
denied_effects: Vec::new(),
}
}
/// Add a global policy rule.
pub fn add_global_policy(&mut self, rule: PolicyRule) {
self.global_policies.push(rule);
}
/// Add an effect type that is always denied.
pub fn deny_effect(&mut self, effect: EffectType) {
self.denied_effects.push(effect);
}
/// Evaluate whether an action is allowed given a lease and capability policies.
pub fn evaluate(
&self,
action: &ActionDef,
lease: &CapabilityLease,
capability_policies: &[PolicyRule],
) -> PolicyDecision {
// 1. Check lease validity
if !lease.is_valid() {
return PolicyDecision::Deny {
reason: format!("lease for {} is expired/revoked", lease.capability_name),
};
}
// 2. Check lease covers this action
if !lease.covers_action(&action.name) {
return PolicyDecision::Deny {
reason: format!(
"lease for {} does not cover action {}",
lease.capability_name, action.name
),
};
}
// 3. Check denied effect types
for effect in &action.effects {
if self.denied_effects.contains(effect) {
return PolicyDecision::Deny {
reason: format!("effect type {effect:?} is denied by global policy"),
};
}
}
// 4. Evaluate global policies
let mut decision = PolicyDecision::Allow;
for rule in &self.global_policies {
if rule_matches(rule, action) {
decision = merge_decision(decision, rule.effect, &rule.name);
}
}
// 5. Evaluate capability-level policies
for rule in capability_policies {
if rule_matches(rule, action) {
decision = merge_decision(decision, rule.effect, &rule.name);
}
}
// 6. Check action-level requires_approval
if action.requires_approval {
decision = merge_decision(
decision,
PolicyEffect::RequireApproval,
"action requires approval",
);
}
decision
}
}
impl Default for PolicyEngine {
fn default() -> Self {
Self::new()
}
}
/// Check whether a policy rule's condition matches the given action.
fn rule_matches(rule: &PolicyRule, action: &ActionDef) -> bool {
match &rule.condition {
PolicyCondition::Always => true,
PolicyCondition::ActionMatches { pattern } => action.name.contains(pattern.as_str()),
PolicyCondition::EffectTypeIs(effect) => action.effects.contains(effect),
}
}
/// Merge a new policy effect into the current decision.
/// Deny > RequireApproval > Allow.
fn merge_decision(current: PolicyDecision, effect: PolicyEffect, source: &str) -> PolicyDecision {
match effect {
PolicyEffect::Deny => PolicyDecision::Deny {
reason: source.to_string(),
},
PolicyEffect::RequireApproval => match current {
PolicyDecision::Deny { .. } => current,
_ => PolicyDecision::RequireApproval {
reason: source.to_string(),
},
},
PolicyEffect::Allow => current,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::capability::LeaseId;
use crate::types::thread::ThreadId;
use chrono::Utc;
fn make_action(name: &str, effects: Vec<EffectType>, requires_approval: bool) -> ActionDef {
ActionDef {
name: name.into(),
description: String::new(),
parameters_schema: serde_json::json!({}),
effects,
requires_approval,
}
}
fn make_lease() -> CapabilityLease {
CapabilityLease {
id: LeaseId::new(),
thread_id: ThreadId::new(),
capability_name: "test".into(),
granted_actions: vec![],
granted_at: Utc::now(),
expires_at: None,
max_uses: None,
uses_remaining: None,
revoked: false,
}
}
#[test]
fn allow_by_default() {
let engine = PolicyEngine::new();
let action = make_action("read_file", vec![EffectType::ReadLocal], false);
let lease = make_lease();
assert_eq!(engine.evaluate(&action, &lease, &[]), PolicyDecision::Allow);
}
#[test]
fn denied_effect_type() {
let mut engine = PolicyEngine::new();
engine.deny_effect(EffectType::Financial);
let action = make_action("transfer", vec![EffectType::Financial], false);
let lease = make_lease();
assert!(matches!(
engine.evaluate(&action, &lease, &[]),
PolicyDecision::Deny { .. }
));
}
#[test]
fn action_requires_approval() {
let engine = PolicyEngine::new();
let action = make_action("deploy", vec![EffectType::WriteExternal], true);
let lease = make_lease();
assert!(matches!(
engine.evaluate(&action, &lease, &[]),
PolicyDecision::RequireApproval { .. }
));
}
#[test]
fn global_policy_deny_overrides_approval() {
let mut engine = PolicyEngine::new();
engine.add_global_policy(PolicyRule {
name: "no external writes".into(),
condition: PolicyCondition::EffectTypeIs(EffectType::WriteExternal),
effect: PolicyEffect::Deny,
});
let action = make_action("deploy", vec![EffectType::WriteExternal], true);
let lease = make_lease();
assert!(matches!(
engine.evaluate(&action, &lease, &[]),
PolicyDecision::Deny { .. }
));
}
#[test]
fn capability_policy_requires_approval() {
let engine = PolicyEngine::new();
let action = make_action("create_issue", vec![EffectType::WriteExternal], false);
let lease = make_lease();
let cap_policies = vec![PolicyRule {
name: "approve writes".into(),
condition: PolicyCondition::EffectTypeIs(EffectType::WriteExternal),
effect: PolicyEffect::RequireApproval,
}];
assert!(matches!(
engine.evaluate(&action, &lease, &cap_policies),
PolicyDecision::RequireApproval { .. }
));
}
#[test]
fn expired_lease_denied() {
let engine = PolicyEngine::new();
let action = make_action("read", vec![EffectType::ReadLocal], false);
let mut lease = make_lease();
lease.revoked = true;
assert!(matches!(
engine.evaluate(&action, &lease, &[]),
PolicyDecision::Deny { .. }
));
}
#[test]
fn lease_not_covering_action_denied() {
let engine = PolicyEngine::new();
let action = make_action("delete_repo", vec![EffectType::WriteExternal], false);
let mut lease = make_lease();
lease.granted_actions = vec!["create_issue".into()];
assert!(matches!(
engine.evaluate(&action, &lease, &[]),
PolicyDecision::Deny { .. }
));
}
#[test]
fn action_matches_pattern() {
let mut engine = PolicyEngine::new();
engine.add_global_policy(PolicyRule {
name: "approve deletes".into(),
condition: PolicyCondition::ActionMatches {
pattern: "delete".into(),
},
effect: PolicyEffect::RequireApproval,
});
let action = make_action("delete_repo", vec![EffectType::WriteExternal], false);
let lease = make_lease();
assert!(matches!(
engine.evaluate(&action, &lease, &[]),
PolicyDecision::RequireApproval { .. }
));
let action2 = make_action("create_issue", vec![EffectType::WriteExternal], false);
assert_eq!(
engine.evaluate(&action2, &lease, &[]),
PolicyDecision::Allow
);
}
}
@@ -0,0 +1,169 @@
//! Capability registry — stores capability definitions available to the system.
use std::collections::HashMap;
use crate::types::capability::{ActionDef, Capability};
/// Registry of all known capabilities.
///
/// Capabilities are registered at startup (from extensions, built-in tools,
/// etc.) and queried when granting leases or resolving action names.
#[derive(Debug, Default)]
pub struct CapabilityRegistry {
capabilities: HashMap<String, Capability>,
}
impl CapabilityRegistry {
pub fn new() -> Self {
Self::default()
}
/// Register a capability. Overwrites any existing capability with the same name.
pub fn register(&mut self, capability: Capability) {
self.capabilities.insert(capability.name.clone(), capability);
}
/// Look up a capability by name.
pub fn get(&self, name: &str) -> Option<&Capability> {
self.capabilities.get(name)
}
/// List all registered capabilities.
pub fn list(&self) -> Vec<&Capability> {
self.capabilities.values().collect()
}
/// Look up a specific action across all capabilities.
///
/// Returns `(capability_name, action_def)` if found.
pub fn find_action(&self, action_name: &str) -> Option<(&str, &ActionDef)> {
for cap in self.capabilities.values() {
if let Some(action) = cap.actions.iter().find(|a| a.name == action_name) {
return Some((&cap.name, action));
}
}
None
}
/// Get an action definition from a specific capability.
pub fn get_action(&self, capability_name: &str, action_name: &str) -> Option<&ActionDef> {
self.capabilities
.get(capability_name)?
.actions
.iter()
.find(|a| a.name == action_name)
}
/// Collect all action definitions across all capabilities.
pub fn all_actions(&self) -> Vec<&ActionDef> {
self.capabilities
.values()
.flat_map(|c| c.actions.iter())
.collect()
}
/// Number of registered capabilities.
pub fn len(&self) -> usize {
self.capabilities.len()
}
pub fn is_empty(&self) -> bool {
self.capabilities.is_empty()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::capability::EffectType;
fn test_capability() -> Capability {
Capability {
name: "github".into(),
description: "GitHub integration".into(),
actions: vec![
ActionDef {
name: "create_issue".into(),
description: "Create a GitHub issue".into(),
parameters_schema: serde_json::json!({"type": "object"}),
effects: vec![EffectType::WriteExternal, EffectType::CredentialedNetwork],
requires_approval: false,
},
ActionDef {
name: "list_prs".into(),
description: "List pull requests".into(),
parameters_schema: serde_json::json!({"type": "object"}),
effects: vec![EffectType::ReadExternal, EffectType::CredentialedNetwork],
requires_approval: false,
},
],
knowledge: vec!["When creating issues, always add labels.".into()],
policies: vec![],
}
}
#[test]
fn register_and_get() {
let mut reg = CapabilityRegistry::new();
reg.register(test_capability());
assert_eq!(reg.len(), 1);
assert!(reg.get("github").is_some());
assert!(reg.get("slack").is_none());
}
#[test]
fn find_action_across_capabilities() {
let mut reg = CapabilityRegistry::new();
reg.register(test_capability());
let (cap_name, action) = reg.find_action("create_issue").unwrap();
assert_eq!(cap_name, "github");
assert_eq!(action.name, "create_issue");
assert!(reg.find_action("nonexistent").is_none());
}
#[test]
fn get_action_from_capability() {
let mut reg = CapabilityRegistry::new();
reg.register(test_capability());
assert!(reg.get_action("github", "list_prs").is_some());
assert!(reg.get_action("github", "delete_repo").is_none());
assert!(reg.get_action("slack", "list_prs").is_none());
}
#[test]
fn all_actions_collects_across_capabilities() {
let mut reg = CapabilityRegistry::new();
reg.register(test_capability());
reg.register(Capability {
name: "memory".into(),
description: "Memory tools".into(),
actions: vec![ActionDef {
name: "memory_search".into(),
description: "Search memory".into(),
parameters_schema: serde_json::json!({"type": "object"}),
effects: vec![EffectType::ReadLocal],
requires_approval: false,
}],
knowledge: vec![],
policies: vec![],
});
assert_eq!(reg.all_actions().len(), 3);
}
#[test]
fn overwrite_on_re_register() {
let mut reg = CapabilityRegistry::new();
reg.register(test_capability());
assert_eq!(reg.get("github").unwrap().actions.len(), 2);
reg.register(Capability {
name: "github".into(),
description: "Updated".into(),
actions: vec![],
knowledge: vec![],
policies: vec![],
});
assert_eq!(reg.get("github").unwrap().actions.len(), 0);
assert_eq!(reg.len(), 1);
}
}
@@ -0,0 +1,24 @@
//! Context building for LLM calls.
//!
//! Assembles the message sequence and action definitions from thread state,
//! active leases, and (Phase 4) project memory docs.
use std::sync::Arc;
use crate::types::capability::{ActionDef, CapabilityLease};
use crate::types::error::EngineError;
use crate::types::message::ThreadMessage;
use crate::traits::effect::EffectExecutor;
/// Build the context for an LLM call: messages and available actions.
///
/// Phase 1: passes through thread messages + resolves actions from leases.
/// Phase 4 will add memory doc retrieval and injection.
pub async fn build_step_context(
messages: &[ThreadMessage],
leases: &[CapabilityLease],
effects: &Arc<dyn EffectExecutor>,
) -> Result<(Vec<ThreadMessage>, Vec<ActionDef>), EngineError> {
let actions = effects.available_actions(leases).await?;
Ok((messages.to_vec(), actions))
}
@@ -0,0 +1,98 @@
//! Tool intent nudge detection.
//!
//! Detects when the LLM expresses intent to use a tool without actually
//! producing action calls (e.g. "Let me search..." or "I'll fetch...").
//! Mirrors the logic in `src/agent/agentic_loop.rs` `llm_signals_tool_intent`.
/// Check if a text response signals tool intent without actual action calls.
///
/// Returns `true` if the text contains phrases like "Let me search...",
/// "I'll fetch...", etc. that indicate the LLM wanted to call a tool.
pub fn signals_tool_intent(response: &str) -> bool {
let lower = response.to_lowercase();
// Skip false positives
let false_positive_phrases = [
"let me explain",
"let me think",
"let me know",
"let me summarize",
"let me clarify",
];
for phrase in &false_positive_phrases {
if lower.contains(phrase) {
return false;
}
}
let intent_prefixes = ["let me ", "i'll ", "i will ", "i'm going to "];
let action_verbs = [
"search", "look up", "check", "fetch", "find", "query", "read", "run", "execute", "call",
"use", "invoke",
];
for prefix in &intent_prefixes {
if let Some(after) = lower.strip_prefix(prefix) {
for verb in &action_verbs {
if after.starts_with(verb) {
return true;
}
}
}
// Also check if the prefix appears mid-sentence (after period or newline)
for sep in [". ", ".\n", "\n"] {
for part in lower.split(sep) {
let trimmed = part.trim();
if let Some(after) = trimmed.strip_prefix(prefix) {
for verb in &action_verbs {
if after.starts_with(verb) {
return true;
}
}
}
}
}
}
false
}
/// The nudge message injected into context when tool intent is detected.
pub const TOOL_INTENT_NUDGE: &str =
"You expressed intent to use a tool but didn't make an action call. \
Please go ahead and call the appropriate action.";
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn detects_let_me_search() {
assert!(signals_tool_intent("Let me search for that"));
}
#[test]
fn detects_ill_fetch() {
assert!(signals_tool_intent("I'll fetch the latest data"));
}
#[test]
fn ignores_let_me_explain() {
assert!(!signals_tool_intent("Let me explain how this works"));
}
#[test]
fn ignores_let_me_know() {
assert!(!signals_tool_intent("Let me know if you need more"));
}
#[test]
fn ignores_plain_text() {
assert!(!signals_tool_intent("The answer is 42."));
}
#[test]
fn detects_after_period() {
assert!(signals_tool_intent("Sure. Let me search for that."));
}
}
@@ -0,0 +1,629 @@
//! Core execution loop — the replacement for `run_agentic_loop()`.
//!
//! The `ExecutionLoop` owns a thread and drives it through LLM call →
//! action execution → result processing → repeat cycles. Unlike the
//! existing delegate pattern, the loop is self-contained: all behavior
//! differences between thread types are handled via capability leases
//! and policy, not delegate implementations.
use std::sync::Arc;
use tracing::{debug, warn};
use crate::capability::lease::LeaseManager;
use crate::capability::policy::PolicyEngine;
use crate::executor::context::build_step_context;
use crate::executor::intent;
use crate::executor::structured::execute_action_calls;
use crate::runtime::messaging::{SignalReceiver, ThreadOutcome, ThreadSignal};
use crate::traits::effect::{EffectExecutor, ThreadExecutionContext};
use crate::traits::llm::{LlmBackend, LlmCallConfig};
use crate::types::error::EngineError;
use crate::types::event::EventKind;
use crate::types::message::ThreadMessage;
use crate::types::step::{LlmResponse, Step, StepStatus};
use crate::types::thread::{Thread, ThreadState};
/// The core execution loop for a thread.
pub struct ExecutionLoop {
pub thread: Thread,
llm: Arc<dyn LlmBackend>,
effects: Arc<dyn EffectExecutor>,
leases: Arc<LeaseManager>,
policy: Arc<PolicyEngine>,
signal_rx: SignalReceiver,
user_id: String,
}
impl ExecutionLoop {
pub fn new(
thread: Thread,
llm: Arc<dyn LlmBackend>,
effects: Arc<dyn EffectExecutor>,
leases: Arc<LeaseManager>,
policy: Arc<PolicyEngine>,
signal_rx: SignalReceiver,
user_id: String,
) -> Self {
Self {
thread,
llm,
effects,
leases,
policy,
signal_rx,
user_id,
}
}
/// Run the execution loop to completion.
pub async fn run(&mut self) -> Result<ThreadOutcome, EngineError> {
// Transition to Running
self.thread.transition_to(ThreadState::Running, None)?;
let max_iterations = self.thread.config.max_iterations;
let max_nudges = self.thread.config.max_tool_intent_nudges;
let nudge_enabled = self.thread.config.enable_tool_intent_nudge;
let mut nudge_count: u32 = 0;
for iteration in 0..max_iterations {
// 1. Check signals
match self.check_signals() {
SignalAction::Continue => {}
SignalAction::Stop => {
self.thread
.transition_to(ThreadState::Completed, Some("stopped by signal".into()))?;
return Ok(ThreadOutcome::Stopped);
}
SignalAction::Inject(msg) => {
self.thread.add_message(msg);
}
}
// 2. Get active leases
let active_leases = self.leases.active_for_thread(self.thread.id).await;
// 3. Build context
let (messages, actions) =
build_step_context(&self.thread.messages, &active_leases, &self.effects).await?;
// 4. Create step
let mut step = Step::new(self.thread.id, iteration + 1);
step.status = StepStatus::LlmCalling;
self.thread.add_event(EventKind::StepStarted {
step_id: step.id,
});
// 5. Call LLM
let force_text = iteration >= max_iterations.saturating_sub(1);
let config = LlmCallConfig {
force_text,
..LlmCallConfig::default()
};
let llm_output = self.llm.complete(&messages, &actions, &config).await?;
step.tokens_used = llm_output.usage;
self.thread.total_tokens_used += llm_output.usage.total();
step.llm_response = Some(llm_output.response.clone());
// 6. Handle response
match llm_output.response {
LlmResponse::Text(text) => {
// Check for tool intent nudge
if nudge_enabled
&& nudge_count < max_nudges
&& intent::signals_tool_intent(&text)
{
nudge_count += 1;
debug!(
thread_id = %self.thread.id,
nudge_count,
"tool intent detected, injecting nudge"
);
self.thread
.add_message(ThreadMessage::assistant(text));
self.thread
.add_message(ThreadMessage::system(intent::TOOL_INTENT_NUDGE));
step.status = StepStatus::Completed;
step.completed_at = Some(chrono::Utc::now());
self.thread.add_event(EventKind::StepCompleted {
step_id: step.id,
tokens: step.tokens_used,
});
self.thread.step_count += 1;
continue;
}
// Final text response
self.thread
.add_message(ThreadMessage::assistant(text.clone()));
step.status = StepStatus::Completed;
step.completed_at = Some(chrono::Utc::now());
self.thread.add_event(EventKind::StepCompleted {
step_id: step.id,
tokens: step.tokens_used,
});
self.thread.step_count += 1;
self.thread.transition_to(
ThreadState::Completed,
Some("text response".into()),
)?;
return Ok(ThreadOutcome::Completed {
response: Some(text),
});
}
LlmResponse::ActionCalls { calls, content } => {
nudge_count = 0;
// Record assistant message with action calls
self.thread
.add_message(ThreadMessage::assistant_with_actions(content, calls.clone()));
step.status = StepStatus::Executing;
// Build execution context
let exec_ctx = ThreadExecutionContext {
thread_id: self.thread.id,
thread_type: self.thread.thread_type,
project_id: self.thread.project_id,
user_id: self.user_id.clone(),
step_id: step.id,
};
// Execute actions
let batch = execute_action_calls(
&calls,
&self.thread,
&self.effects,
&self.leases,
&self.policy,
&exec_ctx,
&[], // capability-level policies (TODO: resolve from registry)
)
.await?;
// Record events
for event_kind in batch.events {
self.thread.add_event(event_kind);
}
// Add action results as messages
for result in &batch.results {
self.thread.add_message(ThreadMessage::action_result(
&result.call_id,
&result.action_name,
serde_json::to_string(&result.output).unwrap_or_default(),
));
}
step.action_results = batch.results;
step.status = StepStatus::Completed;
step.completed_at = Some(chrono::Utc::now());
self.thread.add_event(EventKind::StepCompleted {
step_id: step.id,
tokens: step.tokens_used,
});
self.thread.step_count += 1;
// Check if approval is needed
if let Some(outcome) = batch.need_approval {
self.thread
.transition_to(ThreadState::Waiting, Some("awaiting approval".into()))?;
return Ok(outcome);
}
}
}
}
// Max iterations reached
warn!(
thread_id = %self.thread.id,
max_iterations,
"max iterations reached"
);
self.thread.transition_to(
ThreadState::Completed,
Some("max iterations reached".into()),
)?;
Ok(ThreadOutcome::MaxIterations)
}
/// Check for pending signals without blocking.
fn check_signals(&mut self) -> SignalAction {
match self.signal_rx.try_recv() {
Ok(ThreadSignal::Stop) => SignalAction::Stop,
Ok(ThreadSignal::InjectMessage(msg)) => SignalAction::Inject(msg),
Ok(ThreadSignal::Suspend) => {
// For now, treat suspend as stop. Phase 3 adds proper suspend/resume.
SignalAction::Stop
}
Ok(ThreadSignal::Resume) | Ok(ThreadSignal::ChildCompleted { .. }) => {
SignalAction::Continue
}
Err(tokio::sync::mpsc::error::TryRecvError::Empty) => SignalAction::Continue,
Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
// Channel closed — the manager dropped our sender. Treat as stop.
SignalAction::Stop
}
}
}
}
enum SignalAction {
Continue,
Stop,
Inject(ThreadMessage),
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::capability::{ActionDef, CapabilityLease, EffectType};
use crate::types::project::ProjectId;
use crate::types::step::{ActionResult, TokenUsage};
use crate::types::thread::{ThreadConfig, ThreadType};
use crate::traits::llm::{LlmCallConfig, LlmOutput};
use std::sync::Mutex;
use std::time::Duration;
// ── Mock LLM ────────────────────────────────────────────
struct MockLlm {
responses: Mutex<Vec<LlmOutput>>,
}
impl MockLlm {
fn new(responses: Vec<LlmOutput>) -> Self {
Self {
responses: Mutex::new(responses),
}
}
}
#[async_trait::async_trait]
impl LlmBackend for MockLlm {
async fn complete(
&self,
_messages: &[ThreadMessage],
_actions: &[ActionDef],
_config: &LlmCallConfig,
) -> Result<LlmOutput, EngineError> {
let mut responses = self.responses.lock().unwrap();
if responses.is_empty() {
Ok(LlmOutput {
response: LlmResponse::Text("(no more responses)".into()),
usage: TokenUsage::default(),
})
} else {
Ok(responses.remove(0))
}
}
fn model_name(&self) -> &str {
"mock"
}
}
// ── Mock EffectExecutor ─────────────────────────────────
struct MockEffects {
results: Mutex<Vec<Result<ActionResult, EngineError>>>,
actions: Vec<ActionDef>,
}
impl MockEffects {
fn new(actions: Vec<ActionDef>, results: Vec<Result<ActionResult, EngineError>>) -> Self {
Self {
results: Mutex::new(results),
actions,
}
}
}
#[async_trait::async_trait]
impl EffectExecutor for MockEffects {
async fn execute_action(
&self,
_action_name: &str,
_parameters: serde_json::Value,
_lease: &CapabilityLease,
_context: &ThreadExecutionContext,
) -> Result<ActionResult, EngineError> {
let mut results = self.results.lock().unwrap();
if results.is_empty() {
Ok(ActionResult {
call_id: String::new(),
action_name: String::new(),
output: serde_json::json!({"result": "ok"}),
is_error: false,
duration: Duration::from_millis(1),
})
} else {
results.remove(0)
}
}
async fn available_actions(
&self,
_leases: &[CapabilityLease],
) -> Result<Vec<ActionDef>, EngineError> {
Ok(self.actions.clone())
}
}
// ── Helpers ─────────────────────────────────────────────
fn text_response(text: &str) -> LlmOutput {
LlmOutput {
response: LlmResponse::Text(text.into()),
usage: TokenUsage {
input_tokens: 100,
output_tokens: 50,
..Default::default()
},
}
}
fn action_response(action_name: &str, call_id: &str) -> LlmOutput {
LlmOutput {
response: LlmResponse::ActionCalls {
calls: vec![crate::types::step::ActionCall {
id: call_id.into(),
action_name: action_name.into(),
parameters: serde_json::json!({}),
}],
content: None,
},
usage: TokenUsage {
input_tokens: 100,
output_tokens: 50,
..Default::default()
},
}
}
fn test_action() -> ActionDef {
ActionDef {
name: "test_tool".into(),
description: "A test tool".into(),
parameters_schema: serde_json::json!({"type": "object"}),
effects: vec![EffectType::ReadLocal],
requires_approval: false,
}
}
async fn make_loop(
llm_responses: Vec<LlmOutput>,
effect_results: Vec<Result<ActionResult, EngineError>>,
config: ThreadConfig,
) -> (ExecutionLoop, crate::runtime::messaging::SignalSender) {
let project_id = ProjectId::new();
let thread = Thread::new("test goal", ThreadType::Foreground, project_id, config);
let tid = thread.id;
let llm = Arc::new(MockLlm::new(llm_responses));
let effects = Arc::new(MockEffects::new(vec![test_action()], effect_results));
let leases = Arc::new(LeaseManager::new());
let policy = Arc::new(PolicyEngine::new());
// Grant a default lease
leases
.grant(tid, "test_cap", vec![], None, None)
.await;
let (tx, rx) = crate::runtime::messaging::signal_channel(16);
let exec = ExecutionLoop::new(
thread,
llm,
effects,
leases,
policy,
rx,
"test-user".into(),
);
(exec, tx)
}
// ── Tests ───────────────────────────────────────────────
#[tokio::test]
async fn text_response_completes() {
let (mut exec, _tx) = make_loop(
vec![text_response("Hello!")],
vec![],
ThreadConfig::default(),
)
.await;
let outcome = exec.run().await.unwrap();
assert!(matches!(outcome, ThreadOutcome::Completed { response: Some(r) } if r == "Hello!"));
assert!(exec.thread.state.is_terminal() || exec.thread.state == ThreadState::Completed);
assert_eq!(exec.thread.step_count, 1);
assert!(exec.thread.total_tokens_used > 0);
}
#[tokio::test]
async fn action_then_text() {
let (mut exec, _tx) = make_loop(
vec![
action_response("test_tool", "call_1"),
text_response("Done!"),
],
vec![Ok(ActionResult {
call_id: "call_1".into(),
action_name: "test_tool".into(),
output: serde_json::json!({"data": "result"}),
is_error: false,
duration: Duration::from_millis(5),
})],
ThreadConfig::default(),
)
.await;
let outcome = exec.run().await.unwrap();
assert!(matches!(outcome, ThreadOutcome::Completed { response: Some(r) } if r == "Done!"));
assert_eq!(exec.thread.step_count, 2);
// Should have: system(nudge not counted), assistant+actions, action_result, assistant
assert!(exec.thread.messages.len() >= 3);
}
#[tokio::test]
async fn max_iterations_reached() {
// LLM always returns actions, so it never exits naturally
let many_actions: Vec<LlmOutput> = (0..5)
.map(|i| action_response("test_tool", &format!("call_{i}")))
.collect();
let many_results: Vec<Result<ActionResult, EngineError>> = (0..5)
.map(|i| {
Ok(ActionResult {
call_id: format!("call_{i}"),
action_name: "test_tool".into(),
output: serde_json::json!({"i": i}),
is_error: false,
duration: Duration::from_millis(1),
})
})
.collect();
let config = ThreadConfig {
max_iterations: 3,
..ThreadConfig::default()
};
let (mut exec, _tx) = make_loop(many_actions, many_results, config).await;
let outcome = exec.run().await.unwrap();
// The last iteration forces text mode, and MockLlm returns action_response
// which gets treated as the 3rd iteration, then on the 3rd iteration force_text
// is set. But MockLlm ignores force_text. So we get MaxIterations after 3 iterations.
// Actually, max_iterations=3, and force_text is set when iteration >= max-1 = 2,
// so iteration 2 (0-indexed) has force_text. The MockLlm still returns action calls,
// so we loop 3 times and exit.
assert!(matches!(
outcome,
ThreadOutcome::MaxIterations | ThreadOutcome::Completed { .. }
));
assert!(exec.thread.step_count <= 3);
}
#[tokio::test]
async fn stop_signal_exits() {
// LLM would loop forever, but we send a stop signal
let many_actions: Vec<LlmOutput> = (0..100)
.map(|i| action_response("test_tool", &format!("call_{i}")))
.collect();
let many_results: Vec<Result<ActionResult, EngineError>> = (0..100)
.map(|i| {
Ok(ActionResult {
call_id: format!("call_{i}"),
action_name: "test_tool".into(),
output: serde_json::json!({}),
is_error: false,
duration: Duration::from_millis(1),
})
})
.collect();
let (mut exec, tx) = make_loop(many_actions, many_results, ThreadConfig::default()).await;
// Send stop before first iteration
tx.send(ThreadSignal::Stop).await.unwrap();
let outcome = exec.run().await.unwrap();
assert!(matches!(outcome, ThreadOutcome::Stopped));
}
#[tokio::test]
async fn inject_message_appears_in_context() {
let (mut exec, tx) = make_loop(
vec![text_response("Got your message")],
vec![],
ThreadConfig::default(),
)
.await;
tx.send(ThreadSignal::InjectMessage(ThreadMessage::user(
"injected!",
)))
.await
.unwrap();
let outcome = exec.run().await.unwrap();
assert!(matches!(outcome, ThreadOutcome::Completed { .. }));
assert!(exec
.thread
.messages
.iter()
.any(|m| m.content == "injected!"));
}
#[tokio::test]
async fn tool_intent_nudge_injected() {
let (mut exec, _tx) = make_loop(
vec![
text_response("Let me search for that"),
text_response("The answer is 42"),
],
vec![],
ThreadConfig {
enable_tool_intent_nudge: true,
max_tool_intent_nudges: 2,
..ThreadConfig::default()
},
)
.await;
let outcome = exec.run().await.unwrap();
assert!(
matches!(outcome, ThreadOutcome::Completed { response: Some(r) } if r == "The answer is 42")
);
assert_eq!(exec.thread.step_count, 2);
// Should have nudge system message
assert!(exec
.thread
.messages
.iter()
.any(|m| m.content.contains("didn't make an action call")));
}
#[tokio::test]
async fn events_are_recorded() {
let (mut exec, _tx) = make_loop(
vec![text_response("Hello!")],
vec![],
ThreadConfig::default(),
)
.await;
exec.run().await.unwrap();
let _event_kinds: Vec<String> = exec
.thread
.events
.iter()
.map(|e| format!("{:?}", std::mem::discriminant(&e.kind)))
.collect();
// Should have: StateChanged(Created->Running), StepStarted, MessageAdded,
// StepCompleted, StateChanged(Running->Completed)
assert!(exec.thread.events.len() >= 4);
// Verify first event is state change to Running
assert!(matches!(
&exec.thread.events[0].kind,
EventKind::StateChanged {
from: ThreadState::Created,
to: ThreadState::Running,
..
}
));
}
}
+11 -3
View File
@@ -1,5 +1,13 @@
//! Step execution.
//!
//! The core execution loop that replaces `run_agentic_loop()`, plus
//! tier-specific executors (structured tool calls, embedded scripting, etc.).
//! Implemented in Phase 2.
//! - [`ExecutionLoop`] — core loop replacing `run_agentic_loop()`
//! - [`structured`] — Tier 0 action execution (structured tool calls)
//! - [`context`] — context building for LLM calls
//! - [`intent`] — tool intent nudge detection
pub mod context;
pub mod intent;
pub mod loop_engine;
pub mod structured;
pub use loop_engine::ExecutionLoop;
@@ -0,0 +1,162 @@
//! Tier 0 executor: structured tool calls.
//!
//! Executes action calls by delegating to the `EffectExecutor` trait,
//! checking leases and policies for each call.
use std::sync::Arc;
use crate::capability::lease::LeaseManager;
use crate::capability::policy::{PolicyDecision, PolicyEngine};
use crate::runtime::messaging::ThreadOutcome;
use crate::traits::effect::{EffectExecutor, ThreadExecutionContext};
use crate::types::error::EngineError;
use crate::types::event::EventKind;
use crate::types::step::{ActionCall, ActionResult};
use crate::types::thread::Thread;
/// Result of executing a batch of action calls.
pub struct ActionBatchResult {
/// Results for each action call (in order).
pub results: Vec<ActionResult>,
/// Events generated during execution.
pub events: Vec<EventKind>,
/// If set, execution was interrupted and the thread needs approval.
pub need_approval: Option<ThreadOutcome>,
}
/// Execute a batch of action calls using the Tier 0 (structured) approach.
///
/// For each action call:
/// 1. Find the lease that grants this action
/// 2. Check policy (deny/allow/approve)
/// 3. Consume a lease use
/// 4. Call `EffectExecutor::execute_action()`
/// 5. Record result and emit event
///
/// Stops at the first action that requires approval.
pub async fn execute_action_calls(
calls: &[ActionCall],
thread: &Thread,
effects: &Arc<dyn EffectExecutor>,
leases: &LeaseManager,
policy: &PolicyEngine,
context: &ThreadExecutionContext,
capability_policies: &[crate::types::capability::PolicyRule],
) -> Result<ActionBatchResult, EngineError> {
let mut results = Vec::with_capacity(calls.len());
let mut events = Vec::new();
for call in calls {
// 1. Find the lease for this action
let lease = match leases.find_lease_for_action(thread.id, &call.action_name).await {
Some(l) => l,
None => {
let error_result = ActionResult {
call_id: call.id.clone(),
action_name: call.action_name.clone(),
output: serde_json::json!({"error": format!(
"no active lease covers action '{}'", call.action_name
)}),
is_error: true,
duration: std::time::Duration::ZERO,
};
events.push(EventKind::ActionFailed {
step_id: context.step_id,
action_name: call.action_name.clone(),
call_id: call.id.clone(),
error: format!("no lease for action '{}'", call.action_name),
});
results.push(error_result);
continue;
}
};
// 2. Find the action definition and check policy
let action_def = effects
.available_actions(std::slice::from_ref(&lease))
.await?
.into_iter()
.find(|a| a.name == call.action_name);
if let Some(ref action_def) = action_def {
let decision = policy.evaluate(action_def, &lease, capability_policies);
match decision {
PolicyDecision::Deny { reason } => {
let error_result = ActionResult {
call_id: call.id.clone(),
action_name: call.action_name.clone(),
output: serde_json::json!({"error": format!("denied: {reason}")}),
is_error: true,
duration: std::time::Duration::ZERO,
};
events.push(EventKind::ActionFailed {
step_id: context.step_id,
action_name: call.action_name.clone(),
call_id: call.id.clone(),
error: reason,
});
results.push(error_result);
continue;
}
PolicyDecision::RequireApproval { .. } => {
events.push(EventKind::ApprovalRequested {
action_name: call.action_name.clone(),
call_id: call.id.clone(),
});
return Ok(ActionBatchResult {
results,
events,
need_approval: Some(ThreadOutcome::NeedApproval {
action_name: call.action_name.clone(),
call_id: call.id.clone(),
parameters: call.parameters.clone(),
}),
});
}
PolicyDecision::Allow => {}
}
}
// 3. Consume a lease use
leases.consume_use(lease.id).await?;
// 4. Execute the action
let result = effects
.execute_action(&call.action_name, call.parameters.clone(), &lease, context)
.await;
match result {
Ok(action_result) => {
events.push(EventKind::ActionExecuted {
step_id: context.step_id,
action_name: call.action_name.clone(),
call_id: call.id.clone(),
duration_ms: action_result.duration.as_millis() as u64,
});
results.push(action_result);
}
Err(e) => {
let error_result = ActionResult {
call_id: call.id.clone(),
action_name: call.action_name.clone(),
output: serde_json::json!({"error": e.to_string()}),
is_error: true,
duration: std::time::Duration::ZERO,
};
events.push(EventKind::ActionFailed {
step_id: context.step_id,
action_name: call.action_name.clone(),
call_id: call.id.clone(),
error: e.to_string(),
});
results.push(error_result);
}
}
}
Ok(ActionBatchResult {
results,
events,
need_approval: None,
})
}
+21
View File
@@ -44,3 +44,24 @@ pub use types::thread::{Thread, ThreadConfig, ThreadId, ThreadState, ThreadType}
pub use traits::effect::{EffectExecutor, ThreadExecutionContext};
pub use traits::llm::{LlmBackend, LlmCallConfig, LlmOutput};
pub use traits::store::Store;
// ── Re-exports: capability ────────────────────────────────────
pub use capability::registry::CapabilityRegistry;
pub use capability::lease::LeaseManager;
pub use capability::policy::{PolicyDecision, PolicyEngine};
// ── Re-exports: runtime ───────────────────────────────────────
pub use runtime::manager::ThreadManager;
pub use runtime::messaging::ThreadOutcome;
pub use runtime::tree::ThreadTree;
// ── Re-exports: executor ──────────────────────────────────────
pub use executor::ExecutionLoop;
// ── Re-exports: memory ────────────────────────────────────────
pub use memory::MemoryStore;
pub use memory::RetrievalEngine;
+8 -3
View File
@@ -1,5 +1,10 @@
//! Memory document system.
//!
//! Project-scoped document storage and retrieval engine for building
//! thread context from durable knowledge.
//! Implemented in Phase 2 (store) and Phase 4 (retrieval).
//! - [`MemoryStore`] — project-scoped document CRUD
//! - [`RetrievalEngine`] — context building from project docs (Phase 4)
pub mod retrieval;
pub mod store;
pub use retrieval::RetrievalEngine;
pub use store::MemoryStore;
@@ -0,0 +1,35 @@
//! Context retrieval engine.
//!
//! Builds context for thread steps by retrieving relevant memory docs
//! from the project. Phase 1: stub. Phase 4 implements keyword + semantic search.
use crate::types::error::EngineError;
use crate::types::memory::MemoryDoc;
use crate::types::project::ProjectId;
/// Retrieves relevant memory docs for a thread's context.
pub struct RetrievalEngine;
impl RetrievalEngine {
pub fn new() -> Self {
Self
}
/// Retrieve relevant memory docs for the given query.
///
/// Phase 1: returns empty vec. Phase 4 implements search.
pub async fn retrieve_context(
&self,
_project_id: ProjectId,
_query: &str,
_max_docs: usize,
) -> Result<Vec<MemoryDoc>, EngineError> {
Ok(Vec::new())
}
}
impl Default for RetrievalEngine {
fn default() -> Self {
Self::new()
}
}
@@ -0,0 +1,66 @@
//! Project-scoped memory document operations.
use std::sync::Arc;
use crate::traits::store::Store;
use crate::types::error::EngineError;
use crate::types::memory::{DocId, DocType, MemoryDoc};
use crate::types::project::ProjectId;
use crate::types::thread::ThreadId;
/// Thin wrapper over the [`Store`] trait for project-scoped doc operations.
pub struct MemoryStore {
store: Arc<dyn Store>,
}
impl MemoryStore {
pub fn new(store: Arc<dyn Store>) -> Self {
Self { store }
}
/// Create a new memory document.
pub async fn create_doc(
&self,
project_id: ProjectId,
doc_type: DocType,
title: &str,
content: &str,
) -> Result<MemoryDoc, EngineError> {
let doc = MemoryDoc::new(project_id, doc_type, title, content);
self.store.save_memory_doc(&doc).await?;
Ok(doc)
}
/// Create a doc linked to a source thread.
pub async fn create_doc_from_thread(
&self,
project_id: ProjectId,
doc_type: DocType,
title: &str,
content: &str,
source_thread_id: ThreadId,
) -> Result<MemoryDoc, EngineError> {
let doc = MemoryDoc::new(project_id, doc_type, title, content)
.with_source_thread(source_thread_id);
self.store.save_memory_doc(&doc).await?;
Ok(doc)
}
/// Load a single doc by ID.
pub async fn get_doc(&self, id: DocId) -> Result<Option<MemoryDoc>, EngineError> {
self.store.load_memory_doc(id).await
}
/// List all docs in a project, optionally filtered by type.
pub async fn list_docs(
&self,
project_id: ProjectId,
doc_type: Option<DocType>,
) -> Result<Vec<MemoryDoc>, EngineError> {
let all = self.store.list_memory_docs(project_id).await?;
match doc_type {
Some(dt) => Ok(all.into_iter().filter(|d| d.doc_type == dt).collect()),
None => Ok(all),
}
}
}
@@ -0,0 +1,422 @@
//! Thread manager — top-level orchestrator for thread lifecycle.
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, error};
use crate::capability::lease::LeaseManager;
use crate::capability::policy::PolicyEngine;
use crate::capability::registry::CapabilityRegistry;
use crate::executor::ExecutionLoop;
use crate::runtime::messaging::{self, SignalSender, ThreadOutcome, ThreadSignal};
use crate::runtime::tree::ThreadTree;
use crate::traits::effect::EffectExecutor;
use crate::traits::llm::LlmBackend;
use crate::traits::store::Store;
use crate::types::error::EngineError;
use crate::types::message::ThreadMessage;
use crate::types::project::ProjectId;
use crate::types::thread::{Thread, ThreadConfig, ThreadId, ThreadType};
/// Handle to a running thread for checking results.
struct RunningThread {
signal_tx: SignalSender,
handle: tokio::task::JoinHandle<Result<ThreadOutcome, EngineError>>,
}
/// Top-level orchestrator for thread lifecycle.
///
/// Manages thread spawning, supervision, signaling, and tree relationships.
pub struct ThreadManager {
llm: Arc<dyn LlmBackend>,
effects: Arc<dyn EffectExecutor>,
store: Arc<dyn Store>,
pub capabilities: Arc<CapabilityRegistry>,
pub leases: Arc<LeaseManager>,
pub policy: Arc<PolicyEngine>,
tree: RwLock<ThreadTree>,
running: RwLock<HashMap<ThreadId, RunningThread>>,
}
impl ThreadManager {
pub fn new(
llm: Arc<dyn LlmBackend>,
effects: Arc<dyn EffectExecutor>,
store: Arc<dyn Store>,
capabilities: Arc<CapabilityRegistry>,
leases: Arc<LeaseManager>,
policy: Arc<PolicyEngine>,
) -> Self {
Self {
llm,
effects,
store,
capabilities,
leases,
policy,
tree: RwLock::new(ThreadTree::new()),
running: RwLock::new(HashMap::new()),
}
}
/// Spawn a new thread and start executing it.
///
/// Grants default capability leases for all registered capabilities.
/// Returns the thread ID immediately; the thread runs in a background task.
pub async fn spawn_thread(
&self,
goal: impl Into<String>,
thread_type: ThreadType,
project_id: ProjectId,
config: ThreadConfig,
parent_id: Option<ThreadId>,
user_id: impl Into<String>,
) -> Result<ThreadId, EngineError> {
let mut thread = Thread::new(goal, thread_type, project_id, config);
if let Some(pid) = parent_id {
thread = thread.with_parent(pid);
}
let thread_id = thread.id;
let user_id = user_id.into();
// Register in tree
if let Some(pid) = parent_id {
self.tree.write().await.add_child(pid, thread_id);
}
// Grant leases for all registered capabilities
for cap in self.capabilities.list() {
let lease = self
.leases
.grant(thread_id, &cap.name, vec![], None, None)
.await;
thread.capability_leases.push(lease.id);
}
// Persist
self.store.save_thread(&thread).await?;
// Create signal channel
let (tx, rx) = messaging::signal_channel(32);
// Build execution loop
let llm = Arc::clone(&self.llm);
let effects = Arc::clone(&self.effects);
let leases = Arc::clone(&self.leases);
let policy = Arc::clone(&self.policy);
let exec_loop = ExecutionLoop::new(thread, llm, effects, leases, policy, rx, user_id);
// Spawn background task
let handle = tokio::spawn(async move {
let mut exec = exec_loop;
let result = exec.run().await;
debug!(thread_id = %thread_id, "thread execution finished");
result
});
self.running.write().await.insert(
thread_id,
RunningThread {
signal_tx: tx,
handle,
},
);
Ok(thread_id)
}
/// Send a stop signal to a running thread.
pub async fn stop_thread(&self, thread_id: ThreadId) -> Result<(), EngineError> {
let running = self.running.read().await;
if let Some(rt) = running.get(&thread_id) {
let _ = rt.signal_tx.send(ThreadSignal::Stop).await;
Ok(())
} else {
Err(EngineError::ThreadNotFound(thread_id))
}
}
/// Inject a user message into a running thread.
pub async fn inject_message(
&self,
thread_id: ThreadId,
message: ThreadMessage,
) -> Result<(), EngineError> {
let running = self.running.read().await;
if let Some(rt) = running.get(&thread_id) {
let _ = rt
.signal_tx
.send(ThreadSignal::InjectMessage(message))
.await;
Ok(())
} else {
Err(EngineError::ThreadNotFound(thread_id))
}
}
/// Check if a thread is still running.
pub async fn is_running(&self, thread_id: ThreadId) -> bool {
let running = self.running.read().await;
running
.get(&thread_id)
.is_some_and(|rt| !rt.handle.is_finished())
}
/// Wait for a thread to finish and return its outcome.
/// Removes the thread from the running set.
pub async fn join_thread(
&self,
thread_id: ThreadId,
) -> Result<ThreadOutcome, EngineError> {
let rt = {
let mut running = self.running.write().await;
running.remove(&thread_id)
};
match rt {
Some(rt) => match rt.handle.await {
Ok(result) => result,
Err(e) => {
error!(thread_id = %thread_id, "thread task panicked: {e}");
Ok(ThreadOutcome::Failed {
error: format!("thread task panicked: {e}"),
})
}
},
None => Err(EngineError::ThreadNotFound(thread_id)),
}
}
/// Get children of a thread.
pub async fn children_of(&self, thread_id: ThreadId) -> Vec<ThreadId> {
let tree = self.tree.read().await;
tree.children_of(thread_id).to_vec()
}
/// Get the parent of a thread.
pub async fn parent_of(&self, thread_id: ThreadId) -> Option<ThreadId> {
let tree = self.tree.read().await;
tree.parent_of(thread_id)
}
/// Clean up finished threads from the running set.
pub async fn cleanup_finished(&self) -> Vec<ThreadId> {
let mut running = self.running.write().await;
let finished: Vec<ThreadId> = running
.iter()
.filter(|(_, rt)| rt.handle.is_finished())
.map(|(id, _)| *id)
.collect();
for id in &finished {
running.remove(id);
}
finished
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::capability::{ActionDef, Capability, CapabilityLease, EffectType};
use crate::types::event::ThreadEvent;
use crate::types::memory::{DocId, MemoryDoc};
use crate::types::project::Project;
use crate::types::step::{ActionResult, LlmResponse, Step, TokenUsage};
use crate::types::thread::ThreadState;
use crate::traits::llm::{LlmCallConfig, LlmOutput};
use std::sync::Mutex;
use std::time::Duration;
// ── Mocks ───────────────────────────────────────────────
struct MockLlm {
responses: Mutex<Vec<LlmOutput>>,
}
impl MockLlm {
fn text(msg: &str) -> Arc<Self> {
Arc::new(Self {
responses: Mutex::new(vec![LlmOutput {
response: LlmResponse::Text(msg.into()),
usage: TokenUsage::default(),
}]),
})
}
}
#[async_trait::async_trait]
impl LlmBackend for MockLlm {
async fn complete(
&self,
_: &[crate::types::message::ThreadMessage],
_: &[ActionDef],
_: &LlmCallConfig,
) -> Result<LlmOutput, EngineError> {
let mut r = self.responses.lock().unwrap();
if r.is_empty() {
Ok(LlmOutput {
response: LlmResponse::Text("done".into()),
usage: TokenUsage::default(),
})
} else {
Ok(r.remove(0))
}
}
fn model_name(&self) -> &str {
"mock"
}
}
struct MockEffects;
#[async_trait::async_trait]
impl EffectExecutor for MockEffects {
async fn execute_action(
&self,
_: &str,
_: serde_json::Value,
_: &CapabilityLease,
_: &crate::traits::effect::ThreadExecutionContext,
) -> Result<ActionResult, EngineError> {
Ok(ActionResult {
call_id: String::new(),
action_name: String::new(),
output: serde_json::json!({}),
is_error: false,
duration: Duration::from_millis(1),
})
}
async fn available_actions(
&self,
_: &[CapabilityLease],
) -> Result<Vec<ActionDef>, EngineError> {
Ok(vec![])
}
}
struct MockStore;
#[async_trait::async_trait]
impl Store for MockStore {
async fn save_thread(&self, _: &Thread) -> Result<(), EngineError> { Ok(()) }
async fn load_thread(&self, _: ThreadId) -> Result<Option<Thread>, EngineError> { Ok(None) }
async fn list_threads(&self, _: ProjectId) -> Result<Vec<Thread>, EngineError> { Ok(vec![]) }
async fn update_thread_state(&self, _: ThreadId, _: ThreadState) -> Result<(), EngineError> { Ok(()) }
async fn save_step(&self, _: &Step) -> Result<(), EngineError> { Ok(()) }
async fn load_steps(&self, _: ThreadId) -> Result<Vec<Step>, EngineError> { Ok(vec![]) }
async fn append_events(&self, _: &[ThreadEvent]) -> Result<(), EngineError> { Ok(()) }
async fn load_events(&self, _: ThreadId) -> Result<Vec<ThreadEvent>, EngineError> { Ok(vec![]) }
async fn save_project(&self, _: &Project) -> Result<(), EngineError> { Ok(()) }
async fn load_project(&self, _: ProjectId) -> Result<Option<Project>, EngineError> { Ok(None) }
async fn save_memory_doc(&self, _: &MemoryDoc) -> Result<(), EngineError> { Ok(()) }
async fn load_memory_doc(&self, _: DocId) -> Result<Option<MemoryDoc>, EngineError> { Ok(None) }
async fn list_memory_docs(&self, _: ProjectId) -> Result<Vec<MemoryDoc>, EngineError> { Ok(vec![]) }
async fn save_lease(&self, _: &CapabilityLease) -> Result<(), EngineError> { Ok(()) }
async fn load_active_leases(&self, _: ThreadId) -> Result<Vec<CapabilityLease>, EngineError> { Ok(vec![]) }
async fn revoke_lease(&self, _: crate::types::capability::LeaseId, _: &str) -> Result<(), EngineError> { Ok(()) }
}
fn make_manager(llm: Arc<dyn LlmBackend>) -> ThreadManager {
let mut caps = CapabilityRegistry::new();
caps.register(Capability {
name: "test".into(),
description: "Test capability".into(),
actions: vec![ActionDef {
name: "test_tool".into(),
description: "Test".into(),
parameters_schema: serde_json::json!({}),
effects: vec![EffectType::ReadLocal],
requires_approval: false,
}],
knowledge: vec![],
policies: vec![],
});
ThreadManager::new(
llm,
Arc::new(MockEffects),
Arc::new(MockStore),
Arc::new(caps),
Arc::new(LeaseManager::new()),
Arc::new(PolicyEngine::new()),
)
}
// ── Tests ───────────────────────────────────────────────
#[tokio::test]
async fn spawn_and_join() {
let mgr = make_manager(MockLlm::text("Hello!"));
let project = ProjectId::new();
let tid = mgr
.spawn_thread("test", ThreadType::Foreground, project, ThreadConfig::default(), None, "user")
.await
.unwrap();
let outcome = mgr.join_thread(tid).await.unwrap();
assert!(matches!(outcome, ThreadOutcome::Completed { response: Some(r) } if r == "Hello!"));
}
#[tokio::test]
async fn stop_thread_works() {
// LLM that returns many action responses
let responses: Vec<LlmOutput> = (0..100)
.map(|i| LlmOutput {
response: LlmResponse::ActionCalls {
calls: vec![crate::types::step::ActionCall {
id: format!("c{i}"),
action_name: "test_tool".into(),
parameters: serde_json::json!({}),
}],
content: None,
},
usage: TokenUsage::default(),
})
.collect();
let mgr = make_manager(Arc::new(MockLlm {
responses: Mutex::new(responses),
}));
let project = ProjectId::new();
let tid = mgr
.spawn_thread("test", ThreadType::Foreground, project, ThreadConfig::default(), None, "user")
.await
.unwrap();
// Give it a moment to start, then stop
tokio::time::sleep(Duration::from_millis(10)).await;
mgr.stop_thread(tid).await.unwrap();
let outcome = mgr.join_thread(tid).await.unwrap();
assert!(matches!(
outcome,
ThreadOutcome::Stopped | ThreadOutcome::Completed { .. } | ThreadOutcome::MaxIterations
));
}
#[tokio::test]
async fn parent_child_tree() {
let mgr = make_manager(MockLlm::text("parent done"));
let project = ProjectId::new();
let parent = mgr
.spawn_thread("parent", ThreadType::Foreground, project, ThreadConfig::default(), None, "user")
.await
.unwrap();
let child = mgr
.spawn_thread("child", ThreadType::Research, project, ThreadConfig::default(), Some(parent), "user")
.await
.unwrap();
assert_eq!(mgr.parent_of(child).await, Some(parent));
assert_eq!(mgr.children_of(parent).await, vec![child]);
}
}
@@ -0,0 +1,53 @@
//! Thread-to-thread messaging via channels.
use crate::types::message::ThreadMessage;
use crate::types::thread::ThreadId;
/// Signal sent to a running thread via its mailbox.
#[derive(Debug)]
pub enum ThreadSignal {
/// Stop the thread gracefully.
Stop,
/// Pause execution (can be resumed later).
Suspend,
/// Resume a suspended thread.
Resume,
/// Inject a user message into the thread's context.
InjectMessage(ThreadMessage),
/// Notification that a child thread completed.
ChildCompleted {
child_id: ThreadId,
outcome: ThreadOutcome,
},
}
/// Final outcome of a thread's execution.
#[derive(Debug, Clone)]
pub enum ThreadOutcome {
/// Completed with an optional text response.
Completed { response: Option<String> },
/// Thread was stopped by a signal.
Stopped,
/// Max iterations reached without completing.
MaxIterations,
/// Terminal failure.
Failed { error: String },
/// A capability action requires user approval before continuing.
NeedApproval {
action_name: String,
call_id: String,
parameters: serde_json::Value,
},
}
/// A mailbox for sending signals to a running thread.
///
/// Each thread gets a `(sender, receiver)` pair. The `ThreadManager` holds
/// the sender; the `ExecutionLoop` holds the receiver.
pub type SignalSender = tokio::sync::mpsc::Sender<ThreadSignal>;
pub type SignalReceiver = tokio::sync::mpsc::Receiver<ThreadSignal>;
/// Create a new signal channel with the given buffer size.
pub fn signal_channel(buffer: usize) -> (SignalSender, SignalReceiver) {
tokio::sync::mpsc::channel(buffer)
}
+11 -2
View File
@@ -1,4 +1,13 @@
//! Thread lifecycle management.
//!
//! ThreadManager, thread tree, and inter-thread messaging.
//! Implemented in Phase 2.
//! - [`ThreadManager`] — top-level orchestrator for spawning and supervising threads
//! - [`ThreadTree`] — parent-child relationship tracking
//! - [`messaging`] — inter-thread signal channel
pub mod manager;
pub mod messaging;
pub mod tree;
pub use manager::ThreadManager;
pub use messaging::ThreadOutcome;
pub use tree::ThreadTree;
+129
View File
@@ -0,0 +1,129 @@
//! Thread tree — parent-child relationship tracking.
use std::collections::HashMap;
use crate::types::thread::ThreadId;
/// Manages parent-child thread relationships.
///
/// Simple in-memory tree. Threads form a forest (multiple roots).
#[derive(Debug, Default)]
pub struct ThreadTree {
/// child → parent
parents: HashMap<ThreadId, ThreadId>,
/// parent → children (ordered by insertion)
children: HashMap<ThreadId, Vec<ThreadId>>,
}
impl ThreadTree {
pub fn new() -> Self {
Self::default()
}
/// Register a parent-child relationship.
pub fn add_child(&mut self, parent_id: ThreadId, child_id: ThreadId) {
self.parents.insert(child_id, parent_id);
self.children.entry(parent_id).or_default().push(child_id);
}
/// Get the parent of a thread, if any.
pub fn parent_of(&self, thread_id: ThreadId) -> Option<ThreadId> {
self.parents.get(&thread_id).copied()
}
/// Get the children of a thread.
pub fn children_of(&self, thread_id: ThreadId) -> &[ThreadId] {
self.children
.get(&thread_id)
.map(Vec::as_slice)
.unwrap_or(&[])
}
/// Walk up the tree to collect all ancestors (parent, grandparent, ...).
pub fn ancestors(&self, thread_id: ThreadId) -> Vec<ThreadId> {
let mut result = Vec::new();
let mut current = thread_id;
while let Some(parent) = self.parents.get(&current) {
result.push(*parent);
current = *parent;
}
result
}
/// Remove a thread from the tree. Does not remove its children.
pub fn remove(&mut self, thread_id: ThreadId) {
if let Some(parent) = self.parents.remove(&thread_id)
&& let Some(siblings) = self.children.get_mut(&parent)
{
siblings.retain(|id| *id != thread_id);
}
// Orphan any children (their parent_id entries become stale)
self.children.remove(&thread_id);
}
/// Check if a thread is a root (no parent).
pub fn is_root(&self, thread_id: ThreadId) -> bool {
!self.parents.contains_key(&thread_id)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn add_and_query() {
let mut tree = ThreadTree::new();
let parent = ThreadId::new();
let child1 = ThreadId::new();
let child2 = ThreadId::new();
tree.add_child(parent, child1);
tree.add_child(parent, child2);
assert_eq!(tree.parent_of(child1), Some(parent));
assert_eq!(tree.parent_of(child2), Some(parent));
assert_eq!(tree.children_of(parent).len(), 2);
assert!(tree.is_root(parent));
assert!(!tree.is_root(child1));
}
#[test]
fn ancestors_walk_up() {
let mut tree = ThreadTree::new();
let root = ThreadId::new();
let mid = ThreadId::new();
let leaf = ThreadId::new();
tree.add_child(root, mid);
tree.add_child(mid, leaf);
let ancestors = tree.ancestors(leaf);
assert_eq!(ancestors, vec![mid, root]);
}
#[test]
fn remove_detaches_from_parent() {
let mut tree = ThreadTree::new();
let parent = ThreadId::new();
let child = ThreadId::new();
tree.add_child(parent, child);
tree.remove(child);
assert_eq!(tree.parent_of(child), None);
assert!(tree.children_of(parent).is_empty());
}
#[test]
fn children_of_unknown_returns_empty() {
let tree = ThreadTree::new();
assert!(tree.children_of(ThreadId::new()).is_empty());
}
#[test]
fn ancestors_of_root_is_empty() {
let tree = ThreadTree::new();
assert!(tree.ancestors(ThreadId::new()).is_empty());
}
}