use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::fs;
use std::path::Path;
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::workflow_projection::{event_seq_from_id, read_run_events_pure};
use crate::{RunChatBinding, WorkflowEventEnvelope, WorkflowOutputRef};
const BLOB_PREVIEW_MAX_BYTES: usize = 64 * 1024;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct BlobPreviewDTO {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub output_hash: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub output_bytes: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub content_type: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub truncated: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub value: Option<Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub text: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub redacted: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AttemptTerminalDTO {
pub session_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cli_session_id: Option<String>,
pub web_port: u16,
pub status: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub lark_app_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub bot_name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cli_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_dir: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub log_path: Option<String>,
pub started_at: u64,
pub updated_at: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub closed_at: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub has_terminal_log: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AttemptIODTO {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub input: Option<BlobPreviewDTO>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resolved_input: Option<BlobPreviewDTO>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub output: Option<BlobPreviewDTO>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub log: Option<BlobPreviewDTO>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub terminal: Option<AttemptTerminalDTO>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wait_prompt: Option<BlobPreviewDTO>,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum RunStatus {
Pending,
Running,
Waiting,
Succeeded,
Failed,
Cancelled,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum NodeStatus {
Idle,
Triggered,
Running,
Waiting,
Retrying,
Succeeded,
Failed,
Skipped,
Cancelled,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum ActivityStatus {
Pending,
Acquired,
Running,
Waiting,
EffectAttempting,
Succeeded,
Failed,
TimedOut,
Cancelled,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum LoopIterationStatus {
Running,
Approved,
Rejected,
Failed,
Cancelled,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum LoopStatus {
Running,
Succeeded,
Failed,
Cancelled,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct EffectAttemptedState {
pub idempotency_key: String,
pub input_hash: String,
pub idempotency_ttl_ms: u64,
pub provider: String,
pub attempted_at_event_id: String,
pub attempted_at_ms: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReconcileResultState {
pub decision: String,
pub capability: String,
pub evidence: Value,
pub event_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct WaitResolutionState {
pub kind: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resolution: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub by: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub comment: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub event_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub deadline_at: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub exceeded_at_ms: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct WaitState {
pub wait_kind: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub deadline_at: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub prompt: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub prompt_ref: Option<WorkflowOutputRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub prompt_preview: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub approvers: Option<Vec<String>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub on_timeout: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resolution: Option<WaitResolutionState>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct CancelRequestState {
pub cancel_origin_event_id: String,
pub requested_by: String,
pub reason: String,
pub delivered: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AttemptState {
pub attempt_id: String,
pub attempt_number: u64,
pub input_ref: WorkflowOutputRef,
pub status: ActivityStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub lease_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timeout_ms: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_output_bytes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub effect_attempted: Option<EffectAttemptedState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub latest_reconcile_result: Option<ReconcileResultState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cancel_request: Option<CancelRequestState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wait: Option<WaitState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub output: Option<WorkflowOutputRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub external_refs: Option<Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub running_ms: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cancel_origin_event_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ActivityState {
pub activity_id: String,
pub attempts: Vec<AttemptState>,
pub status: ActivityStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub current_attempt_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub owner_node_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct NodeState {
pub node_id: String,
pub status: NodeStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub activity_id: Option<String>,
pub retry_count: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_attempt_at: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error_class: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub condition_event_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cancel_origin_event_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct LoopIterationState {
pub iteration: u64,
pub status: LoopIterationStatus,
pub body_activity_ids: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub decision_activity_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wait_resolved_event_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub decision_by: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub decision_comment: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timed_out: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct LoopState {
pub loop_id: String,
pub status: LoopStatus,
pub iteration: u64,
pub max_iterations: u64,
pub iterations: Vec<LoopIterationState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub output: Option<WorkflowOutputRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error_code: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error_class: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct RunState {
pub run_id: String,
pub status: RunStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workflow_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub revision_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub initiator: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub input: Option<WorkflowOutputRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub output: Option<WorkflowOutputRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub failed_node_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub root_cause_event_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cancel_origin_event_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub bot_snapshots: Option<BTreeMap<String, BotSnapshot>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cancelled_run_intent: Option<CancelIntent>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub cancelled_node_intents: BTreeMap<String, CancelIntent>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct CancelIntent {
pub cancel_origin_event_id: String,
pub requested_by: String,
pub reason: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct BotSnapshot {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub lark_app_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cli_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub display_name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_dir: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct LoopSnapshotDTO {
pub loop_id: String,
pub status: LoopStatus,
pub iteration: u64,
pub max_iterations: u64,
pub iterations: Vec<LoopIterationState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub output: Option<WorkflowOutputRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error_code: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error_class: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct RunSnapshotDTO {
pub run_id: String,
pub run: RunState,
pub last_seq: u64,
pub nodes: Vec<NodeState>,
pub activities: Vec<ActivityState>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub loops: Option<BTreeMap<String, LoopSnapshotDTO>>,
pub dangling: DanglingSnapshot,
pub outputs: BTreeMap<String, WorkflowOutputRef>,
pub attempt_io: BTreeMap<String, AttemptIODTO>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub chat_binding: Option<RunChatBinding>,
pub updated_at: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct DanglingSnapshot {
pub activities: Vec<String>,
pub effect_attempted: Vec<String>,
pub waits: Vec<String>,
pub wait_resolutions: Vec<String>,
pub cancels: Vec<String>,
}
pub async fn read_run_snapshot(run_dir: &Path) -> Result<Option<RunSnapshotDTO>> {
let events = match read_run_events_pure(run_dir)? {
Some(events) if !events.is_empty() => events,
_ => return Ok(None),
};
let snap = replay_events(&events)?;
let binding = read_chat_binding_pure(run_dir).await?;
let def = read_workflow_definition_pure(run_dir).await?;
let attempt_io = build_attempt_io(run_dir, &snap, def.as_ref())?;
Ok(Some(RunSnapshotDTO {
run_id: snap.run.run_id.clone(),
run: snap.run,
last_seq: snap.last_seq,
nodes: snap.nodes.into_values().collect(),
activities: snap.activities.into_values().collect(),
loops: if snap.loops.is_empty() {
None
} else {
Some(
snap.loops
.into_iter()
.map(|(key, value)| (key, value))
.collect(),
)
},
dangling: DanglingSnapshot {
activities: snap.dangling_activities,
effect_attempted: snap.dangling_effect_attempted,
waits: snap.dangling_waits,
wait_resolutions: snap.dangling_wait_resolutions,
cancels: snap.dangling_cancels,
},
outputs: snap.outputs,
attempt_io,
chat_binding: binding,
updated_at: events.last().map(|ev| ev.timestamp).unwrap_or_default(),
}))
}
#[derive(Debug, Clone)]
struct ReplaySnapshot {
run: RunState,
nodes: BTreeMap<String, NodeState>,
activities: BTreeMap<String, ActivityState>,
loops: BTreeMap<String, LoopSnapshotDTO>,
outputs: BTreeMap<String, WorkflowOutputRef>,
last_seq: u64,
dangling_activities: Vec<String>,
dangling_effect_attempted: Vec<String>,
dangling_waits: Vec<String>,
dangling_wait_resolutions: Vec<String>,
dangling_cancels: Vec<String>,
}
fn replay_events(events: &[WorkflowEventEnvelope]) -> Result<ReplaySnapshot> {
let first = events.first().context("replay: empty event log")?;
if first.event_type != "runCreated" {
anyhow::bail!(
"replay: first event must be runCreated, got {}",
first.event_type
);
}
let mut snap = ReplaySnapshot {
run: RunState {
run_id: first.run_id.clone(),
status: RunStatus::Pending,
workflow_id: None,
revision_id: None,
initiator: None,
input: None,
output: None,
failed_node_id: None,
root_cause_event_id: None,
cancel_origin_event_id: None,
bot_snapshots: None,
cancelled_run_intent: None,
cancelled_node_intents: BTreeMap::new(),
},
nodes: BTreeMap::new(),
activities: BTreeMap::new(),
loops: BTreeMap::new(),
outputs: BTreeMap::new(),
last_seq: 0,
dangling_activities: Vec::new(),
dangling_effect_attempted: Vec::new(),
dangling_waits: Vec::new(),
dangling_wait_resolutions: Vec::new(),
dangling_cancels: Vec::new(),
};
let mut waits_open = BTreeSet::new();
let mut run_cancel_intent: Option<(String, String, String)> = None;
let mut node_cancel_intents: BTreeMap<String, (String, String, String)> = BTreeMap::new();
for ev in events {
if ev.run_id != snap.run.run_id {
anyhow::bail!(
"replay: runId mismatch at {} — log is {}, event has {}",
ev.event_id,
snap.run.run_id,
ev.run_id
);
}
snap.last_seq = snap.last_seq.max(event_seq_from_id(&ev.event_id));
if is_payload_ref(&ev.payload) {
continue;
}
let payload = &ev.payload;
match ev.event_type.as_str() {
"runCreated" => apply_run_created(&mut snap.run, payload),
"runStarted" => snap.run.status = RunStatus::Running,
"runSucceeded" => apply_run_succeeded(&mut snap.run, payload),
"runFailed" => apply_run_failed(&mut snap.run, payload),
"runCanceled" => apply_run_canceled(&mut snap.run, payload),
"nodeWaiting" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
get_node(&mut snap.nodes, &node_id).status = NodeStatus::Waiting;
}
}
"nodeRetrying" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
let node = get_node(&mut snap.nodes, &node_id);
node.status = NodeStatus::Retrying;
node.retry_count = node.retry_count.saturating_add(1);
}
}
"nodeSucceeded" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
let node = get_node(&mut snap.nodes, &node_id);
node.status = NodeStatus::Succeeded;
node.activity_id = payload_str(payload, "lastActivityId");
}
}
"nodeFailed" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
let node = get_node(&mut snap.nodes, &node_id);
node.status = NodeStatus::Failed;
node.activity_id = payload_str(payload, "lastActivityId");
node.error_class = payload_str(payload, "errorClass");
}
}
"nodeSkipped" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
let node = get_node(&mut snap.nodes, &node_id);
node.status = NodeStatus::Skipped;
node.condition_event_id = Some(ev.event_id.clone());
}
}
"nodeCanceled" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
let node = get_node(&mut snap.nodes, &node_id);
node.status = NodeStatus::Cancelled;
node.cancel_origin_event_id = payload_str(payload, "cancelOriginEventId");
}
}
"loopStarted" => {
if let Some(loop_id) = payload_str(payload, "loopId") {
let loop_state = get_loop(&mut snap.loops, &loop_id);
loop_state.status = LoopStatus::Running;
loop_state.max_iterations = payload_u64(payload, "maxIterations").unwrap_or(0);
}
}
"loopIterationStarted" => {
if let Some(loop_id) = payload_str(payload, "loopId") {
let iteration = payload_u64(payload, "iteration").unwrap_or(0);
let loop_state = get_loop(&mut snap.loops, &loop_id);
loop_state.status = LoopStatus::Running;
loop_state.iteration = iteration;
let it = get_loop_iteration(loop_state, iteration);
it.status = LoopIterationStatus::Running;
}
}
"loopIterationFinished" => {
if let Some(loop_id) = payload_str(payload, "loopId") {
let iteration = payload_u64(payload, "iteration").unwrap_or(0);
let loop_state = get_loop(&mut snap.loops, &loop_id);
loop_state.iteration = loop_state.iteration.max(iteration);
let it = get_loop_iteration(loop_state, iteration);
it.status = match payload_str(payload, "resolution")
.as_deref()
.unwrap_or("failed")
{
"approved" => LoopIterationStatus::Approved,
"rejected" => LoopIterationStatus::Rejected,
"cancelled" => LoopIterationStatus::Cancelled,
_ => LoopIterationStatus::Failed,
};
it.decision_activity_id = payload_str(payload, "decisionActivityId");
it.wait_resolved_event_id = payload_str(payload, "waitResolvedEventId");
it.decision_by = payload_str(payload, "by");
it.decision_comment = payload_str(payload, "comment");
it.timed_out = payload_bool(payload, "timedOut");
}
}
"loopFinished" => {
if let Some(loop_id) = payload_str(payload, "loopId") {
let loop_state = get_loop(&mut snap.loops, &loop_id);
loop_state.iteration = payload_u64(payload, "finalIteration").unwrap_or(0);
loop_state.status = match payload_str(payload, "resolution")
.as_deref()
.unwrap_or("failed")
{
"approved" => LoopStatus::Succeeded,
"cancelled" => LoopStatus::Cancelled,
_ => LoopStatus::Failed,
};
loop_state.output = payload_workflow_output_ref(payload, "outputRef");
loop_state.error_code = payload_str(payload, "errorCode");
loop_state.error_class = payload_str(payload, "errorClass");
if let Some(output_ref) = loop_state.output.clone() {
snap.outputs
.insert(format!("{}::work::{}", ev.run_id, loop_id), output_ref);
}
if loop_state.status != LoopStatus::Succeeded {
if let Some(inflight) = loop_state
.iterations
.iter_mut()
.find(|it| matches!(it.status, LoopIterationStatus::Running))
{
inflight.status = if loop_state.status == LoopStatus::Cancelled {
LoopIterationStatus::Cancelled
} else {
LoopIterationStatus::Failed
};
}
}
}
}
"conditionEvaluated" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
let node = get_node(&mut snap.nodes, &node_id);
node.condition_event_id = Some(ev.event_id.clone());
}
}
"attemptCreated" => {
if let (Some(activity_id), Some(node_id), Some(attempt_id)) = (
payload_str(payload, "activityId"),
payload_str(payload, "nodeId"),
payload_str(payload, "attemptId"),
) {
let attempt_number = payload_u64(payload, "attemptNumber").unwrap_or(0);
let input_ref = match payload_workflow_output_ref(payload, "inputRef") {
Some(value) => value,
None => continue,
};
let activity = get_activity(&mut snap.activities, &activity_id);
activity.attempts.push(AttemptState {
attempt_id: attempt_id.clone(),
attempt_number,
input_ref,
status: ActivityStatus::Pending,
lease_id: None,
timeout_ms: None,
max_output_bytes: None,
effect_attempted: None,
latest_reconcile_result: None,
cancel_request: None,
wait: None,
output: None,
external_refs: None,
error: None,
running_ms: None,
cancel_origin_event_id: None,
});
activity.current_attempt_id = Some(attempt_id.clone());
activity.status = ActivityStatus::Pending;
activity.owner_node_id = Some(node_id.clone());
let node = get_node(&mut snap.nodes, &node_id);
node.activity_id = Some(activity_id.clone());
if attempt_number == 1 && matches!(node.status, NodeStatus::Idle) {
node.status = NodeStatus::Triggered;
}
if let Some((loop_id, iteration)) = parse_loop_activity_id(&activity_id) {
if let Some(loop_state) = snap.loops.get_mut(&loop_id) {
let it = get_loop_iteration(loop_state, iteration);
if !it.body_activity_ids.contains(&activity_id) {
it.body_activity_ids.push(activity_id.clone());
}
}
}
}
}
"leaseSigned" => {
if let (Some(activity_id), Some(attempt_id)) = (
payload_str(payload, "activityId"),
payload_str(payload, "attemptId"),
) {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.lease_id = payload_str(payload, "leaseId");
attempt.timeout_ms = payload_u64(payload, "timeoutMs");
attempt.max_output_bytes = payload_u64(payload, "maxOutputBytes");
}
}
}
"backoffScheduled" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
get_node(&mut snap.nodes, &node_id).next_attempt_at =
payload_u64(payload, "nextAttemptAt");
}
}
"backoffElapsed" => {
if let Some(node_id) = payload_str(payload, "nodeId") {
get_node(&mut snap.nodes, &node_id).next_attempt_at = None;
}
}
"effectAttempted" => {
if let (Some(activity_id), Some(attempt_id)) = (
payload_str(payload, "activityId"),
payload_str(payload, "attemptId"),
) {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.effect_attempted = Some(EffectAttemptedState {
idempotency_key: payload_str(payload, "idempotencyKey")
.unwrap_or_default(),
input_hash: payload_str(payload, "inputHash").unwrap_or_default(),
idempotency_ttl_ms: payload_u64(payload, "idempotencyTtlMs")
.unwrap_or_default(),
provider: payload_str(payload, "provider").unwrap_or_default(),
attempted_at_event_id: ev.event_id.clone(),
attempted_at_ms: ev.timestamp,
});
attempt.status = ActivityStatus::EffectAttempting;
if let Some(activity) = snap.activities.get_mut(&activity_id) {
activity.status = ActivityStatus::EffectAttempting;
}
}
}
}
"activitySucceeded" => {
if let (Some(activity_id), Some(attempt_id)) = (
payload_str(payload, "activityId"),
payload_str(payload, "attemptId"),
) {
let output_ref = if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.status = ActivityStatus::Succeeded;
attempt.output = payload_workflow_output_ref(payload, "outputRef");
attempt.external_refs = payload.get("externalRefs").cloned();
attempt.output.clone()
} else {
None
};
if let Some(activity) = snap.activities.get_mut(&activity_id) {
activity.status = ActivityStatus::Succeeded;
}
if let Some(output_ref) = output_ref {
snap.outputs.insert(activity_id.clone(), output_ref);
}
waits_open.remove(&activity_id);
}
}
"activityFailed" => {
if let (Some(activity_id), Some(attempt_id)) = (
payload_str(payload, "activityId"),
payload_str(payload, "attemptId"),
) {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.status = ActivityStatus::Failed;
attempt.error = payload.get("error").cloned();
if let Some(activity) = snap.activities.get_mut(&activity_id) {
activity.status = ActivityStatus::Failed;
}
waits_open.remove(&activity_id);
}
}
}
"activityTimedOut" => {
if let (Some(activity_id), Some(attempt_id)) = (
payload_str(payload, "activityId"),
payload_str(payload, "attemptId"),
) {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.status = ActivityStatus::TimedOut;
attempt.running_ms = payload_u64(payload, "runningMs");
if let Some(activity) = snap.activities.get_mut(&activity_id) {
activity.status = ActivityStatus::TimedOut;
}
waits_open.remove(&activity_id);
}
}
}
"activityRunning" => {
if let (Some(activity_id), Some(attempt_id)) = (
payload_str(payload, "activityId"),
payload_str(payload, "attemptId"),
) {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.status = ActivityStatus::Running;
}
let owner_node_id = snap
.activities
.get(&activity_id)
.and_then(|activity| activity.owner_node_id.clone());
if let Some(activity) = snap.activities.get_mut(&activity_id) {
activity.status = ActivityStatus::Running;
}
if let Some(owner) = owner_node_id {
let node = get_node(&mut snap.nodes, &owner);
if matches!(node.status, NodeStatus::Triggered | NodeStatus::Retrying) {
node.status = NodeStatus::Running;
}
}
}
}
"activityWaiting" => {
if let Some(activity_id) = payload_str(payload, "activityId") {
let attempt_id = snap
.activities
.get(&activity_id)
.and_then(|activity| activity.current_attempt_id.clone());
if let Some(attempt_id) = attempt_id {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.status = ActivityStatus::Waiting;
}
if let Some(activity) = snap.activities.get_mut(&activity_id) {
activity.status = ActivityStatus::Waiting;
}
}
}
}
"activityCanceled" => {
if let (Some(activity_id), Some(attempt_id)) = (
payload_str(payload, "activityId"),
payload_str(payload, "attemptId"),
) {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.status = ActivityStatus::Cancelled;
attempt.cancel_origin_event_id =
payload_str(payload, "cancelOriginEventId");
if let Some(activity) = snap.activities.get_mut(&activity_id) {
activity.status = ActivityStatus::Cancelled;
}
waits_open.remove(&activity_id);
}
}
}
"waitCreated" => {
if let Some(activity_id) = payload_str(payload, "activityId") {
waits_open.insert(activity_id.clone());
let attempt_id = snap
.activities
.get(&activity_id)
.and_then(|activity| activity.current_attempt_id.clone());
if let Some(attempt_id) = attempt_id {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
attempt.wait = Some(WaitState {
wait_kind: payload_str(payload, "waitKind").unwrap_or_default(),
deadline_at: payload_u64(payload, "deadlineAt"),
prompt: payload_str(payload, "prompt"),
prompt_ref: payload_workflow_output_ref(payload, "promptRef"),
prompt_preview: payload_str(payload, "promptPreview"),
approvers: payload_string_array(payload, "approvers"),
on_timeout: payload_str(payload, "onTimeout"),
resolution: None,
});
}
}
}
}
"waitResolved" => {
if let Some(activity_id) = payload_str(payload, "activityId") {
waits_open.remove(&activity_id);
let attempt_id = snap
.activities
.get(&activity_id)
.and_then(|activity| activity.current_attempt_id.clone());
if let Some(attempt_id) = attempt_id {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
if let Some(wait) = attempt.wait.as_mut() {
wait.resolution = Some(WaitResolutionState {
kind: "resolved".to_string(),
resolution: payload_str(payload, "resolution"),
by: payload_str(payload, "by"),
comment: payload_str(payload, "comment"),
event_id: Some(ev.event_id.clone()),
deadline_at: None,
exceeded_at_ms: None,
});
}
}
}
}
}
"waitDeadlineExceeded" => {
if let Some(activity_id) = payload_str(payload, "activityId") {
waits_open.remove(&activity_id);
let attempt_id = snap
.activities
.get(&activity_id)
.and_then(|activity| activity.current_attempt_id.clone());
if let Some(attempt_id) = attempt_id {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
if let Some(wait) = attempt.wait.as_mut() {
wait.resolution = Some(WaitResolutionState {
kind: "deadlineExceeded".to_string(),
resolution: None,
by: None,
comment: None,
event_id: Some(ev.event_id.clone()),
deadline_at: payload_u64(payload, "deadlineAt"),
exceeded_at_ms: payload_u64(payload, "exceededAtMs"),
});
}
}
}
}
}
"cancelRequested" => {
if let Some(target) = payload.get("target") {
if let Some(kind) = target.get("kind").and_then(Value::as_str) {
match kind {
"activity" => {
if let Some(activity_id) =
target.get("activityId").and_then(Value::as_str)
{
mark_activity_cancel(
&mut snap.activities,
activity_id,
&ev,
payload,
);
}
}
"node" => {
if let Some(node_id) = target.get("nodeId").and_then(Value::as_str)
{
let node_id = node_id.to_string();
node_cancel_intents.entry(node_id.clone()).or_insert_with(
|| {
(
ev.event_id.clone(),
payload_str(payload, "by").unwrap_or_default(),
payload_str(payload, "reason").unwrap_or_default(),
)
},
);
for activity in snap.activities.values_mut() {
if activity.owner_node_id.as_deref()
== Some(node_id.as_str())
{
mark_attempt_cancel(activity, &ev, payload);
}
}
}
}
"run" => {
if run_cancel_intent.is_none() {
run_cancel_intent = Some((
ev.event_id.clone(),
payload_str(payload, "by").unwrap_or_default(),
payload_str(payload, "reason").unwrap_or_default(),
));
}
for activity in snap.activities.values_mut() {
mark_attempt_cancel(activity, &ev, payload);
}
}
_ => {}
}
}
}
}
"cancelDelivered" => {
if let Some(activity_id) = payload_str(payload, "activityId") {
let attempt_id = snap
.activities
.get(&activity_id)
.and_then(|activity| activity.current_attempt_id.clone());
if let Some(attempt_id) = attempt_id {
if let Some(attempt) =
get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
{
if let Some(cancel) = attempt.cancel_request.as_mut() {
cancel.delivered = true;
}
}
}
}
}
"workerLost" | "resumeStarted" => {}
"reconcileResult" => {
if let Some(idempotency_key) = payload_str(payload, "idempotencyKey") {
for activity in snap.activities.values_mut() {
if let Some(attempt) = activity.attempts.iter_mut().find(|candidate| {
candidate
.effect_attempted
.as_ref()
.map(|x| x.idempotency_key.as_str())
== Some(idempotency_key.as_str())
}) {
attempt.latest_reconcile_result = Some(ReconcileResultState {
decision: payload_str(payload, "decision").unwrap_or_default(),
capability: payload_str(payload, "capability").unwrap_or_default(),
evidence: payload.get("evidence").cloned().unwrap_or(Value::Null),
event_id: ev.event_id.clone(),
});
break;
}
}
}
}
_ => {}
}
}
let mut dangling_activities = Vec::new();
let mut dangling_effect_attempted = Vec::new();
let mut dangling_wait_resolutions = Vec::new();
let mut dangling_cancels = Vec::new();
for activity in snap.activities.values() {
let Some(latest) = activity.attempts.last() else {
continue;
};
let is_terminal = matches!(
latest.status,
ActivityStatus::Succeeded
| ActivityStatus::Failed
| ActivityStatus::TimedOut
| ActivityStatus::Cancelled
);
if !is_terminal {
dangling_activities.push(activity.activity_id.clone());
if latest.effect_attempted.is_some() {
dangling_effect_attempted.push(activity.activity_id.clone());
}
if latest
.wait
.as_ref()
.and_then(|w| w.resolution.as_ref())
.is_some()
{
dangling_wait_resolutions.push(activity.activity_id.clone());
}
if latest.cancel_request.is_some() {
dangling_cancels.push(activity.activity_id.clone());
}
}
}
dangling_activities.sort();
dangling_effect_attempted.sort();
dangling_wait_resolutions.sort();
dangling_cancels.sort();
if !matches!(snap.run.status, RunStatus::Cancelled) {
if let Some((cancel_origin_event_id, requested_by, reason)) = run_cancel_intent {
snap.run.cancelled_run_intent = Some(CancelIntent {
cancel_origin_event_id,
requested_by,
reason,
});
}
}
for (node_id, intent) in node_cancel_intents {
if matches!(
snap.nodes.get(&node_id).map(|node| node.status),
Some(NodeStatus::Cancelled)
) {
continue;
}
snap.run.cancelled_node_intents.insert(
node_id,
CancelIntent {
cancel_origin_event_id: intent.0,
requested_by: intent.1,
reason: intent.2,
},
);
}
Ok(ReplaySnapshot {
run: snap.run,
nodes: snap.nodes,
activities: snap.activities,
loops: snap.loops,
outputs: snap.outputs,
last_seq: snap.last_seq,
dangling_activities,
dangling_effect_attempted,
dangling_waits: waits_open.into_iter().collect(),
dangling_wait_resolutions,
dangling_cancels,
})
}
fn get_node<'a>(nodes: &'a mut BTreeMap<String, NodeState>, node_id: &str) -> &'a mut NodeState {
nodes
.entry(node_id.to_string())
.or_insert_with(|| NodeState {
node_id: node_id.to_string(),
status: NodeStatus::Idle,
activity_id: None,
retry_count: 0,
next_attempt_at: None,
error_class: None,
condition_event_id: None,
cancel_origin_event_id: None,
})
}
fn get_activity<'a>(
activities: &'a mut BTreeMap<String, ActivityState>,
activity_id: &str,
) -> &'a mut ActivityState {
activities
.entry(activity_id.to_string())
.or_insert_with(|| ActivityState {
activity_id: activity_id.to_string(),
attempts: Vec::new(),
status: ActivityStatus::Pending,
current_attempt_id: None,
owner_node_id: None,
})
}
fn get_attempt_mut<'a>(
activities: &'a mut BTreeMap<String, ActivityState>,
activity_id: &str,
attempt_id: &str,
) -> Option<&'a mut AttemptState> {
activities
.get_mut(activity_id)?
.attempts
.iter_mut()
.find(|attempt| attempt.attempt_id == attempt_id)
}
fn get_loop<'a>(
loops: &'a mut BTreeMap<String, LoopSnapshotDTO>,
loop_id: &str,
) -> &'a mut LoopSnapshotDTO {
loops
.entry(loop_id.to_string())
.or_insert_with(|| LoopSnapshotDTO {
loop_id: loop_id.to_string(),
status: LoopStatus::Running,
iteration: 0,
max_iterations: 0,
iterations: Vec::new(),
output: None,
error_code: None,
error_class: None,
})
}
fn get_loop_iteration<'a>(
loop_state: &'a mut LoopSnapshotDTO,
iteration: u64,
) -> &'a mut LoopIterationState {
if let Some(idx) = loop_state
.iterations
.iter()
.position(|candidate| candidate.iteration == iteration)
{
return &mut loop_state.iterations[idx];
}
loop_state.iterations.push(LoopIterationState {
iteration,
status: LoopIterationStatus::Running,
body_activity_ids: Vec::new(),
decision_activity_id: None,
wait_resolved_event_id: None,
decision_by: None,
decision_comment: None,
timed_out: None,
});
let idx = loop_state.iterations.len() - 1;
&mut loop_state.iterations[idx]
}
fn apply_run_created(run: &mut RunState, payload: &Value) {
run.workflow_id = payload_str(payload, "workflowId");
run.revision_id = payload_str(payload, "revisionId");
run.initiator = payload_str(payload, "initiator");
run.input = payload_workflow_output_ref(payload, "inputRef");
if let Some(bot_snapshots) = payload.get("botSnapshots").and_then(|value| {
serde_json::from_value::<BTreeMap<String, BotSnapshot>>(value.clone()).ok()
}) {
run.bot_snapshots = Some(bot_snapshots);
}
}
fn apply_run_succeeded(run: &mut RunState, payload: &Value) {
run.status = RunStatus::Succeeded;
run.output = payload_workflow_output_ref(payload, "outputRef");
}
fn apply_run_failed(run: &mut RunState, payload: &Value) {
run.status = RunStatus::Failed;
run.failed_node_id = payload_str(payload, "failedNodeId");
run.root_cause_event_id = payload_str(payload, "rootCauseEventId");
}
fn apply_run_canceled(run: &mut RunState, payload: &Value) {
run.status = RunStatus::Cancelled;
run.cancel_origin_event_id = payload_str(payload, "cancelOriginEventId");
}
fn mark_attempt_cancel(activity: &mut ActivityState, ev: &WorkflowEventEnvelope, payload: &Value) {
let Some(attempt_id) = activity.current_attempt_id.clone() else {
return;
};
let Some(attempt) = activity
.attempts
.iter_mut()
.find(|attempt| attempt.attempt_id == attempt_id)
else {
return;
};
let is_terminal = matches!(
attempt.status,
ActivityStatus::Succeeded
| ActivityStatus::Failed
| ActivityStatus::TimedOut
| ActivityStatus::Cancelled
);
if is_terminal || attempt.cancel_request.is_some() {
return;
}
attempt.cancel_request = Some(CancelRequestState {
cancel_origin_event_id: ev.event_id.clone(),
requested_by: payload_str(payload, "by").unwrap_or_default(),
reason: payload_str(payload, "reason").unwrap_or_default(),
delivered: false,
});
}
fn mark_activity_cancel(
activities: &mut BTreeMap<String, ActivityState>,
activity_id: &str,
ev: &WorkflowEventEnvelope,
payload: &Value,
) {
if let Some(activity) = activities.get_mut(activity_id) {
mark_attempt_cancel(activity, ev, payload);
}
}
fn payload_str(payload: &Value, key: &str) -> Option<String> {
payload
.get(key)
.and_then(Value::as_str)
.map(ToOwned::to_owned)
}
fn payload_u64(payload: &Value, key: &str) -> Option<u64> {
payload.get(key).and_then(Value::as_u64)
}
fn payload_bool(payload: &Value, key: &str) -> Option<bool> {
payload.get(key).and_then(Value::as_bool)
}
fn payload_string_array(payload: &Value, key: &str) -> Option<Vec<String>> {
let arr = payload.get(key)?.as_array()?;
Some(
arr.iter()
.filter_map(|item| item.as_str().map(ToOwned::to_owned))
.collect(),
)
}
fn payload_workflow_output_ref(payload: &Value, key: &str) -> Option<WorkflowOutputRef> {
payload
.get(key)
.cloned()
.and_then(|value| serde_json::from_value::<WorkflowOutputRef>(value).ok())
}
fn is_payload_ref(payload: &Value) -> bool {
let Some(obj) = payload.as_object() else {
return false;
};
obj.get("ref").and_then(Value::as_str).is_some()
}
async fn read_chat_binding_pure(run_dir: &Path) -> Result<Option<RunChatBinding>> {
let path = run_dir.join("chat-binding.json");
let raw = match fs::read_to_string(&path) {
Ok(raw) => raw,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(err) => return Err(err.into()),
};
let binding = serde_json::from_str::<RunChatBinding>(&raw)
.with_context(|| format!("failed to parse {}", path.display()))?;
Ok(Some(binding))
}
async fn read_workflow_definition_pure(run_dir: &Path) -> Result<Option<Value>> {
let path = run_dir.join("workflow.json");
let raw = match fs::read_to_string(&path) {
Ok(raw) => raw,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(err) => return Err(err.into()),
};
let def = serde_json::from_str::<Value>(&raw)
.with_context(|| format!("failed to parse {}", path.display()))?;
Ok(Some(def))
}
fn build_attempt_io(
run_dir: &Path,
snap: &ReplaySnapshot,
def: Option<&Value>,
) -> Result<BTreeMap<String, AttemptIODTO>> {
let mut out = BTreeMap::new();
let mut cache = HashMap::new();
for activity in snap.activities.values() {
for attempt in &activity.attempts {
let mut io = AttemptIODTO {
input: Some(preview_ref(run_dir, &attempt.input_ref, &mut cache)?),
resolved_input: None,
output: None,
log: preview_attempt_log(run_dir, &activity.activity_id, &attempt.attempt_id)?,
terminal: read_attempt_terminal(
run_dir,
&activity.activity_id,
&attempt.attempt_id,
)?,
wait_prompt: None,
};
if let Some(output_ref) = attempt.output.as_ref() {
io.output = Some(preview_ref(run_dir, output_ref, &mut cache)?);
}
if let Some(wait) = attempt.wait.as_ref() {
if let Some(prompt_ref) = wait.prompt_ref.as_ref() {
io.wait_prompt = Some(preview_ref(run_dir, prompt_ref, &mut cache)?);
}
}
if let Some(input_value) = io.input.as_ref().and_then(|preview| preview.value.clone()) {
if let Some(def) = def {
io.resolved_input = Some(preview_resolved_input(
run_dir,
snap,
def,
input_value,
&mut cache,
)?);
}
}
out.insert(attempt.attempt_id.clone(), io);
}
}
Ok(out)
}
fn preview_ref(
run_dir: &Path,
ref_value: &WorkflowOutputRef,
cache: &mut HashMap<String, BlobPreviewDTO>,
) -> Result<BlobPreviewDTO> {
if let Some(cached) = cache.get(&ref_value.output_hash) {
return Ok(cached.clone());
}
let base = BlobPreviewDTO {
output_hash: Some(ref_value.output_hash.clone()),
output_bytes: Some(ref_value.output_bytes),
content_type: ref_value.content_type.clone(),
truncated: None,
value: None,
text: None,
error: None,
redacted: None,
};
let Some(output_path) = Some(&ref_value.output_path) else {
return Ok(base);
};
if !is_path_inside(run_dir, Path::new(output_path)) {
let mut preview = base.clone();
preview.error = Some("outputPath is outside run directory".to_string());
cache.insert(ref_value.output_hash.clone(), preview.clone());
return Ok(preview);
}
let bytes = match fs::read(output_path) {
Ok(bytes) => bytes,
Err(err) => {
let mut preview = base.clone();
preview.error = Some(err.to_string());
cache.insert(ref_value.output_hash.clone(), preview.clone());
return Ok(preview);
}
};
let mut preview = base;
preview.output_bytes = Some(bytes.len());
preview.truncated = Some(bytes.len() > BLOB_PREVIEW_MAX_BYTES);
let slice = if bytes.len() > BLOB_PREVIEW_MAX_BYTES {
&bytes[..BLOB_PREVIEW_MAX_BYTES]
} else {
&bytes[..]
};
let text = String::from_utf8_lossy(slice).to_string();
if !preview.truncated.unwrap_or(false) && is_json_content(ref_value.content_type.as_deref()) {
match serde_json::from_slice::<Value>(slice) {
Ok(value) => preview.value = Some(value),
Err(err) => {
preview.text = Some(text.clone());
preview.error = Some(format!("invalid JSON: {}", err));
}
}
} else {
preview.text = Some(text);
}
cache.insert(ref_value.output_hash.clone(), preview.clone());
Ok(preview)
}
fn preview_attempt_log(
run_dir: &Path,
activity_id: &str,
attempt_id: &str,
) -> Result<Option<BlobPreviewDTO>> {
let path = run_dir
.join("attempts")
.join(activity_id)
.join(attempt_id)
.join("terminal.log");
if !is_path_inside(run_dir, &path) {
return Ok(Some(BlobPreviewDTO {
output_hash: None,
output_bytes: None,
content_type: Some("text/plain".to_string()),
truncated: None,
value: None,
text: None,
error: Some("attempt log is outside run directory".to_string()),
redacted: None,
}));
}
let raw = match fs::read(&path) {
Ok(raw) => raw,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(err) => {
return Ok(Some(BlobPreviewDTO {
output_hash: None,
output_bytes: None,
content_type: Some("text/plain".to_string()),
truncated: None,
value: None,
text: None,
error: Some(err.to_string()),
redacted: None,
}));
}
};
let bytes = raw.len();
let start = bytes.saturating_sub(BLOB_PREVIEW_MAX_BYTES);
let text = String::from_utf8_lossy(&raw[start..]).to_string();
Ok(Some(BlobPreviewDTO {
output_hash: None,
output_bytes: Some(bytes),
content_type: Some("text/plain".to_string()),
truncated: Some(bytes > BLOB_PREVIEW_MAX_BYTES),
value: None,
text: Some(text),
error: None,
redacted: None,
}))
}
fn read_attempt_terminal(
run_dir: &Path,
activity_id: &str,
attempt_id: &str,
) -> Result<Option<AttemptTerminalDTO>> {
let path = run_dir
.join("attempts")
.join(activity_id)
.join(attempt_id)
.join("terminal.json");
if !is_path_inside(run_dir, &path) {
return Ok(Some(AttemptTerminalDTO {
session_id: String::new(),
cli_session_id: None,
web_port: 0,
status: "closed".to_string(),
lark_app_id: None,
bot_name: None,
cli_id: None,
working_dir: None,
log_path: None,
started_at: 0,
updated_at: 0,
closed_at: None,
error: Some("terminal sidecar is outside run directory".to_string()),
has_terminal_log: None,
}));
}
let raw = match fs::read_to_string(&path) {
Ok(raw) => raw,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(err) => {
return Ok(Some(AttemptTerminalDTO {
session_id: String::new(),
cli_session_id: None,
web_port: 0,
status: "closed".to_string(),
lark_app_id: None,
bot_name: None,
cli_id: None,
working_dir: None,
log_path: None,
started_at: 0,
updated_at: 0,
closed_at: None,
error: Some(err.to_string()),
has_terminal_log: None,
}));
}
};
let parsed: Value = match serde_json::from_str(&raw) {
Ok(value) => value,
Err(err) => {
return Ok(Some(AttemptTerminalDTO {
session_id: String::new(),
cli_session_id: None,
web_port: 0,
status: "closed".to_string(),
lark_app_id: None,
bot_name: None,
cli_id: None,
working_dir: None,
log_path: None,
started_at: 0,
updated_at: 0,
closed_at: None,
error: Some(format!("invalid terminal sidecar: {}", err)),
has_terminal_log: None,
}));
}
};
if payload_u64(&parsed, "schemaVersion") != Some(1)
|| payload_str(&parsed, "sessionId").is_none()
|| payload_u64(&parsed, "webPort").is_none()
|| !matches!(
payload_str(&parsed, "status").as_deref(),
Some("live" | "closed")
)
|| payload_u64(&parsed, "startedAt").is_none()
|| payload_u64(&parsed, "updatedAt").is_none()
{
return Ok(Some(AttemptTerminalDTO {
session_id: String::new(),
cli_session_id: None,
web_port: 0,
status: "closed".to_string(),
lark_app_id: None,
bot_name: None,
cli_id: None,
working_dir: None,
log_path: None,
started_at: 0,
updated_at: 0,
closed_at: None,
error: Some("invalid terminal sidecar".to_string()),
has_terminal_log: None,
}));
}
let terminal_log = path.with_file_name("terminal.log");
let has_terminal_log = fs::metadata(&terminal_log)
.map(|meta| meta.is_file() && meta.len() > 0)
.unwrap_or(false);
Ok(Some(AttemptTerminalDTO {
session_id: payload_str(&parsed, "sessionId").unwrap_or_default(),
cli_session_id: payload_str(&parsed, "cliSessionId"),
web_port: payload_u64(&parsed, "webPort").unwrap_or_default() as u16,
status: payload_str(&parsed, "status").unwrap_or_else(|| "closed".to_string()),
lark_app_id: payload_str(&parsed, "larkAppId"),
bot_name: payload_str(&parsed, "botName"),
cli_id: payload_str(&parsed, "cliId"),
working_dir: payload_str(&parsed, "workingDir"),
log_path: payload_str(&parsed, "logPath"),
started_at: payload_u64(&parsed, "startedAt").unwrap_or_default(),
updated_at: payload_u64(&parsed, "updatedAt").unwrap_or_default(),
closed_at: payload_u64(&parsed, "closedAt"),
error: None,
has_terminal_log: Some(has_terminal_log),
}))
}
fn preview_resolved_input(
run_dir: &Path,
snap: &ReplaySnapshot,
def: &Value,
raw_input: Value,
cache: &mut HashMap<String, BlobPreviewDTO>,
) -> Result<BlobPreviewDTO> {
match resolve_dashboard_bindings(&raw_input, run_dir, snap, def, cache) {
Ok(value) => Ok(BlobPreviewDTO {
output_hash: None,
output_bytes: Some(serde_json::to_vec(&value)?.len()),
content_type: Some("application/json".to_string()),
truncated: None,
value: Some(value),
text: None,
error: None,
redacted: None,
}),
Err(err) => Ok(BlobPreviewDTO {
output_hash: None,
output_bytes: Some(serde_json::to_vec(&raw_input)?.len()),
content_type: Some("application/json".to_string()),
truncated: None,
value: Some(raw_input),
text: None,
error: Some(format!("failed to resolve bindings: {}", err)),
redacted: None,
}),
}
}
fn resolve_dashboard_bindings(
value: &Value,
run_dir: &Path,
snap: &ReplaySnapshot,
def: &Value,
cache: &mut HashMap<String, BlobPreviewDTO>,
) -> Result<Value> {
if let Some(ref_spec) = value.as_object().and_then(|obj| {
if obj.len() == 1 {
obj.get("$ref").and_then(Value::as_str)
} else {
None
}
}) {
return resolve_dashboard_ref(ref_spec, run_dir, snap, def, cache);
}
if let Some(s) = value.as_str() {
return interpolate_dashboard_string_refs(s, run_dir, snap, def, cache).map(Value::String);
}
if let Some(arr) = value.as_array() {
let mut out = Vec::with_capacity(arr.len());
for item in arr {
out.push(resolve_dashboard_bindings(item, run_dir, snap, def, cache)?);
}
return Ok(Value::Array(out));
}
if let Some(obj) = value.as_object() {
let mut out = serde_json::Map::new();
for (key, item) in obj {
out.insert(
key.clone(),
resolve_dashboard_bindings(item, run_dir, snap, def, cache)?,
);
}
return Ok(Value::Object(out));
}
Ok(value.clone())
}
fn interpolate_dashboard_string_refs(
value: &str,
run_dir: &Path,
snap: &ReplaySnapshot,
def: &Value,
cache: &mut HashMap<String, BlobPreviewDTO>,
) -> Result<String> {
if !value.contains("${") {
return Ok(value.to_string());
}
let mut out = String::new();
let mut cursor = 0;
while let Some(start) = value[cursor..].find("${") {
let start = cursor + start;
out.push_str(&value[cursor..start]);
let end = value[start + 2..]
.find('}')
.ok_or_else(|| anyhow::anyhow!("unterminated string ref interpolation in '{value}'"))?
+ start
+ 2;
let ref_spec = &value[start + 2..end];
if ref_spec.is_empty() {
anyhow::bail!("empty string ref interpolation in '{value}'");
}
let resolved = resolve_dashboard_ref(ref_spec, run_dir, snap, def, cache)?;
out.push_str(&stringify_dashboard_interpolated_value(ref_spec, resolved)?);
cursor = end + 1;
}
out.push_str(&value[cursor..]);
Ok(out)
}
fn stringify_dashboard_interpolated_value(ref_spec: &str, value: Value) -> Result<String> {
match value {
Value::Null => Ok("null".to_string()),
Value::String(s) => Ok(s),
Value::Number(n) => Ok(n.to_string()),
Value::Bool(b) => Ok(b.to_string()),
other => anyhow::bail!(
"string interpolation '${{{}}}' resolved to {}",
ref_spec,
if other.is_array() { "array" } else { "object" }
),
}
}
fn resolve_dashboard_ref(
ref_spec: &str,
run_dir: &Path,
snap: &ReplaySnapshot,
def: &Value,
cache: &mut HashMap<String, BlobPreviewDTO>,
) -> Result<Value> {
if let Some(rest) = ref_spec.strip_prefix("params.") {
let input_ref = snap
.run
.input
.as_ref()
.context(format!("$ref '{ref_spec}' requires run input"))?;
let preview = preview_ref(run_dir, input_ref, cache)?;
let params = preview.value.clone().context(format!(
"$ref '{ref_spec}' output preview has no JSON value"
))?;
return walk_preview_path(params, rest.split('.').collect::<Vec<_>>(), ref_spec);
}
let Some(sep_idx) = ref_spec.find(".output.") else {
anyhow::bail!("$ref '{ref_spec}' missing '.output.' separator");
};
let node_id = &ref_spec[..sep_idx];
let path = &ref_spec[sep_idx + ".output.".len()..];
let node = def
.get("nodes")
.and_then(|nodes| nodes.get(node_id))
.context(format!(
"$ref '{ref_spec}' targets unknown node '{node_id}'"
))?;
let output_ref = snap
.outputs
.get(&format!("{}::work::{}", snap.run.run_id, node_id))
.context(format!("$ref '{ref_spec}' has no successful output yet"))?;
let preview = preview_ref(run_dir, output_ref, cache)?;
let value = preview.value.clone().context(format!(
"$ref '{ref_spec}' output preview has no JSON value"
))?;
let root = if node.get("type").and_then(Value::as_str) == Some("hostExecutor")
&& value
.as_object()
.map(|obj| obj.contains_key("output"))
.unwrap_or(false)
{
value.get("output").cloned().unwrap_or(Value::Null)
} else {
value
};
walk_preview_path(root, path.split('.').collect::<Vec<_>>(), ref_spec)
}
fn walk_preview_path(value: Value, segments: Vec<&str>, ref_spec: &str) -> Result<Value> {
let mut cursor = value;
for seg in segments {
if cursor.is_null() {
anyhow::bail!("$ref '{ref_spec}' hit null at '{seg}'");
}
if let Some(arr) = cursor.as_array() {
let idx: usize = seg
.parse()
.map_err(|_| anyhow::anyhow!("$ref '{ref_spec}' array index '{seg}' invalid"))?;
cursor = arr.get(idx).cloned().context(format!(
"$ref '{ref_spec}' array index '{seg}' out of bounds"
))?;
continue;
}
let obj = cursor
.as_object()
.context(format!("$ref '{ref_spec}' segment '{seg}' not found"))?;
cursor = obj
.get(seg)
.cloned()
.context(format!("$ref '{ref_spec}' segment '{seg}' not found"))?;
}
Ok(cursor)
}
fn is_json_content(content_type: Option<&str>) -> bool {
content_type
.unwrap_or_default()
.to_ascii_lowercase()
.contains("json")
}
fn parse_loop_activity_id(activity_id: &str) -> Option<(String, u64)> {
let loop_start = activity_id.find("::loop::")?;
let after_loop = &activity_id[loop_start + "::loop::".len()..];
let iter_end = after_loop.find("::")?;
let loop_part = &after_loop[..iter_end];
let (loop_id, iteration) = loop_part.rsplit_once('.')?;
let iteration = iteration.parse().ok()?;
Some((loop_id.to_string(), iteration))
}
fn is_path_inside(parent: &Path, child: &Path) -> bool {
let parent = parent
.canonicalize()
.unwrap_or_else(|_| parent.to_path_buf());
let child = child.canonicalize().unwrap_or_else(|_| child.to_path_buf());
child.starts_with(&parent)
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
fn temp_run_dir(label: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"beam-workflow-snapshot-{label}-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos()
))
}
fn env(
event_id: &str,
run_id: &str,
event_type: &str,
payload: Value,
) -> WorkflowEventEnvelope {
WorkflowEventEnvelope {
event_id: event_id.to_string(),
run_id: run_id.to_string(),
timestamp: 1,
schema_version: 1,
actor: crate::WorkflowActor::System,
event_type: event_type.to_string(),
payload,
payload_hash: None,
}
}
#[test]
fn replay_projects_basic_state_and_outputs() {
let run_id = "run-1";
let events = vec![
env(
"run-1-1",
run_id,
"runCreated",
serde_json::json!({
"workflowId": "flow-a",
"revisionId": "rev-a",
"inputRef": {
"outputHash": "sha256:input",
"outputPath": "/tmp/input.json",
"outputBytes": 2,
"outputSchemaVersion": 1,
"contentType": "application/json"
},
"initiator": "cli"
}),
),
env("run-1-2", run_id, "runStarted", serde_json::json!({})),
env(
"run-1-3",
run_id,
"attemptCreated",
serde_json::json!({
"activityId": "run-1::work::node-a",
"attemptId": "run-1::work::node-a::att-1",
"attemptNumber": 1,
"nodeId": "node-a",
"inputRef": {
"outputHash": "sha256:input",
"outputPath": "/tmp/input.json",
"outputBytes": 2,
"outputSchemaVersion": 1,
"contentType": "application/json"
}
}),
),
env(
"run-1-4",
run_id,
"activitySucceeded",
serde_json::json!({
"activityId": "run-1::work::node-a",
"attemptId": "run-1::work::node-a::att-1",
"outputRef": {
"outputHash": "sha256:output",
"outputPath": "/tmp/output.json",
"outputBytes": 17,
"outputSchemaVersion": 1,
"contentType": "application/json"
},
"externalRefs": {"ok": true}
}),
),
];
let snap = replay_events(&events).expect("replay");
assert_eq!(snap.run.workflow_id.as_deref(), Some("flow-a"));
assert_eq!(snap.run.status, RunStatus::Running);
assert_eq!(snap.activities.len(), 1);
assert_eq!(snap.outputs.len(), 1);
assert_eq!(snap.dangling_activities, Vec::<String>::new());
}
#[tokio::test]
async fn read_run_snapshot_replays_outputs_and_binding() {
let run_dir = temp_run_dir("read");
fs::create_dir_all(run_dir.join("blobs")).unwrap();
fs::create_dir_all(
run_dir
.join("attempts")
.join("run-1::work::node-a")
.join("run-1::work::node-a::att-1"),
)
.unwrap();
fs::write(
run_dir.join("workflow.json"),
r#"{"workflowId":"flow-a","nodes":{"node-a":{"type":"hostExecutor"}}}"#,
)
.unwrap();
fs::write(
run_dir.join("chat-binding.json"),
r#"{"chatId":"chat-1","larkAppId":"app-1"}"#,
)
.unwrap();
fs::write(run_dir.join("blobs").join("input"), br#"{"foo":"bar"}"#).unwrap();
fs::write(
run_dir.join("blobs").join("output"),
br#"{"output":{"hello":"world"},"externalRefs":{"ok":true}}"#,
)
.unwrap();
fs::write(
run_dir
.join("attempts")
.join("run-1::work::node-a")
.join("run-1::work::node-a::att-1")
.join("terminal.log"),
"hello world",
)
.unwrap();
fs::write(
run_dir.join("attempts").join("run-1::work::node-a").join("run-1::work::node-a::att-1").join("terminal.json"),
r#"{"schemaVersion":1,"sessionId":"sess-1","webPort":8080,"status":"live","startedAt":1,"updatedAt":2}"#,
)
.unwrap();
let events = vec![
env(
"run-1-1",
"run-1",
"runCreated",
serde_json::json!({
"workflowId":"flow-a",
"revisionId":"rev-a",
"inputRef": {
"outputHash": "sha256:input",
"outputPath": run_dir.join("blobs").join("input"),
"outputBytes": 13,
"outputSchemaVersion": 1,
"contentType": "application/json"
}
}),
),
env("run-1-2", "run-1", "runStarted", serde_json::json!({})),
env(
"run-1-3",
"run-1",
"attemptCreated",
serde_json::json!({
"activityId": "run-1::work::node-a",
"attemptId": "run-1::work::node-a::att-1",
"attemptNumber": 1,
"nodeId": "node-a",
"inputRef": {
"outputHash": "sha256:input",
"outputPath": run_dir.join("blobs").join("input"),
"outputBytes": 13,
"outputSchemaVersion": 1,
"contentType": "application/json"
}
}),
),
env(
"run-1-4",
"run-1",
"activitySucceeded",
serde_json::json!({
"activityId": "run-1::work::node-a",
"attemptId": "run-1::work::node-a::att-1",
"outputRef": {
"outputHash": "sha256:output",
"outputPath": run_dir.join("blobs").join("output"),
"outputBytes": 55,
"outputSchemaVersion": 1,
"contentType": "application/json"
},
"externalRefs": {"ok": true}
}),
),
];
let events_json = events
.iter()
.map(|ev| serde_json::to_string(ev).unwrap())
.collect::<Vec<_>>()
.join("\n")
+ "\n";
fs::write(run_dir.join("events.ndjson"), events_json).unwrap();
let snapshot = read_run_snapshot(&run_dir)
.await
.expect("snapshot")
.expect("some");
assert_eq!(snapshot.run.workflow_id.as_deref(), Some("flow-a"));
assert_eq!(
snapshot.chat_binding.as_ref().map(|b| b.chat_id.as_str()),
Some("chat-1")
);
assert_eq!(snapshot.activities.len(), 1);
assert_eq!(snapshot.outputs.len(), 1);
assert!(
snapshot
.attempt_io
.contains_key("run-1::work::node-a::att-1")
);
let io = snapshot
.attempt_io
.get("run-1::work::node-a::att-1")
.expect("attempt io");
assert!(io.input.as_ref().and_then(|p| p.value.as_ref()).is_some());
assert!(io.output.as_ref().and_then(|p| p.value.as_ref()).is_some());
let _ = fs::remove_dir_all(&run_dir);
}
}