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 format!("{:x}", hasher.finalize())
337}
338
339#[cfg(test)]
340mod tests {
341 use super::*;
342 use crate::{
343 BootstrapWorkflowRunInput, CreateTaskInput, ParsedSchedule, ParsedScheduleKind,
344 RunChatBinding, bootstrap_workflow_run, create_task,
345 };
346 use std::collections::BTreeMap;
347 use std::time::{SystemTime, UNIX_EPOCH};
348
349 fn temp_paths(label: &str) -> BeamPaths {
350 let nanos = SystemTime::now()
351 .duration_since(UNIX_EPOCH)
352 .unwrap_or_default()
353 .as_nanos();
354 BeamPaths::from_root(std::env::temp_dir().join(format!(
355 "beam-workflow-resume-{label}-{nanos}-{}",
356 std::process::id()
357 )))
358 }
359
360 #[tokio::test]
361 async fn schedule_resume_reconciles_found_task() {
362 let paths = temp_paths("found");
363 let _ = std::fs::remove_dir_all(paths.root());
364 let params: BTreeMap<String, Value> = BTreeMap::from([
365 (String::from("name"), Value::String("beam".to_string())),
366 ]);
367 let run_id = "run-1";
368 bootstrap_workflow_run(
369 &paths,
370 BootstrapWorkflowRunInput {
371 run_id,
372 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}}}"#,
373 expected_workflow_id: Some("flow-a"),
374 params: ¶ms,
375 initiator: "cli",
376 chat_binding: Some(RunChatBinding {
377 chat_id: "chat-1".to_string(),
378 lark_app_id: "app-1".to_string(),
379 }),
380 },
381 )
382 .unwrap();
383 {
384 let mut log = EventLog::new(run_id, paths.workflow_runs_dir()).unwrap();
385 let _ = log
386 .append(EventDraft {
387 event_type: "attemptCreated".to_string(),
388 actor: WorkflowActor::Scheduler,
389 payload: serde_json::json!({
390 "nodeId": "a",
391 "activityId": "act-1",
392 "attemptId": "act-1::att-1",
393 "attemptNumber": 1,
394 "inputRef": {
395 "outputHash": "sha256:dummy",
396 "outputPath": "dummy",
397 "outputBytes": 1,
398 "outputSchemaVersion": 1,
399 "contentType": "application/json",
400 }
401 }),
402 timestamp: None,
403 payload_hash: None,
404 })
405 .unwrap();
406 let _ = log
407 .append(EventDraft {
408 event_type: "effectAttempted".to_string(),
409 actor: WorkflowActor::HostExecutor,
410 payload: serde_json::json!({
411 "activityId": "act-1",
412 "attemptId": "act-1::att-1",
413 "idempotencyKey": "wf_key",
414 "inputHash": "sha256:1",
415 "idempotencyTtlMs": 9999999u64,
416 "provider": "beam-schedule",
417 }),
418 timestamp: None,
419 payload_hash: None,
420 })
421 .unwrap();
422 create_task(
423 &paths,
424 CreateTaskInput {
425 id: Some("wf_key".to_string()),
426 name: "schedule-demo daily 9am".to_string(),
427 schedule: "0 9 * * *".to_string(),
428 parsed: ParsedSchedule {
429 kind: ParsedScheduleKind::Cron,
430 run_at: None,
431 minutes: None,
432 expr: Some("0 9 * * *".to_string()),
433 display: "0 9 * * *".to_string(),
434 },
435 prompt: "Schedule demo".to_string(),
436 working_dir: "/tmp/beam-schedule-demo".to_string(),
437 chat_id: "oc_workflow_demo".to_string(),
438 root_message_id: None,
439 scope: Some("thread".to_string()),
440 chat_type: None,
441 lark_app_id: None,
442 creator_chat_id: None,
443 creator_root_message_id: None,
444 creator_lark_app_id: None,
445 next_run_at: None,
446 repeat: None,
447 deliver: None,
448 },
449 )
450 .unwrap();
451 }
452 let mut log = EventLog::new(run_id, paths.workflow_runs_dir()).unwrap();
453 let result = resume_schedule_dangling_effects(&mut log, &paths, "daemon-1", None)
454 .await
455 .unwrap();
456 assert_eq!(result.reconciled.len(), 1);
457 assert_eq!(result.reconciled[0].decision, "completedByIdempotentSubmit");
458 let events = log.read_all().unwrap();
459 assert!(events.iter().any(|e| e.event_type == "reconcileResult"));
460 assert!(events.iter().any(|e| e.event_type == "activitySucceeded"));
461 let _ = std::fs::remove_dir_all(paths.root());
462 }
463
464 #[tokio::test]
465 async fn recover_prior_reconcile_result_materializes_terminal_without_recalling_provider() {
466 let paths = temp_paths("prior-reconcile");
467 let _ = std::fs::remove_dir_all(paths.root());
468 let run_id = "run-recover-1";
469 bootstrap_workflow_run(
470 &paths,
471 BootstrapWorkflowRunInput {
472 run_id,
473 workflow_json: r#"{"workflowId":"flow-a","version":1,"nodes":{"a":{"type":"hostExecutor","executor":"feishu-im","input":{"larkAppId":"app-1","chatId":"chat-1","content":"hello"}}}}"#,
474 expected_workflow_id: Some("flow-a"),
475 params: &BTreeMap::new(),
476 initiator: "cli",
477 chat_binding: None,
478 },
479 )
480 .unwrap();
481 {
482 let mut log = EventLog::new(run_id, paths.workflow_runs_dir()).unwrap();
483 let _ = log
484 .append(EventDraft {
485 event_type: "attemptCreated".to_string(),
486 actor: WorkflowActor::Scheduler,
487 payload: serde_json::json!({
488 "nodeId": "a",
489 "activityId": "act-1",
490 "attemptId": "act-1::att-1",
491 "attemptNumber": 1,
492 "inputRef": {
493 "outputHash": "sha256:dummy",
494 "outputPath": paths.workflow_run_dir(run_id).join("blobs").join("dummy").display().to_string(),
495 "outputBytes": 2,
496 "outputSchemaVersion": 1,
497 "contentType": "application/json",
498 }
499 }),
500 timestamp: None,
501 payload_hash: None,
502 })
503 .unwrap();
504 let _ = log
505 .append(EventDraft {
506 event_type: "effectAttempted".to_string(),
507 actor: WorkflowActor::HostExecutor,
508 payload: serde_json::json!({
509 "activityId": "act-1",
510 "attemptId": "act-1::att-1",
511 "idempotencyKey": "wf_key",
512 "inputHash": "sha256:1",
513 "idempotencyTtlMs": 9999999u64,
514 "provider": "feishu-im",
515 }),
516 timestamp: None,
517 payload_hash: None,
518 })
519 .unwrap();
520 let _ = log
521 .append(EventDraft {
522 event_type: "reconcileResult".to_string(),
523 actor: WorkflowActor::System,
524 payload: serde_json::json!({
525 "activityId": "act-1",
526 "attemptId": "act-1::att-1",
527 "idempotencyKey": "wf_key",
528 "capability": "idempotentSubmit",
529 "decision": "completedByIdempotentSubmit",
530 "evidence": {
531 "externalRefs": { "messageId": "msg-1" }
532 },
533 }),
534 timestamp: None,
535 payload_hash: None,
536 })
537 .unwrap();
538
539 let snapshot = read_run_snapshot(&log.run_dir).await.unwrap().unwrap();
540 let attempt = &snapshot.activities[0].attempts[0];
541 let outcome = recover_prior_reconcile_result(&mut log, "act-1", attempt)
542 .await
543 .unwrap()
544 .expect("recovery");
545 match outcome {
546 PriorReconcileRecoveryOutcome::Recovered { decision, .. } => {
547 assert_eq!(decision, "completedByIdempotentSubmit");
548 }
549 other => panic!("unexpected recovery outcome: {:?}", other),
550 }
551 }
552 let events = EventLog::new(run_id, paths.workflow_runs_dir())
553 .unwrap()
554 .read_all()
555 .unwrap();
556 assert!(events.iter().any(|e| e.event_type == "activitySucceeded"));
557 let _ = std::fs::remove_dir_all(paths.root());
558 }
559}