feat(wflow): sleep and messages

[dman-os]

Summary by CodeRabbit

• New Features

  • Jobs can pause (sleep) and wait for JSON messages; hosts can send messages to running jobs. SDK exposes sleep/recv APIs with decoding support. Runtime and scheduler now record timer/message wait states and timer firings.

• Bug Fixes

  • Safer dispatch failure handling with rollback attempts on queued workflow jobs.
  • Platform-aware keyring usage and stricter error propagation for credential operations.

• Tests

  • New integration tests validating sleep and recv message behaviors.

• Chores

  • CI/dev environment updated to include system packages needed for message/timer support.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: e2fdd362-817a-457a-8215-ff026d8b19f1

📥 Commits

Reviewing files that changed from the base of the PR and between 913aa44 and 43ca1a2.

📒 Files selected for processing (1)

• flake.nix

✅ Files skipped from review due to trivial changes (1)

• flake.nix

---

📝 Walkthrough

Walkthrough

Added step-waiting and message delivery to workflows: new wait/result types, host APIs (sleep/recv-message/send-message), partition log events/effects, timer handling in effect worker, queueing/resuming logic in partition reducer, SDK recv/sleep helpers, tests, and platform-specific keyring and CI/nix updates.

Changes

Cohort / File(s)	Summary
Cargo / build src/daybook_core/Cargo.toml, .github/workflows/builds.yml, flake.nix	Platform-specific keyring features enabled; CI apt packages and Nix inputs updated to include dbus, pkg-config, libdbus headers.
Runtime dispatch src/daybook_core/rt.rs	Treat StepWait as non-terminal; on dispatch add failure attempt to cancel queued workflow job via wflow_ingress.cancel_job and return original add error.
Secrets / keyring src/daybook_core/secrets.rs	Gate keyring usage via should_use_keyring(), make keyring entry creation/read/write failures propagate instead of silently logging; stricter error handling for repair/backfill paths.
WASM guest & WIT host surface src/test_wflows/lib.rs, src/wash_plugin_wflow/wit/main.wit	Add sleep_then_succeed and recv_message guest actions; add host APIs sleep and recv-message, and partition-host send-message import.
Wash plugin (host/traps) src/wash_plugin_wflow/lib.rs	Add JobTrap::WaitStep + WaitTrapDeets; implement sleep and recv_message host methods yielding WaitStep; map traps to JobRunResult::StepWait; add stub send_message.
Wflow ingress & partition APIs src/wflow/ingress.rs	Extend WflowIngress trait with send_message and implement it for PartitionLogIngress to append JobMessage entries.
Partition job events & log src/wflow_core/partition/job_events.rs, src/wflow_core/partition/log.rs	New JobMessageEvent, JobTimerFiredEvent, JobWaitResult/JobWaitResultDeets; JobRunResult::StepWait added; partition log gains JobMessage and JobTimerFired variants.
Partition state & reducer src/wflow_core/partition/state.rs, src/wflow_core/partition/reduce.rs	Job state gains pending_messages queue and active_wait (JobWaitState); reducers added for JobMessage and JobTimerFired, lifecycle changes to handle StepWait (enqueue/consume messages, schedule/clear waits, CancelWait emission).
Partition effects & reducer integration src/wflow_core/partition/effects.rs, src/wflow_tokio/partition/reducer.rs	New effect variants: WaitTimer, WaitMessage, CancelWait; reschedule logic now preserves WaitTimer; log entry handling extended to route message/timer-fired entries into job reducer.
Effect worker (timers & cancellations) src/wflow_tokio/partition/effect_worker.rs	Add pending timer tracking, periodic tick to fire due timers and emit JobTimerFired entries, handle WaitTimer/CancelWait effects and retain sessions for StepWait.
Wflow SDK additions src/wflow_sdk/lib.rs	Add RecvCodec trait and impl for Json<T>; add WflowCtx::sleep(Duration) and WflowCtx::recv<O: RecvCodec>() to interact with host wait/message APIs.
Tests src/wflow/test.rs, src/wflow/test/recv_message.rs, src/wflow/test/sleep_then_succeed.rs, src/test_wflows/lib.rs	New tests exercising sleep and recv-message flows; test harness gains send_job_message helper; guest test wasm handlers added for sleep and receive actions.

Sequence Diagram(s)

