1use std::fs;
2use std::path::PathBuf;
3
4use anyhow::{Context, Result};
5use serde_json::Value;
6use sha2::{Digest, Sha256};
7
8use crate::workflow_snapshot::ActivityStatus;
9use crate::{
10 AttemptState, BeamPaths, EventDraft, EventLog, WorkflowActor, WorkflowOutputRef, get_task,
11 read_run_snapshot,
12};
13
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct ScheduleResumeOutcome {
16 pub activity_id: String,
17 pub attempt_id: String,
18 pub decision: String,
19}
20
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct ScheduleResumeResult {
23 pub reconciled: Vec<ScheduleResumeOutcome>,
24 pub fresh_retry: Vec<ScheduleResumeOutcome>,
25 pub skipped: Vec<String>,
26}
27
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub enum PriorReconcileRecoveryOutcome {
30 Recovered {
31 activity_id: String,
32 attempt_id: String,
33 decision: String,
34 },
35 FreshRetry {
36 activity_id: String,
37 attempt_id: String,
38 },
39}
40
41pub async fn recover_prior_reconcile_result(
42 log: &mut EventLog,
43 activity_id: &str,
44 attempt: &AttemptState,
45) -> Result<Option<PriorReconcileRecoveryOutcome>> {
46 let Some(rr) = attempt.latest_reconcile_result.as_ref() else {
47 return Ok(None);
48 };
49 if matches!(
50 attempt.status,
51 ActivityStatus::Succeeded
52 | ActivityStatus::Failed
53 | ActivityStatus::TimedOut
54 | ActivityStatus::Cancelled
55 ) {
56 return Ok(None);
57 }
58
59 match rr.decision.as_str() {
60 "completedByIdempotentSubmit" => {
61 let external_refs = rr
62 .evidence
63 .get("externalRefs")
64 .cloned()
65 .and_then(|value| value.as_object().cloned().map(serde_json::Value::Object));
66 let Some(external_refs) = external_refs else {
67 let _ = log.append(EventDraft {
68 event_type: "activityFailed".to_string(),
69 actor: WorkflowActor::System,
70 payload: serde_json::json!({
71 "activityId": activity_id,
72 "attemptId": attempt.attempt_id,
73 "error": {
74 "errorCode": "CorruptLog",
75 "errorClass": "manual",
76 "errorMessage": "Prior reconcileResult{decision=completedByIdempotentSubmit} is missing evidence.externalRefs (or it is not an object) — refusing to fabricate an activitySucceeded from empty refs.",
77 }
78 }),
79 timestamp: None,
80 payload_hash: None,
81 })?;
82 return Ok(Some(PriorReconcileRecoveryOutcome::Recovered {
83 activity_id: activity_id.to_string(),
84 attempt_id: attempt.attempt_id.clone(),
85 decision: "manual".to_string(),
86 }));
87 };
88 let output_ref = write_json_blob(&mut *log, serde_json::json!(&external_refs))?;
89 let _ = log.append(EventDraft {
90 event_type: "activitySucceeded".to_string(),
91 actor: WorkflowActor::System,
92 payload: serde_json::json!({
93 "activityId": activity_id,
94 "attemptId": attempt.attempt_id,
95 "outputRef": output_ref,
96 "externalRefs": external_refs,
97 }),
98 timestamp: None,
99 payload_hash: None,
100 })?;
101 Ok(Some(PriorReconcileRecoveryOutcome::Recovered {
102 activity_id: activity_id.to_string(),
103 attempt_id: attempt.attempt_id.clone(),
104 decision: "completedByIdempotentSubmit".to_string(),
105 }))
106 }
107 "manual" => {
108 let error_code = rr
109 .evidence
110 .get("errorCode")
111 .and_then(serde_json::Value::as_str)
112 .unwrap_or("UnknownProviderError");
113 let _ = log.append(EventDraft {
114 event_type: "activityFailed".to_string(),
115 actor: WorkflowActor::System,
116 payload: serde_json::json!({
117 "activityId": activity_id,
118 "attemptId": attempt.attempt_id,
119 "error": {
120 "errorCode": error_code,
121 "errorClass": "manual",
122 "errorMessage": format!(
123 "Recovered from prior crashed reconcile cycle (decision=manual, errorCode={}).",
124 error_code
125 ),
126 }
127 }),
128 timestamp: None,
129 payload_hash: None,
130 })?;
131 Ok(Some(PriorReconcileRecoveryOutcome::Recovered {
132 activity_id: activity_id.to_string(),
133 attempt_id: attempt.attempt_id.clone(),
134 decision: "manual".to_string(),
135 }))
136 }
137 "freshRetry" => Ok(Some(PriorReconcileRecoveryOutcome::FreshRetry {
138 activity_id: activity_id.to_string(),
139 attempt_id: attempt.attempt_id.clone(),
140 })),
141 "replayed" => {
142 let _ = log.append(EventDraft {
143 event_type: "activityFailed".to_string(),
144 actor: WorkflowActor::System,
145 payload: serde_json::json!({
146 "activityId": activity_id,
147 "attemptId": attempt.attempt_id,
148 "error": {
149 "errorCode": "CorruptLog",
150 "errorClass": "manual",
151 "errorMessage": "Prior reconcileResult decision=replayed but no terminal event present — log inconsistency.",
152 }
153 }),
154 timestamp: None,
155 payload_hash: None,
156 })?;
157 Ok(Some(PriorReconcileRecoveryOutcome::Recovered {
158 activity_id: activity_id.to_string(),
159 attempt_id: attempt.attempt_id.clone(),
160 decision: "manual".to_string(),
161 }))
162 }
163 _ => Ok(None),
164 }
165}
166
167pub async fn resume_schedule_dangling_effects(
168 log: &mut EventLog,
169 paths: &BeamPaths,
170 daemon_id: &str,
171 reason: Option<&str>,
172) -> Result<ScheduleResumeResult> {
173 let last_seen_event_id = log
174 .read_all()?
175 .last()
176 .map(|event| event.event_id.clone())
177 .unwrap_or_default();
178 let snapshot = read_run_snapshot(&log.run_dir)
179 .await?
180 .context("workflow resume requires an existing run snapshot")?;
181 let _ = log.append(EventDraft {
182 event_type: "resumeStarted".to_string(),
183 actor: WorkflowActor::System,
184 payload: serde_json::json!({
185 "daemonId": daemon_id,
186 "lastSeenEventId": last_seen_event_id,
187 "reason": reason,
188 }),
189 timestamp: None,
190 payload_hash: None,
191 })?;
192
193 let mut reconciled = Vec::new();
194 let mut fresh_retry = Vec::new();
195 let mut skipped = Vec::new();
196
197 for activity_id in &snapshot.dangling.effect_attempted {
198 let Some(activity) = snapshot
199 .activities
200 .iter()
201 .find(|a| &a.activity_id == activity_id)
202 else {
203 skipped.push(activity_id.clone());
204 continue;
205 };
206 let Some(latest) = activity.attempts.last() else {
207 skipped.push(activity_id.clone());
208 continue;
209 };
210 let Some(effect_attempted) = latest.effect_attempted.as_ref() else {
211 skipped.push(activity_id.clone());
212 continue;
213 };
214 if effect_attempted.provider != "beam-schedule" {
215 skipped.push(activity_id.clone());
216 continue;
217 }
218
219 if let Some(recovery) =
220 recover_prior_reconcile_result(&mut *log, activity_id, latest).await?
221 {
222 match recovery {
223 PriorReconcileRecoveryOutcome::Recovered {
224 activity_id,
225 attempt_id,
226 decision,
227 } => {
228 reconciled.push(ScheduleResumeOutcome {
229 activity_id,
230 attempt_id,
231 decision,
232 });
233 }
234 PriorReconcileRecoveryOutcome::FreshRetry {
235 activity_id,
236 attempt_id,
237 } => {
238 fresh_retry.push(ScheduleResumeOutcome {
239 activity_id,
240 attempt_id,
241 decision: "freshRetry".to_string(),
242 });
243 }
244 }
245 continue;
246 }
247
248 match get_task(paths, &effect_attempted.idempotency_key)? {
249 Some(task) => {
250 let external_refs = serde_json::json!({ "taskId": task.id });
251 let output_ref = write_json_blob(&mut *log, external_refs.clone())?;
252 let _ = log.append(EventDraft {
253 event_type: "reconcileResult".to_string(),
254 actor: WorkflowActor::System,
255 payload: serde_json::json!({
256 "activityId": activity_id,
257 "idempotencyKey": effect_attempted.idempotency_key,
258 "capability": "readOnlyLookup",
259 "decision": "completedByIdempotentSubmit",
260 "evidence": {
261 "source": "getTask",
262 "externalRefs": external_refs,
263 },
264 }),
265 timestamp: None,
266 payload_hash: None,
267 })?;
268 let _ = log.append(EventDraft {
269 event_type: "activitySucceeded".to_string(),
270 actor: WorkflowActor::System,
271 payload: serde_json::json!({
272 "activityId": activity_id,
273 "attemptId": latest.attempt_id,
274 "outputRef": output_ref,
275 "externalRefs": { "taskId": task.id },
276 }),
277 timestamp: None,
278 payload_hash: None,
279 })?;
280 reconciled.push(ScheduleResumeOutcome {
281 activity_id: activity_id.clone(),
282 attempt_id: latest.attempt_id.clone(),
283 decision: "completedByIdempotentSubmit".to_string(),
284 });
285 }
286 None => {
287 let _ = log.append(EventDraft {
288 event_type: "reconcileResult".to_string(),
289 actor: WorkflowActor::System,
290 payload: serde_json::json!({
291 "activityId": activity_id,
292 "idempotencyKey": effect_attempted.idempotency_key,
293 "capability": "readOnlyLookup",
294 "decision": "freshRetry",
295 "evidence": {
296 "source": "getTask",
297 "returned": "undefined",
298 },
299 }),
300 timestamp: None,
301 payload_hash: None,
302 })?;
303 fresh_retry.push(ScheduleResumeOutcome {
304 activity_id: activity_id.clone(),
305 attempt_id: latest.attempt_id.clone(),
306 decision: "freshRetry".to_string(),
307 });
308 }
309 }
310 }
311
312 Ok(ScheduleResumeResult {
313 reconciled,
314 fresh_retry,
315 skipped,
316 })
317}
318
319fn write_json_blob(log: &mut EventLog, value: Value) -> Result<WorkflowOutputRef> {
320 let bytes = serde_json::to_vec(&value)?;
321 let hash = sha256_hex(&bytes);
322 let path = PathBuf::from(&log.blob_dir).join(&hash);
323 fs::write(&path, &bytes)?;
324 Ok(WorkflowOutputRef {
325 output_hash: format!("sha256:{hash}"),
326 output_path: path.display().to_string(),
327 output_bytes: bytes.len(),
328 output_schema_version: 1,
329 content_type: Some("application/json".to_string()),
330 })
331}
332
333fn sha256_hex(bytes: &[u8]) -> String {
334 let mut hasher = Sha256::new();
335 hasher.update(bytes);
336 lower_hex(&hasher.finalize())
337}
338
339fn lower_hex(bytes: &[u8]) -> String {
340 const HEX: &[u8; 16] = b"0123456789abcdef";
341 let mut out = String::with_capacity(bytes.len() * 2);
342 for byte in bytes {
343 out.push(HEX[(byte >> 4) as usize] as char);
344 out.push(HEX[(byte & 0x0f) as usize] as char);
345 }
346 out
347}
348
349#[cfg(test)]
350mod tests {
351 use super::*;
352 use crate::{
353 BootstrapWorkflowRunInput, CreateTaskInput, ParsedSchedule, ParsedScheduleKind,
354 RunChatBinding, bootstrap_workflow_run, create_task,
355 };
356 use std::collections::BTreeMap;
357 use std::time::{SystemTime, UNIX_EPOCH};
358
359 fn temp_paths(label: &str) -> BeamPaths {
360 let nanos = SystemTime::now()
361 .duration_since(UNIX_EPOCH)
362 .unwrap_or_default()
363 .as_nanos();
364 BeamPaths::from_root(std::env::temp_dir().join(format!(
365 "beam-workflow-resume-{label}-{nanos}-{}",
366 std::process::id()
367 )))
368 }
369
370 #[tokio::test]
371 async fn schedule_resume_reconciles_found_task() {
372 let paths = temp_paths("found");
373 let _ = std::fs::remove_dir_all(paths.root());
374 let params: BTreeMap<String, Value> =
375 BTreeMap::from([(String::from("name"), Value::String("beam".to_string()))]);
376 let run_id = "run-1";
377 bootstrap_workflow_run(
378 &paths,
379 BootstrapWorkflowRunInput {
380 run_id,
381 workflow_json: r#"{"workflowId":"flow-a","version":1,"params":{"name":{"type":"string"}},"nodes":{"a":{"type":"hostExecutor","executor":"beam-schedule","input":{"name":"schedule-demo daily 9am","schedule":"0 9 * * *","parsed":{"kind":"cron","expr":"0 9 * * *","display":"0 9 * * *"},"prompt":"Schedule demo","workingDir":"/tmp/beam-schedule-demo","chatId":"oc_workflow_demo","scope":"thread"},"unsafeAllowUngated":true}}}"#,
382 expected_workflow_id: Some("flow-a"),
383 params: ¶ms,
384 initiator: "cli",
385 chat_binding: Some(RunChatBinding {
386 chat_id: "chat-1".to_string(),
387 lark_app_id: "app-1".to_string(),
388 }),
389 },
390 )
391 .unwrap();
392 {
393 let mut log = EventLog::new(run_id, paths.workflow_runs_dir()).unwrap();
394 let _ = log
395 .append(EventDraft {
396 event_type: "attemptCreated".to_string(),
397 actor: WorkflowActor::Scheduler,
398 payload: serde_json::json!({
399 "nodeId": "a",
400 "activityId": "act-1",
401 "attemptId": "act-1::att-1",
402 "attemptNumber": 1,
403 "inputRef": {
404 "outputHash": "sha256:dummy",
405 "outputPath": "dummy",
406 "outputBytes": 1,
407 "outputSchemaVersion": 1,
408 "contentType": "application/json",
409 }
410 }),
411 timestamp: None,
412 payload_hash: None,
413 })
414 .unwrap();
415 let _ = log
416 .append(EventDraft {
417 event_type: "effectAttempted".to_string(),
418 actor: WorkflowActor::HostExecutor,
419 payload: serde_json::json!({
420 "activityId": "act-1",
421 "attemptId": "act-1::att-1",
422 "idempotencyKey": "wf_key",
423 "inputHash": "sha256:1",
424 "idempotencyTtlMs": 9999999u64,
425 "provider": "beam-schedule",
426 }),
427 timestamp: None,
428 payload_hash: None,
429 })
430 .unwrap();
431 create_task(
432 &paths,
433 CreateTaskInput {
434 id: Some("wf_key".to_string()),
435 name: "schedule-demo daily 9am".to_string(),
436 schedule: "0 9 * * *".to_string(),
437 parsed: ParsedSchedule {
438 kind: ParsedScheduleKind::Cron,
439 run_at: None,
440 minutes: None,
441 expr: Some("0 9 * * *".to_string()),
442 display: "0 9 * * *".to_string(),
443 },
444 prompt: "Schedule demo".to_string(),
445 working_dir: "/tmp/beam-schedule-demo".to_string(),
446 chat_id: "oc_workflow_demo".to_string(),
447 root_message_id: None,
448 scope: Some("thread".to_string()),
449 chat_type: None,
450 lark_app_id: None,
451 creator_chat_id: None,
452 creator_root_message_id: None,
453 creator_lark_app_id: None,
454 next_run_at: None,
455 repeat: None,
456 deliver: None,
457 },
458 )
459 .unwrap();
460 }
461 let mut log = EventLog::new(run_id, paths.workflow_runs_dir()).unwrap();
462 let result = resume_schedule_dangling_effects(&mut log, &paths, "daemon-1", None)
463 .await
464 .unwrap();
465 assert_eq!(result.reconciled.len(), 1);
466 assert_eq!(result.reconciled[0].decision, "completedByIdempotentSubmit");
467 let events = log.read_all().unwrap();
468 assert!(events.iter().any(|e| e.event_type == "reconcileResult"));
469 assert!(events.iter().any(|e| e.event_type == "activitySucceeded"));
470 let _ = std::fs::remove_dir_all(paths.root());
471 }
472
473 #[tokio::test]
474 async fn recover_prior_reconcile_result_materializes_terminal_without_recalling_provider() {
475 let paths = temp_paths("prior-reconcile");
476 let _ = std::fs::remove_dir_all(paths.root());
477 let run_id = "run-recover-1";
478 bootstrap_workflow_run(
479 &paths,
480 BootstrapWorkflowRunInput {
481 run_id,
482 workflow_json: r#"{"workflowId":"flow-a","version":1,"nodes":{"a":{"type":"hostExecutor","executor":"feishu-im","input":{"larkAppId":"app-1","chatId":"chat-1","content":"hello"}}}}"#,
483 expected_workflow_id: Some("flow-a"),
484 params: &BTreeMap::new(),
485 initiator: "cli",
486 chat_binding: None,
487 },
488 )
489 .unwrap();
490 {
491 let mut log = EventLog::new(run_id, paths.workflow_runs_dir()).unwrap();
492 let _ = log
493 .append(EventDraft {
494 event_type: "attemptCreated".to_string(),
495 actor: WorkflowActor::Scheduler,
496 payload: serde_json::json!({
497 "nodeId": "a",
498 "activityId": "act-1",
499 "attemptId": "act-1::att-1",
500 "attemptNumber": 1,
501 "inputRef": {
502 "outputHash": "sha256:dummy",
503 "outputPath": paths.workflow_run_dir(run_id).join("blobs").join("dummy").display().to_string(),
504 "outputBytes": 2,
505 "outputSchemaVersion": 1,
506 "contentType": "application/json",
507 }
508 }),
509 timestamp: None,
510 payload_hash: None,
511 })
512 .unwrap();
513 let _ = log
514 .append(EventDraft {
515 event_type: "effectAttempted".to_string(),
516 actor: WorkflowActor::HostExecutor,
517 payload: serde_json::json!({
518 "activityId": "act-1",
519 "attemptId": "act-1::att-1",
520 "idempotencyKey": "wf_key",
521 "inputHash": "sha256:1",
522 "idempotencyTtlMs": 9999999u64,
523 "provider": "feishu-im",
524 }),
525 timestamp: None,
526 payload_hash: None,
527 })
528 .unwrap();
529 let _ = log
530 .append(EventDraft {
531 event_type: "reconcileResult".to_string(),
532 actor: WorkflowActor::System,
533 payload: serde_json::json!({
534 "activityId": "act-1",
535 "attemptId": "act-1::att-1",
536 "idempotencyKey": "wf_key",
537 "capability": "idempotentSubmit",
538 "decision": "completedByIdempotentSubmit",
539 "evidence": {
540 "externalRefs": { "messageId": "msg-1" }
541 },
542 }),
543 timestamp: None,
544 payload_hash: None,
545 })
546 .unwrap();
547
548 let snapshot = read_run_snapshot(&log.run_dir).await.unwrap().unwrap();
549 let attempt = &snapshot.activities[0].attempts[0];
550 let outcome = recover_prior_reconcile_result(&mut log, "act-1", attempt)
551 .await
552 .unwrap()
553 .expect("recovery");
554 match outcome {
555 PriorReconcileRecoveryOutcome::Recovered { decision, .. } => {
556 assert_eq!(decision, "completedByIdempotentSubmit");
557 }
558 other => panic!("unexpected recovery outcome: {:?}", other),
559 }
560 }
561 let events = EventLog::new(run_id, paths.workflow_runs_dir())
562 .unwrap()
563 .read_all()
564 .unwrap();
565 assert!(events.iter().any(|e| e.event_type == "activitySucceeded"));
566 let _ = std::fs::remove_dir_all(paths.root());
567 }
568}