a3s_code_core/
dynamic_workflow.rs

1//! A3S Flow-backed dynamic workflow runtime.
2//!
3//! `DynamicWorkflowRuntime` lets hosts run a sandboxed PTC script as an A3S
4//! Flow runtime. Flow owns durable replay and step lifecycle; A3S Code's
5//! existing `program` tool remains the sandbox and tool-call boundary.
6
7use crate::tools::{
8    registry_tool_invoker, Tool, ToolContext, ToolInvocation, ToolInvoker, ToolOutput,
9    ToolRegistry, ToolResult,
10};
11use crate::{
12    agent::AgentEvent,
13    flow_graph::FlowGraphObserver,
14    planning::{Complexity, ExecutionPlan, Task, TaskStatus},
15};
16use a3s_flow::{
17    FanoutFlowEventObserver, FlowEngine, FlowEvent, FlowEventEnvelope, FlowEventObserver,
18    FlowEventStore, FlowRuntime, InMemoryEventStore, LocalFileEventStore, RuntimeCommand,
19    StepInvocation, StepStatus, WorkflowInvocation, WorkflowRunSnapshot, WorkflowRunStatus,
20    WorkflowSpec,
21};
22use anyhow::{Context, Result};
23use async_trait::async_trait;
24use chrono::Utc;
25use serde::{Deserialize, Serialize};
26use serde_json::{json, Map, Value};
27use std::collections::BTreeSet;
28use std::path::{Path, PathBuf};
29use std::sync::Arc;
30use std::time::Duration;
31use tokio::sync::{broadcast, Mutex};
32
33const DYNAMIC_WORKFLOW_TOOL: &str = "dynamic_workflow";
34const PROGRAM_TOOL: &str = "program";
35const PARALLEL_TASK_TOOL: &str = "parallel_task";
36const MAX_INLINE_RETRY_RESUMES: usize = 8;
37const MAX_INLINE_RETRY_DELAY: Duration = Duration::from_secs(5);
38
39/// Project-relative directory used for durable dynamic workflow history.
40pub const DYNAMIC_WORKFLOW_STORE_RELATIVE_PATH: &str = ".a3s/workflow";
41
42/// Resolve the durable dynamic workflow history directory for a local workspace.
43pub fn dynamic_workflow_store_path(workspace_root: impl AsRef<Path>) -> PathBuf {
44    workspace_root
45        .as_ref()
46        .join(DYNAMIC_WORKFLOW_STORE_RELATIVE_PATH)
47}
48
49/// Limits forwarded to the underlying PTC `program` tool.
50#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
51#[serde(rename_all = "camelCase")]
52pub struct DynamicWorkflowScriptLimits {
53    #[serde(skip_serializing_if = "Option::is_none")]
54    pub timeout_ms: Option<u64>,
55    #[serde(skip_serializing_if = "Option::is_none")]
56    pub max_tool_calls: Option<usize>,
57    #[serde(skip_serializing_if = "Option::is_none")]
58    pub max_output_bytes: Option<usize>,
59}
60
61/// Runs A3S Flow workflow and step invocations through a sandboxed PTC script.
62#[derive(Clone)]
63pub struct DynamicWorkflowRuntime {
64    invoker: Arc<dyn ToolInvoker>,
65    context: ToolContext,
66    source: Arc<str>,
67    allowed_tools: Vec<String>,
68    limits: DynamicWorkflowScriptLimits,
69}
70
71impl DynamicWorkflowRuntime {
72    pub fn new(
73        registry: Arc<ToolRegistry>,
74        context: ToolContext,
75        source: impl Into<String>,
76    ) -> Self {
77        let allowed_tools = default_allowed_tools(&registry);
78        // Session/agent callers install the governed gateway in ToolContext.
79        // The raw registry adapter is retained only for explicit low-level
80        // callers that construct this public runtime outside an AgentSession.
81        let invoker = context
82            .tool_invoker()
83            .unwrap_or_else(|| registry_tool_invoker(registry));
84        Self {
85            invoker,
86            context,
87            source: Arc::from(source.into()),
88            allowed_tools,
89            limits: DynamicWorkflowScriptLimits::default(),
90        }
91    }
92
93    pub fn with_allowed_tools(mut self, allowed_tools: impl IntoIterator<Item = String>) -> Self {
94        self.allowed_tools = sanitize_allowed_tools(allowed_tools);
95        self
96    }
97
98    pub fn with_limits(mut self, limits: DynamicWorkflowScriptLimits) -> Self {
99        self.limits = limits;
100        self
101    }
102
103    async fn run_script(&self, payload: Value) -> a3s_flow::Result<ToolResult> {
104        let mut args = json!({
105            "type": "script",
106            "language": "javascript",
107            "source": self.source.as_ref(),
108            "inputs": payload,
109            "allowed_tools": self.allowed_tools,
110        });
111        if let Some(object) = args.as_object_mut() {
112            if let Ok(Value::Object(limits)) = serde_json::to_value(&self.limits) {
113                if !limits.is_empty() {
114                    object.insert("limits".to_string(), Value::Object(limits));
115                }
116            }
117        }
118
119        let result = self
120            .invoker
121            .invoke(ToolInvocation::nested(PROGRAM_TOOL, args), &self.context)
122            .await;
123        if result.exit_code != 0 {
124            return Err(a3s_flow::FlowError::Runtime(result.output));
125        }
126        Ok(result)
127    }
128
129    async fn run_tool_step(&self, tool_name: &str, args: Value) -> a3s_flow::Result<Value> {
130        let result = self
131            .invoker
132            .invoke(
133                ToolInvocation::nested(tool_name.to_string(), args),
134                &self.context,
135            )
136            .await;
137        if result.exit_code != 0 {
138            return Err(a3s_flow::FlowError::Runtime(result.output));
139        }
140        Ok(json!({
141            "tool": result.name,
142            "output": result.output,
143            "exit_code": result.exit_code,
144            "metadata": result.metadata,
145        }))
146    }
147}
148
149#[async_trait]
150impl FlowRuntime for DynamicWorkflowRuntime {
151    async fn run_workflow(
152        &self,
153        invocation: WorkflowInvocation,
154    ) -> a3s_flow::Result<RuntimeCommand> {
155        let payload = invocation_payload("workflow", &invocation.run_id, &invocation.history)
156            .with("input", invocation.input);
157        let result = self.run_script(payload.into_value()).await?;
158        serde_json::from_value(script_result(&result)?).map_err(a3s_flow::FlowError::from)
159    }
160
161    async fn run_step(&self, invocation: StepInvocation) -> a3s_flow::Result<Value> {
162        if invocation.step_name == PARALLEL_TASK_TOOL {
163            return self
164                .run_tool_step(PARALLEL_TASK_TOOL, invocation.input)
165                .await;
166        }
167
168        let payload = invocation_payload("step", &invocation.run_id, &invocation.history)
169            .with("step_id", invocation.step_id)
170            .with("step_name", invocation.step_name)
171            .with("input", invocation.input);
172        let result = self.run_script(payload.into_value()).await?;
173        script_result(&result)
174    }
175}
176
177struct WorkflowProgressState {
178    tasks: Vec<Task>,
179}
180
181impl WorkflowProgressState {
182    fn new() -> Self {
183        Self { tasks: Vec::new() }
184    }
185
186    fn upsert_step(
187        &mut self,
188        step_id: &str,
189        step_name: &str,
190        input: Option<&Value>,
191        status: TaskStatus,
192    ) {
193        let content = workflow_step_description(step_id, step_name, input);
194        if let Some(task) = self.tasks.iter_mut().find(|task| task.id == step_id) {
195            task.content = content;
196            task.status = status;
197            task.tool = Some(step_name.to_string());
198        } else {
199            self.tasks
200                .push(Task::new(step_id.to_string(), content).with_tool(step_name));
201            if let Some(task) = self.tasks.last_mut() {
202                task.status = status;
203            }
204        }
205    }
206
207    fn mark_status(&mut self, step_id: &str, status: TaskStatus) {
208        if let Some(task) = self.tasks.iter_mut().find(|task| task.id == step_id) {
209            task.status = status;
210        }
211    }
212
213    fn step_position(&self, step_id: &str) -> (usize, usize) {
214        let total = self.tasks.len().max(1);
215        let number = self
216            .tasks
217            .iter()
218            .position(|task| task.id == step_id)
219            .map(|idx| idx + 1)
220            .unwrap_or(total);
221        (number, total)
222    }
223
224    fn step_description(&self, step_id: &str) -> String {
225        self.tasks
226            .iter()
227            .find(|task| task.id == step_id)
228            .map(|task| task.content.clone())
229            .unwrap_or_else(|| step_id.to_string())
230    }
231}
232
233struct AgentEventFlowObserver {
234    tx: broadcast::Sender<AgentEvent>,
235    session_id: String,
236    state: Mutex<WorkflowProgressState>,
237}
238
239impl AgentEventFlowObserver {
240    fn new(tx: broadcast::Sender<AgentEvent>, session_id: String) -> Self {
241        Self {
242            tx,
243            session_id,
244            state: Mutex::new(WorkflowProgressState::new()),
245        }
246    }
247
248    fn emit_task_update(&self, tasks: &[Task]) {
249        let _ = self.tx.send(AgentEvent::TaskUpdated {
250            session_id: self.session_id.clone(),
251            tasks: tasks.to_vec(),
252        });
253    }
254}
255
256#[async_trait]
257impl FlowEventObserver for AgentEventFlowObserver {
258    async fn observe(&self, envelope: FlowEventEnvelope) {
259        match envelope.event {
260            FlowEvent::RunStarted => {
261                let _ = self.tx.send(AgentEvent::PlanningStart {
262                    prompt: "dynamic_workflow".to_string(),
263                });
264            }
265            FlowEvent::StepCreated {
266                step_id,
267                step_name,
268                input,
269                ..
270            } => {
271                let mut state = self.state.lock().await;
272                state.upsert_step(&step_id, &step_name, Some(&input), TaskStatus::Pending);
273                self.emit_task_update(&state.tasks);
274                let mut plan = ExecutionPlan::new("dynamic workflow", Complexity::Medium);
275                for task in state.tasks.iter().cloned() {
276                    plan.add_step(task);
277                }
278                let _ = self.tx.send(AgentEvent::PlanningEnd {
279                    estimated_steps: plan.steps.len(),
280                    plan,
281                });
282            }
283            FlowEvent::StepStarted { step_id, .. } => {
284                let mut state = self.state.lock().await;
285                state.mark_status(&step_id, TaskStatus::InProgress);
286                self.emit_task_update(&state.tasks);
287                let (step_number, total_steps) = state.step_position(&step_id);
288                let _ = self.tx.send(AgentEvent::StepStart {
289                    description: state.step_description(&step_id),
290                    step_id,
291                    step_number,
292                    total_steps,
293                });
294            }
295            FlowEvent::StepCompleted { step_id, .. } => {
296                let mut state = self.state.lock().await;
297                state.mark_status(&step_id, TaskStatus::Completed);
298                self.emit_task_update(&state.tasks);
299                let (step_number, total_steps) = state.step_position(&step_id);
300                let _ = self.tx.send(AgentEvent::StepEnd {
301                    step_id,
302                    status: TaskStatus::Completed,
303                    step_number,
304                    total_steps,
305                });
306            }
307            FlowEvent::StepRetrying { step_id, .. } => {
308                let mut state = self.state.lock().await;
309                state.mark_status(&step_id, TaskStatus::InProgress);
310                self.emit_task_update(&state.tasks);
311            }
312            FlowEvent::StepFailed { step_id, .. } => {
313                let mut state = self.state.lock().await;
314                state.mark_status(&step_id, TaskStatus::Failed);
315                self.emit_task_update(&state.tasks);
316                let (step_number, total_steps) = state.step_position(&step_id);
317                let _ = self.tx.send(AgentEvent::StepEnd {
318                    step_id,
319                    status: TaskStatus::Failed,
320                    step_number,
321                    total_steps,
322                });
323            }
324            FlowEvent::RunFailed { .. } => {
325                let mut state = self.state.lock().await;
326                for task in &mut state.tasks {
327                    if task.status.is_active() {
328                        task.status = TaskStatus::Failed;
329                    }
330                }
331                self.emit_task_update(&state.tasks);
332            }
333            FlowEvent::RunCancelled { .. } => {
334                let mut state = self.state.lock().await;
335                for task in &mut state.tasks {
336                    if task.status.is_active() {
337                        task.status = TaskStatus::Cancelled;
338                    }
339                }
340                self.emit_task_update(&state.tasks);
341            }
342            _ => {}
343        }
344    }
345}
346
347fn workflow_step_description(step_id: &str, step_name: &str, input: Option<&Value>) -> String {
348    if step_name == PARALLEL_TASK_TOOL {
349        let count = input
350            .and_then(|value| value.get("tasks"))
351            .and_then(Value::as_array)
352            .map(Vec::len)
353            .unwrap_or(0);
354        if count > 0 {
355            return format!("Fan out {count} parallel subagent task(s)");
356        }
357    }
358
359    input
360        .and_then(|value| value.get("description").or_else(|| value.get("title")))
361        .and_then(Value::as_str)
362        .map(ToString::to_string)
363        .unwrap_or_else(|| {
364            if step_name == step_id {
365                step_id.to_string()
366            } else {
367                format!("{step_name}: {step_id}")
368            }
369        })
370}
371
372/// Model-visible tool that executes a dynamic workflow through A3S Flow.
373pub struct DynamicWorkflowTool {
374    registry: Arc<ToolRegistry>,
375    graph_observer: Option<FlowGraphObserver>,
376}
377
378impl DynamicWorkflowTool {
379    pub fn new(registry: Arc<ToolRegistry>) -> Self {
380        Self {
381            registry,
382            graph_observer: None,
383        }
384    }
385
386    /// Project committed Flow events into an optional reactive state graph.
387    /// A3S Flow remains the workflow execution source of truth.
388    pub fn with_graph_observer(mut self, observer: FlowGraphObserver) -> Self {
389        self.graph_observer = Some(observer);
390        self
391    }
392}
393
394#[async_trait]
395impl Tool for DynamicWorkflowTool {
396    fn name(&self) -> &str {
397        DYNAMIC_WORKFLOW_TOOL
398    }
399
400    fn description(&self) -> &str {
401        "Run a local dynamic workflow with A3S Flow. The workflow source is a sandboxed JavaScript PTC script that may call allowed ctx tools; A3S Flow records workflow and step history."
402    }
403
404    fn parameters(&self) -> Value {
405        json!({
406            "type": "object",
407            "additionalProperties": false,
408            "properties": {
409                "source": {
410                    "type": "string",
411                    "description": "JavaScript PTC source defining async function run(ctx, inputs). For inputs.kind='workflow', return a Flow command: {type:'complete', output}, {type:'fail', error}, {type:'schedule_step', step_id, step_name, input, retry?}, or {type:'schedule_steps', steps:[...]}. For inputs.kind='step', return the step JSON output. A scheduled step with step_name='parallel_task' bypasses QuickJS and calls the host parallel_task tool directly with input as its arguments."
412                },
413                "input": {
414                    "type": "object",
415                    "description": "Initial workflow input."
416                },
417                "run_id": {
418                    "type": "string",
419                    "description": "Optional durable run id. Reusing it with the same source and input is idempotent."
420                },
421                "allowed_tools": {
422                    "type": "array",
423                    "description": "Tool names the workflow script may call through ctx. Defaults to all registered tools except program, dynamic_workflow, and parallel_task. Login-registered tools such as runtime are allowed when present.",
424                    "items": { "type": "string" }
425                },
426                "limits": {
427                    "type": "object",
428                    "additionalProperties": false,
429                    "properties": {
430                        "timeoutMs": { "type": "integer", "minimum": 1 },
431                        "maxToolCalls": { "type": "integer", "minimum": 1 },
432                        "maxOutputBytes": { "type": "integer", "minimum": 1 }
433                    }
434                }
435            },
436            "required": ["source"]
437        })
438    }
439
440    async fn execute(&self, args: &Value, ctx: &ToolContext) -> Result<ToolOutput> {
441        let Some(source) = args.get("source").and_then(Value::as_str) else {
442            return Ok(ToolOutput::error("dynamic_workflow requires source"));
443        };
444        let input = args.get("input").cloned().unwrap_or_else(|| json!({}));
445        let allowed_tools = args
446            .get("allowed_tools")
447            .and_then(Value::as_array)
448            .map(|items| {
449                items
450                    .iter()
451                    .filter_map(Value::as_str)
452                    .map(ToString::to_string)
453                    .collect::<Vec<_>>()
454            })
455            .unwrap_or_else(|| default_allowed_tools(&self.registry));
456        let limits = args
457            .get("limits")
458            .cloned()
459            .and_then(|value| serde_json::from_value(value).ok())
460            .unwrap_or_default();
461
462        let runtime = Arc::new(
463            DynamicWorkflowRuntime::new(Arc::clone(&self.registry), ctx.clone(), source)
464                .with_allowed_tools(allowed_tools)
465                .with_limits(limits),
466        );
467        let requested_run_id = args.get("run_id").and_then(Value::as_str);
468        let store = match flow_store_for_context(ctx, requested_run_id).await {
469            Ok(store) => store,
470            Err(error) => return Ok(ToolOutput::error(error.to_string())),
471        };
472        let mut observers: Vec<Arc<dyn FlowEventObserver>> = Vec::new();
473        if let Some(tx) = ctx.agent_event_tx.clone() {
474            observers.push(Arc::new(AgentEventFlowObserver::new(
475                tx,
476                ctx.session_id.clone().unwrap_or_default(),
477            )));
478        }
479        if let Some(observer) = &self.graph_observer {
480            observers.push(Arc::new(observer.clone()));
481        }
482        let engine = if observers.is_empty() {
483            FlowEngine::new(store, runtime)
484        } else {
485            FlowEngine::builder(runtime)
486                .with_store(store)
487                .with_observer(Arc::new(FanoutFlowEventObserver::from_observers(observers)))
488                .build()
489        };
490        let source_hash = source_hash(source);
491        let spec = WorkflowSpec::rust_embedded(
492            "a3s-code.dynamic-workflow",
493            source_hash.as_str(),
494            "ptc",
495            "run",
496        );
497
498        let run_id = match requested_run_id {
499            Some(run_id) => match engine.start_with_id(run_id, spec, input).await {
500                Ok(run_id) => run_id,
501                Err(err) => return Ok(ToolOutput::error(err.to_string())),
502            },
503            None => match engine.start(spec, input).await {
504                Ok(run_id) => run_id,
505                Err(err) => return Ok(ToolOutput::error(err.to_string())),
506            },
507        };
508
509        let snapshot = match drive_inline_retries(&engine, &run_id, ctx).await {
510            Ok(snapshot) => snapshot,
511            Err(err) => return Ok(ToolOutput::error(err.to_string())),
512        };
513        let history = match engine.history(&run_id).await {
514            Ok(history) => history,
515            Err(err) => return Ok(ToolOutput::error(err.to_string())),
516        };
517
518        let output = match &snapshot.output {
519            Some(output) => {
520                serde_json::to_string_pretty(output).unwrap_or_else(|_| output.to_string())
521            }
522            None => snapshot
523                .error
524                .clone()
525                .unwrap_or_else(|| format!("workflow status: {:?}", snapshot.status)),
526        };
527
528        let status = snapshot.status;
529        let metadata = json!({
530            "dynamic_workflow": {
531                "run_id": run_id,
532                "status": format!("{:?}", snapshot.status),
533                "last_sequence": snapshot.last_sequence,
534                "source_hash": source_hash,
535                "snapshot": snapshot,
536                "history": history,
537            }
538        });
539        let output = match status {
540            WorkflowRunStatus::Completed => ToolOutput::success(output),
541            WorkflowRunStatus::Failed | WorkflowRunStatus::Cancelled => ToolOutput::error(output),
542            _ => ToolOutput::error(format!(
543                "dynamic_workflow ended without a terminal result: {status:?}; {output}"
544            )),
545        };
546
547        Ok(output.with_metadata(metadata))
548    }
549}
550
551/// Drive short, persisted step retries inside the originating tool call.
552///
553/// A3S Flow deliberately suspends at a delayed retry boundary. Interactive
554/// waits and hooks must remain suspended for an external host, but a bounded
555/// retry delay is ordinary fault recovery: returning it as a terminal tool
556/// error forces every caller to reimplement the scheduler and previously made
557/// DeepResearch abandon its event-sourced run. Retry attempts and their delay
558/// remain authoritative in the Flow journal; this helper only waits for the
559/// due time and asks the engine to replay the same run.
560async fn drive_inline_retries(
561    engine: &FlowEngine,
562    run_id: &str,
563    ctx: &ToolContext,
564) -> Result<WorkflowRunSnapshot> {
565    for _ in 0..MAX_INLINE_RETRY_RESUMES {
566        let snapshot = engine.snapshot(run_id).await?;
567        if snapshot.status.is_terminal() {
568            return Ok(snapshot);
569        }
570        let Some(retry_after) = snapshot
571            .steps
572            .values()
573            .filter(|step| step.status == StepStatus::Pending)
574            .filter_map(|step| step.retry_after)
575            .min()
576        else {
577            return Ok(snapshot);
578        };
579        let delay = retry_after
580            .signed_duration_since(Utc::now())
581            .to_std()
582            .unwrap_or_default();
583        if delay > MAX_INLINE_RETRY_DELAY {
584            return Ok(snapshot);
585        }
586        let cancellation = ctx.cancellation_token();
587        tokio::select! {
588            biased;
589            _ = cancellation.cancelled() => {
590                anyhow::bail!("dynamic_workflow cancelled while waiting for a scheduled retry");
591            }
592            _ = tokio::time::sleep(delay) => {}
593        }
594        engine.drive(run_id).await?;
595    }
596    engine.snapshot(run_id).await.map_err(Into::into)
597}
598
599pub fn register_dynamic_workflow(registry: &Arc<ToolRegistry>) {
600    registry.register(Arc::new(DynamicWorkflowTool::new(Arc::clone(registry))));
601}
602
603async fn flow_store_for_context(
604    ctx: &ToolContext,
605    requested_run_id: Option<&str>,
606) -> Result<Arc<dyn FlowEventStore>> {
607    match ctx.workspace_services.local_root() {
608        Some(root) => {
609            let store = dynamic_workflow_store_path(root);
610            validate_dynamic_workflow_directory(&root.join(".a3s"), ".a3s").await?;
611            validate_dynamic_workflow_directory(&store, ".a3s/workflow").await?;
612            if let Some(run_id) = requested_run_id.filter(|run_id| safe_workflow_run_id(run_id)) {
613                validate_dynamic_workflow_log(&store.join(format!("{run_id}.jsonl"))).await?;
614            }
615            Ok(Arc::new(LocalFileEventStore::new(store)))
616        }
617        None => Ok(Arc::new(InMemoryEventStore::new())),
618    }
619}
620
621async fn validate_dynamic_workflow_directory(path: &Path, label: &str) -> Result<()> {
622    match tokio::fs::symlink_metadata(path).await {
623        Ok(metadata) if metadata.file_type().is_symlink() => {
624            anyhow::bail!("refusing to use symlinked dynamic workflow directory {label}")
625        }
626        Ok(metadata) if !metadata.is_dir() => {
627            anyhow::bail!("dynamic workflow path {label} exists but is not a directory")
628        }
629        Ok(_) => Ok(()),
630        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
631        Err(error) => Err(error).with_context(|| format!("inspect dynamic workflow path {label}")),
632    }
633}
634
635async fn validate_dynamic_workflow_log(path: &Path) -> Result<()> {
636    match tokio::fs::symlink_metadata(path).await {
637        Ok(metadata) if metadata.file_type().is_symlink() => anyhow::bail!(
638            "refusing to read or append symlinked dynamic workflow history {}",
639            path.display()
640        ),
641        Ok(metadata) if !metadata.is_file() => anyhow::bail!(
642            "dynamic workflow history path {} exists but is not a file",
643            path.display()
644        ),
645        Ok(_) => Ok(()),
646        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
647        Err(error) => Err(error)
648            .with_context(|| format!("inspect dynamic workflow history {}", path.display())),
649    }
650}
651
652fn safe_workflow_run_id(run_id: &str) -> bool {
653    !run_id.is_empty()
654        && run_id
655            .chars()
656            .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
657}
658
659struct PayloadBuilder {
660    value: Map<String, Value>,
661}
662
663impl PayloadBuilder {
664    fn with(mut self, key: &str, value: impl Serialize) -> Self {
665        self.value.insert(
666            key.to_string(),
667            serde_json::to_value(value).unwrap_or(Value::Null),
668        );
669        self
670    }
671
672    fn into_value(self) -> Value {
673        Value::Object(self.value)
674    }
675}
676
677fn invocation_payload(kind: &str, run_id: &str, history: &[FlowEventEnvelope]) -> PayloadBuilder {
678    let mut value = Map::new();
679    value.insert("kind".to_string(), json!(kind));
680    value.insert("run_id".to_string(), json!(run_id));
681    value.insert("history".to_string(), json!(history));
682    value.insert("step_outputs".to_string(), completed_step_outputs(history));
683    value.insert("step_failures".to_string(), failed_step_outputs(history));
684    PayloadBuilder { value }
685}
686
687fn completed_step_outputs(history: &[FlowEventEnvelope]) -> Value {
688    let mut outputs = Map::new();
689    for envelope in history {
690        if let FlowEvent::StepCompleted { step_id, output } = &envelope.event {
691            outputs.insert(step_id.clone(), output.clone());
692        }
693    }
694    Value::Object(outputs)
695}
696
697fn failed_step_outputs(history: &[FlowEventEnvelope]) -> Value {
698    let mut outputs = Map::new();
699    for envelope in history {
700        if let FlowEvent::StepFailed {
701            step_id,
702            attempt,
703            error,
704        } = &envelope.event
705        {
706            outputs.insert(
707                step_id.clone(),
708                json!({
709                    "attempt": attempt,
710                    "error": error,
711                }),
712            );
713        }
714    }
715    Value::Object(outputs)
716}
717
718fn script_result(result: &ToolResult) -> a3s_flow::Result<Value> {
719    result
720        .metadata
721        .as_ref()
722        .and_then(|metadata| metadata.get("script_result"))
723        .cloned()
724        .ok_or_else(|| {
725            a3s_flow::FlowError::Runtime(
726                "PTC program result did not include script_result metadata".to_string(),
727            )
728        })
729}
730
731fn default_allowed_tools(registry: &ToolRegistry) -> Vec<String> {
732    sanitize_allowed_tools(registry.list())
733}
734
735fn sanitize_allowed_tools(items: impl IntoIterator<Item = String>) -> Vec<String> {
736    let mut tools = items.into_iter().collect::<BTreeSet<_>>();
737    tools.remove(PROGRAM_TOOL);
738    tools.remove(DYNAMIC_WORKFLOW_TOOL);
739    tools.remove(PARALLEL_TASK_TOOL);
740    tools.into_iter().collect()
741}
742
743fn source_hash(source: &str) -> String {
744    sha256::digest(source.as_bytes())
745}
746
747#[cfg(test)]
748#[path = "dynamic_workflow/tests.rs"]
749mod tests;
a3s_code_core/dynamic_workflow.rs

a3s_code_core/
dynamic_workflow.rs