sequenceDiagram
    participant Guest as WASM Guest (wflow)
    participant SDK as Wflow SDK (WflowCtx)
    participant Host as wash_plugin_wflow (Host)
    participant Ingress as WflowIngress / PartitionLog
    participant Worker as TokioEffectWorker
    participant Reducer as Partition Reducer

    Guest->>SDK: call sleep(ms) / recv()
    SDK->>Host: next_step() -> op_id
    alt sleep
        SDK->>Host: sleep(job_id, step_id, duration_ms) -> yields WaitStep (Timer)
    else recv
        SDK->>Host: recv-message(job_id, step_id) -> yields WaitStep (Message)
    end
    Host->>Ingress: append PartitionEffect WaitTimer/WaitMessage or JobMessage (if pre-buffered)
    Worker->>Worker: if WaitTimer store pending_timers (no immediate Run)
    Note over Reducer,Ingress: Partition log includes Wait effects / JobMessage
    Ingress->>Reducer: handle JobTimerFired or JobMessage entries
    Reducer->>Reducer: consume pending_messages or match timer -> produce RunJob effect
    Reducer->>Worker: emit RunJob effect
    Worker->>Guest: resume job/run attempt (effect worker executes)
    Guest->>Host: completes step (StepEffect / Success)
    Host->>Ingress: append JobEffectResult (Success)
    Reducer->>Reducer: finalize job run state

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• feat: plug_plabels #18 — touches runtime dispatch/wflow logic in src/daybook_core/rt.rs, closely related to dispatch and cancel behavior changes.
• feat(wflow): log and resume for wflow jobs #10 — affects trap/session and translation surfaces in wash_plugin_wflow, related to JobTrap ↔ JobRunResult mappings.

"I hopped through logs with whiskers bright,
Timers and messages danced in the night,
I queued a note, I let time sleep,
Then watched the workflow wake from its keep — 🐇✨"

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 22.45% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title 'feat(wflow): sleep and messages' directly describes the main changes: adding sleep (timing) and message handling capabilities to the workflow system. It is specific and clearly summarizes the primary feature additions across multiple files.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/core/sleep-and-messages

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

🧹 Nitpick comments (1)

src/wflow_sdk/lib.rs (1)

99-120: Consider returning T directly for API consistency.

The effect() method returns Result<O, JobErrorX> (unwrapped value), while recv() returns Result<Json<T>, JobErrorX> (wrapped in Json). This inconsistency may surprise users. Consider unwrapping like effect() does:

♻️ Suggested change for consistency

-    pub fn recv<T>(&self) -> Result<Json<T>, JobErrorX>
+    pub fn recv<T>(&self) -> Result<T, JobErrorX>
     where
         T: serde::de::DeserializeOwned,
     {
         // ... existing code ...
-        Ok(Json(value))
+        Ok(value)
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/wflow_sdk/lib.rs` around lines 99 - 120, The recv function currently
returns Result<Json<T>, JobErrorX> which is inconsistent with effect() that
returns Result<O, JobErrorX>; change recv<T>(&self) to return Result<T,
JobErrorX>, deserialize value_json into T and return the inner value (not
wrapped in Json). Update the function signature and the final Ok(...) to return
the deserialized T, keeping existing error mapping (JobErrorX) and using
std::any::type_name::<T>() in the error message as before; ensure Json<T> usages
(if any callers or types) are adjusted accordingly.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Nitpick comments:
In `@src/wflow_sdk/lib.rs`:
- Around line 99-120: The recv function currently returns Result<Json<T>,
JobErrorX> which is inconsistent with effect() that returns Result<O,
JobErrorX>; change recv<T>(&self) to return Result<T, JobErrorX>, deserialize
value_json into T and return the inner value (not wrapped in Json). Update the
function signature and the final Ok(...) to return the deserialized T, keeping
existing error mapping (JobErrorX) and using std::any::type_name::<T>() in the
error message as before; ensure Json<T> usages (if any callers or types) are
adjusted accordingly.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: d28102a4-8140-4486-85e1-615637fbefd3

📥 Commits

Reviewing files that changed from the base of the PR and between 83465e9 and c325d5d.

⛔ Files ignored due to path filters (1)

• Cargo.lock is excluded by !**/*.lock and included by **/*

📒 Files selected for processing (18)

• src/daybook_core/Cargo.toml
• src/daybook_core/rt.rs
• src/daybook_core/secrets.rs
• src/test_wflows/lib.rs
• src/wash_plugin_wflow/lib.rs
• src/wash_plugin_wflow/wit/main.wit
• src/wflow/ingress.rs
• src/wflow/test.rs
• src/wflow/test/recv_message.rs
• src/wflow/test/sleep_then_succeed.rs
• src/wflow_core/partition/effects.rs
• src/wflow_core/partition/job_events.rs
• src/wflow_core/partition/log.rs
• src/wflow_core/partition/reduce.rs
• src/wflow_core/partition/state.rs
• src/wflow_sdk/lib.rs
• src/wflow_tokio/partition/effect_worker.rs
• src/wflow_tokio/partition/reducer.rs