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