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