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