use std::collections::{BTreeMap, 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, WorkflowBudgetUsage, WorkflowBudgets, WorkflowDefinitionBundle,
WorkflowFailure, WorkflowFailureCode, WorkflowPlan, WorkflowProgress, WorkflowRunDefinition,
WorkflowRunEvent, WorkflowRunEventKind, WorkflowRunSnapshot, WorkflowRunStatus,
WorkflowStepKind, WorkflowStepSnapshot, WorkflowStepStatus, WorkflowSuspensionContext,
};
use bamboo_engine::{
AgentStepPort, AgentStepResult, FileWorkflowRunRepository, NamedAgentSpec, PermissionDecision,
WorkflowDefinitionPort, WorkflowPolicyPort, WorkflowPolicyTarget, WorkflowRunEngine,
WorkflowRunError, WorkflowSecretMaterial, WorkflowSecretResolverPort,
WorkflowSessionPermissionPort,
};
use bamboo_skills::SkillManager;
use serde::{Deserialize, Serialize};
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?;
self.index_started_run_or_compensate(session_id, snapshot)
.await
}
async fn index_started_run_or_compensate(
&self,
session_id: &str,
snapshot: WorkflowRunSnapshot,
) -> Result<WorkflowRunSnapshot, WorkflowRunError> {
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> {
let progress = self
.progress_for_session(session_id, run_id, u64::MAX)
.await?;
ensure_workflow_cancel_allowed(&progress.snapshot)?;
self.engine.cancel(run_id).await
}
pub async fn restart_for_session(
&self,
session_id: &str,
run_id: &str,
) -> Result<WorkflowRunSnapshot, WorkflowRunError> {
let progress = self
.progress_for_session(session_id, run_id, u64::MAX)
.await?;
ensure_workflow_restart_as_new_run_allowed(&progress.snapshot)?;
self.ensure_run_index_capacity(session_id).await?;
let (_, workspace_trusted) = self.session_context(session_id).await?;
let snapshot = self
.engine
.restart(run_id, workspace_trusted, vec!["read".to_string()])
.await?;
self.index_started_run_or_compensate(session_id, snapshot)
.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 ensure_workflow_cancel_allowed(snapshot: &WorkflowRunSnapshot) -> Result<(), WorkflowRunError> {
match snapshot.status {
WorkflowRunStatus::Succeeded | WorkflowRunStatus::Failed => Err(WorkflowRunError::Terminal),
WorkflowRunStatus::Queued
| WorkflowRunStatus::Running
| WorkflowRunStatus::Suspended
| WorkflowRunStatus::Cancelled => Ok(()),
}
}
fn ensure_workflow_restart_as_new_run_allowed(
snapshot: &WorkflowRunSnapshot,
) -> Result<(), WorkflowRunError> {
match (snapshot.status, snapshot.suspension.as_ref()) {
(WorkflowRunStatus::Suspended, Some(WorkflowSuspensionContext::Recovery { .. })) => Ok(()),
(status, _) if status.is_terminal() => Err(WorkflowRunError::Terminal),
_ => Err(WorkflowRunError::Preflight(
"only recovery-suspended workflows can restart as a new run".to_string(),
)),
}
}
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(())
}
/// Stable metadata-only projection of a durable workflow run.
///
/// The internal snapshot deliberately persists the complete definition bundle,
/// validated arguments, input hashes, tool outputs, and diagnostic strings for
/// deterministic recovery. None of those values are public API data. HTTP and
/// model-facing tools must serialize this projection instead of the durable
/// snapshot itself.
#[derive(Debug, Clone, Serialize, PartialEq)]
pub(crate) struct PublicWorkflowRunSnapshot {
pub run_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub parent_run_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub parent_step_id: Option<String>,
pub session_id: String,
pub workflow_id: String,
pub workflow_revision: u64,
pub definition_bundle_hash: String,
pub status: WorkflowRunStatus,
pub can_cancel: bool,
pub can_restart_as_new_run: bool,
pub planned_steps: BTreeMap<String, PublicWorkflowPlannedStep>,
pub plan: PublicWorkflowPlan,
pub steps: BTreeMap<String, PublicWorkflowStepSnapshot>,
pub budget: WorkflowBudgets,
pub usage: WorkflowBudgetUsage,
pub child_agent_count: u32,
pub last_sequence: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub failure: Option<PublicWorkflowFailure>,
#[serde(skip_serializing_if = "Option::is_none")]
pub suspension: Option<PublicWorkflowSuspension>,
pub created_at: chrono::DateTime<chrono::Utc>,
pub updated_at: chrono::DateTime<chrono::Utc>,
}
#[derive(Debug, Clone, Serialize, PartialEq)]
pub(crate) struct PublicWorkflowStepSnapshot {
pub id: String,
pub status: WorkflowStepStatus,
#[serde(skip_serializing_if = "Option::is_none")]
pub failure: Option<PublicWorkflowFailure>,
pub attempts: u32,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub(crate) struct PublicWorkflowPlannedStep {
pub id: String,
pub kind: PublicWorkflowPlannedStepKind,
}
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(crate) enum PublicWorkflowPlannedStepKind {
Tool,
Agent,
Workflow,
}
/// Safe plan topology. Data bindings, map item names, delays, tool arguments,
/// prompts, schemas, and capabilities remain internal.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[serde(tag = "type", rename_all = "snake_case")]
pub(crate) enum PublicWorkflowPlan {
Step {
step: String,
},
Sequence {
nodes: Vec<PublicWorkflowPlan>,
},
Parallel {
nodes: Vec<PublicWorkflowPlan>,
},
Map {
body: Box<PublicWorkflowPlan>,
},
Retry {
node: Box<PublicWorkflowPlan>,
max_attempts: u32,
},
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub(crate) struct PublicWorkflowFailure {
pub code: WorkflowFailureCode,
pub message: &'static str,
pub retryable: bool,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[serde(tag = "type", rename_all = "snake_case")]
pub(crate) enum PublicWorkflowSuspension {
ToolApproval { step_id: String },
ToolRunning { step_id: String, killed: bool },
Recovery,
}
/// Durable reconnect events use the same metadata-only boundary as snapshots.
/// Raw step/run outputs, suspension reasons, and backend diagnostics are never
/// serialized to HTTP, SSE consumers, or model-facing tools.
#[derive(Debug, Clone, Serialize, PartialEq)]
pub(crate) struct PublicWorkflowRunEvent {
pub run_id: String,
pub sequence: u64,
pub at: chrono::DateTime<chrono::Utc>,
#[serde(skip_serializing_if = "Option::is_none")]
pub step_id: Option<String>,
#[serde(flatten)]
pub kind: PublicWorkflowRunEventKind,
}
#[derive(Debug, Clone, Serialize, PartialEq)]
#[serde(tag = "type", rename_all = "snake_case")]
pub(crate) enum PublicWorkflowRunEventKind {
RunQueued,
RunStarted,
Phase { name: &'static str },
StepQueued,
StepStarted,
StepSuspended,
StepCompleted,
StepFailed { failure: PublicWorkflowFailure },
StepCancelled,
StepSkipped,
RunSuspended,
RunSucceeded,
RunFailed { failure: PublicWorkflowFailure },
RunCancelled,
}
fn public_workflow_plan(plan: &WorkflowPlan) -> PublicWorkflowPlan {
match plan {
WorkflowPlan::Step { step } => PublicWorkflowPlan::Step { step: step.clone() },
WorkflowPlan::Sequence { nodes } => PublicWorkflowPlan::Sequence {
nodes: nodes.iter().map(public_workflow_plan).collect(),
},
WorkflowPlan::Parallel { nodes } => PublicWorkflowPlan::Parallel {
nodes: nodes.iter().map(public_workflow_plan).collect(),
},
WorkflowPlan::Map { body, .. } => PublicWorkflowPlan::Map {
body: Box::new(public_workflow_plan(body)),
},
WorkflowPlan::Retry {
node, max_attempts, ..
} => PublicWorkflowPlan::Retry {
node: Box::new(public_workflow_plan(node)),
max_attempts: *max_attempts,
},
}
}
fn public_planned_steps(
definition: &WorkflowRunDefinition,
) -> BTreeMap<String, PublicWorkflowPlannedStep> {
definition
.steps
.iter()
.map(|step| {
let kind = match &step.kind {
WorkflowStepKind::Tool { .. } => PublicWorkflowPlannedStepKind::Tool,
WorkflowStepKind::Agent { .. } => PublicWorkflowPlannedStepKind::Agent,
WorkflowStepKind::Workflow { .. } => PublicWorkflowPlannedStepKind::Workflow,
};
(
step.id.clone(),
PublicWorkflowPlannedStep {
id: step.id.clone(),
kind,
},
)
})
.collect()
}
fn public_workflow_failure(failure: WorkflowFailure) -> PublicWorkflowFailure {
let message = match failure.code {
WorkflowFailureCode::InvalidDefinition => "Workflow definition is invalid",
WorkflowFailureCode::InvalidInput => "Workflow input is invalid",
WorkflowFailureCode::InvalidOutput => "Workflow output is invalid",
WorkflowFailureCode::UnknownReference => "Workflow reference is unavailable",
WorkflowFailureCode::PermissionDenied => "Workflow permission was denied",
WorkflowFailureCode::UntrustedWorkspace => "Workflow workspace is not trusted",
WorkflowFailureCode::BudgetExceeded => "Workflow execution budget was exceeded",
WorkflowFailureCode::RetryExhausted => "Workflow retry budget was exhausted",
WorkflowFailureCode::ExecutionFailed => "Workflow execution failed",
WorkflowFailureCode::Cancelled => "Workflow execution was cancelled",
WorkflowFailureCode::RecoverySuspended => "Workflow recovery requires attention",
WorkflowFailureCode::Suspended => "Workflow execution is suspended",
WorkflowFailureCode::DependencySkipped => "Workflow dependency was skipped",
WorkflowFailureCode::Storage => "Workflow storage is unavailable",
};
PublicWorkflowFailure {
code: failure.code,
message,
retryable: failure.retryable,
}
}
fn public_workflow_step(step: WorkflowStepSnapshot) -> PublicWorkflowStepSnapshot {
PublicWorkflowStepSnapshot {
id: step.id,
status: step.status,
failure: step.failure.map(public_workflow_failure),
attempts: step.attempts,
}
}
fn public_workflow_suspension(suspension: WorkflowSuspensionContext) -> PublicWorkflowSuspension {
match suspension {
WorkflowSuspensionContext::ToolApproval { step_id, .. } => {
PublicWorkflowSuspension::ToolApproval { step_id }
}
WorkflowSuspensionContext::ToolRunning {
step_id, killed, ..
} => PublicWorkflowSuspension::ToolRunning { step_id, killed },
WorkflowSuspensionContext::Recovery { .. } => PublicWorkflowSuspension::Recovery,
}
}
fn public_workflow_phase(name: &str) -> &'static str {
match name {
"retry_reserved" => "retry_reserved",
"step_reserved" => "step_reserved",
"agent_reserved" => "agent_reserved",
"agent_usage_recorded" => "agent_usage_recorded",
"suspension_context_persisted" => "suspension_context_persisted",
_ => "workflow_progressed",
}
}
pub(crate) fn public_workflow_snapshot(snapshot: WorkflowRunSnapshot) -> PublicWorkflowRunSnapshot {
let can_cancel = ensure_workflow_cancel_allowed(&snapshot).is_ok();
let can_restart_as_new_run = ensure_workflow_restart_as_new_run_allowed(&snapshot).is_ok();
let planned_steps = public_planned_steps(&snapshot.definition);
let plan = public_workflow_plan(&snapshot.definition.plan);
let child_agent_count = snapshot.usage.agents;
PublicWorkflowRunSnapshot {
run_id: snapshot.run_id,
parent_run_id: snapshot.parent_run_id,
parent_step_id: snapshot.parent_step_id,
session_id: snapshot.session_id,
workflow_id: snapshot.definition.id,
workflow_revision: snapshot.definition.revision,
definition_bundle_hash: snapshot.definition_bundle_hash,
status: snapshot.status,
can_cancel,
can_restart_as_new_run,
planned_steps,
plan,
steps: snapshot
.steps
.into_iter()
.map(|(id, step)| (id, public_workflow_step(step)))
.collect(),
budget: snapshot.definition.budgets,
usage: snapshot.usage,
child_agent_count,
last_sequence: snapshot.last_sequence,
failure: snapshot.failure.map(public_workflow_failure),
suspension: snapshot.suspension.map(public_workflow_suspension),
created_at: snapshot.created_at,
updated_at: snapshot.updated_at,
}
}
pub(crate) fn public_workflow_event(event: WorkflowRunEvent) -> PublicWorkflowRunEvent {
let kind = match event.kind {
WorkflowRunEventKind::RunQueued => PublicWorkflowRunEventKind::RunQueued,
WorkflowRunEventKind::RunStarted => PublicWorkflowRunEventKind::RunStarted,
WorkflowRunEventKind::Phase { name } => PublicWorkflowRunEventKind::Phase {
name: public_workflow_phase(&name),
},
WorkflowRunEventKind::StepQueued => PublicWorkflowRunEventKind::StepQueued,
WorkflowRunEventKind::StepStarted => PublicWorkflowRunEventKind::StepStarted,
WorkflowRunEventKind::StepSuspended { .. } => PublicWorkflowRunEventKind::StepSuspended,
WorkflowRunEventKind::StepCompleted { .. } => PublicWorkflowRunEventKind::StepCompleted,
WorkflowRunEventKind::StepFailed { failure } => PublicWorkflowRunEventKind::StepFailed {
failure: public_workflow_failure(failure),
},
WorkflowRunEventKind::StepCancelled => PublicWorkflowRunEventKind::StepCancelled,
WorkflowRunEventKind::StepSkipped { .. } => PublicWorkflowRunEventKind::StepSkipped,
WorkflowRunEventKind::RunSuspended { .. } => PublicWorkflowRunEventKind::RunSuspended,
WorkflowRunEventKind::RunSucceeded { .. } => PublicWorkflowRunEventKind::RunSucceeded,
WorkflowRunEventKind::RunFailed { failure } => PublicWorkflowRunEventKind::RunFailed {
failure: public_workflow_failure(failure),
},
WorkflowRunEventKind::RunCancelled => PublicWorkflowRunEventKind::RunCancelled,
};
PublicWorkflowRunEvent {
run_id: event.run_id,
sequence: event.sequence,
at: event.at,
step_id: event.step_id,
kind,
}
}
/// 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
.into_iter()
.map(public_workflow_event)
.collect::<Vec<_>>(),
)
}
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(_) => {
ToolError::InvalidArguments("workflow input is invalid".to_string())
}
WorkflowRunError::Compile(_) => {
ToolError::InvalidArguments("workflow definition is invalid".to_string())
}
WorkflowRunError::Preflight(_) => {
ToolError::InvalidArguments("workflow preflight failed".to_string())
}
WorkflowRunError::Storage(details) => {
tracing::error!(%details, "workflow tool storage unavailable");
ToolError::Execution("workflow storage unavailable".to_string())
}
WorkflowRunError::NotFound => ToolError::Execution("workflow run not found".to_string()),
WorkflowRunError::Terminal => {
ToolError::Execution("workflow run is already terminal".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_engine::WorkflowRunRepository;
use bamboo_llm::protocol::{gemini::GeminiTool, ToProvider};
use std::collections::HashMap;
use tokio::sync::{RwLock, Semaphore};
#[derive(Default)]
struct WorkflowTestStorage {
sessions: RwLock<HashMap<String, Session>>,
fail_saves: AtomicBool,
}
impl WorkflowTestStorage {
fn set_fail_saves(&self, fail: bool) {
self.fail_saves.store(fail, Ordering::SeqCst);
}
}
#[async_trait]
impl Storage for WorkflowTestStorage {
async fn save_session(&self, session: &Session) -> std::io::Result<()> {
if self.fail_saves.load(Ordering::SeqCst) {
return Err(std::io::Error::other(
"injected workflow session persistence failure",
));
}
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"}),
},
}]
}
}
struct BlockingWorkflowReadTool {
entered: Arc<Semaphore>,
}
#[async_trait]
impl ToolExecutor for BlockingWorkflowReadTool {
async fn execute(
&self,
_call: &ToolCall,
) -> bamboo_agent_core::tools::executor::Result<ToolResult> {
self.entered.add_permits(1);
std::future::pending().await
}
fn list_tools(&self) -> Vec<ToolSchema> {
WorkflowReadTool.list_tools()
}
}
async fn workflow_test_access_with_tools(
tools: Arc<dyn ToolExecutor>,
) -> (
WorkflowRunAccess,
bamboo_engine::SessionRepository,
tempfile::TempDir,
) {
let (access, repo, directory, _) = workflow_test_access_with_tools_and_storage(tools).await;
(access, repo, directory)
}
async fn workflow_test_access_with_tools_and_storage(
tools: Arc<dyn ToolExecutor>,
) -> (
WorkflowRunAccess,
bamboo_engine::SessionRepository,
tempfile::TempDir,
Arc<WorkflowTestStorage>,
) {
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::new(WorkflowTestStorage::default());
let storage_port: Arc<dyn Storage> = storage.clone();
let persistence = Arc::new(bamboo_storage::LockedSessionStore::new(
storage_port.clone(),
));
let cache = Arc::default();
let repo = bamboo_engine::SessionRepository::new(cache, storage_port, persistence);
let access = WorkflowRunAccess::new(directory.path(), tools, skills, repo.clone())
.await
.expect("workflow access");
(access, repo, directory, storage)
}
async fn workflow_test_access() -> (
WorkflowRunAccess,
bamboo_engine::SessionRepository,
tempfile::TempDir,
) {
workflow_test_access_with_tools(Arc::new(WorkflowReadTool)).await
}
async fn seed_durable_workflow_run(
access: &WorkflowRunAccess,
directory: &Path,
workspace: &Path,
session_id: &str,
run_id: &str,
status: WorkflowRunStatus,
suspension: Option<WorkflowSuspensionContext>,
) -> WorkflowRunSnapshot {
let bundle = access
.skills
.pin_workflow_definition_bundle(Some(workspace), "review-flow", 42)
.await
.expect("pin workflow test bundle");
let definition = bundle.root().cloned().expect("pinned root definition");
let step_status = match status {
WorkflowRunStatus::Queued => WorkflowStepStatus::Queued,
WorkflowRunStatus::Running => WorkflowStepStatus::Running,
WorkflowRunStatus::Suspended => WorkflowStepStatus::Suspended,
WorkflowRunStatus::Succeeded => WorkflowStepStatus::Succeeded,
WorkflowRunStatus::Failed => WorkflowStepStatus::Failed,
WorkflowRunStatus::Cancelled => WorkflowStepStatus::Cancelled,
};
let now = chrono::Utc::now();
let failure = (status == WorkflowRunStatus::Failed).then(|| WorkflowFailure {
code: WorkflowFailureCode::ExecutionFailed,
message: "seeded workflow failure".to_string(),
retryable: false,
});
let snapshot = WorkflowRunSnapshot {
run_id: run_id.to_string(),
parent_run_id: None,
parent_step_id: None,
session_id: session_id.to_string(),
definition: definition.clone(),
definition_bundle: bundle,
definition_bundle_hash: "seeded-public-bundle-hash".to_string(),
validated_args: json!({}),
status,
steps: definition
.steps
.iter()
.map(|step| {
(
step.id.clone(),
WorkflowStepSnapshot {
id: step.id.clone(),
status: step_status,
input_hash: "seeded-input-hash".to_string(),
output: None,
failure: failure.clone(),
attempts: 0,
},
)
})
.collect(),
usage: WorkflowBudgetUsage::default(),
last_sequence: 1,
output: None,
failure: failure.clone(),
suspension,
created_at: now,
updated_at: now,
};
persist_durable_workflow_run(directory, &snapshot).await;
snapshot
}
async fn persist_durable_workflow_run(directory: &Path, snapshot: &WorkflowRunSnapshot) {
let kind = match snapshot.status {
WorkflowRunStatus::Queued => WorkflowRunEventKind::RunQueued,
WorkflowRunStatus::Running => WorkflowRunEventKind::RunStarted,
WorkflowRunStatus::Suspended => WorkflowRunEventKind::RunSuspended {
reason: "seeded suspension".to_string(),
},
WorkflowRunStatus::Succeeded => WorkflowRunEventKind::RunSucceeded {
output: Value::Null,
},
WorkflowRunStatus::Failed => WorkflowRunEventKind::RunFailed {
failure: snapshot.failure.clone().expect("failed run has failure"),
},
WorkflowRunStatus::Cancelled => WorkflowRunEventKind::RunCancelled,
};
let repository = FileWorkflowRunRepository::new(directory.join("workflow-runs"))
.expect("open workflow test repository");
repository
.create(
snapshot,
&WorkflowRunEvent {
run_id: snapshot.run_id.clone(),
sequence: snapshot.last_sequence,
at: snapshot.updated_at,
step_id: None,
kind,
},
)
.await
.expect("seed durable workflow run");
}
async fn replace_workflow_run_index(
repo: &bamboo_engine::SessionRepository,
session_id: &str,
run_ids: Vec<String>,
) {
repo.update_runtime_session(
session_id,
&[bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY],
move |session| {
session.metadata.insert(
bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY.to_string(),
serde_json::to_string(&run_ids).expect("run index json"),
);
},
)
.await
.expect("replace workflow run index")
.expect("workflow session");
}
async fn wait_for_workflow_run_to_settle(
access: &WorkflowRunAccess,
run_id: &str,
) -> WorkflowRunSnapshot {
tokio::time::timeout(std::time::Duration::from_secs(2), async {
loop {
let snapshot = access
.engine
.progress(run_id, u64::MAX)
.await
.expect("workflow progress")
.snapshot;
if snapshot.status.is_terminal() && !access.engine.is_run_active(run_id) {
break snapshot;
}
tokio::task::yield_now().await;
}
})
.await
.expect("workflow run settles")
}
fn private_workflow_snapshot(status: WorkflowRunStatus) -> WorkflowRunSnapshot {
let definition = WorkflowRunDefinition {
workflow_schema: 1,
id: "review-flow".to_string(),
revision: 42,
input_schema: json!({"private_schema": "PRIVATE-SCHEMA-SENTINEL"}),
output_schema: Some(json!({"private_output": "PRIVATE-SCHEMA-SENTINEL"})),
steps: vec![bamboo_domain::WorkflowStepDefinition {
id: "inspect".to_string(),
kind: WorkflowStepKind::Tool {
tool: "PRIVATE-TOOL-SENTINEL".to_string(),
args: json!({"credential": "PRIVATE-ARG-SENTINEL"}),
capabilities: vec!["PRIVATE-CAPABILITY-SENTINEL".to_string()],
},
failure: bamboo_domain::FailurePolicy::FailFast,
output_schema: Some(json!({"private": "PRIVATE-STEP-SCHEMA-SENTINEL"})),
}],
plan: WorkflowPlan::Retry {
node: Box::new(WorkflowPlan::Map {
source: bamboo_domain::ValueRef::Literal {
value: json!("PRIVATE-BINDING-SENTINEL"),
},
item: "PRIVATE-ITEM-SENTINEL".to_string(),
body: Box::new(WorkflowPlan::Step {
step: "inspect".to_string(),
}),
}),
max_attempts: 3,
delay_ms: 987_654,
},
budgets: WorkflowBudgets {
max_concurrency: 2,
max_agents: 4,
max_steps: 8,
max_retries: 3,
max_nesting_depth: 2,
wall_time_ms: 10_000,
max_tokens: Some(1_000),
max_cost_micros: Some(2_000),
},
};
let definition_bundle = WorkflowDefinitionBundle {
publication_revision: 7,
root_id: definition.id.clone(),
root_revision: definition.revision,
root_invocation_policy: json!({"private": "PRIVATE-POLICY-SENTINEL"}),
definitions: BTreeMap::from([(
WorkflowDefinitionBundle::key(&definition.id, definition.revision),
definition.clone(),
)]),
};
let now = chrono::Utc::now();
WorkflowRunSnapshot {
run_id: "public-run".to_string(),
parent_run_id: Some("public-parent-run".to_string()),
parent_step_id: Some("public-parent-step".to_string()),
session_id: "public-session".to_string(),
definition,
definition_bundle,
definition_bundle_hash: "public-bundle-hash".to_string(),
validated_args: json!({"password": "PRIVATE-VALIDATED-ARG-SENTINEL"}),
status,
steps: BTreeMap::from([(
"inspect".to_string(),
WorkflowStepSnapshot {
id: "inspect".to_string(),
status: WorkflowStepStatus::Failed,
input_hash: "PRIVATE-INPUT-HASH-SENTINEL".to_string(),
output: Some(json!({"raw_tool_output": "PRIVATE-OUTPUT-SENTINEL"})),
failure: Some(WorkflowFailure {
code: WorkflowFailureCode::Storage,
message: "/private/workspace/PRIVATE-DIAGNOSTIC-SENTINEL".to_string(),
retryable: true,
}),
attempts: 2,
},
)]),
usage: WorkflowBudgetUsage {
steps: 1,
retries: 1,
agents: 3,
tokens: 40,
cost_micros: 50,
},
last_sequence: 9,
output: Some(json!({"raw_run_output": "PRIVATE-RUN-OUTPUT-SENTINEL"})),
failure: Some(WorkflowFailure {
code: WorkflowFailureCode::ExecutionFailed,
message: "credential PRIVATE-RUN-FAILURE-SENTINEL".to_string(),
retryable: false,
}),
suspension: Some(WorkflowSuspensionContext::ToolApproval {
step_id: "inspect".to_string(),
tool: "PRIVATE-SUSPENSION-TOOL-SENTINEL".to_string(),
tool_call_id: "PRIVATE-TOOL-CALL-SENTINEL".to_string(),
}),
created_at: now,
updated_at: now,
}
}
#[test]
fn public_workflow_snapshot_is_stable_metadata_only_for_every_status() {
let statuses = [
(WorkflowRunStatus::Queued, "queued", true),
(WorkflowRunStatus::Running, "running", true),
(WorkflowRunStatus::Suspended, "suspended", true),
(WorkflowRunStatus::Succeeded, "succeeded", false),
(WorkflowRunStatus::Failed, "failed", false),
(WorkflowRunStatus::Cancelled, "cancelled", true),
];
for (status, wire_status, can_cancel) in statuses {
let public =
serde_json::to_value(public_workflow_snapshot(private_workflow_snapshot(status)))
.expect("public snapshot serializes");
let text = public.to_string();
assert_eq!(public["status"], wire_status);
assert_eq!(public["can_cancel"], can_cancel);
assert_eq!(public["can_restart_as_new_run"], false);
assert_eq!(public["workflow_id"], "review-flow");
assert_eq!(public["workflow_revision"], 42);
assert_eq!(public["definition_bundle_hash"], "public-bundle-hash");
assert_eq!(public["planned_steps"]["inspect"]["kind"], "tool");
assert_eq!(public["plan"]["type"], "retry");
assert_eq!(public["steps"]["inspect"]["attempts"], 2);
assert_eq!(public["budget"]["max_steps"], 8);
assert_eq!(public["usage"]["agents"], 3);
assert_eq!(public["child_agent_count"], 3);
assert_eq!(public["last_sequence"], 9);
assert_eq!(public["failure"]["message"], "Workflow execution failed");
assert_eq!(public["suspension"]["type"], "tool_approval");
for internal_field in [
"definition",
"definition_bundle",
"validated_args",
"output",
] {
assert!(
public.get(internal_field).is_none(),
"public snapshot exposed internal field {internal_field}: {text}"
);
}
for private in [
"PRIVATE-",
"validated_args",
"input_hash",
"output_schema",
"root_invocation_policy",
"delay_ms",
"tool_call_id",
"raw_tool_output",
"raw_run_output",
] {
assert!(
!text.contains(private),
"public snapshot leaked {private}: {text}"
);
}
}
let mut recovery = private_workflow_snapshot(WorkflowRunStatus::Suspended);
recovery.suspension = Some(WorkflowSuspensionContext::Recovery {
reason: "PRIVATE-RECOVERY-REASON-SENTINEL".to_string(),
});
let public = serde_json::to_value(public_workflow_snapshot(recovery))
.expect("public recovery snapshot serializes");
let text = public.to_string();
assert_eq!(public["can_cancel"], true);
assert_eq!(public["can_restart_as_new_run"], true);
assert_eq!(public["suspension"]["type"], "recovery");
assert!(!text.contains("PRIVATE-RECOVERY-REASON-SENTINEL"));
}
#[tokio::test]
async fn public_workflow_actions_match_cancel_and_restart_endpoint_acceptance() {
let (access, repo, directory) = workflow_test_access().await;
let workspace = directory.path().join("action-workspace");
std::fs::create_dir_all(&workspace).expect("action workspace");
let session_id = "workflow-action-matrix";
let mut session = Session::new(session_id, "model");
session.workspace = Some(workspace.to_string_lossy().into_owned());
repo.save(&mut session).await.expect("save action session");
let cases = vec![
("queued", WorkflowRunStatus::Queued, None, true, false),
("running", WorkflowRunStatus::Running, None, true, false),
(
"suspended_without_context",
WorkflowRunStatus::Suspended,
None,
true,
false,
),
(
"tool_approval",
WorkflowRunStatus::Suspended,
Some(WorkflowSuspensionContext::ToolApproval {
step_id: "inspect".to_string(),
tool: "Read".to_string(),
tool_call_id: "approval-call".to_string(),
}),
true,
false,
),
(
"tool_running",
WorkflowRunStatus::Suspended,
Some(WorkflowSuspensionContext::ToolRunning {
step_id: "inspect".to_string(),
tool: "Read".to_string(),
tool_call_id: "running-call".to_string(),
killed: true,
}),
true,
false,
),
(
"recovery",
WorkflowRunStatus::Suspended,
Some(WorkflowSuspensionContext::Recovery {
reason: "process restarted".to_string(),
}),
true,
true,
),
(
"succeeded",
WorkflowRunStatus::Succeeded,
None,
false,
false,
),
("failed", WorkflowRunStatus::Failed, None, false, false),
("cancelled", WorkflowRunStatus::Cancelled, None, true, false),
];
for (name, status, suspension, can_cancel, can_restart_as_new_run) in cases {
let cancel_run_id = format!("action-cancel-{name}");
let cancel_snapshot = seed_durable_workflow_run(
&access,
directory.path(),
&workspace,
session_id,
&cancel_run_id,
status,
suspension.clone(),
)
.await;
let cancel_public = public_workflow_snapshot(cancel_snapshot);
assert_eq!(
cancel_public.can_cancel, can_cancel,
"cancel projection mismatch for {name}"
);
assert_eq!(
access
.cancel_for_session(session_id, &cancel_run_id)
.await
.is_ok(),
can_cancel,
"cancel endpoint mismatch for {name}"
);
let restart_run_id = format!("action-restart-{name}");
let restart_snapshot = seed_durable_workflow_run(
&access,
directory.path(),
&workspace,
session_id,
&restart_run_id,
status,
suspension,
)
.await;
let restart_public = public_workflow_snapshot(restart_snapshot);
assert_eq!(
restart_public.can_restart_as_new_run, can_restart_as_new_run,
"restart projection mismatch for {name}"
);
let restarted = access
.restart_for_session(session_id, &restart_run_id)
.await;
assert_eq!(
restarted.is_ok(),
can_restart_as_new_run,
"restart endpoint mismatch for {name}: {restarted:?}"
);
if let Ok(restarted) = restarted {
let _ = access
.cancel_for_session(session_id, &restarted.run_id)
.await;
}
}
}
#[tokio::test]
async fn restart_as_new_run_is_indexed_isolated_and_survives_reconstruction() {
let (access, repo, directory, storage) =
workflow_test_access_with_tools_and_storage(Arc::new(WorkflowReadTool)).await;
let workspace = directory.path().join("restart-workspace");
std::fs::create_dir_all(&workspace).expect("restart workspace");
let session_id = "restart-owner";
let mut session = Session::new(session_id, "model");
session.workspace = Some(workspace.to_string_lossy().into_owned());
repo.save(&mut session).await.expect("save restart owner");
let mut other = Session::new("restart-other", "model");
other.workspace = Some(workspace.to_string_lossy().into_owned());
repo.save(&mut other).await.expect("save other session");
let original = seed_durable_workflow_run(
&access,
directory.path(),
&workspace,
session_id,
"recovery-original",
WorkflowRunStatus::Suspended,
Some(WorkflowSuspensionContext::Recovery {
reason: "process restarted".to_string(),
}),
)
.await;
replace_workflow_run_index(&repo, session_id, vec![original.run_id.clone()]).await;
let restarted = access
.restart_for_session(session_id, &original.run_id)
.await
.expect("restart recovery suspension as a new run");
assert_ne!(restarted.run_id, original.run_id);
assert_eq!(restarted.session_id, session_id);
assert_eq!(
access
.progress_for_session(session_id, &original.run_id, u64::MAX)
.await
.expect("original progress")
.snapshot,
original,
"restart-as-new must not mutate the original suspended run"
);
let immediate = access
.list_for_session(session_id)
.await
.expect("immediate owner list");
let immediate_ids = immediate
.iter()
.map(|snapshot| snapshot.run_id.as_str())
.collect::<BTreeSet<_>>();
assert_eq!(
immediate_ids,
BTreeSet::from([original.run_id.as_str(), restarted.run_id.as_str()])
);
assert!(access
.list_for_session("restart-other")
.await
.expect("isolated other list")
.is_empty());
assert!(matches!(
access
.progress_for_session("restart-other", &restarted.run_id, u64::MAX)
.await,
Err(WorkflowRunError::NotFound)
));
assert!(matches!(
access
.restart_for_session("restart-other", &original.run_id)
.await,
Err(WorkflowRunError::NotFound)
));
let settled = wait_for_workflow_run_to_settle(&access, &restarted.run_id).await;
assert_eq!(settled.status, WorkflowRunStatus::Succeeded);
let skills = access.skills.clone();
drop(access);
drop(repo);
let storage_port: Arc<dyn Storage> = storage;
let reopened_repo = bamboo_engine::SessionRepository::new(
Arc::default(),
storage_port.clone(),
Arc::new(bamboo_storage::LockedSessionStore::new(storage_port)),
);
let reopened = WorkflowRunAccess::new(
directory.path(),
Arc::new(WorkflowReadTool),
skills,
reopened_repo,
)
.await
.expect("reconstruct workflow access");
let reconstructed = reopened
.list_for_session(session_id)
.await
.expect("list after reconstruction");
let reconstructed_ids = reconstructed
.iter()
.map(|snapshot| snapshot.run_id.as_str())
.collect::<BTreeSet<_>>();
assert_eq!(
reconstructed_ids,
BTreeSet::from([original.run_id.as_str(), restarted.run_id.as_str()])
);
assert_eq!(
reopened
.progress_for_session(session_id, &original.run_id, u64::MAX)
.await
.expect("reconstructed original")
.snapshot,
original
);
}
#[tokio::test]
async fn restart_capacity_preflight_creates_no_new_run_or_active_orphan() {
let (access, repo, directory) = workflow_test_access().await;
let workspace = directory.path().join("restart-capacity-workspace");
std::fs::create_dir_all(&workspace).expect("restart capacity workspace");
let session_id = "restart-capacity";
let mut session = Session::new(session_id, "model");
session.workspace = Some(workspace.to_string_lossy().into_owned());
repo.save(&mut session)
.await
.expect("save restart capacity session");
let original = seed_durable_workflow_run(
&access,
directory.path(),
&workspace,
session_id,
"capacity-run-0",
WorkflowRunStatus::Suspended,
Some(WorkflowSuspensionContext::Recovery {
reason: "process restarted".to_string(),
}),
)
.await;
let mut run_ids = vec![original.run_id.clone()];
for index in 1..MAX_WORKFLOW_RUN_IDS_PER_SESSION {
let mut snapshot = original.clone();
snapshot.run_id = format!("capacity-run-{index}");
snapshot.created_at = chrono::Utc::now();
snapshot.updated_at = snapshot.created_at;
persist_durable_workflow_run(directory.path(), &snapshot).await;
run_ids.push(snapshot.run_id);
}
replace_workflow_run_index(&repo, session_id, run_ids).await;
let before = access
.engine
.list_run_ids()
.await
.expect("run ids before capacity rejection")
.into_iter()
.collect::<BTreeSet<_>>();
assert!(matches!(
access
.restart_for_session(session_id, &original.run_id)
.await,
Err(WorkflowRunError::Preflight(message))
if message == "workflow run index is full of active runs"
));
let after = access
.engine
.list_run_ids()
.await
.expect("run ids after capacity rejection")
.into_iter()
.collect::<BTreeSet<_>>();
assert_eq!(
after, before,
"capacity rejection must precede run creation"
);
assert!(after
.iter()
.all(|run_id| !access.engine.is_run_active(run_id)));
assert_eq!(
access
.progress_for_session(session_id, &original.run_id, u64::MAX)
.await
.expect("original after capacity rejection")
.snapshot,
original
);
}
#[tokio::test]
async fn restart_index_persistence_failure_cancels_the_unindexed_new_run() {
let (access, repo, directory, storage) =
workflow_test_access_with_tools_and_storage(Arc::new(BlockingWorkflowReadTool {
entered: Arc::new(Semaphore::new(0)),
}))
.await;
let workspace = directory.path().join("restart-failure-workspace");
std::fs::create_dir_all(&workspace).expect("restart failure workspace");
let session_id = "restart-index-failure";
let mut session = Session::new(session_id, "model");
session.workspace = Some(workspace.to_string_lossy().into_owned());
repo.save(&mut session)
.await
.expect("save restart failure session");
let original = seed_durable_workflow_run(
&access,
directory.path(),
&workspace,
session_id,
"failure-original",
WorkflowRunStatus::Suspended,
Some(WorkflowSuspensionContext::Recovery {
reason: "process restarted".to_string(),
}),
)
.await;
replace_workflow_run_index(&repo, session_id, vec![original.run_id.clone()]).await;
let before = access
.engine
.list_run_ids()
.await
.expect("run ids before injected failure")
.into_iter()
.collect::<BTreeSet<_>>();
storage.set_fail_saves(true);
let error = access
.restart_for_session(session_id, &original.run_id)
.await
.expect_err("injected index persistence failure");
storage.set_fail_saves(false);
assert!(matches!(error, WorkflowRunError::Storage(_)));
let after = access
.engine
.list_run_ids()
.await
.expect("run ids after injected failure")
.into_iter()
.collect::<BTreeSet<_>>();
let created = after.difference(&before).cloned().collect::<Vec<_>>();
assert_eq!(created.len(), 1, "restart should have created one new run");
let unindexed_run_id = &created[0];
let compensated = wait_for_workflow_run_to_settle(&access, unindexed_run_id).await;
assert_eq!(compensated.status, WorkflowRunStatus::Cancelled);
assert!(!access.engine.is_run_active(unindexed_run_id));
assert_eq!(
access
.progress_for_session(session_id, &original.run_id, u64::MAX)
.await
.expect("original after index failure")
.snapshot,
original
);
let listed = access
.list_for_session(session_id)
.await
.expect("owner list after index failure");
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].run_id, original.run_id);
let durable = storage
.load_session(session_id)
.await
.expect("load durable owner")
.expect("durable owner");
let durable_ids = durable
.metadata
.get(bamboo_skills::WORKFLOW_RUN_IDS_METADATA_KEY)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
.expect("durable run index");
assert_eq!(durable_ids, vec![original.run_id]);
assert!(!durable_ids.contains(unindexed_run_id));
}
#[test]
fn public_workflow_events_preserve_sequence_and_drop_private_payloads() {
let private = "PRIVATE-EVENT-SENTINEL";
let failure = WorkflowFailure {
code: WorkflowFailureCode::ExecutionFailed,
message: format!("/private/workspace/{private}"),
retryable: true,
};
let kinds = vec![
WorkflowRunEventKind::RunQueued,
WorkflowRunEventKind::RunStarted,
WorkflowRunEventKind::Phase {
name: private.to_string(),
},
WorkflowRunEventKind::StepQueued,
WorkflowRunEventKind::StepStarted,
WorkflowRunEventKind::StepSuspended {
reason: private.to_string(),
},
WorkflowRunEventKind::StepCompleted {
output: json!({"raw": private}),
},
WorkflowRunEventKind::StepFailed {
failure: failure.clone(),
},
WorkflowRunEventKind::StepCancelled,
WorkflowRunEventKind::StepSkipped {
reason: private.to_string(),
},
WorkflowRunEventKind::RunSuspended {
reason: private.to_string(),
},
WorkflowRunEventKind::RunSucceeded {
output: json!({"raw": private}),
},
WorkflowRunEventKind::RunFailed { failure },
WorkflowRunEventKind::RunCancelled,
];
let at = chrono::Utc::now();
let events = kinds
.into_iter()
.enumerate()
.map(|(index, kind)| {
public_workflow_event(WorkflowRunEvent {
run_id: "public-run".to_string(),
sequence: index as u64 + 1,
at,
step_id: Some("inspect".to_string()),
kind,
})
})
.collect::<Vec<_>>();
let public = serde_json::to_value(&events).expect("public events serialize");
let text = public.to_string();
assert!(!text.contains(private), "public events leaked: {text}");
assert_eq!(public[2]["name"], "workflow_progressed");
assert_eq!(public[6]["type"], "step_completed");
assert!(public[6].get("output").is_none());
assert_eq!(public[7]["failure"]["message"], "Workflow execution failed");
assert_eq!(public[11]["type"], "run_succeeded");
assert!(public[11].get("output").is_none());
assert_eq!(
events
.iter()
.map(|event| event.sequence)
.collect::<Vec<_>>(),
(1..=14).collect::<Vec<_>>()
);
}
#[test]
fn workflow_tool_errors_never_expose_backend_diagnostics() {
let sentinel = "/private/workspace/credentials-PRIVATE-SENTINEL";
let cases = [
WorkflowRunError::Storage(sentinel.to_string()),
WorkflowRunError::InvalidInput(sentinel.to_string()),
WorkflowRunError::Preflight(sentinel.to_string()),
];
for error in cases {
assert!(!workflow_tool_error(error).to_string().contains(sentinel));
}
assert!(!workflow_tool_error(WorkflowRunError::Compile(
bamboo_domain::WorkflowCompileError::InvalidSchema(sentinel.to_string())
))
.to_string()
.contains(sentinel));
}
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)
));
// Recreate the adapter over the same durable journal, then combine the
// snapshot with only the tail after sequence 4. This is the reconnect
// contract used by Lotus after a process or transport restart.
let run_id = started.run_id.clone();
let skills = access.skills.clone();
drop(access);
let reopened = WorkflowRunAccess::new(
directory.path(),
Arc::new(WorkflowReadTool),
skills,
repo.clone(),
)
.await
.expect("reopen workflow adapter");
let reconnected = reopened
.progress_for_session("workflow-session", &run_id, 4)
.await
.expect("durable reconnect");
assert_eq!(reconnected.snapshot.status, WorkflowRunStatus::Succeeded);
assert_eq!(
reconnected
.events
.iter()
.map(|event| event.sequence)
.collect::<Vec<_>>(),
vec![5, 6, 7]
);
assert_eq!(
public_workflow_snapshot(reconnected.snapshot).last_sequence,
7
);
assert!(matches!(
public_workflow_event(reconnected.events.last().cloned().expect("tail event")).kind,
PublicWorkflowRunEventKind::RunSucceeded
));
assert_eq!(
reopened
.list_for_session("workflow-session")
.await
.expect("reconnected list")
.len(),
1
);
}
#[tokio::test]
async fn workflow_cancel_is_idempotent_and_reconnects_to_one_terminal_event() {
let entered = Arc::new(Semaphore::new(0));
let (access, repo, directory) =
workflow_test_access_with_tools(Arc::new(BlockingWorkflowReadTool {
entered: entered.clone(),
}))
.await;
let workspace = directory.path().join("workspace");
std::fs::create_dir_all(&workspace).expect("workspace");
let mut session = Session::new("cancel-session", "model");
session.workspace = Some(workspace.to_string_lossy().into_owned());
repo.save(&mut session).await.expect("save session");
let started = access
.start("cancel-session", "review-flow", 42, json!({}), None)
.await
.expect("start blocking run");
let _entered = tokio::time::timeout(std::time::Duration::from_secs(2), entered.acquire())
.await
.expect("blocking step entered")
.expect("semaphore open");
let first = access
.cancel_for_session("cancel-session", &started.run_id)
.await
.expect("first cancel");
let second = access
.cancel_for_session("cancel-session", &started.run_id)
.await
.expect("idempotent cancel");
assert_eq!(first.status, WorkflowRunStatus::Cancelled);
assert_eq!(second.status, WorkflowRunStatus::Cancelled);
assert_eq!(first.last_sequence, second.last_sequence);
let progress = access
.progress_for_session("cancel-session", &started.run_id, 0)
.await
.expect("cancelled journal");
assert_eq!(progress.snapshot.status, WorkflowRunStatus::Cancelled);
assert_eq!(progress.snapshot.last_sequence, first.last_sequence);
assert_eq!(
progress
.events
.iter()
.filter(|event| matches!(event.kind, WorkflowRunEventKind::RunCancelled))
.count(),
1
);
assert!(!progress
.events
.iter()
.any(|event| matches!(event.kind, WorkflowRunEventKind::RunSucceeded { .. })));
assert_eq!(
progress
.events
.iter()
.map(|event| event.sequence)
.collect::<Vec<_>>(),
(1..=progress.snapshot.last_sequence).collect::<Vec<_>>()
);
let tail = access
.progress_for_session(
"cancel-session",
&started.run_id,
progress.snapshot.last_sequence,
)
.await
.expect("terminal reconnect tail");
assert!(tail.events.is_empty());
assert!(matches!(
public_workflow_snapshot(second).status,
WorkflowRunStatus::Cancelled
));
}
#[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());
}
}