use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use async_trait::async_trait;
use bamboo_agent_core::tools::{
Tool, ToolClass, ToolCtx, ToolError, ToolExecutionSessionFlags, ToolOutcome, ToolResult,
};
use bamboo_domain::{
StartWorkflowRun, WorkflowBudgets, WorkflowDefinitionBundle, WorkflowProgress,
WorkflowRunDefinition, WorkflowRunSnapshot,
};
use bamboo_engine::{
AgentStepPort, AgentStepResult, FileWorkflowRunRepository, NamedAgentSpec, PermissionDecision,
WorkflowDefinitionPort, WorkflowPolicyPort, WorkflowPolicyTarget, WorkflowRunEngine,
WorkflowRunError, WorkflowSecretMaterial, WorkflowSecretResolverPort,
WorkflowSessionPermissionPort,
};
use bamboo_skills::SkillManager;
use serde::Deserialize;
use serde_json::{json, Value};
const MAX_CONCURRENCY: usize = 8;
const MAX_AGENTS: u32 = 16;
const MAX_STEPS: u32 = 512;
const MAX_RETRIES: u32 = 16;
const MAX_NESTING_DEPTH: u32 = 8;
const MAX_WALL_TIME_MS: u64 = 60 * 60 * 1000;
const MAX_TOKENS: u64 = 2_000_000;
const MAX_COST_MICROS: u64 = 100_000_000;
const MAX_PINNED_DEFINITIONS_PER_RUN: usize = 32;
const MAX_PINNED_BUNDLE_BYTES_PER_RUN: usize = 512 * 1024;
const MAX_WORKFLOW_RUN_IDS_PER_SESSION: usize = 256;
const SAFE_UNTRUSTED_WORKFLOW_TOOLS: &[&str] = &[
"Read",
"read_file",
"GetFileInfo",
"Glob",
"list_directory",
"Grep",
];
/// Server-owned access boundary for workflow runs. Session, workspace trust and
/// capabilities are derived here rather than accepted from HTTP/tool callers.
#[derive(Clone)]
pub struct WorkflowRunAccess {
engine: Arc<WorkflowRunEngine>,
skills: Arc<SkillManager>,
sessions: bamboo_engine::SessionRepository,
}
struct ServerWorkflowSessionPermissions {
sessions: bamboo_engine::SessionRepository,
permission_config: Arc<bamboo_tools::permission::PermissionConfig>,
}
#[async_trait]
impl WorkflowSessionPermissionPort for ServerWorkflowSessionPermissions {
async fn flags_for_session(
&self,
session_id: &str,
) -> Result<ToolExecutionSessionFlags, String> {
let session = self
.sessions
.try_load(session_id)
.await
.map_err(|error| error.to_string())?
.ok_or_else(|| format!("workflow session '{session_id}' does not exist"))?;
let configured = if session
.agent_runtime_state
.as_ref()
.is_some_and(|state| state.plan_mode.is_some())
{
bamboo_domain::PermissionMode::Plan
} else {
self.permission_config.mode()
};
Ok(ToolExecutionSessionFlags::from_session_and_configured_mode(
&session, configured,
))
}
}
impl WorkflowRunAccess {
pub async fn new(
data_dir: &Path,
tools: Arc<dyn bamboo_agent_core::tools::ToolExecutor>,
skills: Arc<SkillManager>,
sessions: bamboo_engine::SessionRepository,
) -> Result<Self, String> {
Self::new_with_permission_config(data_dir, tools, skills, sessions, None).await
}
pub async fn new_with_permission_config(
data_dir: &Path,
tools: Arc<dyn bamboo_agent_core::tools::ToolExecutor>,
skills: Arc<SkillManager>,
sessions: bamboo_engine::SessionRepository,
permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
) -> Result<Self, String> {
let repository = Arc::new(
FileWorkflowRunRepository::new(data_dir.join("workflow-runs"))
.map_err(|error| format!("failed to initialize workflow journal: {error}"))?,
);
let engine = WorkflowRunEngine::new(
repository,
tools,
Arc::new(UnavailableAgentPort),
Arc::new(ExternallyPinnedDefinitions),
Arc::new(ServerWorkflowPolicy),
Arc::new(UnavailableSecretResolver),
WorkflowBudgets {
max_concurrency: MAX_CONCURRENCY,
max_agents: MAX_AGENTS,
max_steps: MAX_STEPS,
max_retries: MAX_RETRIES,
max_nesting_depth: MAX_NESTING_DEPTH,
wall_time_ms: MAX_WALL_TIME_MS,
max_tokens: Some(MAX_TOKENS),
max_cost_micros: Some(MAX_COST_MICROS),
},
);
if let Some(permission_config) = permission_config {
engine.set_session_permission_port(Arc::new(ServerWorkflowSessionPermissions {
sessions: sessions.clone(),
permission_config,
}));
}
engine
.recover()
.await
.map_err(|error| format!("failed to recover workflow journal: {error}"))?;
Ok(Self {
engine,
skills,
sessions,
})
}
async fn session_context(
&self,
session_id: &str,
) -> Result<(Option<PathBuf>, bool), WorkflowRunError> {
let session =
self.sessions.try_load(session_id).await.map_err(|_| {
WorkflowRunError::Preflight("session state is unavailable".to_string())
})?;
let session = session.ok_or_else(|| {
WorkflowRunError::Preflight("workflow session does not exist".to_string())
})?;
let preferred = session.workspace.map(PathBuf::from);
let workspace =
bamboo_agent_core::workspace_state::ensure_session_workspace(session_id, preferred)
.or_else(|| {
// Server bootstrap registers the workspace-root provider, so this
// yields a persistent session-scoped directory under data_dir when
// the session has no explicit/configured workspace (#217).
Some(
bamboo_agent_core::workspace_state::workspace_or_process_cwd(Some(
session_id,
)),
)
});
// Path resolution and workspace trust are separate authorities. Until
// #601 provides an explicit server-owned trust decision, never infer
// trust merely because the resolved path is absolute. Read-only runs
// remain available through `ServerWorkflowPolicy`; every stronger
// capability stays fail-closed.
Ok((workspace, false))
}
pub async fn start(
&self,
session_id: &str,
workflow_id: &str,
revision: u64,
args: Value,
budget: Option<WorkflowBudgets>,
) -> Result<WorkflowRunSnapshot, WorkflowRunError> {
self.start_for_invoker(session_id, workflow_id, revision, args, budget, false)
.await
}
pub async fn start_from_tool(
&self,
session_id: &str,
workflow_id: &str,
revision: u64,
args: Value,
budget: Option<WorkflowBudgets>,
) -> Result<WorkflowRunSnapshot, WorkflowRunError> {
self.start_for_invoker(session_id, workflow_id, revision, args, budget, true)
.await
}
async fn start_for_invoker(
&self,
session_id: &str,
workflow_id: &str,
revision: u64,
args: Value,
budget: Option<WorkflowBudgets>,
model_started: bool,
) -> Result<WorkflowRunSnapshot, WorkflowRunError> {
self.ensure_run_index_capacity(session_id).await?;
let (workspace, workspace_trusted) = self.session_context(session_id).await?;
let store = self
.skills
.store_for_workspace(workspace.as_deref())
.await
.map_err(|_| {
WorkflowRunError::Preflight("workflow catalog is unavailable".to_string())
})?;
let catalog = store.workflow_catalog_snapshot().await;
let entry = catalog
.entries
.iter()
.find(|entry| {
entry.winner
&& entry.id == workflow_id
&& entry.revision == revision
&& entry.status == bamboo_skills::WorkflowStatus::Valid
})
.ok_or_else(|| {
WorkflowRunError::Preflight(
"requested workflow revision is unavailable".to_string(),
)
})?;
if entry.kind != bamboo_skills::WorkflowKind::Orchestration {
return Err(WorkflowRunError::Preflight(
"instruction workflows must be activated with load_skill".to_string(),
));
}
if model_started {
let session = self
.sessions
.try_load(session_id)
.await
.map_err(|_| {
WorkflowRunError::Preflight("session state is unavailable".to_string())
})?
.ok_or(WorkflowRunError::NotFound)?;
let opted_in = session
.metadata
.get(bamboo_skills::WORKFLOW_ORCHESTRATION_OPT_IN_METADATA_KEY)
.is_some_and(|value| value.eq_ignore_ascii_case("true"));
if !opted_in {
return Err(WorkflowRunError::Preflight(
"model-started orchestration requires explicit session opt-in".to_string(),
));
}
}
let mut bundle = self
.skills
.pin_workflow_definition_bundle(workspace.as_deref(), workflow_id, revision)
.await
.map_err(|_| WorkflowRunError::Preflight("workflow catalog pin failed".to_string()))?;
let policy = if model_started {
"automatic"
} else {
"explicit"
};
if bundle.root_invocation_policy[policy].as_bool() != Some(true) {
return Err(WorkflowRunError::Preflight(format!(
"pinned workflow invocation policy denies {policy} start"
)));
}
let bundle_bytes = serde_json::to_vec(&bundle)
.map_err(|_| WorkflowRunError::Preflight("workflow bundle is invalid".to_string()))?
.len();
enforce_pinned_bundle_limits(bundle.definitions.len(), bundle_bytes)?;
let mut definition = bundle.root().cloned().ok_or_else(|| {
WorkflowRunError::Preflight("pinned workflow root is missing".to_string())
})?;
if let Some(requested) = budget {
validate_requested_budget(&requested)?;
definition.budgets = tighten_workflow_budget(&definition.budgets, &requested);
let root_key = WorkflowDefinitionBundle::key(&definition.id, definition.revision);
bundle.definitions.insert(root_key, definition.clone());
}
let snapshot = self
.engine
.start_pinned(
StartWorkflowRun {
definition,
args,
session_id: session_id.to_string(),
workspace_trusted,
// #601 is not a production capability authority yet. Grant
// only the server-owned read-only class needed by the review
// dogfood workflow; the concrete base ToolExecutor still
// performs its normal per-tool/per-resource permission gate.
allowed_capabilities: vec!["read".to_string()],
},
bundle,
)
.await?;
if let Err(error) = self.remember_run_id(session_id, &snapshot.run_id).await {
return match self.engine.cancel(&snapshot.run_id).await {
Ok(cancelled) if cancelled.status.is_terminal() => Err(WorkflowRunError::Storage(
format!(
"run index persistence failed; run {} reached terminal {:?}: {error}",
snapshot.run_id, cancelled.status
),
)),
Ok(cancelled) => Err(WorkflowRunError::Storage(format!(
"run index persistence failed; orphan run {} remains {:?}; repair with this run id: {error}",
snapshot.run_id, cancelled.status
))),
Err(cancel_error) => Err(WorkflowRunError::Storage(format!(
"run index persistence failed; orphan run {} could not be cancelled ({cancel_error}); repair with this run id: {error}",
snapshot.run_id
))),
};
}
Ok(snapshot)
}
async fn ensure_run_index_capacity(&self, session_id: &str) -> Result<(), WorkflowRunError> {
let session = self
.sessions
.try_load(session_id)
.await
.map_err(|_| WorkflowRunError::Storage("session state unavailable".to_string()))?
.ok_or(WorkflowRunError::NotFound)?;
let ids = session
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
.unwrap_or_default();
if ids.len() < MAX_WORKFLOW_RUN_IDS_PER_SESSION {
return Ok(());
}
let mut evictable = BTreeSet::new();
for run_id in &ids {
match self.engine.progress(run_id, u64::MAX).await {
Ok(progress) if progress.snapshot.status.is_terminal() => {
evictable.insert(run_id.clone());
}
Err(WorkflowRunError::NotFound) => {
evictable.insert(run_id.clone());
}
Ok(_) => {}
Err(error) => return Err(error),
}
}
if !evictable.is_empty() {
self.sessions
.update_runtime_session(
session_id,
&[bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY],
move |session| {
let mut ids = session
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
.unwrap_or_default();
ids.retain(|id| !evictable.contains(id));
session.metadata.insert(
bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY.to_string(),
serde_json::to_string(&ids)
.expect("string vector serialization cannot fail"),
);
},
)
.await
.map_err(|_| WorkflowRunError::Storage("run index pruning failed".to_string()))?
.ok_or(WorkflowRunError::NotFound)?;
}
let remaining = self
.sessions
.try_load(session_id)
.await
.map_err(|_| WorkflowRunError::Storage("session state unavailable".to_string()))?
.and_then(|session| {
session
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
})
.unwrap_or_default()
.len();
if remaining >= MAX_WORKFLOW_RUN_IDS_PER_SESSION {
return Err(WorkflowRunError::Preflight(
"workflow run index is full of active runs".to_string(),
));
}
Ok(())
}
async fn remember_run_id(
&self,
session_id: &str,
run_id: &str,
) -> Result<(), WorkflowRunError> {
let run_id = run_id.to_string();
let index_full = Arc::new(AtomicBool::new(false));
let index_full_in_transaction = index_full.clone();
self.sessions
.update_runtime_session(
session_id,
&[bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY],
move |session| {
let mut ids = session
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
.unwrap_or_default();
ids.retain(|existing| existing != &run_id);
if ids.len() >= MAX_WORKFLOW_RUN_IDS_PER_SESSION {
index_full_in_transaction.store(true, Ordering::SeqCst);
return;
}
ids.push(run_id);
session.metadata.insert(
bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY.to_string(),
serde_json::to_string(&ids)
.expect("string vector serialization cannot fail"),
);
},
)
.await
.map_err(|_| WorkflowRunError::Storage("run index persistence failed".to_string()))?
.ok_or(WorkflowRunError::NotFound)
.and_then(|_| {
if index_full.load(Ordering::SeqCst) {
Err(WorkflowRunError::Storage(
"workflow run index reached its active-run capacity".to_string(),
))
} else {
Ok(())
}
})
}
pub async fn list_for_session(
&self,
session_id: &str,
) -> Result<Vec<WorkflowRunSnapshot>, WorkflowRunError> {
self.session_context(session_id).await?;
let session = self
.sessions
.try_load(session_id)
.await
.map_err(|_| WorkflowRunError::Storage("session state unavailable".to_string()))?
.ok_or(WorkflowRunError::NotFound)?;
let run_ids = session
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
.unwrap_or_default();
let mut snapshots = Vec::new();
let mut stale = BTreeSet::new();
for run_id in run_ids {
match self.engine.progress(&run_id, u64::MAX).await {
Ok(progress) if progress.snapshot.session_id == session_id => {
snapshots.push(progress.snapshot);
}
Ok(_) | Err(WorkflowRunError::NotFound) => {
stale.insert(run_id);
}
Err(error) => return Err(error),
}
}
if !stale.is_empty() {
let _ = self
.sessions
.update_runtime_session(
session_id,
&[bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY],
move |session| {
let mut ids = session
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
.unwrap_or_default();
ids.retain(|id| !stale.contains(id));
session.metadata.insert(
bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY.to_string(),
serde_json::to_string(&ids)
.expect("string vector serialization cannot fail"),
);
},
)
.await;
}
snapshots.sort_by_key(|snapshot| std::cmp::Reverse(snapshot.created_at));
Ok(snapshots)
}
pub async fn progress_for_session(
&self,
session_id: &str,
run_id: &str,
since: u64,
) -> Result<WorkflowProgress, WorkflowRunError> {
let progress = self.engine.progress(run_id, since).await?;
if progress.snapshot.session_id != session_id {
return Err(WorkflowRunError::NotFound);
}
Ok(progress)
}
pub async fn cancel_for_session(
&self,
session_id: &str,
run_id: &str,
) -> Result<WorkflowRunSnapshot, WorkflowRunError> {
self.progress_for_session(session_id, run_id, u64::MAX)
.await?;
self.engine.cancel(run_id).await
}
pub async fn restart_for_session(
&self,
session_id: &str,
run_id: &str,
) -> Result<WorkflowRunSnapshot, WorkflowRunError> {
self.progress_for_session(session_id, run_id, u64::MAX)
.await?;
let (_, workspace_trusted) = self.session_context(session_id).await?;
self.engine
.restart(run_id, workspace_trusted, vec!["read".to_string()])
.await
}
pub async fn restart_from_tool(
&self,
session_id: &str,
run_id: &str,
) -> Result<WorkflowRunSnapshot, WorkflowRunError> {
let progress = self
.progress_for_session(session_id, run_id, u64::MAX)
.await?;
if progress.snapshot.definition_bundle.root_invocation_policy["automatic"].as_bool()
!= Some(true)
{
return Err(WorkflowRunError::Preflight(
"pinned workflow invocation policy denies automatic restart".to_string(),
));
}
let session = self
.sessions
.try_load(session_id)
.await
.map_err(|_| WorkflowRunError::Preflight("session state is unavailable".to_string()))?
.ok_or(WorkflowRunError::NotFound)?;
let opted_in = session
.metadata
.get(bamboo_skills::WORKFLOW_ORCHESTRATION_OPT_IN_METADATA_KEY)
.is_some_and(|value| value.eq_ignore_ascii_case("true"));
if !opted_in {
return Err(WorkflowRunError::Preflight(
"model-started orchestration restart requires explicit session opt-in".to_string(),
));
}
self.restart_for_session(session_id, run_id).await
}
}
fn tighten_workflow_budget(
definition: &WorkflowBudgets,
requested: &WorkflowBudgets,
) -> WorkflowBudgets {
WorkflowBudgets {
max_concurrency: definition.max_concurrency.min(requested.max_concurrency),
max_agents: definition.max_agents.min(requested.max_agents),
max_steps: definition.max_steps.min(requested.max_steps),
max_retries: definition.max_retries.min(requested.max_retries),
max_nesting_depth: definition
.max_nesting_depth
.min(requested.max_nesting_depth),
wall_time_ms: definition.wall_time_ms.min(requested.wall_time_ms),
max_tokens: match (definition.max_tokens, requested.max_tokens) {
(Some(left), Some(right)) => Some(left.min(right)),
(Some(value), None) | (None, Some(value)) => Some(value),
(None, None) => None,
},
max_cost_micros: match (definition.max_cost_micros, requested.max_cost_micros) {
(Some(left), Some(right)) => Some(left.min(right)),
(Some(value), None) | (None, Some(value)) => Some(value),
(None, None) => None,
},
}
}
fn validate_requested_budget(requested: &WorkflowBudgets) -> Result<(), WorkflowRunError> {
if requested.max_concurrency == 0
|| requested.max_steps == 0
|| requested.max_nesting_depth == 0
|| requested.wall_time_ms == 0
{
return Err(WorkflowRunError::InvalidInput(
"workflow execution limits must be positive".to_string(),
));
}
Ok(())
}
fn enforce_pinned_bundle_limits(
definition_count: usize,
serialized_bytes: usize,
) -> Result<(), WorkflowRunError> {
if definition_count > MAX_PINNED_DEFINITIONS_PER_RUN {
return Err(WorkflowRunError::Preflight(
"workflow dependency count exceeds the server limit".to_string(),
));
}
if serialized_bytes > MAX_PINNED_BUNDLE_BYTES_PER_RUN {
return Err(WorkflowRunError::Preflight(
"workflow definition bundle exceeds the server size limit".to_string(),
));
}
Ok(())
}
/// The durable snapshot keeps the complete pinned bundle for deterministic
/// restart, but clients need only the root definition, bundle identity/hash,
/// step tree and events. Avoid echoing every nested definition over HTTP/tools.
pub(crate) fn public_workflow_snapshot(mut snapshot: WorkflowRunSnapshot) -> WorkflowRunSnapshot {
snapshot.definition_bundle.definitions.clear();
snapshot
}
/// The catalog adapter above always calls `start_pinned`. Generic engine starts
/// remain unavailable so no future server call site can accidentally re-read a
/// live definition or mix publications.
struct ExternallyPinnedDefinitions;
#[async_trait]
impl WorkflowDefinitionPort for ExternallyPinnedDefinitions {
async fn pin_bundle(
&self,
_root: &WorkflowRunDefinition,
) -> Result<WorkflowDefinitionBundle, String> {
Err("server workflows must be pinned through SkillManager".to_string())
}
}
/// #563 named-agent registry integration is not complete. Unknown and named
/// agents therefore fail preflight rather than falling back to a dynamic agent.
struct UnavailableAgentPort;
#[async_trait]
impl AgentStepPort for UnavailableAgentPort {
async fn resolve(&self, _name: &str) -> Result<Option<NamedAgentSpec>, String> {
Ok(None)
}
async fn execute(
&self,
_spec: &NamedAgentSpec,
_prompt: Value,
_model: Option<&str>,
_effort: Option<&str>,
_capabilities: &BTreeSet<String>,
_session_id: &str,
) -> Result<AgentStepResult, String> {
Err("named-agent execution is not available".to_string())
}
}
struct ServerWorkflowPolicy;
#[async_trait]
impl WorkflowPolicyPort for ServerWorkflowPolicy {
async fn authorize(
&self,
_session_id: &str,
target: &WorkflowPolicyTarget,
requested: &BTreeSet<String>,
_workspace_trusted: bool,
) -> PermissionDecision {
// Definition-declared capabilities are descriptive input, not an
// authority. Until #601 provides a server-owned capability registry,
// bind authorization to this strict server-owned target allowlist too.
// Unknown/MCP/network/script/write targets therefore fail closed even
// when hostile YAML claims `capabilities: []` or `[read]`.
match target {
// The reference itself belongs to the same immutable, validated
// bundle. Every concrete step in the nested definition is still
// authorized independently during this preflight walk.
WorkflowPolicyTarget::Workflow { .. } if requested.is_empty() => {
PermissionDecision::Allow
}
WorkflowPolicyTarget::Tool(name) => {
let safe_target = SAFE_UNTRUSTED_WORKFLOW_TOOLS
.iter()
.any(|candidate| candidate.eq_ignore_ascii_case(name));
if safe_target && requested.iter().all(|capability| capability == "read") {
PermissionDecision::Allow
} else {
PermissionDecision::Deny(
"workflow capability authority is not available".to_string(),
)
}
}
WorkflowPolicyTarget::Agent(_) | WorkflowPolicyTarget::Workflow { .. } => {
PermissionDecision::Deny(
"workflow capability authority is not available".to_string(),
)
}
}
}
}
struct UnavailableSecretResolver;
#[async_trait]
impl WorkflowSecretResolverPort for UnavailableSecretResolver {
async fn resolve(
&self,
_session_id: &str,
_capability: &str,
) -> Result<WorkflowSecretMaterial, String> {
Err("workflow secret capability resolver is not available".to_string())
}
}
#[derive(Debug, Deserialize)]
#[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)]
enum WorkflowToolInput {
Start {
workflow_id: String,
revision: u64,
#[serde(default = "empty_object")]
args: Value,
#[serde(default)]
budget: Option<WorkflowBudgets>,
},
List {},
Get {
run_id: String,
},
Events {
run_id: String,
#[serde(default)]
since: u64,
},
Cancel {
run_id: String,
},
Restart {
run_id: String,
},
}
fn empty_object() -> Value {
json!({})
}
pub struct WorkflowRunTool {
access: WorkflowRunAccess,
}
impl WorkflowRunTool {
pub fn new(access: WorkflowRunAccess) -> Self {
Self { access }
}
}
#[async_trait]
impl Tool for WorkflowRunTool {
fn name(&self) -> &str {
"workflow_run"
}
fn description(&self) -> &str {
"Start, inspect, cancel, or safely restart a catalog-pinned workflow run"
}
fn parameters_schema(&self) -> Value {
json!({
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["start", "list", "get", "events", "cancel", "restart"]
},
"workflow_id": {"type": "string", "minLength": 1},
"revision": {"type": "integer", "minimum": 1},
"args": {"type": "object", "default": {}},
"budget": {
"type": "object",
"properties": {
"max_concurrency": {"type": "integer", "minimum": 1},
"max_agents": {"type": "integer", "minimum": 0},
"max_steps": {"type": "integer", "minimum": 1},
"max_retries": {"type": "integer", "minimum": 0},
"max_nesting_depth": {"type": "integer", "minimum": 1},
"wall_time_ms": {"type": "integer", "minimum": 1},
"max_tokens": {"type": "integer", "minimum": 0},
"max_cost_micros": {"type": "integer", "minimum": 0}
},
"required": [
"max_concurrency",
"max_agents",
"max_steps",
"max_retries",
"max_nesting_depth",
"wall_time_ms"
],
"additionalProperties": false
},
"run_id": {"type": "string"},
"since": {"type": "integer", "minimum": 0}
},
"required": ["action"],
"additionalProperties": false
})
}
fn classify(&self, args: &Value) -> ToolClass {
match args.get("action").and_then(Value::as_str) {
Some("get" | "list" | "events") => ToolClass::READONLY_PARALLEL,
_ => ToolClass::MUTATING_SERIAL,
}
}
async fn invoke(&self, args: Value, ctx: ToolCtx) -> Result<ToolOutcome, ToolError> {
let input: WorkflowToolInput = serde_json::from_value(args)
.map_err(|error| ToolError::InvalidArguments(error.to_string()))?;
let session_id = ctx.session_id().ok_or_else(|| {
ToolError::InvalidArguments("workflow_run requires a session".to_string())
})?;
let result = match input {
WorkflowToolInput::Start {
workflow_id,
revision,
args,
budget,
} => serde_json::to_value(public_workflow_snapshot(
self.access
.start_from_tool(session_id, &workflow_id, revision, args, budget)
.await
.map_err(workflow_tool_error)?,
)),
WorkflowToolInput::List {} => serde_json::to_value(
self.access
.list_for_session(session_id)
.await
.map_err(workflow_tool_error)?
.into_iter()
.map(public_workflow_snapshot)
.collect::<Vec<_>>(),
),
WorkflowToolInput::Get { run_id } => {
let progress = self
.access
.progress_for_session(session_id, &run_id, u64::MAX)
.await
.map_err(workflow_tool_error)?;
serde_json::to_value(public_workflow_snapshot(progress.snapshot))
}
WorkflowToolInput::Events { run_id, since } => {
let progress = self
.access
.progress_for_session(session_id, &run_id, since)
.await
.map_err(workflow_tool_error)?;
serde_json::to_value(progress.events)
}
WorkflowToolInput::Cancel { run_id } => serde_json::to_value(public_workflow_snapshot(
self.access
.cancel_for_session(session_id, &run_id)
.await
.map_err(workflow_tool_error)?,
)),
WorkflowToolInput::Restart { run_id } => {
serde_json::to_value(public_workflow_snapshot(
self.access
.restart_from_tool(session_id, &run_id)
.await
.map_err(workflow_tool_error)?,
))
}
}
.map_err(|error| ToolError::Execution(error.to_string()))?;
Ok(ToolOutcome::Completed(ToolResult::text(
true,
serde_json::to_string(&result)
.map_err(|error| ToolError::Execution(error.to_string()))?,
)))
}
}
fn workflow_tool_error(error: WorkflowRunError) -> ToolError {
match error {
WorkflowRunError::InvalidInput(message) => ToolError::InvalidArguments(message),
WorkflowRunError::Compile(error) => ToolError::InvalidArguments(error.to_string()),
other => ToolError::Execution(other.to_string()),
}
}
#[cfg(test)]
mod tests {
use super::*;
use bamboo_agent_core::storage::Storage;
use bamboo_agent_core::tools::{
FunctionSchema, ToolCall, ToolExecutor, ToolResult, ToolSchema,
};
use bamboo_agent_core::Session;
use bamboo_llm::protocol::{gemini::GeminiTool, ToProvider};
use std::collections::HashMap;
use tokio::sync::RwLock;
#[derive(Default)]
struct WorkflowTestStorage {
sessions: RwLock<HashMap<String, Session>>,
}
#[async_trait]
impl Storage for WorkflowTestStorage {
async fn save_session(&self, session: &Session) -> std::io::Result<()> {
self.sessions
.write()
.await
.insert(session.id.clone(), session.clone());
Ok(())
}
async fn load_session(&self, session_id: &str) -> std::io::Result<Option<Session>> {
Ok(self.sessions.read().await.get(session_id).cloned())
}
async fn delete_session(&self, session_id: &str) -> std::io::Result<bool> {
Ok(self.sessions.write().await.remove(session_id).is_some())
}
}
struct WorkflowReadTool;
#[async_trait]
impl ToolExecutor for WorkflowReadTool {
async fn execute(
&self,
_call: &ToolCall,
) -> bamboo_agent_core::tools::executor::Result<ToolResult> {
Ok(ToolResult::text(true, r#"{"ok":true}"#))
}
fn list_tools(&self) -> Vec<ToolSchema> {
vec![ToolSchema {
schema_type: "function".to_string(),
function: FunctionSchema {
name: "Read".to_string(),
description: "read".to_string(),
parameters: serde_json::json!({"type":"object"}),
},
}]
}
}
async fn workflow_test_access() -> (
WorkflowRunAccess,
bamboo_engine::SessionRepository,
tempfile::TempDir,
) {
let directory = tempfile::tempdir().expect("tempdir");
let skills_dir = directory.path().join("skills");
let root = skills_dir.join("review-flow");
std::fs::create_dir_all(&root).expect("workflow dir");
std::fs::write(
root.join("SKILL.md"),
"---\nname: review-flow\ndescription: Review flow\n---\nRun review flow.\n",
)
.expect("skill");
std::fs::write(
root.join("workflow.yaml"),
"workflow_schema: 1\nid: review-flow\nrevision: 42\ninvocation_policy: {explicit: true, automatic: true}\ninput_schema:\n type: object\n additionalProperties: true\nsteps:\n - id: inspect\n type: tool\n tool: Read\n args: {}\n capabilities: [read]\n output_schema:\n type: object\n additionalProperties: true\nplan:\n type: step\n step: inspect\nbudgets:\n max_concurrency: 2\n max_agents: 1\n max_steps: 4\n max_retries: 2\n max_nesting_depth: 2\n wall_time_ms: 10000\n max_tokens: 1000\n max_cost_micros: 1000\n",
)
.expect("workflow");
let skills = Arc::new(SkillManager::with_config(bamboo_skills::SkillStoreConfig {
skills_dir,
..Default::default()
}));
skills.initialize().await.expect("skills initialize");
let storage: Arc<dyn Storage> = Arc::new(WorkflowTestStorage::default());
let persistence = Arc::new(bamboo_storage::LockedSessionStore::new(storage.clone()));
let cache = Arc::new(dashmap::DashMap::new());
let repo = bamboo_engine::SessionRepository::new(cache, storage, persistence);
let access = WorkflowRunAccess::new(
directory.path(),
Arc::new(WorkflowReadTool),
skills,
repo.clone(),
)
.await
.expect("workflow access");
(access, repo, directory)
}
fn canonical_workflow_run_schema() -> Value {
json!({
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["start", "list", "get", "events", "cancel", "restart"]
},
"workflow_id": {"type": "string", "minLength": 1},
"revision": {"type": "integer", "minimum": 1},
"args": {"type": "object", "default": {}},
"budget": {
"type": "object",
"properties": {
"max_concurrency": {"type": "integer", "minimum": 1},
"max_agents": {"type": "integer", "minimum": 0},
"max_steps": {"type": "integer", "minimum": 1},
"max_retries": {"type": "integer", "minimum": 0},
"max_nesting_depth": {"type": "integer", "minimum": 1},
"wall_time_ms": {"type": "integer", "minimum": 1},
"max_tokens": {"type": "integer", "minimum": 0},
"max_cost_micros": {"type": "integer", "minimum": 0}
},
"required": [
"max_concurrency",
"max_agents",
"max_steps",
"max_retries",
"max_nesting_depth",
"wall_time_ms"
],
"additionalProperties": false
},
"run_id": {"type": "string"},
"since": {"type": "integer", "minimum": 0}
},
"required": ["action"],
"additionalProperties": false
})
}
#[tokio::test]
async fn workflow_run_schema_is_flat_complete_and_canonical() {
let (access, _, _) = workflow_test_access().await;
let schema = WorkflowRunTool::new(access).parameters_schema();
for combinator in ["oneOf", "anyOf", "allOf"] {
assert!(
schema.get(combinator).is_none(),
"workflow_run must not advertise root {combinator}"
);
}
assert_eq!(schema, canonical_workflow_run_schema());
}
#[tokio::test]
async fn workflow_run_schema_survives_openai_sanitization_with_all_properties() {
let (access, _, _) = workflow_test_access().await;
let schema = WorkflowRunTool::new(access).parameters_schema();
let sanitized =
bamboo_llm::providers::common::tool_schema::sanitize_openai_function_parameters_schema(
&schema,
);
let properties = sanitized["properties"]
.as_object()
.expect("sanitized workflow_run properties");
assert!(!properties.is_empty());
assert_eq!(properties.len(), 7);
assert_eq!(sanitized, canonical_workflow_run_schema());
}
#[tokio::test]
async fn workflow_run_schema_reaches_gemini_unchanged() {
let (access, _, _) = workflow_test_access().await;
let direct = WorkflowRunTool::new(access).to_schema();
let gemini: GeminiTool = direct.to_provider().expect("Gemini tool conversion");
let declaration = gemini
.function_declarations
.first()
.expect("workflow_run declaration");
assert_eq!(declaration.name, "workflow_run");
assert_eq!(
declaration.parameters_json_schema.as_ref(),
Some(&canonical_workflow_run_schema())
);
assert!(declaration.parameters.is_none());
}
#[tokio::test]
async fn workflow_run_enforces_opt_in_tightens_budget_lists_and_isolates_sessions() {
let (access, repo, directory) = workflow_test_access().await;
let workspace = directory.path().join("workspace");
std::fs::create_dir_all(&workspace).expect("workspace");
let mut session = Session::new("workflow-session", "model");
session.workspace = Some(workspace.to_string_lossy().into_owned());
repo.save(&mut session).await.expect("save session");
let denied = access
.start_from_tool("workflow-session", "review-flow", 42, json!({}), None)
.await
.expect_err("model start defaults off without session opt-in");
assert!(denied.to_string().contains("opt-in"));
session.metadata.insert(
bamboo_skills::WORKFLOW_ORCHESTRATION_OPT_IN_METADATA_KEY.to_string(),
"true".to_string(),
);
repo.save(&mut session).await.expect("save opt-in");
let requested = WorkflowBudgets {
max_concurrency: 1,
max_agents: 0,
max_steps: 2,
max_retries: 0,
max_nesting_depth: 1,
wall_time_ms: 5_000,
max_tokens: Some(500),
max_cost_micros: Some(500),
};
let started = access
.start_from_tool(
"workflow-session",
"review-flow",
42,
json!({}),
Some(requested.clone()),
)
.await
.expect("opted-in model start");
assert_eq!(started.definition.budgets, requested);
let listed = access
.list_for_session("workflow-session")
.await
.expect("session run list");
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].run_id, started.run_id);
let progress = access
.progress_for_session("workflow-session", &started.run_id, 0)
.await
.expect("run events");
assert!(progress
.events
.first()
.is_some_and(|event| event.kind == bamboo_domain::WorkflowRunEventKind::RunQueued));
let completed = tokio::time::timeout(std::time::Duration::from_secs(2), async {
loop {
let progress = access
.progress_for_session("workflow-session", &started.run_id, 0)
.await
.expect("terminal run progress");
if progress.snapshot.status.is_terminal() {
break progress;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await
.expect("workflow reaches terminal state");
assert_eq!(
completed.snapshot.status,
bamboo_domain::WorkflowRunStatus::Succeeded
);
assert_eq!(
completed
.events
.iter()
.map(|event| event.sequence)
.collect::<Vec<_>>(),
(1..=7).collect::<Vec<_>>()
);
assert!(matches!(
completed.events.as_slice(),
[
bamboo_domain::WorkflowRunEvent {
kind: bamboo_domain::WorkflowRunEventKind::RunQueued,
..
},
bamboo_domain::WorkflowRunEvent {
kind: bamboo_domain::WorkflowRunEventKind::RunStarted,
..
},
bamboo_domain::WorkflowRunEvent {
kind: bamboo_domain::WorkflowRunEventKind::Phase { ref name },
..
},
bamboo_domain::WorkflowRunEvent {
kind: bamboo_domain::WorkflowRunEventKind::StepQueued,
..
},
bamboo_domain::WorkflowRunEvent {
kind: bamboo_domain::WorkflowRunEventKind::StepStarted,
..
},
bamboo_domain::WorkflowRunEvent {
kind: bamboo_domain::WorkflowRunEventKind::StepCompleted { .. },
..
},
bamboo_domain::WorkflowRunEvent {
kind: bamboo_domain::WorkflowRunEventKind::RunSucceeded { .. },
..
}
] if name == "step_reserved"
));
assert_eq!(
completed.snapshot.last_sequence,
completed.events.last().expect("terminal event").sequence
);
let invalid_budget = WorkflowBudgets {
max_steps: 0,
..requested.clone()
};
assert!(matches!(
access
.start_from_tool(
"workflow-session",
"review-flow",
42,
json!({}),
Some(invalid_budget),
)
.await,
Err(WorkflowRunError::InvalidInput(_))
));
// Restart authority is the original immutable publication, not the
// current catalog policy.
let workflow_path = directory.path().join("skills/review-flow/workflow.yaml");
let original_workflow = std::fs::read_to_string(&workflow_path).expect("workflow yaml");
std::fs::write(
&workflow_path,
original_workflow.replace(
"invocation_policy: {explicit: true, automatic: true}",
"invocation_policy: {explicit: true, automatic: false}",
),
)
.expect("disable automatic live policy");
access.skills.store().reload().await.expect("reload policy");
let restart = access
.restart_from_tool("workflow-session", &started.run_id)
.await
.expect_err("succeeded runs are terminal");
assert!(matches!(restart, WorkflowRunError::Terminal));
let mut other = Session::new("other-session", "model");
other.workspace = Some(workspace.to_string_lossy().into_owned());
repo.save(&mut other).await.expect("save other session");
assert!(access
.list_for_session("other-session")
.await
.expect("isolated list")
.is_empty());
assert!(matches!(
access
.progress_for_session("other-session", &started.run_id, 0)
.await,
Err(WorkflowRunError::NotFound)
));
}
#[test]
fn tool_input_rejects_security_context_spoofing() {
let error = serde_json::from_value::<WorkflowToolInput>(json!({
"action": "start",
"workflow_id": "safe",
"revision": 1,
"workspace_trusted": true
}))
.unwrap_err();
assert!(error.to_string().contains("unknown field"));
}
#[test]
fn tool_input_rejects_fields_from_other_actions() {
let invalid = [
json!({"action": "list", "run_id": "run-1"}),
json!({"action": "get", "run_id": "run-1", "since": 1}),
json!({"action": "events", "run_id": "run-1", "workflow_id": "flow"}),
json!({"action": "cancel", "run_id": "run-1", "revision": 1}),
json!({"action": "restart", "run_id": "run-1", "budget": {}}),
json!({
"action": "start",
"workflow_id": "flow",
"revision": 1,
"run_id": "run-1"
}),
];
for input in invalid {
let error = serde_json::from_value::<WorkflowToolInput>(input.clone())
.expect_err("action-specific fields must remain authoritative at runtime");
assert!(
error.to_string().contains("unknown field"),
"unexpected error for {input}: {error}"
);
}
assert!(matches!(
serde_json::from_value::<WorkflowToolInput>(json!({"action": "list"}))
.expect("fieldless list action"),
WorkflowToolInput::List {}
));
}
#[tokio::test]
async fn omitted_start_args_and_zero_budgets_match_schema() {
let WorkflowToolInput::Start { args, .. } =
serde_json::from_value::<WorkflowToolInput>(json!({
"action": "start",
"workflow_id": "safe",
"revision": 1
}))
.expect("tool args default")
else {
panic!("start input")
};
assert_eq!(args, json!({}));
let http: crate::handlers::workflow_runs::StartWorkflowRunRequest =
serde_json::from_value(json!({"workflow_id":"safe", "revision":1}))
.expect("http args default");
assert_eq!(http.args, json!({}));
let (access, _, _) = workflow_test_access().await;
let schema = WorkflowRunTool { access }.parameters_schema();
let properties = &schema["properties"];
assert_eq!(properties["args"]["default"], json!({}));
assert_eq!(
properties["budget"]["properties"]["max_agents"]["minimum"],
0
);
assert_eq!(
properties["budget"]["properties"]["max_retries"]["minimum"],
0
);
assert_eq!(
properties["budget"]["properties"]["max_tokens"]["minimum"],
0
);
assert_eq!(
properties["budget"]["properties"]["max_cost_micros"]["minimum"],
0
);
}
#[tokio::test]
async fn run_index_updates_are_concurrent_and_never_evict_at_active_capacity() {
let (access, repo, _) = workflow_test_access().await;
let mut session = Session::new("run-index", "model");
repo.save(&mut session).await.expect("seed session");
let results = futures::future::join_all((0..32).map(|index| {
let access = access.clone();
async move {
let run_id = format!("run-{index}");
access.remember_run_id("run-index", &run_id).await
}
}))
.await;
assert!(results.into_iter().all(|result| result.is_ok()));
let concurrent = repo
.try_load("run-index")
.await
.expect("load")
.expect("session");
let ids = serde_json::from_str::<Vec<String>>(
concurrent
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.expect("run ids"),
)
.expect("ids json");
assert_eq!(ids.len(), 32);
assert_eq!(ids.iter().collect::<BTreeSet<_>>().len(), 32);
let capacity_ids = (0..MAX_WORKFLOW_RUN_IDS_PER_SESSION)
.map(|index| format!("active-{index}"))
.collect::<Vec<_>>();
repo.update_runtime_session(
"run-index",
&[bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY],
{
let capacity_ids = capacity_ids.clone();
move |session| {
session.metadata.insert(
bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY.to_string(),
serde_json::to_string(&capacity_ids).expect("ids json"),
);
}
},
)
.await
.expect("fill index")
.expect("session");
assert!(matches!(
access.remember_run_id("run-index", "new-run").await,
Err(WorkflowRunError::Storage(_))
));
let retained = repo
.try_load("run-index")
.await
.expect("load")
.expect("session");
let retained = serde_json::from_str::<Vec<String>>(
retained
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.expect("run ids"),
)
.expect("ids json");
assert_eq!(
retained, capacity_ids,
"oldest active id must not be evicted"
);
}
#[tokio::test]
async fn real_model_workflow_run_index_survives_tool_result_and_final_session_save() {
use bamboo_agent_core::storage::AttachmentReader;
use bamboo_engine::{Agent, ExecuteRequestBuilder};
use bamboo_llm::{LLMChunk, LLMProvider, LLMStream};
use futures::stream;
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;
struct NoAttachments;
#[async_trait]
impl AttachmentReader for NoAttachments {
async fn read_attachment(
&self,
_session_id: &str,
_attachment_id: &str,
) -> std::io::Result<Option<(Vec<u8>, String)>> {
Ok(None)
}
}
struct QueueProvider {
queue: Mutex<Vec<Vec<bamboo_llm::provider::Result<LLMChunk>>>>,
}
#[async_trait]
impl LLMProvider for QueueProvider {
async fn chat_stream(
&self,
_messages: &[bamboo_agent_core::Message],
_tools: &[ToolSchema],
_max_output_tokens: Option<u32>,
_model: &str,
) -> bamboo_llm::provider::Result<LLMStream> {
Ok(Box::pin(stream::iter(self.queue.lock().await.remove(0))))
}
}
let (access, repo, directory) = workflow_test_access().await;
let session_id = "real-model-workflow-run";
let mut session = Session::new(session_id, "test-model");
session.metadata.insert(
bamboo_skills::WORKFLOW_ORCHESTRATION_OPT_IN_METADATA_KEY.to_string(),
"true".to_string(),
);
session
.metadata
.insert("external.metadata".to_string(), "preserve".to_string());
session.add_message(bamboo_agent_core::Message::system("system"));
session.add_message(bamboo_agent_core::Message::user("run review workflow"));
repo.save(&mut session).await.expect("seed session");
let call = ToolCall {
id: "call-workflow-run".to_string(),
tool_type: "function".to_string(),
function: bamboo_agent_core::tools::FunctionCall {
name: "workflow_run".to_string(),
arguments: json!({
"action":"start",
"workflow_id":"review-flow",
"revision":42
})
.to_string(),
},
};
let provider = Arc::new(QueueProvider {
queue: Mutex::new(vec![
vec![Ok(LLMChunk::ToolCalls(vec![call])), Ok(LLMChunk::Done)],
vec![Ok(LLMChunk::Token("done".to_string())), Ok(LLMChunk::Done)],
]),
});
let tools = Arc::new(
bamboo_tools::BuiltinToolExecutorBuilder::new()
.with_tool(WorkflowRunTool::new(access.clone()))
.expect("workflow tool")
.build(),
);
let metrics = bamboo_metrics::MetricsCollector::spawn(
Arc::new(bamboo_metrics::SqliteMetricsStorage::new(
directory.path().join("runner-metrics.db"),
)),
7,
);
let agent = Agent::builder()
.storage(repo.storage().clone())
.persistence(Arc::new(repo.clone()))
.attachment_reader(Arc::new(NoAttachments))
.skill_manager(access.skills.clone())
.metrics_collector(metrics)
.config(Arc::new(RwLock::new(bamboo_llm::Config::default())))
.provider(provider)
.default_tools(tools)
.build()
.expect("agent");
let (event_tx, _event_rx) = tokio::sync::mpsc::channel(128);
agent
.execute(
&mut session,
ExecuteRequestBuilder::new(
"run review workflow",
event_tx,
CancellationToken::new(),
)
.model("test-model")
.build(),
)
.await
.expect("real model workflow run");
let saved = repo
.storage()
.load_session(session_id)
.await
.expect("load")
.expect("saved");
assert_eq!(
saved.metadata.get("external.metadata").map(String::as_str),
Some("preserve")
);
assert!(saved
.metadata
.contains_key(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY));
let listed = access
.list_for_session(session_id)
.await
.expect("list after final save");
assert_eq!(listed.len(), 1);
assert!(saved.messages.iter().any(|message| {
message.tool_calls.as_ref().is_some_and(|calls| {
calls
.iter()
.any(|call| call.function.name == "workflow_run")
})
}));
}
#[tokio::test]
async fn http_start_survives_concurrent_stale_runner_final_save_and_server_restart() {
use bamboo_agent_core::storage::AttachmentReader;
use bamboo_engine::{Agent, ExecuteRequestBuilder};
use bamboo_llm::{LLMChunk, LLMProvider, LLMStream};
use futures::stream;
use tokio::sync::{oneshot, Mutex};
use tokio_util::sync::CancellationToken;
struct NoAttachments;
#[async_trait]
impl AttachmentReader for NoAttachments {
async fn read_attachment(
&self,
_session_id: &str,
_attachment_id: &str,
) -> std::io::Result<Option<(Vec<u8>, String)>> {
Ok(None)
}
}
struct PausingProvider {
entered: Mutex<Option<oneshot::Sender<()>>>,
resume: Mutex<Option<oneshot::Receiver<()>>>,
}
#[async_trait]
impl LLMProvider for PausingProvider {
async fn chat_stream(
&self,
_messages: &[bamboo_agent_core::Message],
_tools: &[ToolSchema],
_max_output_tokens: Option<u32>,
_model: &str,
) -> bamboo_llm::provider::Result<LLMStream> {
if let Some(entered) = self.entered.lock().await.take() {
let _ = entered.send(());
}
if let Some(resume) = self.resume.lock().await.take() {
let _ = resume.await;
}
Ok(Box::pin(stream::iter(vec![
Ok(LLMChunk::Token("done".to_string())),
Ok(LLMChunk::Done),
])))
}
}
let (access, repo, directory) = workflow_test_access().await;
let session_id = "http-start-concurrent-runner-save";
let mut session = Session::new(session_id, "test-model");
session.add_message(bamboo_agent_core::Message::system("system"));
session.add_message(bamboo_agent_core::Message::user("keep running"));
repo.save(&mut session).await.expect("seed session");
let mut runner_session = repo
.try_load(session_id)
.await
.expect("load runner session")
.expect("runner session");
let (entered_tx, entered_rx) = oneshot::channel();
let (resume_tx, resume_rx) = oneshot::channel();
let provider = Arc::new(PausingProvider {
entered: Mutex::new(Some(entered_tx)),
resume: Mutex::new(Some(resume_rx)),
});
let metrics = bamboo_metrics::MetricsCollector::spawn(
Arc::new(bamboo_metrics::SqliteMetricsStorage::new(
directory.path().join("http-runner-metrics.db"),
)),
7,
);
let agent = Agent::builder()
.storage(repo.storage().clone())
.persistence(Arc::new(repo.clone()))
.attachment_reader(Arc::new(NoAttachments))
.skill_manager(access.skills.clone())
.metrics_collector(metrics)
.config(Arc::new(RwLock::new(bamboo_llm::Config::default())))
.provider(provider)
.default_tools(Arc::new(
bamboo_tools::BuiltinToolExecutorBuilder::new().build(),
))
.build()
.expect("agent");
let (event_tx, _event_rx) = tokio::sync::mpsc::channel(128);
let runner = tokio::spawn(async move {
agent
.execute(
&mut runner_session,
ExecuteRequestBuilder::new("keep running", event_tx, CancellationToken::new())
.model("test-model")
.build(),
)
.await
});
tokio::time::timeout(std::time::Duration::from_secs(2), entered_rx)
.await
.expect("runner enters model round")
.expect("runner entry signal");
let started = access
.start(session_id, "review-flow", 42, json!({}), None)
.await
.expect("HTTP-equivalent explicit start");
let durable_during_round = repo
.storage()
.load_session(session_id)
.await
.expect("load during round")
.expect("session during round");
assert!(durable_during_round
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.is_some_and(|raw| raw.contains(&started.run_id)));
resume_tx.send(()).expect("resume runner");
tokio::time::timeout(std::time::Duration::from_secs(2), runner)
.await
.expect("runner completes")
.expect("runner task")
.expect("runner execution");
let durable_after_final_save = repo
.storage()
.load_session(session_id)
.await
.expect("load after final save")
.expect("saved session");
assert!(durable_after_final_save
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.is_some_and(|raw| raw.contains(&started.run_id)));
tokio::time::timeout(std::time::Duration::from_secs(2), async {
loop {
let progress = access
.progress_for_session(session_id, &started.run_id, u64::MAX)
.await
.expect("workflow progress before restart");
if progress.snapshot.status.is_terminal()
&& !access.engine.is_run_active(&started.run_id)
{
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("workflow reaches terminal state before restart");
let skills = access.skills.clone();
drop(access);
let restarted =
WorkflowRunAccess::new(directory.path(), Arc::new(WorkflowReadTool), skills, repo)
.await
.expect("restart workflow access");
let listed = restarted
.list_for_session(session_id)
.await
.expect("list after restart");
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].run_id, started.run_id);
}
#[tokio::test]
async fn production_policy_allows_read_without_fabricating_workspace_trust() {
let read = BTreeSet::from(["read".to_string()]);
assert_eq!(
ServerWorkflowPolicy
.authorize(
"session",
&WorkflowPolicyTarget::Tool("read_file".to_string()),
&read,
false,
)
.await,
PermissionDecision::Allow
);
let write = BTreeSet::from(["write".to_string()]);
assert!(matches!(
ServerWorkflowPolicy
.authorize(
"session",
&WorkflowPolicyTarget::Tool("write_file".to_string()),
&write,
false,
)
.await,
PermissionDecision::Deny(_)
));
for hostile_target in [
"Write",
"write_file",
"WebFetch",
"mcp::remote_tool",
"Bash",
] {
for claimed in [BTreeSet::new(), read.clone()] {
assert!(matches!(
ServerWorkflowPolicy
.authorize(
"session",
&WorkflowPolicyTarget::Tool(hostile_target.to_string()),
&claimed,
false,
)
.await,
PermissionDecision::Deny(_)
));
}
}
assert_eq!(
ServerWorkflowPolicy
.authorize(
"session",
&WorkflowPolicyTarget::Workflow {
id: "nested-review".to_string(),
revision: 1,
},
&BTreeSet::new(),
false,
)
.await,
PermissionDecision::Allow
);
}
#[test]
fn pinned_bundle_limits_reject_oversized_runs_before_engine_start() {
assert!(enforce_pinned_bundle_limits(
MAX_PINNED_DEFINITIONS_PER_RUN,
MAX_PINNED_BUNDLE_BYTES_PER_RUN
)
.is_ok());
assert!(enforce_pinned_bundle_limits(
MAX_PINNED_DEFINITIONS_PER_RUN + 1,
MAX_PINNED_BUNDLE_BYTES_PER_RUN
)
.is_err());
assert!(enforce_pinned_bundle_limits(
MAX_PINNED_DEFINITIONS_PER_RUN,
MAX_PINNED_BUNDLE_BYTES_PER_RUN + 1
)
.is_err());
}
}