1use std::collections::{BTreeMap, BTreeSet, HashMap};
2use std::fs;
3use std::path::Path;
4
5use anyhow::{Context, Result};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9use crate::workflow_projection::{event_seq_from_id, read_run_events_pure};
10use crate::{RunChatBinding, WorkflowEventEnvelope, WorkflowOutputRef};
11
12const BLOB_PREVIEW_MAX_BYTES: usize = 64 * 1024;
13
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
15#[serde(rename_all = "camelCase")]
16pub struct BlobPreviewDTO {
17 #[serde(default, skip_serializing_if = "Option::is_none")]
18 pub output_hash: Option<String>,
19 #[serde(default, skip_serializing_if = "Option::is_none")]
20 pub output_bytes: Option<usize>,
21 #[serde(default, skip_serializing_if = "Option::is_none")]
22 pub content_type: Option<String>,
23 #[serde(default, skip_serializing_if = "Option::is_none")]
24 pub truncated: Option<bool>,
25 #[serde(default, skip_serializing_if = "Option::is_none")]
26 pub value: Option<Value>,
27 #[serde(default, skip_serializing_if = "Option::is_none")]
28 pub text: Option<String>,
29 #[serde(default, skip_serializing_if = "Option::is_none")]
30 pub error: Option<String>,
31 #[serde(default, skip_serializing_if = "Option::is_none")]
32 pub redacted: Option<bool>,
33}
34
35#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
36#[serde(rename_all = "camelCase")]
37pub struct AttemptTerminalDTO {
38 pub session_id: String,
39 #[serde(default, skip_serializing_if = "Option::is_none")]
40 pub cli_session_id: Option<String>,
41 pub web_port: u16,
42 pub status: String,
43 #[serde(default, skip_serializing_if = "Option::is_none")]
44 pub lark_app_id: Option<String>,
45 #[serde(default, skip_serializing_if = "Option::is_none")]
46 pub bot_name: Option<String>,
47 #[serde(default, skip_serializing_if = "Option::is_none")]
48 pub cli_id: Option<String>,
49 #[serde(default, skip_serializing_if = "Option::is_none")]
50 pub working_dir: Option<String>,
51 #[serde(default, skip_serializing_if = "Option::is_none")]
52 pub log_path: Option<String>,
53 pub started_at: u64,
54 pub updated_at: u64,
55 #[serde(default, skip_serializing_if = "Option::is_none")]
56 pub closed_at: Option<u64>,
57 #[serde(default, skip_serializing_if = "Option::is_none")]
58 pub error: Option<String>,
59 #[serde(default, skip_serializing_if = "Option::is_none")]
60 pub has_terminal_log: Option<bool>,
61}
62
63#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
64#[serde(rename_all = "camelCase")]
65pub struct AttemptIODTO {
66 #[serde(default, skip_serializing_if = "Option::is_none")]
67 pub input: Option<BlobPreviewDTO>,
68 #[serde(default, skip_serializing_if = "Option::is_none")]
69 pub resolved_input: Option<BlobPreviewDTO>,
70 #[serde(default, skip_serializing_if = "Option::is_none")]
71 pub output: Option<BlobPreviewDTO>,
72 #[serde(default, skip_serializing_if = "Option::is_none")]
73 pub log: Option<BlobPreviewDTO>,
74 #[serde(default, skip_serializing_if = "Option::is_none")]
75 pub terminal: Option<AttemptTerminalDTO>,
76 #[serde(default, skip_serializing_if = "Option::is_none")]
77 pub wait_prompt: Option<BlobPreviewDTO>,
78}
79
80#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
81#[serde(rename_all = "camelCase")]
82pub enum RunStatus {
83 Pending,
84 Running,
85 Waiting,
86 Succeeded,
87 Failed,
88 Cancelled,
89}
90
91#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
92#[serde(rename_all = "camelCase")]
93pub enum NodeStatus {
94 Idle,
95 Triggered,
96 Running,
97 Waiting,
98 Retrying,
99 Succeeded,
100 Failed,
101 Skipped,
102 Cancelled,
103}
104
105#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
106#[serde(rename_all = "camelCase")]
107pub enum ActivityStatus {
108 Pending,
109 Acquired,
110 Running,
111 Waiting,
112 EffectAttempting,
113 Succeeded,
114 Failed,
115 TimedOut,
116 Cancelled,
117}
118
119#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
120#[serde(rename_all = "camelCase")]
121pub enum LoopIterationStatus {
122 Running,
123 Approved,
124 Rejected,
125 Failed,
126 Cancelled,
127}
128
129#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
130#[serde(rename_all = "camelCase")]
131pub enum LoopStatus {
132 Running,
133 Succeeded,
134 Failed,
135 Cancelled,
136}
137
138#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
139#[serde(rename_all = "camelCase")]
140pub struct EffectAttemptedState {
141 pub idempotency_key: String,
142 pub input_hash: String,
143 pub idempotency_ttl_ms: u64,
144 pub provider: String,
145 pub attempted_at_event_id: String,
146 pub attempted_at_ms: u64,
147}
148
149#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
150#[serde(rename_all = "camelCase")]
151pub struct ReconcileResultState {
152 pub decision: String,
153 pub capability: String,
154 pub evidence: Value,
155 pub event_id: String,
156}
157
158#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
159#[serde(rename_all = "camelCase")]
160pub struct WaitResolutionState {
161 pub kind: String,
162 #[serde(default, skip_serializing_if = "Option::is_none")]
163 pub resolution: Option<String>,
164 #[serde(default, skip_serializing_if = "Option::is_none")]
165 pub by: Option<String>,
166 #[serde(default, skip_serializing_if = "Option::is_none")]
167 pub comment: Option<String>,
168 #[serde(default, skip_serializing_if = "Option::is_none")]
169 pub event_id: Option<String>,
170 #[serde(default, skip_serializing_if = "Option::is_none")]
171 pub deadline_at: Option<u64>,
172 #[serde(default, skip_serializing_if = "Option::is_none")]
173 pub exceeded_at_ms: Option<u64>,
174}
175
176#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
177#[serde(rename_all = "camelCase")]
178pub struct WaitState {
179 pub wait_kind: String,
180 #[serde(default, skip_serializing_if = "Option::is_none")]
181 pub deadline_at: Option<u64>,
182 #[serde(default, skip_serializing_if = "Option::is_none")]
183 pub prompt: Option<String>,
184 #[serde(default, skip_serializing_if = "Option::is_none")]
185 pub prompt_ref: Option<WorkflowOutputRef>,
186 #[serde(default, skip_serializing_if = "Option::is_none")]
187 pub prompt_preview: Option<String>,
188 #[serde(default, skip_serializing_if = "Option::is_none")]
189 pub approvers: Option<Vec<String>>,
190 #[serde(default, skip_serializing_if = "Option::is_none")]
191 pub on_timeout: Option<String>,
192 #[serde(default, skip_serializing_if = "Option::is_none")]
193 pub resolution: Option<WaitResolutionState>,
194}
195
196#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
197#[serde(rename_all = "camelCase")]
198pub struct CancelRequestState {
199 pub cancel_origin_event_id: String,
200 pub requested_by: String,
201 pub reason: String,
202 pub delivered: bool,
203}
204
205#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
206#[serde(rename_all = "camelCase")]
207pub struct AttemptState {
208 pub attempt_id: String,
209 pub attempt_number: u64,
210 pub input_ref: WorkflowOutputRef,
211 pub status: ActivityStatus,
212 #[serde(default, skip_serializing_if = "Option::is_none")]
213 pub lease_id: Option<String>,
214 #[serde(default, skip_serializing_if = "Option::is_none")]
215 pub timeout_ms: Option<u64>,
216 #[serde(default, skip_serializing_if = "Option::is_none")]
217 pub max_output_bytes: Option<u64>,
218 #[serde(default, skip_serializing_if = "Option::is_none")]
219 pub effect_attempted: Option<EffectAttemptedState>,
220 #[serde(default, skip_serializing_if = "Option::is_none")]
221 pub latest_reconcile_result: Option<ReconcileResultState>,
222 #[serde(default, skip_serializing_if = "Option::is_none")]
223 pub cancel_request: Option<CancelRequestState>,
224 #[serde(default, skip_serializing_if = "Option::is_none")]
225 pub wait: Option<WaitState>,
226 #[serde(default, skip_serializing_if = "Option::is_none")]
227 pub output: Option<WorkflowOutputRef>,
228 #[serde(default, skip_serializing_if = "Option::is_none")]
229 pub external_refs: Option<Value>,
230 #[serde(default, skip_serializing_if = "Option::is_none")]
231 pub error: Option<Value>,
232 #[serde(default, skip_serializing_if = "Option::is_none")]
233 pub running_ms: Option<u64>,
234 #[serde(default, skip_serializing_if = "Option::is_none")]
235 pub cancel_origin_event_id: Option<String>,
236}
237
238#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
239#[serde(rename_all = "camelCase")]
240pub struct ActivityState {
241 pub activity_id: String,
242 pub attempts: Vec<AttemptState>,
243 pub status: ActivityStatus,
244 #[serde(default, skip_serializing_if = "Option::is_none")]
245 pub current_attempt_id: Option<String>,
246 #[serde(default, skip_serializing_if = "Option::is_none")]
247 pub owner_node_id: Option<String>,
248}
249
250#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
251#[serde(rename_all = "camelCase")]
252pub struct NodeState {
253 pub node_id: String,
254 pub status: NodeStatus,
255 #[serde(default, skip_serializing_if = "Option::is_none")]
256 pub activity_id: Option<String>,
257 pub retry_count: u64,
258 #[serde(default, skip_serializing_if = "Option::is_none")]
259 pub next_attempt_at: Option<u64>,
260 #[serde(default, skip_serializing_if = "Option::is_none")]
261 pub error_class: Option<String>,
262 #[serde(default, skip_serializing_if = "Option::is_none")]
263 pub condition_event_id: Option<String>,
264 #[serde(default, skip_serializing_if = "Option::is_none")]
265 pub cancel_origin_event_id: Option<String>,
266}
267
268#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
269#[serde(rename_all = "camelCase")]
270pub struct LoopIterationState {
271 pub iteration: u64,
272 pub status: LoopIterationStatus,
273 pub body_activity_ids: Vec<String>,
274 #[serde(default, skip_serializing_if = "Option::is_none")]
275 pub decision_activity_id: Option<String>,
276 #[serde(default, skip_serializing_if = "Option::is_none")]
277 pub wait_resolved_event_id: Option<String>,
278 #[serde(default, skip_serializing_if = "Option::is_none")]
279 pub decision_by: Option<String>,
280 #[serde(default, skip_serializing_if = "Option::is_none")]
281 pub decision_comment: Option<String>,
282 #[serde(default, skip_serializing_if = "Option::is_none")]
283 pub timed_out: Option<bool>,
284}
285
286#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
287#[serde(rename_all = "camelCase")]
288pub struct LoopState {
289 pub loop_id: String,
290 pub status: LoopStatus,
291 pub iteration: u64,
292 pub max_iterations: u64,
293 pub iterations: Vec<LoopIterationState>,
294 #[serde(default, skip_serializing_if = "Option::is_none")]
295 pub output: Option<WorkflowOutputRef>,
296 #[serde(default, skip_serializing_if = "Option::is_none")]
297 pub error_code: Option<String>,
298 #[serde(default, skip_serializing_if = "Option::is_none")]
299 pub error_class: Option<String>,
300}
301
302#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
303#[serde(rename_all = "camelCase")]
304pub struct RunState {
305 pub run_id: String,
306 pub status: RunStatus,
307 #[serde(default, skip_serializing_if = "Option::is_none")]
308 pub workflow_id: Option<String>,
309 #[serde(default, skip_serializing_if = "Option::is_none")]
310 pub revision_id: Option<String>,
311 #[serde(default, skip_serializing_if = "Option::is_none")]
312 pub initiator: Option<String>,
313 #[serde(default, skip_serializing_if = "Option::is_none")]
314 pub input: Option<WorkflowOutputRef>,
315 #[serde(default, skip_serializing_if = "Option::is_none")]
316 pub output: Option<WorkflowOutputRef>,
317 #[serde(default, skip_serializing_if = "Option::is_none")]
318 pub failed_node_id: Option<String>,
319 #[serde(default, skip_serializing_if = "Option::is_none")]
320 pub root_cause_event_id: Option<String>,
321 #[serde(default, skip_serializing_if = "Option::is_none")]
322 pub cancel_origin_event_id: Option<String>,
323 #[serde(default, skip_serializing_if = "Option::is_none")]
324 pub bot_snapshots: Option<BTreeMap<String, BotSnapshot>>,
325 #[serde(default, skip_serializing_if = "Option::is_none")]
326 pub cancelled_run_intent: Option<CancelIntent>,
327 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
328 pub cancelled_node_intents: BTreeMap<String, CancelIntent>,
329}
330
331#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
332#[serde(rename_all = "camelCase")]
333pub struct CancelIntent {
334 pub cancel_origin_event_id: String,
335 pub requested_by: String,
336 pub reason: String,
337}
338
339#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
340#[serde(rename_all = "camelCase")]
341pub struct BotSnapshot {
342 #[serde(default, skip_serializing_if = "Option::is_none")]
343 pub lark_app_id: Option<String>,
344 #[serde(default, skip_serializing_if = "Option::is_none")]
345 pub cli_id: Option<String>,
346 #[serde(default, skip_serializing_if = "Option::is_none")]
347 pub display_name: Option<String>,
348 #[serde(default, skip_serializing_if = "Option::is_none")]
349 pub working_dir: Option<String>,
350}
351
352#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
353#[serde(rename_all = "camelCase")]
354pub struct LoopSnapshotDTO {
355 pub loop_id: String,
356 pub status: LoopStatus,
357 pub iteration: u64,
358 pub max_iterations: u64,
359 pub iterations: Vec<LoopIterationState>,
360 #[serde(default, skip_serializing_if = "Option::is_none")]
361 pub output: Option<WorkflowOutputRef>,
362 #[serde(default, skip_serializing_if = "Option::is_none")]
363 pub error_code: Option<String>,
364 #[serde(default, skip_serializing_if = "Option::is_none")]
365 pub error_class: Option<String>,
366}
367
368#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
369#[serde(rename_all = "camelCase")]
370pub struct RunSnapshotDTO {
371 pub run_id: String,
372 pub run: RunState,
373 pub last_seq: u64,
374 pub nodes: Vec<NodeState>,
375 pub activities: Vec<ActivityState>,
376 #[serde(default, skip_serializing_if = "Option::is_none")]
377 pub loops: Option<BTreeMap<String, LoopSnapshotDTO>>,
378 pub dangling: DanglingSnapshot,
379 pub outputs: BTreeMap<String, WorkflowOutputRef>,
380 pub attempt_io: BTreeMap<String, AttemptIODTO>,
381 #[serde(default, skip_serializing_if = "Option::is_none")]
382 pub chat_binding: Option<RunChatBinding>,
383 pub updated_at: u64,
384}
385
386#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
387#[serde(rename_all = "camelCase")]
388pub struct DanglingSnapshot {
389 pub activities: Vec<String>,
390 pub effect_attempted: Vec<String>,
391 pub waits: Vec<String>,
392 pub wait_resolutions: Vec<String>,
396 pub cancels: Vec<String>,
397}
398
399pub async fn read_run_snapshot(run_dir: &Path) -> Result<Option<RunSnapshotDTO>> {
400 let events = match read_run_events_pure(run_dir)? {
401 Some(events) if !events.is_empty() => events,
402 _ => return Ok(None),
403 };
404 let snap = replay_events(&events)?;
405 let binding = read_chat_binding_pure(run_dir).await?;
406 let def = read_workflow_definition_pure(run_dir).await?;
407 let attempt_io = build_attempt_io(run_dir, &snap, def.as_ref())?;
408 Ok(Some(RunSnapshotDTO {
409 run_id: snap.run.run_id.clone(),
410 run: snap.run,
411 last_seq: snap.last_seq,
412 nodes: snap.nodes.into_values().collect(),
413 activities: snap.activities.into_values().collect(),
414 loops: if snap.loops.is_empty() {
415 None
416 } else {
417 Some(
418 snap.loops
419 .into_iter()
420 .map(|(key, value)| (key, value))
421 .collect(),
422 )
423 },
424 dangling: DanglingSnapshot {
425 activities: snap.dangling_activities,
426 effect_attempted: snap.dangling_effect_attempted,
427 waits: snap.dangling_waits,
428 wait_resolutions: snap.dangling_wait_resolutions,
429 cancels: snap.dangling_cancels,
430 },
431 outputs: snap.outputs,
432 attempt_io,
433 chat_binding: binding,
434 updated_at: events.last().map(|ev| ev.timestamp).unwrap_or_default(),
435 }))
436}
437
438#[derive(Debug, Clone)]
439struct ReplaySnapshot {
440 run: RunState,
441 nodes: BTreeMap<String, NodeState>,
442 activities: BTreeMap<String, ActivityState>,
443 loops: BTreeMap<String, LoopSnapshotDTO>,
444 outputs: BTreeMap<String, WorkflowOutputRef>,
445 last_seq: u64,
446 dangling_activities: Vec<String>,
447 dangling_effect_attempted: Vec<String>,
448 dangling_waits: Vec<String>,
449 dangling_wait_resolutions: Vec<String>,
450 dangling_cancels: Vec<String>,
451}
452
453fn replay_events(events: &[WorkflowEventEnvelope]) -> Result<ReplaySnapshot> {
454 let first = events.first().context("replay: empty event log")?;
455 if first.event_type != "runCreated" {
456 anyhow::bail!(
457 "replay: first event must be runCreated, got {}",
458 first.event_type
459 );
460 }
461 let mut snap = ReplaySnapshot {
462 run: RunState {
463 run_id: first.run_id.clone(),
464 status: RunStatus::Pending,
465 workflow_id: None,
466 revision_id: None,
467 initiator: None,
468 input: None,
469 output: None,
470 failed_node_id: None,
471 root_cause_event_id: None,
472 cancel_origin_event_id: None,
473 bot_snapshots: None,
474 cancelled_run_intent: None,
475 cancelled_node_intents: BTreeMap::new(),
476 },
477 nodes: BTreeMap::new(),
478 activities: BTreeMap::new(),
479 loops: BTreeMap::new(),
480 outputs: BTreeMap::new(),
481 last_seq: 0,
482 dangling_activities: Vec::new(),
483 dangling_effect_attempted: Vec::new(),
484 dangling_waits: Vec::new(),
485 dangling_wait_resolutions: Vec::new(),
486 dangling_cancels: Vec::new(),
487 };
488 let mut waits_open = BTreeSet::new();
489 let mut run_cancel_intent: Option<(String, String, String)> = None;
490 let mut node_cancel_intents: BTreeMap<String, (String, String, String)> = BTreeMap::new();
491
492 for ev in events {
493 if ev.run_id != snap.run.run_id {
494 anyhow::bail!(
495 "replay: runId mismatch at {} — log is {}, event has {}",
496 ev.event_id,
497 snap.run.run_id,
498 ev.run_id
499 );
500 }
501 snap.last_seq = snap.last_seq.max(event_seq_from_id(&ev.event_id));
502 if is_payload_ref(&ev.payload) {
503 continue;
504 }
505 let payload = &ev.payload;
506 match ev.event_type.as_str() {
507 "runCreated" => apply_run_created(&mut snap.run, payload),
508 "runStarted" => snap.run.status = RunStatus::Running,
509 "runSucceeded" => apply_run_succeeded(&mut snap.run, payload),
510 "runFailed" => apply_run_failed(&mut snap.run, payload),
511 "runCanceled" => apply_run_canceled(&mut snap.run, payload),
512 "nodeWaiting" => {
513 if let Some(node_id) = payload_str(payload, "nodeId") {
514 get_node(&mut snap.nodes, &node_id).status = NodeStatus::Waiting;
515 }
516 }
517 "nodeRetrying" => {
518 if let Some(node_id) = payload_str(payload, "nodeId") {
519 let node = get_node(&mut snap.nodes, &node_id);
520 node.status = NodeStatus::Retrying;
521 node.retry_count = node.retry_count.saturating_add(1);
522 }
523 }
524 "nodeSucceeded" => {
525 if let Some(node_id) = payload_str(payload, "nodeId") {
526 let node = get_node(&mut snap.nodes, &node_id);
527 node.status = NodeStatus::Succeeded;
528 node.activity_id = payload_str(payload, "lastActivityId");
529 }
530 }
531 "nodeFailed" => {
532 if let Some(node_id) = payload_str(payload, "nodeId") {
533 let node = get_node(&mut snap.nodes, &node_id);
534 node.status = NodeStatus::Failed;
535 node.activity_id = payload_str(payload, "lastActivityId");
536 node.error_class = payload_str(payload, "errorClass");
537 }
538 }
539 "nodeSkipped" => {
540 if let Some(node_id) = payload_str(payload, "nodeId") {
541 let node = get_node(&mut snap.nodes, &node_id);
542 node.status = NodeStatus::Skipped;
543 node.condition_event_id = Some(ev.event_id.clone());
544 }
545 }
546 "nodeCanceled" => {
547 if let Some(node_id) = payload_str(payload, "nodeId") {
548 let node = get_node(&mut snap.nodes, &node_id);
549 node.status = NodeStatus::Cancelled;
550 node.cancel_origin_event_id = payload_str(payload, "cancelOriginEventId");
551 }
552 }
553 "loopStarted" => {
554 if let Some(loop_id) = payload_str(payload, "loopId") {
555 let loop_state = get_loop(&mut snap.loops, &loop_id);
556 loop_state.status = LoopStatus::Running;
557 loop_state.max_iterations = payload_u64(payload, "maxIterations").unwrap_or(0);
558 }
559 }
560 "loopIterationStarted" => {
561 if let Some(loop_id) = payload_str(payload, "loopId") {
562 let iteration = payload_u64(payload, "iteration").unwrap_or(0);
563 let loop_state = get_loop(&mut snap.loops, &loop_id);
564 loop_state.status = LoopStatus::Running;
565 loop_state.iteration = iteration;
566 let it = get_loop_iteration(loop_state, iteration);
567 it.status = LoopIterationStatus::Running;
568 }
569 }
570 "loopIterationFinished" => {
571 if let Some(loop_id) = payload_str(payload, "loopId") {
572 let iteration = payload_u64(payload, "iteration").unwrap_or(0);
573 let loop_state = get_loop(&mut snap.loops, &loop_id);
574 loop_state.iteration = loop_state.iteration.max(iteration);
575 let it = get_loop_iteration(loop_state, iteration);
576 it.status = match payload_str(payload, "resolution")
577 .as_deref()
578 .unwrap_or("failed")
579 {
580 "approved" => LoopIterationStatus::Approved,
581 "rejected" => LoopIterationStatus::Rejected,
582 "cancelled" => LoopIterationStatus::Cancelled,
583 _ => LoopIterationStatus::Failed,
584 };
585 it.decision_activity_id = payload_str(payload, "decisionActivityId");
586 it.wait_resolved_event_id = payload_str(payload, "waitResolvedEventId");
587 it.decision_by = payload_str(payload, "by");
588 it.decision_comment = payload_str(payload, "comment");
589 it.timed_out = payload_bool(payload, "timedOut");
590 }
591 }
592 "loopFinished" => {
593 if let Some(loop_id) = payload_str(payload, "loopId") {
594 let loop_state = get_loop(&mut snap.loops, &loop_id);
595 loop_state.iteration = payload_u64(payload, "finalIteration").unwrap_or(0);
596 loop_state.status = match payload_str(payload, "resolution")
597 .as_deref()
598 .unwrap_or("failed")
599 {
600 "approved" => LoopStatus::Succeeded,
601 "cancelled" => LoopStatus::Cancelled,
602 _ => LoopStatus::Failed,
603 };
604 loop_state.output = payload_workflow_output_ref(payload, "outputRef");
605 loop_state.error_code = payload_str(payload, "errorCode");
606 loop_state.error_class = payload_str(payload, "errorClass");
607 if let Some(output_ref) = loop_state.output.clone() {
608 snap.outputs
609 .insert(format!("{}::work::{}", ev.run_id, loop_id), output_ref);
610 }
611 if loop_state.status != LoopStatus::Succeeded {
612 if let Some(inflight) = loop_state
613 .iterations
614 .iter_mut()
615 .find(|it| matches!(it.status, LoopIterationStatus::Running))
616 {
617 inflight.status = if loop_state.status == LoopStatus::Cancelled {
618 LoopIterationStatus::Cancelled
619 } else {
620 LoopIterationStatus::Failed
621 };
622 }
623 }
624 }
625 }
626 "conditionEvaluated" => {
627 if let Some(node_id) = payload_str(payload, "nodeId") {
628 let node = get_node(&mut snap.nodes, &node_id);
629 node.condition_event_id = Some(ev.event_id.clone());
630 }
631 }
632 "attemptCreated" => {
633 if let (Some(activity_id), Some(node_id), Some(attempt_id)) = (
634 payload_str(payload, "activityId"),
635 payload_str(payload, "nodeId"),
636 payload_str(payload, "attemptId"),
637 ) {
638 let attempt_number = payload_u64(payload, "attemptNumber").unwrap_or(0);
639 let input_ref = match payload_workflow_output_ref(payload, "inputRef") {
640 Some(value) => value,
641 None => continue,
642 };
643 let activity = get_activity(&mut snap.activities, &activity_id);
644 activity.attempts.push(AttemptState {
645 attempt_id: attempt_id.clone(),
646 attempt_number,
647 input_ref,
648 status: ActivityStatus::Pending,
649 lease_id: None,
650 timeout_ms: None,
651 max_output_bytes: None,
652 effect_attempted: None,
653 latest_reconcile_result: None,
654 cancel_request: None,
655 wait: None,
656 output: None,
657 external_refs: None,
658 error: None,
659 running_ms: None,
660 cancel_origin_event_id: None,
661 });
662 activity.current_attempt_id = Some(attempt_id.clone());
663 activity.status = ActivityStatus::Pending;
664 activity.owner_node_id = Some(node_id.clone());
665 let node = get_node(&mut snap.nodes, &node_id);
666 node.activity_id = Some(activity_id.clone());
667 if attempt_number == 1 && matches!(node.status, NodeStatus::Idle) {
668 node.status = NodeStatus::Triggered;
669 }
670 if let Some((loop_id, iteration)) = parse_loop_activity_id(&activity_id) {
671 if let Some(loop_state) = snap.loops.get_mut(&loop_id) {
672 let it = get_loop_iteration(loop_state, iteration);
673 if !it.body_activity_ids.contains(&activity_id) {
674 it.body_activity_ids.push(activity_id.clone());
675 }
676 }
677 }
678 }
679 }
680 "leaseSigned" => {
681 if let (Some(activity_id), Some(attempt_id)) = (
682 payload_str(payload, "activityId"),
683 payload_str(payload, "attemptId"),
684 ) {
685 if let Some(attempt) =
686 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
687 {
688 attempt.lease_id = payload_str(payload, "leaseId");
689 attempt.timeout_ms = payload_u64(payload, "timeoutMs");
690 attempt.max_output_bytes = payload_u64(payload, "maxOutputBytes");
691 }
692 }
693 }
694 "backoffScheduled" => {
695 if let Some(node_id) = payload_str(payload, "nodeId") {
696 get_node(&mut snap.nodes, &node_id).next_attempt_at =
697 payload_u64(payload, "nextAttemptAt");
698 }
699 }
700 "backoffElapsed" => {
701 if let Some(node_id) = payload_str(payload, "nodeId") {
702 get_node(&mut snap.nodes, &node_id).next_attempt_at = None;
703 }
704 }
705 "effectAttempted" => {
706 if let (Some(activity_id), Some(attempt_id)) = (
707 payload_str(payload, "activityId"),
708 payload_str(payload, "attemptId"),
709 ) {
710 if let Some(attempt) =
711 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
712 {
713 attempt.effect_attempted = Some(EffectAttemptedState {
714 idempotency_key: payload_str(payload, "idempotencyKey")
715 .unwrap_or_default(),
716 input_hash: payload_str(payload, "inputHash").unwrap_or_default(),
717 idempotency_ttl_ms: payload_u64(payload, "idempotencyTtlMs")
718 .unwrap_or_default(),
719 provider: payload_str(payload, "provider").unwrap_or_default(),
720 attempted_at_event_id: ev.event_id.clone(),
721 attempted_at_ms: ev.timestamp,
722 });
723 attempt.status = ActivityStatus::EffectAttempting;
724 if let Some(activity) = snap.activities.get_mut(&activity_id) {
725 activity.status = ActivityStatus::EffectAttempting;
726 }
727 }
728 }
729 }
730 "activitySucceeded" => {
731 if let (Some(activity_id), Some(attempt_id)) = (
732 payload_str(payload, "activityId"),
733 payload_str(payload, "attemptId"),
734 ) {
735 let output_ref = if let Some(attempt) =
736 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
737 {
738 attempt.status = ActivityStatus::Succeeded;
739 attempt.output = payload_workflow_output_ref(payload, "outputRef");
740 attempt.external_refs = payload.get("externalRefs").cloned();
741 attempt.output.clone()
742 } else {
743 None
744 };
745 if let Some(activity) = snap.activities.get_mut(&activity_id) {
746 activity.status = ActivityStatus::Succeeded;
747 }
748 if let Some(output_ref) = output_ref {
749 snap.outputs.insert(activity_id.clone(), output_ref);
750 }
751 waits_open.remove(&activity_id);
752 }
753 }
754 "activityFailed" => {
755 if let (Some(activity_id), Some(attempt_id)) = (
756 payload_str(payload, "activityId"),
757 payload_str(payload, "attemptId"),
758 ) {
759 if let Some(attempt) =
760 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
761 {
762 attempt.status = ActivityStatus::Failed;
763 attempt.error = payload.get("error").cloned();
764 if let Some(activity) = snap.activities.get_mut(&activity_id) {
765 activity.status = ActivityStatus::Failed;
766 }
767 waits_open.remove(&activity_id);
768 }
769 }
770 }
771 "activityTimedOut" => {
772 if let (Some(activity_id), Some(attempt_id)) = (
773 payload_str(payload, "activityId"),
774 payload_str(payload, "attemptId"),
775 ) {
776 if let Some(attempt) =
777 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
778 {
779 attempt.status = ActivityStatus::TimedOut;
780 attempt.running_ms = payload_u64(payload, "runningMs");
781 if let Some(activity) = snap.activities.get_mut(&activity_id) {
782 activity.status = ActivityStatus::TimedOut;
783 }
784 waits_open.remove(&activity_id);
785 }
786 }
787 }
788 "activityRunning" => {
789 if let (Some(activity_id), Some(attempt_id)) = (
790 payload_str(payload, "activityId"),
791 payload_str(payload, "attemptId"),
792 ) {
793 if let Some(attempt) =
794 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
795 {
796 attempt.status = ActivityStatus::Running;
797 }
798 let owner_node_id = snap
799 .activities
800 .get(&activity_id)
801 .and_then(|activity| activity.owner_node_id.clone());
802 if let Some(activity) = snap.activities.get_mut(&activity_id) {
803 activity.status = ActivityStatus::Running;
804 }
805 if let Some(owner) = owner_node_id {
806 let node = get_node(&mut snap.nodes, &owner);
807 if matches!(node.status, NodeStatus::Triggered | NodeStatus::Retrying) {
808 node.status = NodeStatus::Running;
809 }
810 }
811 }
812 }
813 "activityWaiting" => {
814 if let Some(activity_id) = payload_str(payload, "activityId") {
815 let attempt_id = snap
816 .activities
817 .get(&activity_id)
818 .and_then(|activity| activity.current_attempt_id.clone());
819 if let Some(attempt_id) = attempt_id {
820 if let Some(attempt) =
821 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
822 {
823 attempt.status = ActivityStatus::Waiting;
824 }
825 if let Some(activity) = snap.activities.get_mut(&activity_id) {
826 activity.status = ActivityStatus::Waiting;
827 }
828 }
829 }
830 }
831 "activityCanceled" => {
832 if let (Some(activity_id), Some(attempt_id)) = (
833 payload_str(payload, "activityId"),
834 payload_str(payload, "attemptId"),
835 ) {
836 if let Some(attempt) =
837 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
838 {
839 attempt.status = ActivityStatus::Cancelled;
840 attempt.cancel_origin_event_id =
841 payload_str(payload, "cancelOriginEventId");
842 if let Some(activity) = snap.activities.get_mut(&activity_id) {
843 activity.status = ActivityStatus::Cancelled;
844 }
845 waits_open.remove(&activity_id);
846 }
847 }
848 }
849 "waitCreated" => {
850 if let Some(activity_id) = payload_str(payload, "activityId") {
851 waits_open.insert(activity_id.clone());
852 let attempt_id = snap
853 .activities
854 .get(&activity_id)
855 .and_then(|activity| activity.current_attempt_id.clone());
856 if let Some(attempt_id) = attempt_id {
857 if let Some(attempt) =
858 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
859 {
860 attempt.wait = Some(WaitState {
861 wait_kind: payload_str(payload, "waitKind").unwrap_or_default(),
862 deadline_at: payload_u64(payload, "deadlineAt"),
863 prompt: payload_str(payload, "prompt"),
864 prompt_ref: payload_workflow_output_ref(payload, "promptRef"),
865 prompt_preview: payload_str(payload, "promptPreview"),
866 approvers: payload_string_array(payload, "approvers"),
867 on_timeout: payload_str(payload, "onTimeout"),
868 resolution: None,
869 });
870 }
871 }
872 }
873 }
874 "waitResolved" => {
875 if let Some(activity_id) = payload_str(payload, "activityId") {
876 waits_open.remove(&activity_id);
877 let attempt_id = snap
878 .activities
879 .get(&activity_id)
880 .and_then(|activity| activity.current_attempt_id.clone());
881 if let Some(attempt_id) = attempt_id {
882 if let Some(attempt) =
883 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
884 {
885 if let Some(wait) = attempt.wait.as_mut() {
886 wait.resolution = Some(WaitResolutionState {
887 kind: "resolved".to_string(),
888 resolution: payload_str(payload, "resolution"),
889 by: payload_str(payload, "by"),
890 comment: payload_str(payload, "comment"),
891 event_id: Some(ev.event_id.clone()),
892 deadline_at: None,
893 exceeded_at_ms: None,
894 });
895 }
896 }
897 }
898 }
899 }
900 "waitDeadlineExceeded" => {
901 if let Some(activity_id) = payload_str(payload, "activityId") {
902 waits_open.remove(&activity_id);
903 let attempt_id = snap
904 .activities
905 .get(&activity_id)
906 .and_then(|activity| activity.current_attempt_id.clone());
907 if let Some(attempt_id) = attempt_id {
908 if let Some(attempt) =
909 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
910 {
911 if let Some(wait) = attempt.wait.as_mut() {
912 wait.resolution = Some(WaitResolutionState {
913 kind: "deadlineExceeded".to_string(),
914 resolution: None,
915 by: None,
916 comment: None,
917 event_id: Some(ev.event_id.clone()),
918 deadline_at: payload_u64(payload, "deadlineAt"),
919 exceeded_at_ms: payload_u64(payload, "exceededAtMs"),
920 });
921 }
922 }
923 }
924 }
925 }
926 "cancelRequested" => {
927 if let Some(target) = payload.get("target") {
928 if let Some(kind) = target.get("kind").and_then(Value::as_str) {
929 match kind {
930 "activity" => {
931 if let Some(activity_id) =
932 target.get("activityId").and_then(Value::as_str)
933 {
934 mark_activity_cancel(
935 &mut snap.activities,
936 activity_id,
937 &ev,
938 payload,
939 );
940 }
941 }
942 "node" => {
943 if let Some(node_id) = target.get("nodeId").and_then(Value::as_str)
944 {
945 let node_id = node_id.to_string();
946 node_cancel_intents.entry(node_id.clone()).or_insert_with(
947 || {
948 (
949 ev.event_id.clone(),
950 payload_str(payload, "by").unwrap_or_default(),
951 payload_str(payload, "reason").unwrap_or_default(),
952 )
953 },
954 );
955 for activity in snap.activities.values_mut() {
956 if activity.owner_node_id.as_deref()
957 == Some(node_id.as_str())
958 {
959 mark_attempt_cancel(activity, &ev, payload);
960 }
961 }
962 }
963 }
964 "run" => {
965 if run_cancel_intent.is_none() {
966 run_cancel_intent = Some((
967 ev.event_id.clone(),
968 payload_str(payload, "by").unwrap_or_default(),
969 payload_str(payload, "reason").unwrap_or_default(),
970 ));
971 }
972 for activity in snap.activities.values_mut() {
973 mark_attempt_cancel(activity, &ev, payload);
974 }
975 }
976 _ => {}
977 }
978 }
979 }
980 }
981 "cancelDelivered" => {
982 if let Some(activity_id) = payload_str(payload, "activityId") {
983 let attempt_id = snap
984 .activities
985 .get(&activity_id)
986 .and_then(|activity| activity.current_attempt_id.clone());
987 if let Some(attempt_id) = attempt_id {
988 if let Some(attempt) =
989 get_attempt_mut(&mut snap.activities, &activity_id, &attempt_id)
990 {
991 if let Some(cancel) = attempt.cancel_request.as_mut() {
992 cancel.delivered = true;
993 }
994 }
995 }
996 }
997 }
998 "workerLost" | "resumeStarted" => {}
999 "reconcileResult" => {
1000 if let Some(idempotency_key) = payload_str(payload, "idempotencyKey") {
1001 for activity in snap.activities.values_mut() {
1002 if let Some(attempt) = activity.attempts.iter_mut().find(|candidate| {
1003 candidate
1004 .effect_attempted
1005 .as_ref()
1006 .map(|x| x.idempotency_key.as_str())
1007 == Some(idempotency_key.as_str())
1008 }) {
1009 attempt.latest_reconcile_result = Some(ReconcileResultState {
1010 decision: payload_str(payload, "decision").unwrap_or_default(),
1011 capability: payload_str(payload, "capability").unwrap_or_default(),
1012 evidence: payload.get("evidence").cloned().unwrap_or(Value::Null),
1013 event_id: ev.event_id.clone(),
1014 });
1015 break;
1016 }
1017 }
1018 }
1019 }
1020 _ => {}
1021 }
1022 }
1023
1024 let mut dangling_activities = Vec::new();
1025 let mut dangling_effect_attempted = Vec::new();
1026 let mut dangling_wait_resolutions = Vec::new();
1027 let mut dangling_cancels = Vec::new();
1028 for activity in snap.activities.values() {
1029 let Some(latest) = activity.attempts.last() else {
1030 continue;
1031 };
1032 let is_terminal = matches!(
1033 latest.status,
1034 ActivityStatus::Succeeded
1035 | ActivityStatus::Failed
1036 | ActivityStatus::TimedOut
1037 | ActivityStatus::Cancelled
1038 );
1039 if !is_terminal {
1040 dangling_activities.push(activity.activity_id.clone());
1041 if latest.effect_attempted.is_some() {
1042 dangling_effect_attempted.push(activity.activity_id.clone());
1043 }
1044 if latest
1045 .wait
1046 .as_ref()
1047 .and_then(|w| w.resolution.as_ref())
1048 .is_some()
1049 {
1050 dangling_wait_resolutions.push(activity.activity_id.clone());
1051 }
1052 if latest.cancel_request.is_some() {
1053 dangling_cancels.push(activity.activity_id.clone());
1054 }
1055 }
1056 }
1057 dangling_activities.sort();
1058 dangling_effect_attempted.sort();
1059 dangling_wait_resolutions.sort();
1060 dangling_cancels.sort();
1061
1062 if !matches!(snap.run.status, RunStatus::Cancelled) {
1063 if let Some((cancel_origin_event_id, requested_by, reason)) = run_cancel_intent {
1064 snap.run.cancelled_run_intent = Some(CancelIntent {
1065 cancel_origin_event_id,
1066 requested_by,
1067 reason,
1068 });
1069 }
1070 }
1071 for (node_id, intent) in node_cancel_intents {
1072 if matches!(
1073 snap.nodes.get(&node_id).map(|node| node.status),
1074 Some(NodeStatus::Cancelled)
1075 ) {
1076 continue;
1077 }
1078 snap.run.cancelled_node_intents.insert(
1079 node_id,
1080 CancelIntent {
1081 cancel_origin_event_id: intent.0,
1082 requested_by: intent.1,
1083 reason: intent.2,
1084 },
1085 );
1086 }
1087
1088 Ok(ReplaySnapshot {
1089 run: snap.run,
1090 nodes: snap.nodes,
1091 activities: snap.activities,
1092 loops: snap.loops,
1093 outputs: snap.outputs,
1094 last_seq: snap.last_seq,
1095 dangling_activities,
1096 dangling_effect_attempted,
1097 dangling_waits: waits_open.into_iter().collect(),
1098 dangling_wait_resolutions,
1099 dangling_cancels,
1100 })
1101}
1102
1103fn get_node<'a>(nodes: &'a mut BTreeMap<String, NodeState>, node_id: &str) -> &'a mut NodeState {
1104 nodes
1105 .entry(node_id.to_string())
1106 .or_insert_with(|| NodeState {
1107 node_id: node_id.to_string(),
1108 status: NodeStatus::Idle,
1109 activity_id: None,
1110 retry_count: 0,
1111 next_attempt_at: None,
1112 error_class: None,
1113 condition_event_id: None,
1114 cancel_origin_event_id: None,
1115 })
1116}
1117
1118fn get_activity<'a>(
1119 activities: &'a mut BTreeMap<String, ActivityState>,
1120 activity_id: &str,
1121) -> &'a mut ActivityState {
1122 activities
1123 .entry(activity_id.to_string())
1124 .or_insert_with(|| ActivityState {
1125 activity_id: activity_id.to_string(),
1126 attempts: Vec::new(),
1127 status: ActivityStatus::Pending,
1128 current_attempt_id: None,
1129 owner_node_id: None,
1130 })
1131}
1132
1133fn get_attempt_mut<'a>(
1134 activities: &'a mut BTreeMap<String, ActivityState>,
1135 activity_id: &str,
1136 attempt_id: &str,
1137) -> Option<&'a mut AttemptState> {
1138 activities
1139 .get_mut(activity_id)?
1140 .attempts
1141 .iter_mut()
1142 .find(|attempt| attempt.attempt_id == attempt_id)
1143}
1144
1145fn get_loop<'a>(
1146 loops: &'a mut BTreeMap<String, LoopSnapshotDTO>,
1147 loop_id: &str,
1148) -> &'a mut LoopSnapshotDTO {
1149 loops
1150 .entry(loop_id.to_string())
1151 .or_insert_with(|| LoopSnapshotDTO {
1152 loop_id: loop_id.to_string(),
1153 status: LoopStatus::Running,
1154 iteration: 0,
1155 max_iterations: 0,
1156 iterations: Vec::new(),
1157 output: None,
1158 error_code: None,
1159 error_class: None,
1160 })
1161}
1162
1163fn get_loop_iteration<'a>(
1164 loop_state: &'a mut LoopSnapshotDTO,
1165 iteration: u64,
1166) -> &'a mut LoopIterationState {
1167 if let Some(idx) = loop_state
1168 .iterations
1169 .iter()
1170 .position(|candidate| candidate.iteration == iteration)
1171 {
1172 return &mut loop_state.iterations[idx];
1173 }
1174 loop_state.iterations.push(LoopIterationState {
1175 iteration,
1176 status: LoopIterationStatus::Running,
1177 body_activity_ids: Vec::new(),
1178 decision_activity_id: None,
1179 wait_resolved_event_id: None,
1180 decision_by: None,
1181 decision_comment: None,
1182 timed_out: None,
1183 });
1184 let idx = loop_state.iterations.len() - 1;
1185 &mut loop_state.iterations[idx]
1186}
1187
1188fn apply_run_created(run: &mut RunState, payload: &Value) {
1189 run.workflow_id = payload_str(payload, "workflowId");
1190 run.revision_id = payload_str(payload, "revisionId");
1191 run.initiator = payload_str(payload, "initiator");
1192 run.input = payload_workflow_output_ref(payload, "inputRef");
1193 if let Some(bot_snapshots) = payload.get("botSnapshots").and_then(|value| {
1194 serde_json::from_value::<BTreeMap<String, BotSnapshot>>(value.clone()).ok()
1195 }) {
1196 run.bot_snapshots = Some(bot_snapshots);
1197 }
1198}
1199
1200fn apply_run_succeeded(run: &mut RunState, payload: &Value) {
1201 run.status = RunStatus::Succeeded;
1202 run.output = payload_workflow_output_ref(payload, "outputRef");
1203}
1204
1205fn apply_run_failed(run: &mut RunState, payload: &Value) {
1206 run.status = RunStatus::Failed;
1207 run.failed_node_id = payload_str(payload, "failedNodeId");
1208 run.root_cause_event_id = payload_str(payload, "rootCauseEventId");
1209}
1210
1211fn apply_run_canceled(run: &mut RunState, payload: &Value) {
1212 run.status = RunStatus::Cancelled;
1213 run.cancel_origin_event_id = payload_str(payload, "cancelOriginEventId");
1214}
1215
1216fn mark_attempt_cancel(activity: &mut ActivityState, ev: &WorkflowEventEnvelope, payload: &Value) {
1217 let Some(attempt_id) = activity.current_attempt_id.clone() else {
1218 return;
1219 };
1220 let Some(attempt) = activity
1221 .attempts
1222 .iter_mut()
1223 .find(|attempt| attempt.attempt_id == attempt_id)
1224 else {
1225 return;
1226 };
1227 let is_terminal = matches!(
1228 attempt.status,
1229 ActivityStatus::Succeeded
1230 | ActivityStatus::Failed
1231 | ActivityStatus::TimedOut
1232 | ActivityStatus::Cancelled
1233 );
1234 if is_terminal || attempt.cancel_request.is_some() {
1235 return;
1236 }
1237 attempt.cancel_request = Some(CancelRequestState {
1238 cancel_origin_event_id: ev.event_id.clone(),
1239 requested_by: payload_str(payload, "by").unwrap_or_default(),
1240 reason: payload_str(payload, "reason").unwrap_or_default(),
1241 delivered: false,
1242 });
1243}
1244
1245fn mark_activity_cancel(
1246 activities: &mut BTreeMap<String, ActivityState>,
1247 activity_id: &str,
1248 ev: &WorkflowEventEnvelope,
1249 payload: &Value,
1250) {
1251 if let Some(activity) = activities.get_mut(activity_id) {
1252 mark_attempt_cancel(activity, ev, payload);
1253 }
1254}
1255
1256fn payload_str(payload: &Value, key: &str) -> Option<String> {
1257 payload
1258 .get(key)
1259 .and_then(Value::as_str)
1260 .map(ToOwned::to_owned)
1261}
1262
1263fn payload_u64(payload: &Value, key: &str) -> Option<u64> {
1264 payload.get(key).and_then(Value::as_u64)
1265}
1266
1267fn payload_bool(payload: &Value, key: &str) -> Option<bool> {
1268 payload.get(key).and_then(Value::as_bool)
1269}
1270
1271fn payload_string_array(payload: &Value, key: &str) -> Option<Vec<String>> {
1272 let arr = payload.get(key)?.as_array()?;
1273 Some(
1274 arr.iter()
1275 .filter_map(|item| item.as_str().map(ToOwned::to_owned))
1276 .collect(),
1277 )
1278}
1279
1280fn payload_workflow_output_ref(payload: &Value, key: &str) -> Option<WorkflowOutputRef> {
1281 payload
1282 .get(key)
1283 .cloned()
1284 .and_then(|value| serde_json::from_value::<WorkflowOutputRef>(value).ok())
1285}
1286
1287fn is_payload_ref(payload: &Value) -> bool {
1288 let Some(obj) = payload.as_object() else {
1289 return false;
1290 };
1291 obj.get("ref").and_then(Value::as_str).is_some()
1292}
1293
1294async fn read_chat_binding_pure(run_dir: &Path) -> Result<Option<RunChatBinding>> {
1295 let path = run_dir.join("chat-binding.json");
1296 let raw = match fs::read_to_string(&path) {
1297 Ok(raw) => raw,
1298 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
1299 Err(err) => return Err(err.into()),
1300 };
1301 let binding = serde_json::from_str::<RunChatBinding>(&raw)
1302 .with_context(|| format!("failed to parse {}", path.display()))?;
1303 Ok(Some(binding))
1304}
1305
1306async fn read_workflow_definition_pure(run_dir: &Path) -> Result<Option<Value>> {
1307 let path = run_dir.join("workflow.json");
1308 let raw = match fs::read_to_string(&path) {
1309 Ok(raw) => raw,
1310 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
1311 Err(err) => return Err(err.into()),
1312 };
1313 let def = serde_json::from_str::<Value>(&raw)
1314 .with_context(|| format!("failed to parse {}", path.display()))?;
1315 Ok(Some(def))
1316}
1317
1318fn build_attempt_io(
1319 run_dir: &Path,
1320 snap: &ReplaySnapshot,
1321 def: Option<&Value>,
1322) -> Result<BTreeMap<String, AttemptIODTO>> {
1323 let mut out = BTreeMap::new();
1324 let mut cache = HashMap::new();
1325 for activity in snap.activities.values() {
1326 for attempt in &activity.attempts {
1327 let mut io = AttemptIODTO {
1328 input: Some(preview_ref(run_dir, &attempt.input_ref, &mut cache)?),
1329 resolved_input: None,
1330 output: None,
1331 log: preview_attempt_log(run_dir, &activity.activity_id, &attempt.attempt_id)?,
1332 terminal: read_attempt_terminal(
1333 run_dir,
1334 &activity.activity_id,
1335 &attempt.attempt_id,
1336 )?,
1337 wait_prompt: None,
1338 };
1339 if let Some(output_ref) = attempt.output.as_ref() {
1340 io.output = Some(preview_ref(run_dir, output_ref, &mut cache)?);
1341 }
1342 if let Some(wait) = attempt.wait.as_ref() {
1343 if let Some(prompt_ref) = wait.prompt_ref.as_ref() {
1344 io.wait_prompt = Some(preview_ref(run_dir, prompt_ref, &mut cache)?);
1345 }
1346 }
1347 if let Some(input_value) = io.input.as_ref().and_then(|preview| preview.value.clone()) {
1348 if let Some(def) = def {
1349 io.resolved_input = Some(preview_resolved_input(
1350 run_dir,
1351 snap,
1352 def,
1353 input_value,
1354 &mut cache,
1355 )?);
1356 }
1357 }
1358 out.insert(attempt.attempt_id.clone(), io);
1359 }
1360 }
1361 Ok(out)
1362}
1363
1364fn preview_ref(
1365 run_dir: &Path,
1366 ref_value: &WorkflowOutputRef,
1367 cache: &mut HashMap<String, BlobPreviewDTO>,
1368) -> Result<BlobPreviewDTO> {
1369 if let Some(cached) = cache.get(&ref_value.output_hash) {
1370 return Ok(cached.clone());
1371 }
1372 let base = BlobPreviewDTO {
1373 output_hash: Some(ref_value.output_hash.clone()),
1374 output_bytes: Some(ref_value.output_bytes),
1375 content_type: ref_value.content_type.clone(),
1376 truncated: None,
1377 value: None,
1378 text: None,
1379 error: None,
1380 redacted: None,
1381 };
1382 let Some(output_path) = Some(&ref_value.output_path) else {
1383 return Ok(base);
1384 };
1385 if !is_path_inside(run_dir, Path::new(output_path)) {
1386 let mut preview = base.clone();
1387 preview.error = Some("outputPath is outside run directory".to_string());
1388 cache.insert(ref_value.output_hash.clone(), preview.clone());
1389 return Ok(preview);
1390 }
1391 let bytes = match fs::read(output_path) {
1392 Ok(bytes) => bytes,
1393 Err(err) => {
1394 let mut preview = base.clone();
1395 preview.error = Some(err.to_string());
1396 cache.insert(ref_value.output_hash.clone(), preview.clone());
1397 return Ok(preview);
1398 }
1399 };
1400 let mut preview = base;
1401 preview.output_bytes = Some(bytes.len());
1402 preview.truncated = Some(bytes.len() > BLOB_PREVIEW_MAX_BYTES);
1403 let slice = if bytes.len() > BLOB_PREVIEW_MAX_BYTES {
1404 &bytes[..BLOB_PREVIEW_MAX_BYTES]
1405 } else {
1406 &bytes[..]
1407 };
1408 let text = String::from_utf8_lossy(slice).to_string();
1409 if !preview.truncated.unwrap_or(false) && is_json_content(ref_value.content_type.as_deref()) {
1410 match serde_json::from_slice::<Value>(slice) {
1411 Ok(value) => preview.value = Some(value),
1412 Err(err) => {
1413 preview.text = Some(text.clone());
1414 preview.error = Some(format!("invalid JSON: {}", err));
1415 }
1416 }
1417 } else {
1418 preview.text = Some(text);
1419 }
1420 cache.insert(ref_value.output_hash.clone(), preview.clone());
1421 Ok(preview)
1422}
1423
1424fn preview_attempt_log(
1425 run_dir: &Path,
1426 activity_id: &str,
1427 attempt_id: &str,
1428) -> Result<Option<BlobPreviewDTO>> {
1429 let path = run_dir
1430 .join("attempts")
1431 .join(activity_id)
1432 .join(attempt_id)
1433 .join("terminal.log");
1434 if !is_path_inside(run_dir, &path) {
1435 return Ok(Some(BlobPreviewDTO {
1436 output_hash: None,
1437 output_bytes: None,
1438 content_type: Some("text/plain".to_string()),
1439 truncated: None,
1440 value: None,
1441 text: None,
1442 error: Some("attempt log is outside run directory".to_string()),
1443 redacted: None,
1444 }));
1445 }
1446 let raw = match fs::read(&path) {
1447 Ok(raw) => raw,
1448 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
1449 Err(err) => {
1450 return Ok(Some(BlobPreviewDTO {
1451 output_hash: None,
1452 output_bytes: None,
1453 content_type: Some("text/plain".to_string()),
1454 truncated: None,
1455 value: None,
1456 text: None,
1457 error: Some(err.to_string()),
1458 redacted: None,
1459 }));
1460 }
1461 };
1462 let bytes = raw.len();
1463 let start = bytes.saturating_sub(BLOB_PREVIEW_MAX_BYTES);
1464 let text = String::from_utf8_lossy(&raw[start..]).to_string();
1465 Ok(Some(BlobPreviewDTO {
1466 output_hash: None,
1467 output_bytes: Some(bytes),
1468 content_type: Some("text/plain".to_string()),
1469 truncated: Some(bytes > BLOB_PREVIEW_MAX_BYTES),
1470 value: None,
1471 text: Some(text),
1472 error: None,
1473 redacted: None,
1474 }))
1475}
1476
1477fn read_attempt_terminal(
1478 run_dir: &Path,
1479 activity_id: &str,
1480 attempt_id: &str,
1481) -> Result<Option<AttemptTerminalDTO>> {
1482 let path = run_dir
1483 .join("attempts")
1484 .join(activity_id)
1485 .join(attempt_id)
1486 .join("terminal.json");
1487 if !is_path_inside(run_dir, &path) {
1488 return Ok(Some(AttemptTerminalDTO {
1489 session_id: String::new(),
1490 cli_session_id: None,
1491 web_port: 0,
1492 status: "closed".to_string(),
1493 lark_app_id: None,
1494 bot_name: None,
1495 cli_id: None,
1496 working_dir: None,
1497 log_path: None,
1498 started_at: 0,
1499 updated_at: 0,
1500 closed_at: None,
1501 error: Some("terminal sidecar is outside run directory".to_string()),
1502 has_terminal_log: None,
1503 }));
1504 }
1505 let raw = match fs::read_to_string(&path) {
1506 Ok(raw) => raw,
1507 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
1508 Err(err) => {
1509 return Ok(Some(AttemptTerminalDTO {
1510 session_id: String::new(),
1511 cli_session_id: None,
1512 web_port: 0,
1513 status: "closed".to_string(),
1514 lark_app_id: None,
1515 bot_name: None,
1516 cli_id: None,
1517 working_dir: None,
1518 log_path: None,
1519 started_at: 0,
1520 updated_at: 0,
1521 closed_at: None,
1522 error: Some(err.to_string()),
1523 has_terminal_log: None,
1524 }));
1525 }
1526 };
1527 let parsed: Value = match serde_json::from_str(&raw) {
1528 Ok(value) => value,
1529 Err(err) => {
1530 return Ok(Some(AttemptTerminalDTO {
1531 session_id: String::new(),
1532 cli_session_id: None,
1533 web_port: 0,
1534 status: "closed".to_string(),
1535 lark_app_id: None,
1536 bot_name: None,
1537 cli_id: None,
1538 working_dir: None,
1539 log_path: None,
1540 started_at: 0,
1541 updated_at: 0,
1542 closed_at: None,
1543 error: Some(format!("invalid terminal sidecar: {}", err)),
1544 has_terminal_log: None,
1545 }));
1546 }
1547 };
1548 if payload_u64(&parsed, "schemaVersion") != Some(1)
1549 || payload_str(&parsed, "sessionId").is_none()
1550 || payload_u64(&parsed, "webPort").is_none()
1551 || !matches!(
1552 payload_str(&parsed, "status").as_deref(),
1553 Some("live" | "closed")
1554 )
1555 || payload_u64(&parsed, "startedAt").is_none()
1556 || payload_u64(&parsed, "updatedAt").is_none()
1557 {
1558 return Ok(Some(AttemptTerminalDTO {
1559 session_id: String::new(),
1560 cli_session_id: None,
1561 web_port: 0,
1562 status: "closed".to_string(),
1563 lark_app_id: None,
1564 bot_name: None,
1565 cli_id: None,
1566 working_dir: None,
1567 log_path: None,
1568 started_at: 0,
1569 updated_at: 0,
1570 closed_at: None,
1571 error: Some("invalid terminal sidecar".to_string()),
1572 has_terminal_log: None,
1573 }));
1574 }
1575 let terminal_log = path.with_file_name("terminal.log");
1576 let has_terminal_log = fs::metadata(&terminal_log)
1577 .map(|meta| meta.is_file() && meta.len() > 0)
1578 .unwrap_or(false);
1579 Ok(Some(AttemptTerminalDTO {
1580 session_id: payload_str(&parsed, "sessionId").unwrap_or_default(),
1581 cli_session_id: payload_str(&parsed, "cliSessionId"),
1582 web_port: payload_u64(&parsed, "webPort").unwrap_or_default() as u16,
1583 status: payload_str(&parsed, "status").unwrap_or_else(|| "closed".to_string()),
1584 lark_app_id: payload_str(&parsed, "larkAppId"),
1585 bot_name: payload_str(&parsed, "botName"),
1586 cli_id: payload_str(&parsed, "cliId"),
1587 working_dir: payload_str(&parsed, "workingDir"),
1588 log_path: payload_str(&parsed, "logPath"),
1589 started_at: payload_u64(&parsed, "startedAt").unwrap_or_default(),
1590 updated_at: payload_u64(&parsed, "updatedAt").unwrap_or_default(),
1591 closed_at: payload_u64(&parsed, "closedAt"),
1592 error: None,
1593 has_terminal_log: Some(has_terminal_log),
1594 }))
1595}
1596
1597fn preview_resolved_input(
1598 run_dir: &Path,
1599 snap: &ReplaySnapshot,
1600 def: &Value,
1601 raw_input: Value,
1602 cache: &mut HashMap<String, BlobPreviewDTO>,
1603) -> Result<BlobPreviewDTO> {
1604 match resolve_dashboard_bindings(&raw_input, run_dir, snap, def, cache) {
1605 Ok(value) => Ok(BlobPreviewDTO {
1606 output_hash: None,
1607 output_bytes: Some(serde_json::to_vec(&value)?.len()),
1608 content_type: Some("application/json".to_string()),
1609 truncated: None,
1610 value: Some(value),
1611 text: None,
1612 error: None,
1613 redacted: None,
1614 }),
1615 Err(err) => Ok(BlobPreviewDTO {
1616 output_hash: None,
1617 output_bytes: Some(serde_json::to_vec(&raw_input)?.len()),
1618 content_type: Some("application/json".to_string()),
1619 truncated: None,
1620 value: Some(raw_input),
1621 text: None,
1622 error: Some(format!("failed to resolve bindings: {}", err)),
1623 redacted: None,
1624 }),
1625 }
1626}
1627
1628fn resolve_dashboard_bindings(
1629 value: &Value,
1630 run_dir: &Path,
1631 snap: &ReplaySnapshot,
1632 def: &Value,
1633 cache: &mut HashMap<String, BlobPreviewDTO>,
1634) -> Result<Value> {
1635 if let Some(ref_spec) = value.as_object().and_then(|obj| {
1636 if obj.len() == 1 {
1637 obj.get("$ref").and_then(Value::as_str)
1638 } else {
1639 None
1640 }
1641 }) {
1642 return resolve_dashboard_ref(ref_spec, run_dir, snap, def, cache);
1643 }
1644 if let Some(s) = value.as_str() {
1645 return interpolate_dashboard_string_refs(s, run_dir, snap, def, cache).map(Value::String);
1646 }
1647 if let Some(arr) = value.as_array() {
1648 let mut out = Vec::with_capacity(arr.len());
1649 for item in arr {
1650 out.push(resolve_dashboard_bindings(item, run_dir, snap, def, cache)?);
1651 }
1652 return Ok(Value::Array(out));
1653 }
1654 if let Some(obj) = value.as_object() {
1655 let mut out = serde_json::Map::new();
1656 for (key, item) in obj {
1657 out.insert(
1658 key.clone(),
1659 resolve_dashboard_bindings(item, run_dir, snap, def, cache)?,
1660 );
1661 }
1662 return Ok(Value::Object(out));
1663 }
1664 Ok(value.clone())
1665}
1666
1667fn interpolate_dashboard_string_refs(
1668 value: &str,
1669 run_dir: &Path,
1670 snap: &ReplaySnapshot,
1671 def: &Value,
1672 cache: &mut HashMap<String, BlobPreviewDTO>,
1673) -> Result<String> {
1674 if !value.contains("${") {
1675 return Ok(value.to_string());
1676 }
1677 let mut out = String::new();
1678 let mut cursor = 0;
1679 while let Some(start) = value[cursor..].find("${") {
1680 let start = cursor + start;
1681 out.push_str(&value[cursor..start]);
1682 let end = value[start + 2..]
1683 .find('}')
1684 .ok_or_else(|| anyhow::anyhow!("unterminated string ref interpolation in '{value}'"))?
1685 + start
1686 + 2;
1687 let ref_spec = &value[start + 2..end];
1688 if ref_spec.is_empty() {
1689 anyhow::bail!("empty string ref interpolation in '{value}'");
1690 }
1691 let resolved = resolve_dashboard_ref(ref_spec, run_dir, snap, def, cache)?;
1692 out.push_str(&stringify_dashboard_interpolated_value(ref_spec, resolved)?);
1693 cursor = end + 1;
1694 }
1695 out.push_str(&value[cursor..]);
1696 Ok(out)
1697}
1698
1699fn stringify_dashboard_interpolated_value(ref_spec: &str, value: Value) -> Result<String> {
1700 match value {
1701 Value::Null => Ok("null".to_string()),
1702 Value::String(s) => Ok(s),
1703 Value::Number(n) => Ok(n.to_string()),
1704 Value::Bool(b) => Ok(b.to_string()),
1705 other => anyhow::bail!(
1706 "string interpolation '${{{}}}' resolved to {}",
1707 ref_spec,
1708 if other.is_array() { "array" } else { "object" }
1709 ),
1710 }
1711}
1712
1713fn resolve_dashboard_ref(
1714 ref_spec: &str,
1715 run_dir: &Path,
1716 snap: &ReplaySnapshot,
1717 def: &Value,
1718 cache: &mut HashMap<String, BlobPreviewDTO>,
1719) -> Result<Value> {
1720 if let Some(rest) = ref_spec.strip_prefix("params.") {
1721 let input_ref = snap
1722 .run
1723 .input
1724 .as_ref()
1725 .context(format!("$ref '{ref_spec}' requires run input"))?;
1726 let preview = preview_ref(run_dir, input_ref, cache)?;
1727 let params = preview.value.clone().context(format!(
1728 "$ref '{ref_spec}' output preview has no JSON value"
1729 ))?;
1730 return walk_preview_path(params, rest.split('.').collect::<Vec<_>>(), ref_spec);
1731 }
1732 let Some(sep_idx) = ref_spec.find(".output.") else {
1733 anyhow::bail!("$ref '{ref_spec}' missing '.output.' separator");
1734 };
1735 let node_id = &ref_spec[..sep_idx];
1736 let path = &ref_spec[sep_idx + ".output.".len()..];
1737 let node = def
1738 .get("nodes")
1739 .and_then(|nodes| nodes.get(node_id))
1740 .context(format!(
1741 "$ref '{ref_spec}' targets unknown node '{node_id}'"
1742 ))?;
1743 let output_ref = snap
1744 .outputs
1745 .get(&format!("{}::work::{}", snap.run.run_id, node_id))
1746 .context(format!("$ref '{ref_spec}' has no successful output yet"))?;
1747 let preview = preview_ref(run_dir, output_ref, cache)?;
1748 let value = preview.value.clone().context(format!(
1749 "$ref '{ref_spec}' output preview has no JSON value"
1750 ))?;
1751 let root = if node.get("type").and_then(Value::as_str) == Some("hostExecutor")
1752 && value
1753 .as_object()
1754 .map(|obj| obj.contains_key("output"))
1755 .unwrap_or(false)
1756 {
1757 value.get("output").cloned().unwrap_or(Value::Null)
1758 } else {
1759 value
1760 };
1761 walk_preview_path(root, path.split('.').collect::<Vec<_>>(), ref_spec)
1762}
1763
1764fn walk_preview_path(value: Value, segments: Vec<&str>, ref_spec: &str) -> Result<Value> {
1765 let mut cursor = value;
1766 for seg in segments {
1767 if cursor.is_null() {
1768 anyhow::bail!("$ref '{ref_spec}' hit null at '{seg}'");
1769 }
1770 if let Some(arr) = cursor.as_array() {
1771 let idx: usize = seg
1772 .parse()
1773 .map_err(|_| anyhow::anyhow!("$ref '{ref_spec}' array index '{seg}' invalid"))?;
1774 cursor = arr.get(idx).cloned().context(format!(
1775 "$ref '{ref_spec}' array index '{seg}' out of bounds"
1776 ))?;
1777 continue;
1778 }
1779 let obj = cursor
1780 .as_object()
1781 .context(format!("$ref '{ref_spec}' segment '{seg}' not found"))?;
1782 cursor = obj
1783 .get(seg)
1784 .cloned()
1785 .context(format!("$ref '{ref_spec}' segment '{seg}' not found"))?;
1786 }
1787 Ok(cursor)
1788}
1789
1790fn is_json_content(content_type: Option<&str>) -> bool {
1791 content_type
1792 .unwrap_or_default()
1793 .to_ascii_lowercase()
1794 .contains("json")
1795}
1796
1797fn parse_loop_activity_id(activity_id: &str) -> Option<(String, u64)> {
1798 let loop_start = activity_id.find("::loop::")?;
1799 let after_loop = &activity_id[loop_start + "::loop::".len()..];
1800 let iter_end = after_loop.find("::")?;
1801 let loop_part = &after_loop[..iter_end];
1802 let (loop_id, iteration) = loop_part.rsplit_once('.')?;
1803 let iteration = iteration.parse().ok()?;
1804 Some((loop_id.to_string(), iteration))
1805}
1806
1807fn is_path_inside(parent: &Path, child: &Path) -> bool {
1808 let parent = parent
1809 .canonicalize()
1810 .unwrap_or_else(|_| parent.to_path_buf());
1811 let child = child.canonicalize().unwrap_or_else(|_| child.to_path_buf());
1812 child.starts_with(&parent)
1813}
1814
1815#[cfg(test)]
1816mod tests {
1817 use super::*;
1818 use std::path::PathBuf;
1819 use std::time::{SystemTime, UNIX_EPOCH};
1820
1821 fn temp_run_dir(label: &str) -> PathBuf {
1822 std::env::temp_dir().join(format!(
1823 "beam-workflow-snapshot-{label}-{}-{}",
1824 std::process::id(),
1825 SystemTime::now()
1826 .duration_since(UNIX_EPOCH)
1827 .unwrap_or_default()
1828 .as_nanos()
1829 ))
1830 }
1831
1832 fn env(
1833 event_id: &str,
1834 run_id: &str,
1835 event_type: &str,
1836 payload: Value,
1837 ) -> WorkflowEventEnvelope {
1838 WorkflowEventEnvelope {
1839 event_id: event_id.to_string(),
1840 run_id: run_id.to_string(),
1841 timestamp: 1,
1842 schema_version: 1,
1843 actor: crate::WorkflowActor::System,
1844 event_type: event_type.to_string(),
1845 payload,
1846 payload_hash: None,
1847 }
1848 }
1849
1850 #[test]
1851 fn replay_projects_basic_state_and_outputs() {
1852 let run_id = "run-1";
1853 let events = vec![
1854 env(
1855 "run-1-1",
1856 run_id,
1857 "runCreated",
1858 serde_json::json!({
1859 "workflowId": "flow-a",
1860 "revisionId": "rev-a",
1861 "inputRef": {
1862 "outputHash": "sha256:input",
1863 "outputPath": "/tmp/input.json",
1864 "outputBytes": 2,
1865 "outputSchemaVersion": 1,
1866 "contentType": "application/json"
1867 },
1868 "initiator": "cli"
1869 }),
1870 ),
1871 env("run-1-2", run_id, "runStarted", serde_json::json!({})),
1872 env(
1873 "run-1-3",
1874 run_id,
1875 "attemptCreated",
1876 serde_json::json!({
1877 "activityId": "run-1::work::node-a",
1878 "attemptId": "run-1::work::node-a::att-1",
1879 "attemptNumber": 1,
1880 "nodeId": "node-a",
1881 "inputRef": {
1882 "outputHash": "sha256:input",
1883 "outputPath": "/tmp/input.json",
1884 "outputBytes": 2,
1885 "outputSchemaVersion": 1,
1886 "contentType": "application/json"
1887 }
1888 }),
1889 ),
1890 env(
1891 "run-1-4",
1892 run_id,
1893 "activitySucceeded",
1894 serde_json::json!({
1895 "activityId": "run-1::work::node-a",
1896 "attemptId": "run-1::work::node-a::att-1",
1897 "outputRef": {
1898 "outputHash": "sha256:output",
1899 "outputPath": "/tmp/output.json",
1900 "outputBytes": 17,
1901 "outputSchemaVersion": 1,
1902 "contentType": "application/json"
1903 },
1904 "externalRefs": {"ok": true}
1905 }),
1906 ),
1907 ];
1908 let snap = replay_events(&events).expect("replay");
1909 assert_eq!(snap.run.workflow_id.as_deref(), Some("flow-a"));
1910 assert_eq!(snap.run.status, RunStatus::Running);
1911 assert_eq!(snap.activities.len(), 1);
1912 assert_eq!(snap.outputs.len(), 1);
1913 assert_eq!(snap.dangling_activities, Vec::<String>::new());
1914 }
1915
1916 #[tokio::test]
1917 async fn read_run_snapshot_replays_outputs_and_binding() {
1918 let run_dir = temp_run_dir("read");
1919 fs::create_dir_all(run_dir.join("blobs")).unwrap();
1920 fs::create_dir_all(
1921 run_dir
1922 .join("attempts")
1923 .join("run-1::work::node-a")
1924 .join("run-1::work::node-a::att-1"),
1925 )
1926 .unwrap();
1927 fs::write(
1928 run_dir.join("workflow.json"),
1929 r#"{"workflowId":"flow-a","nodes":{"node-a":{"type":"hostExecutor"}}}"#,
1930 )
1931 .unwrap();
1932 fs::write(
1933 run_dir.join("chat-binding.json"),
1934 r#"{"chatId":"chat-1","larkAppId":"app-1"}"#,
1935 )
1936 .unwrap();
1937 fs::write(run_dir.join("blobs").join("input"), br#"{"foo":"bar"}"#).unwrap();
1938 fs::write(
1939 run_dir.join("blobs").join("output"),
1940 br#"{"output":{"hello":"world"},"externalRefs":{"ok":true}}"#,
1941 )
1942 .unwrap();
1943 fs::write(
1944 run_dir
1945 .join("attempts")
1946 .join("run-1::work::node-a")
1947 .join("run-1::work::node-a::att-1")
1948 .join("terminal.log"),
1949 "hello world",
1950 )
1951 .unwrap();
1952 fs::write(
1953 run_dir.join("attempts").join("run-1::work::node-a").join("run-1::work::node-a::att-1").join("terminal.json"),
1954 r#"{"schemaVersion":1,"sessionId":"sess-1","webPort":8080,"status":"live","startedAt":1,"updatedAt":2}"#,
1955 )
1956 .unwrap();
1957 let events = vec![
1958 env(
1959 "run-1-1",
1960 "run-1",
1961 "runCreated",
1962 serde_json::json!({
1963 "workflowId":"flow-a",
1964 "revisionId":"rev-a",
1965 "inputRef": {
1966 "outputHash": "sha256:input",
1967 "outputPath": run_dir.join("blobs").join("input"),
1968 "outputBytes": 13,
1969 "outputSchemaVersion": 1,
1970 "contentType": "application/json"
1971 }
1972 }),
1973 ),
1974 env("run-1-2", "run-1", "runStarted", serde_json::json!({})),
1975 env(
1976 "run-1-3",
1977 "run-1",
1978 "attemptCreated",
1979 serde_json::json!({
1980 "activityId": "run-1::work::node-a",
1981 "attemptId": "run-1::work::node-a::att-1",
1982 "attemptNumber": 1,
1983 "nodeId": "node-a",
1984 "inputRef": {
1985 "outputHash": "sha256:input",
1986 "outputPath": run_dir.join("blobs").join("input"),
1987 "outputBytes": 13,
1988 "outputSchemaVersion": 1,
1989 "contentType": "application/json"
1990 }
1991 }),
1992 ),
1993 env(
1994 "run-1-4",
1995 "run-1",
1996 "activitySucceeded",
1997 serde_json::json!({
1998 "activityId": "run-1::work::node-a",
1999 "attemptId": "run-1::work::node-a::att-1",
2000 "outputRef": {
2001 "outputHash": "sha256:output",
2002 "outputPath": run_dir.join("blobs").join("output"),
2003 "outputBytes": 55,
2004 "outputSchemaVersion": 1,
2005 "contentType": "application/json"
2006 },
2007 "externalRefs": {"ok": true}
2008 }),
2009 ),
2010 ];
2011 let events_json = events
2012 .iter()
2013 .map(|ev| serde_json::to_string(ev).unwrap())
2014 .collect::<Vec<_>>()
2015 .join("\n")
2016 + "\n";
2017 fs::write(run_dir.join("events.ndjson"), events_json).unwrap();
2018
2019 let snapshot = read_run_snapshot(&run_dir)
2020 .await
2021 .expect("snapshot")
2022 .expect("some");
2023 assert_eq!(snapshot.run.workflow_id.as_deref(), Some("flow-a"));
2024 assert_eq!(
2025 snapshot.chat_binding.as_ref().map(|b| b.chat_id.as_str()),
2026 Some("chat-1")
2027 );
2028 assert_eq!(snapshot.activities.len(), 1);
2029 assert_eq!(snapshot.outputs.len(), 1);
2030 assert!(
2031 snapshot
2032 .attempt_io
2033 .contains_key("run-1::work::node-a::att-1")
2034 );
2035 let io = snapshot
2036 .attempt_io
2037 .get("run-1::work::node-a::att-1")
2038 .expect("attempt io");
2039 assert!(io.input.as_ref().and_then(|p| p.value.as_ref()).is_some());
2040 assert!(io.output.as_ref().and_then(|p| p.value.as_ref()).is_some());
2041 let _ = fs::remove_dir_all(&run_dir);
2042 }
2043}