use crate::cancel_token::CancellationFlag;
use crate::flow_execution_event::FlowExecutionEvent;
use crate::ir_nodes::IRFlowNode;
use crate::stream_effect::BackpressurePolicy;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex};
pub mod budget_gate;
pub mod agent_loop;
pub mod pure_shape;
pub mod orchestration;
pub mod parallel;
pub mod effects_bridge;
pub mod effect_handlers;
pub mod cognitive;
pub mod algebraic_handlers;
pub mod wire_integrations;
pub mod pix;
pub mod lambda_tools;
pub mod unified_stream;
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct StoreRowCounts {
pub retrieved: u64,
pub persisted: u64,
pub mutated: u64,
pub purged: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StoreRowKind {
Retrieved,
Persisted,
Mutated,
Purged,
}
#[derive(Clone, Debug)]
pub struct QuantFrame {
pub encoding: crate::quant::EncodingScheme,
pub observable: crate::quant::PauliSum,
pub observable_name: String,
}
#[derive(Clone)]
pub struct DispatchCtx {
pub flow_name: String,
pub backend_name: String,
pub api_key: Option<String>,
pub llm_base_url: Option<String>,
pub llm_chat_path: Option<String>,
pub conversation: std::sync::Arc<std::sync::Mutex<crate::conversation::ConversationHistory>>,
pub context_budget: usize,
pub anchors: std::sync::Arc<Vec<crate::ir_nodes::IRAnchor>>,
pub system_prompt: String,
pub default_now_tz: Option<String>,
pub temporal: std::sync::Arc<std::sync::Mutex<crate::temporal_context::TemporalState>>,
pub credential_minter: Option<std::sync::Arc<dyn crate::credential_minter::CredentialMinter>>,
pub credentials:
std::sync::Arc<std::collections::HashMap<String, crate::ir_nodes::IRCredential>>,
pub secret_custody: Option<std::sync::Arc<dyn crate::secret_custody::SecretCustody>>,
pub dataspace_engine: Option<crate::dataspace_engine::SharedDataspaceEngine>,
pub warden_backend: Option<std::sync::Arc<dyn crate::warden::WardenBackend + Send + Sync>>,
pub scopes: Arc<Vec<crate::ir_nodes::IRScope>>,
pub quant_backend: Option<std::sync::Arc<dyn crate::quant::QuantBackend + Send + Sync>>,
pub observables: Arc<Vec<crate::ir_nodes::IRObservable>>,
pub quant_frame: Option<QuantFrame>,
pub compute_specs: Arc<Vec<crate::ir_nodes::IRCompute>>,
pub agent_specs: Arc<Vec<crate::ir_nodes::IRAgent>>,
pub mandate_specs: Arc<Vec<crate::ir_nodes::IRMandate>>,
pub lambda_data_specs: Arc<Vec<crate::ir_nodes::IRLambdaData>>,
pub ots_specs: Arc<Vec<crate::ir_nodes::IROts>>,
pub cancel: CancellationFlag,
pub tx: mpsc::UnboundedSender<FlowExecutionEvent>,
pub enforcement_summaries: Arc<
Mutex<HashMap<String, crate::execution_result::EnforcementSummaryWire>>,
>,
pub step_audit_records: Arc<
Mutex<Vec<crate::axonendpoint_replay::StepAuditRecord>>,
>,
pub runtime_warnings: Arc<
Mutex<Vec<crate::runtime_warnings::RuntimeWarning>>,
>,
pub store_row_counts: std::sync::Arc<std::sync::Mutex<StoreRowCounts>>,
pub branch_path: Vec<String>,
pub step_counter: usize,
pub pem_backend: Option<std::sync::Arc<dyn crate::pem::PersistenceBackend>>,
pub session_id: String,
pub tenant_id: String,
pub let_bindings: std::collections::HashMap<String, String>,
pub effect_frames: Vec<crate::flow_dispatcher::effect_handlers::EffectFrame>,
pub effect_clause_frame: Option<u32>,
pub pending_effect_policy: Option<BackpressurePolicy>,
pub tool_registry: Option<std::sync::Arc<crate::tool_registry::ToolRegistry>>,
pub mdn_corpora: Option<std::sync::Arc<std::collections::HashMap<String, crate::mdn::Corpus>>>,
pub mdn_adaptive: std::sync::Arc<std::collections::HashSet<String>>,
pub mdn_histories:
std::sync::Arc<std::sync::Mutex<std::collections::HashMap<String, crate::mdn_memory::History>>>,
pub mdn_store_sources:
std::sync::Arc<std::collections::HashMap<String, crate::ir_nodes::IRCorpusStoreSource>>,
pub store_registry: Option<std::sync::Arc<crate::store::registry::StoreRegistry>>,
pub held_capabilities: Option<Vec<String>>,
pub audit_chain:
std::sync::Arc<std::sync::Mutex<crate::store::audit_chain::StoreAuditChain>>,
pub pinned_conns: std::sync::Arc<
std::sync::Mutex<
std::collections::HashMap<
String,
crate::pinned_conn::PinnedConn,
>,
>,
>,
pub budget: Option<
std::sync::Arc<std::sync::Mutex<crate::runtime::budget_kernel::BudgetGate>>,
>,
pub channel_semaphores: Option<std::sync::Arc<crate::channel_semaphore::ChannelSemaphores>>,
pub tool_leases: Option<std::sync::Arc<crate::resource_lease::ResourceLeaseGuard>>,
pub cache_runtime: Option<std::sync::Arc<crate::cache_runtime::CacheRuntime>>,
pub cache_policies:
std::sync::Arc<
std::collections::HashMap<String, crate::cache_runtime::ResolvedCachePolicy>,
>,
pub caches: std::sync::Arc<std::collections::HashMap<String, crate::ir_nodes::IRCache>>,
pub cache_invalidation_channels:
std::sync::Arc<std::collections::HashMap<String, Vec<String>>>,
pub event_bus: Option<std::sync::Arc<crate::runtime::channels::TypedEventBus>>,
pub event_outbox: Option<std::sync::Arc<dyn crate::event_outbox::EventOutbox>>,
pub replay_log: Option<std::sync::Arc<dyn crate::replay_token::ReplayLog>>,
}
impl DispatchCtx {
pub fn new(
flow_name: impl Into<String>,
backend_name: impl Into<String>,
system_prompt: impl Into<String>,
cancel: CancellationFlag,
tx: mpsc::UnboundedSender<FlowExecutionEvent>,
) -> Self {
let flow_name = flow_name.into();
let session_id = flow_name.clone();
Self {
flow_name,
backend_name: backend_name.into(),
api_key: None,
llm_base_url: None,
llm_chat_path: None,
conversation: std::sync::Arc::new(std::sync::Mutex::new(
crate::conversation::ConversationHistory::new(),
)),
context_budget: crate::conversation::ContextWindow::new().max_chars,
anchors: std::sync::Arc::new(Vec::new()),
system_prompt: system_prompt.into(),
default_now_tz: None,
temporal: std::sync::Arc::new(std::sync::Mutex::new(
crate::temporal_context::TemporalState::default(),
)),
credential_minter: None,
credentials: std::sync::Arc::new(std::collections::HashMap::new()),
secret_custody: None,
dataspace_engine: None,
warden_backend: None,
scopes: Arc::new(Vec::new()),
quant_backend: None,
observables: Arc::new(Vec::new()),
quant_frame: None,
compute_specs: Arc::new(Vec::new()),
agent_specs: Arc::new(Vec::new()),
mandate_specs: Arc::new(Vec::new()),
lambda_data_specs: Arc::new(Vec::new()),
ots_specs: Arc::new(Vec::new()),
cancel,
tx,
enforcement_summaries: Arc::new(Mutex::new(HashMap::new())),
step_audit_records: Arc::new(Mutex::new(Vec::new())),
runtime_warnings: Arc::new(Mutex::new(Vec::new())),
store_row_counts: std::sync::Arc::new(std::sync::Mutex::new(
StoreRowCounts::default(),
)),
branch_path: Vec::new(),
step_counter: 0,
pem_backend: None,
session_id,
tenant_id: String::new(),
let_bindings: std::collections::HashMap::new(),
effect_frames: Vec::new(),
effect_clause_frame: None,
pending_effect_policy: None,
tool_registry: None,
mdn_corpora: None,
mdn_adaptive: std::sync::Arc::new(std::collections::HashSet::new()),
mdn_histories: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
mdn_store_sources: std::sync::Arc::new(std::collections::HashMap::new()),
store_registry: None,
held_capabilities: None,
audit_chain: std::sync::Arc::new(std::sync::Mutex::new(
crate::store::audit_chain::StoreAuditChain::new(),
)),
pinned_conns: std::sync::Arc::new(std::sync::Mutex::new(
std::collections::HashMap::new(),
)),
budget: None,
channel_semaphores: None,
tool_leases: None,
cache_runtime: None,
cache_policies: std::sync::Arc::new(std::collections::HashMap::new()),
caches: std::sync::Arc::new(std::collections::HashMap::new()),
cache_invalidation_channels: std::sync::Arc::new(std::collections::HashMap::new()),
event_bus: None,
event_outbox: None,
replay_log: None,
}
}
pub fn with_secret_custody(
mut self,
custody: std::sync::Arc<dyn crate::secret_custody::SecretCustody>,
) -> Self {
self.secret_custody = Some(custody);
self
}
pub fn with_dataspace_engine(
mut self,
engine: crate::dataspace_engine::SharedDataspaceEngine,
) -> Self {
self.dataspace_engine = Some(engine);
self
}
pub fn with_warden(
mut self,
backend: std::sync::Arc<dyn crate::warden::WardenBackend + Send + Sync>,
scopes: Arc<Vec<crate::ir_nodes::IRScope>>,
) -> Self {
self.warden_backend = Some(backend);
self.scopes = scopes;
self
}
pub fn with_computes(mut self, specs: Arc<Vec<crate::ir_nodes::IRCompute>>) -> Self {
self.compute_specs = specs;
self
}
pub fn with_agents(mut self, specs: Arc<Vec<crate::ir_nodes::IRAgent>>) -> Self {
self.agent_specs = specs;
self
}
pub fn with_mandates(mut self, specs: Arc<Vec<crate::ir_nodes::IRMandate>>) -> Self {
self.mandate_specs = specs;
self
}
pub fn with_lambdas(mut self, specs: Arc<Vec<crate::ir_nodes::IRLambdaData>>) -> Self {
self.lambda_data_specs = specs;
self
}
pub fn with_ots(mut self, specs: Arc<Vec<crate::ir_nodes::IROts>>) -> Self {
self.ots_specs = specs;
self
}
pub fn with_quant(
mut self,
backend: std::sync::Arc<dyn crate::quant::QuantBackend + Send + Sync>,
observables: Arc<Vec<crate::ir_nodes::IRObservable>>,
) -> Self {
self.quant_backend = Some(backend);
self.observables = observables;
self
}
pub fn with_event_bus(
mut self,
bus: std::sync::Arc<crate::runtime::channels::TypedEventBus>,
) -> Self {
self.event_bus = Some(bus);
self
}
pub fn with_event_outbox(
mut self,
outbox: std::sync::Arc<dyn crate::event_outbox::EventOutbox>,
) -> Self {
self.event_outbox = Some(outbox);
self
}
pub fn with_replay_log(
mut self,
log: std::sync::Arc<dyn crate::replay_token::ReplayLog>,
) -> Self {
self.replay_log = Some(log);
self
}
pub fn with_budget(
mut self,
budget: std::sync::Arc<std::sync::Mutex<crate::runtime::budget_kernel::BudgetGate>>,
) -> Self {
self.budget = Some(budget);
self
}
pub fn with_channel_semaphores(
mut self,
sems: std::sync::Arc<crate::channel_semaphore::ChannelSemaphores>,
) -> Self {
self.channel_semaphores = Some(sems);
self
}
pub fn with_cache(
mut self,
runtime: std::sync::Arc<crate::cache_runtime::CacheRuntime>,
plan: &crate::cache_runtime::CachePlan,
) -> Self {
self.cache_policies = std::sync::Arc::new(plan.tool_policies.clone());
self.caches = std::sync::Arc::new(plan.caches.clone());
self.cache_invalidation_channels =
std::sync::Arc::new(plan.invalidation_channels.clone());
self.cache_runtime = Some(runtime);
self
}
pub fn with_tool_leases(
mut self,
leases: std::sync::Arc<crate::resource_lease::ResourceLeaseGuard>,
) -> Self {
self.tool_leases = Some(leases);
self
}
pub fn with_pinned_conns(
mut self,
conns: std::sync::Arc<
std::sync::Mutex<
std::collections::HashMap<
String,
crate::pinned_conn::PinnedConn,
>,
>,
>,
) -> Self {
self.pinned_conns = conns;
self
}
pub fn with_api_key(mut self, api_key: Option<String>) -> Self {
self.api_key = api_key;
self
}
pub fn with_llm_endpoint(
mut self,
base_url: Option<String>,
chat_path: Option<String>,
) -> Self {
self.llm_base_url = base_url;
self.llm_chat_path = chat_path;
self
}
pub fn with_context_budget(mut self, max_chars: usize) -> Self {
self.context_budget = max_chars;
self
}
pub fn with_anchors(mut self, anchors: std::sync::Arc<Vec<crate::ir_nodes::IRAnchor>>) -> Self {
self.anchors = anchors;
self
}
pub fn with_store_registry(
mut self,
registry: std::sync::Arc<crate::store::registry::StoreRegistry>,
) -> Self {
self.store_registry = Some(registry);
self
}
pub fn with_held_capabilities(mut self, capabilities: Vec<String>) -> Self {
self.held_capabilities = Some(capabilities);
self
}
pub fn with_tool_registry(
mut self,
registry: std::sync::Arc<crate::tool_registry::ToolRegistry>,
) -> Self {
self.tool_registry = Some(registry);
self
}
pub fn with_mdn_corpora(
mut self,
corpora: std::sync::Arc<std::collections::HashMap<String, crate::mdn::Corpus>>,
) -> Self {
self.mdn_corpora = Some(corpora);
self
}
pub fn with_mdn_adaptive(
mut self,
adaptive: std::sync::Arc<std::collections::HashSet<String>>,
) -> Self {
self.mdn_adaptive = adaptive;
self
}
pub fn with_mdn_store_sources(
mut self,
sources: std::sync::Arc<
std::collections::HashMap<String, crate::ir_nodes::IRCorpusStoreSource>,
>,
) -> Self {
self.mdn_store_sources = sources;
self
}
pub fn with_pem(
mut self,
backend: std::sync::Arc<dyn crate::pem::PersistenceBackend>,
) -> Self {
self.pem_backend = Some(backend);
self
}
pub fn with_session_id(mut self, session_id: impl Into<String>) -> Self {
self.session_id = session_id.into();
self
}
pub fn with_tenant_id(mut self, tenant_id: impl Into<String>) -> Self {
self.tenant_id = tenant_id.into();
self
}
pub fn with_effect_policy(mut self, policy: BackpressurePolicy) -> Self {
self.pending_effect_policy = Some(policy);
self
}
pub fn with_external_side_channels(
mut self,
enforcement_summaries: std::sync::Arc<
tokio::sync::Mutex<
std::collections::HashMap<String, crate::execution_result::EnforcementSummaryWire>,
>,
>,
step_audit_records: std::sync::Arc<
tokio::sync::Mutex<Vec<crate::axonendpoint_replay::StepAuditRecord>>,
>,
runtime_warnings: std::sync::Arc<
tokio::sync::Mutex<Vec<crate::runtime_warnings::RuntimeWarning>>,
>,
store_row_counts: std::sync::Arc<std::sync::Mutex<StoreRowCounts>>,
) -> Self {
self.enforcement_summaries = enforcement_summaries;
self.step_audit_records = step_audit_records;
self.runtime_warnings = runtime_warnings;
self.store_row_counts = store_row_counts;
self
}
pub fn record_store_rows(&self, kind: StoreRowKind, n: u64) {
let mut c = self.store_row_counts.lock().unwrap();
match kind {
StoreRowKind::Retrieved => c.retrieved += n,
StoreRowKind::Persisted => c.persisted += n,
StoreRowKind::Mutated => c.mutated += n,
StoreRowKind::Purged => c.purged += n,
}
}
pub fn take_pending_effect_policy(&mut self) -> Option<BackpressurePolicy> {
self.pending_effect_policy.take()
}
pub fn branch_path_string(&self) -> String {
self.branch_path.join(".")
}
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum NodeOutcome {
Completed {
output: String,
tokens_emitted: u64,
step_index: usize,
},
Break,
LoopContinue,
Return { value: String },
Hibernated {
event_name: String,
timeout: String,
step_index: usize,
},
EffectResumed { value: String },
EffectAborted { frame_id: u32, value: String },
EffectForwarded {
effect: String,
operation: String,
arguments: Vec<String>,
},
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum DispatchError {
BackendError { name: String, message: String },
UpstreamCancelled,
MissingDependency { name: &'static str },
ChannelClosed,
EffectQuotaExhausted { effect: String, retry_at_ms: i64 },
EffectDeferred { effect: String, retry_at_ms: i64 },
}
impl std::fmt::Display for DispatchError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::BackendError { name, message } => {
write!(f, "backend '{name}' stream() failed: {message}")
}
Self::UpstreamCancelled => write!(f, "upstream cancelled mid-dispatch"),
Self::MissingDependency { name } => {
write!(f, "dispatcher missing dependency: {name}")
}
Self::ChannelClosed => write!(f, "channel closed (consumer dropped)"),
Self::EffectQuotaExhausted { effect, retry_at_ms } => write!(
f,
"axon-E0810 EffectQuotaExhausted: budget for Tool({effect}) is exhausted \
(on_exhausted: block); next token at {retry_at_ms}ms — the call was not emitted"
),
Self::EffectDeferred { effect, retry_at_ms } => write!(
f,
"axon-E0811 EffectDeferred: budget for Tool({effect}) is exhausted \
(on_exhausted: defer); the tick reschedules to {retry_at_ms}ms — \
the call was not emitted this run"
),
}
}
}
impl std::error::Error for DispatchError {}
pub async fn dispatch_node(
node: &IRFlowNode,
ctx: &mut DispatchCtx,
) -> Result<NodeOutcome, DispatchError> {
match node {
IRFlowNode::Step(step) => pure_shape::run_step(step, ctx).await,
IRFlowNode::Probe(probe) => pure_shape::run_probe(probe, ctx).await,
IRFlowNode::Reason(reason) => pure_shape::run_reason(reason, ctx).await,
IRFlowNode::Validate(validate) => pure_shape::run_validate(validate, ctx).await,
IRFlowNode::Refine(refine) => pure_shape::run_refine(refine, ctx).await,
IRFlowNode::Weave(weave) => pure_shape::run_weave(weave, ctx).await,
IRFlowNode::UseTool(node) => lambda_tools::run_use_tool(node, ctx).await,
IRFlowNode::Remember(node) => cognitive::run_remember(node, ctx).await,
IRFlowNode::Recall(node) => cognitive::run_recall(node, ctx).await,
IRFlowNode::Conditional(cond) => orchestration::run_conditional(cond, ctx).await,
IRFlowNode::ForIn(for_in) => orchestration::run_for_in(for_in, ctx).await,
IRFlowNode::Let(let_bind) => orchestration::run_let(let_bind, ctx).await,
IRFlowNode::Return(ret) => orchestration::run_return(ret, ctx).await,
IRFlowNode::Break(brk) => orchestration::run_break(brk, ctx).await,
IRFlowNode::Continue(cont) => orchestration::run_continue(cont, ctx).await,
IRFlowNode::LambdaDataApply(node) => lambda_tools::run_lambda_data_apply(node, ctx).await,
IRFlowNode::Par(par) => parallel::run_par(par, ctx).await,
IRFlowNode::Hibernate(node) => pix::run_hibernate(node, ctx).await,
IRFlowNode::Deliberate(node) => wire_integrations::run_deliberate(node, ctx).await,
IRFlowNode::Consensus(node) => wire_integrations::run_consensus(node, ctx).await,
IRFlowNode::Forge(node) => cognitive::run_forge(node, ctx).await,
IRFlowNode::Grad(node) => orchestration::run_grad(node, ctx).await,
IRFlowNode::Focus(node) => cognitive::run_focus(node, ctx).await,
IRFlowNode::Associate(node) => cognitive::run_associate(node, ctx).await,
IRFlowNode::Aggregate(node) => cognitive::run_aggregate(node, ctx).await,
IRFlowNode::Explore(node) => cognitive::run_explore(node, ctx).await,
IRFlowNode::Ingest(node) => cognitive::run_ingest(node, ctx).await,
IRFlowNode::ShieldApply(node) => algebraic_handlers::run_shield_apply(node, ctx).await,
IRFlowNode::Stream(stream) => effects_bridge::run_stream(stream, ctx).await,
IRFlowNode::Handle(node) => effect_handlers::run_handle(node, ctx).await,
IRFlowNode::Perform(node) => effect_handlers::run_perform(node, ctx).await,
IRFlowNode::Resume(node) => effect_handlers::run_resume(node, ctx).await,
IRFlowNode::Abort(node) => effect_handlers::run_abort(node, ctx).await,
IRFlowNode::Forward(node) => effect_handlers::run_forward(node, ctx).await,
IRFlowNode::Navigate(node) => cognitive::run_navigate(node, ctx).await,
IRFlowNode::Drill(node) => pix::run_drill(node, ctx).await,
IRFlowNode::Trail(node) => pix::run_trail(node, ctx).await,
IRFlowNode::Corroborate(node) => cognitive::run_corroborate(node, ctx).await,
IRFlowNode::OtsApply(node) => algebraic_handlers::run_ots_apply(node, ctx).await,
IRFlowNode::MandateApply(node) => algebraic_handlers::run_mandate_apply(node, ctx).await,
IRFlowNode::ComputeApply(node) => algebraic_handlers::run_compute_apply(node, ctx).await,
IRFlowNode::AgentCall(node) => agent_loop::run_agent_call(node, ctx).await,
IRFlowNode::Listen(node) => algebraic_handlers::run_listen(node, ctx).await,
IRFlowNode::DaemonStep(node) => algebraic_handlers::run_daemon_step(node, ctx).await,
IRFlowNode::Mint(node) => wire_integrations::run_mint(node, ctx).await,
IRFlowNode::Rotate(node) => wire_integrations::run_rotate(node, ctx).await,
IRFlowNode::Emit(node) => wire_integrations::run_emit(node, ctx).await,
IRFlowNode::Publish(node) => wire_integrations::run_publish(node, ctx).await,
IRFlowNode::Discover(node) => wire_integrations::run_discover(node, ctx).await,
IRFlowNode::Persist(node) => wire_integrations::run_persist(node, ctx).await,
IRFlowNode::Retrieve(node) => wire_integrations::run_retrieve(node, ctx).await,
IRFlowNode::Mutate(node) => wire_integrations::run_mutate(node, ctx).await,
IRFlowNode::Purge(node) => wire_integrations::run_purge(node, ctx).await,
IRFlowNode::Transact(node) => wire_integrations::run_transact(node, ctx).await,
IRFlowNode::Quant(node) => wire_integrations::run_quant(node, ctx).await,
IRFlowNode::Warden(node) => wire_integrations::run_warden(node, ctx).await,
IRFlowNode::Yield(node) => wire_integrations::run_yield(node, ctx).await,
IRFlowNode::Run(node) => algebraic_handlers::run_run(node, ctx).await,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::cancel_token::CancellationFlag;
#[test]
fn budget_dispatch_errors_carry_their_typed_codes() {
let blocked = DispatchError::EffectQuotaExhausted {
effect: "TelnyxCall".into(),
retry_at_ms: 1_700_000_000_000,
};
let msg = blocked.to_string();
assert!(msg.contains("axon-E0810"), "{msg}");
assert!(msg.contains("EffectQuotaExhausted") && msg.contains("Tool(TelnyxCall)"), "{msg}");
assert!(msg.contains("not emitted"), "{msg}");
let deferred = DispatchError::EffectDeferred {
effect: "TelnyxCall".into(),
retry_at_ms: 1_700_000_000_000,
};
let msg = deferred.to_string();
assert!(msg.contains("axon-E0811"), "{msg}");
assert!(msg.contains("EffectDeferred") && msg.contains("reschedules"), "{msg}");
}
#[test]
fn dispatch_ctx_branch_path_empty_at_root() {
let (tx, _rx) = mpsc::unbounded_channel();
let ctx = DispatchCtx::new(
"F",
"stub",
"",
CancellationFlag::new(),
tx,
);
assert!(ctx.branch_path.is_empty());
assert_eq!(ctx.branch_path_string(), "");
assert_eq!(ctx.step_counter, 0);
}
#[test]
fn dispatch_ctx_with_api_key_carries_per_tenant_key() {
let (tx, _rx) = mpsc::unbounded_channel();
let ctx = DispatchCtx::new("F", "kimi", "", CancellationFlag::new(), tx);
assert_eq!(ctx.api_key, None, "default is env-key (None)");
let (tx2, _rx2) = mpsc::unbounded_channel();
let ctx2 = DispatchCtx::new("F", "kimi", "", CancellationFlag::new(), tx2)
.with_api_key(Some("sk-tenant-42".to_string()));
assert_eq!(ctx2.api_key.as_deref(), Some("sk-tenant-42"));
}
#[test]
fn dispatch_ctx_branch_path_joins_segments() {
let (tx, _rx) = mpsc::unbounded_channel();
let mut ctx = DispatchCtx::new(
"F",
"stub",
"",
CancellationFlag::new(),
tx,
);
ctx.branch_path.push("par[0]".to_string());
ctx.branch_path.push("step[1]".to_string());
assert_eq!(ctx.branch_path_string(), "par[0].step[1]");
}
#[test]
fn dispatch_error_display_surface() {
let cases: Vec<(DispatchError, &str)> = vec![
(
DispatchError::BackendError {
name: "anthropic".to_string(),
message: "rate limited".to_string(),
},
"backend 'anthropic' stream() failed: rate limited",
),
(DispatchError::UpstreamCancelled, "upstream cancelled mid-dispatch"),
(
DispatchError::MissingDependency { name: "pem_async" },
"dispatcher missing dependency: pem_async",
),
(DispatchError::ChannelClosed, "channel closed (consumer dropped)"),
];
for (err, expected) in cases {
assert_eq!(format!("{err}"), expected);
}
}
#[tokio::test]
async fn dispatch_node_step_routes_to_pure_shape_handler() {
use crate::ir_nodes::*;
let step = IRStep {
node_type: "step",
source_line: 0,
source_column: 0,
name: "Generate".to_string(),
persona_ref: String::new(),
given: String::new(),
ask: "hi".to_string(),
use_tool: None,
probe: None,
reason: None,
weave: None,
output_type: String::new(),
confidence_floor: None,
navigate_ref: String::new(),
apply_ref: String::new(),
pix_ops: Vec::new(),
stream: None,
performs: Vec::new(),
guards: Vec::new(),
requires_context: None, now_tz: None, body: Vec::new(),
};
let node = IRFlowNode::Step(step);
let (tx, _rx) = mpsc::unbounded_channel();
let mut ctx = DispatchCtx::new(
"F",
"stub",
"",
CancellationFlag::new(),
tx,
);
let outcome = dispatch_node(&node, &mut ctx).await.unwrap();
match outcome {
NodeOutcome::Completed {
output,
tokens_emitted,
step_index,
} => {
assert_eq!(output, "(stub)");
assert_eq!(tokens_emitted, 1);
assert_eq!(step_index, 0);
}
other => panic!("post-33.y.c: Step routes to pure_shape handler returning Completed; got {other:?}"),
}
}
}
pub fn resume_parked_flow(
parked: crate::hibernation::ParkedFlow,
wake_payload: String,
) -> futures::future::BoxFuture<'static, ()> {
Box::pin(resume_parked_flow_inner(parked, wake_payload))
}
async fn resume_parked_flow_inner(
parked: crate::hibernation::ParkedFlow,
wake_payload: String,
) {
let id = parked.continuation_id.clone();
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
tokio::spawn(async move { while rx.recv().await.is_some() {} });
let mut ctx = DispatchCtx::new(
&parked.flow_name,
&parked.backend_name,
&parked.system_prompt,
crate::cancel_token::CancellationFlag::new(),
tx,
);
ctx.tenant_id = parked.tenant_id.clone();
ctx.session_id = parked.session_id.clone();
ctx.step_counter = parked.step_counter;
ctx.let_bindings = parked.let_bindings.clone();
ctx.let_bindings
.insert(parked.event_name.clone(), wake_payload);
ctx = ctx
.with_computes(parked.compute_specs.clone())
.with_agents(parked.agent_specs.clone())
.with_mandates(parked.mandate_specs.clone())
.with_lambdas(parked.lambda_data_specs.clone())
.with_ots(parked.ots_specs.clone());
let mut last_output = String::new();
for node in &parked.remaining_nodes {
match dispatch_node(node, &mut ctx).await {
Ok(NodeOutcome::Completed { output, .. }) => last_output = output,
Ok(NodeOutcome::Return { value }) => {
last_output = value;
break;
}
Ok(NodeOutcome::Hibernated { .. }) => {
crate::hibernation::parking_lot().record_outcome(
&id,
crate::hibernation::ResumedOutcome::Failed {
error: "a resumed continuation hibernated again; chained \
hibernation is not yet defined"
.to_string(),
},
);
return;
}
Ok(_) => {}
Err(e) => {
tracing::warn!(continuation = id.as_str(), error = ?e,
"resumed continuation failed");
crate::hibernation::parking_lot().record_outcome(
&id,
crate::hibernation::ResumedOutcome::Failed {
error: format!("{e:?}"),
},
);
return;
}
}
}
crate::hibernation::parking_lot().record_outcome(
&id,
crate::hibernation::ResumedOutcome::Completed { output: last_output },
);
}