Skip to content

feat(flare-workflow): embedded durable workflow engine for agent orchestration #472

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
getappz merged 14 commits into from
Aug 13, 2026
Merged

feat(flare-workflow): embedded durable workflow engine for agent orchestration #472

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
6fd550b
feat(flare-workflow): scaffold crate with typed DAG types + journal +…
getappz Aug 12, 2026
5898637
feat(flare-workflow): journaled DAG engine core with retries + event bus
getappz Aug 12, 2026
02c9da3
feat(flare-workflow): OpenFang step semantics — conditional/loop/fano…
getappz Aug 12, 2026
e4a92f5
feat(flare-workflow): durable waits — Sleep timers + WaitEvent promis…
getappz Aug 12, 2026
09c17e0
feat(flare-workflow): recovery pass — crash-resume with exactly-once …
getappz Aug 12, 2026
d02b648
feat(flare-workflow): OpenFang JSON schema + compile + example workflows
getappz Aug 12, 2026
0812c41
docs(flare-workflow): crate README — usage, durability, JSON workflows
getappz Aug 12, 2026
e996e40
chore: update Cargo.lock for flare-workflow crate
getappz Aug 12, 2026
b1c3562
feat(workflow): mcp__flare__workflow + agentflare workflow CLI + real…
getappz Aug 12, 2026
c856e20
feat(workflow): MCP handler tests + async cores for daemon-safe invoc…
getappz Aug 12, 2026
08b8dc8
fix(flare-workflow): resolve code-review findings on task/447
getappz Aug 12, 2026
f07942f
fix(flare-workflow): drop unmaintained backoff crate, regen hakari wo…
getappz Aug 13, 2026
8bbd85d
Merge remote-tracking branch 'origin/master' into task/447
getappz Aug 13, 2026
96d3e4e
fix(flare-workflow): resolve CodeRabbit findings — durable sleep, upd…
getappz Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 2 additions & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
[workspace]
members = ["agentflare-workspace-hack","crates/flare-code", "crates/agent-registry", "crates/skill-registry", "crates/gateway-registry", "crates/flare-output", "crates/agentflare-artifacts", "crates/agentflare-backend", "crates/agentflare-db-kit", "crates/flare-search-kit", "crates/agentflare-store", "crates/flare-proxy", "crates/agentflare-shim", "crates/flare-git-core", "crates/flare-git-shim", "crates/agentflare-jobs", "crates/flare-docs", "crates/flare-vault", "crates/agentflare-resource-gate"]
members = ["agentflare-workspace-hack","crates/flare-code", "crates/agent-registry", "crates/skill-registry", "crates/gateway-registry", "crates/flare-output", "crates/agentflare-artifacts", "crates/agentflare-backend", "crates/agentflare-db-kit", "crates/flare-search-kit", "crates/agentflare-store", "crates/flare-proxy", "crates/agentflare-shim", "crates/flare-git-core", "crates/flare-git-shim", "crates/agentflare-jobs", "crates/flare-docs", "crates/flare-vault", "crates/agentflare-resource-gate", "crates/flare-workflow"]
resolver = "2"

[package]
Expand Down Expand Up @@ -88,6 +88,7 @@ flare-docs = { path = "crates/flare-docs" }
flare-git-core = { path = "crates/flare-git-core" }
flare-proxy = { path = "crates/flare-proxy" }
flare-vault = { path = "crates/flare-vault" }
flare-workflow = { path = "crates/flare-workflow" }
skill = { version = "0.8", default-features = false, features = ["network"] }
agentflare-jobs = { path = "crates/agentflare-jobs" }
agentflare-resource-gate = { path = "crates/agentflare-resource-gate" }
Expand Down
1 change: 1 addition & 0 deletions agentflare-workspace-hack/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ tracing = { version = "0.1", features = ["log"] }
tracing-core = { version = "0.1" }
typenum = { version = "1", default-features = false, features = ["const-generics"] }
ureq = { version = "2", features = ["json"] }
uuid = { version = "1", features = ["serde", "v4", "v7"] }
zeroize = { version = "1", features = ["derive"] }

[build-dependencies]
Expand Down
30 changes: 30 additions & 0 deletions crates/flare-workflow/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
[package]
name = "flare-workflow"
version = "0.1.0"
edition = "2024"
rust-version = "1.91"
license = "Apache-2.0"
description = "Embedded durable workflow engine for agent orchestration — typed DAG steps with journaled execution, durable waits, and human-in-the-loop events."
publish = false

[dependencies]
async-trait = "0.1"
chrono = { version = "0.4", features = ["serde"] }
parking_lot = "0.12"
rand = "0.8"
rusqlite = { version = "0.40", features = ["bundled"] }
rusqlite_migration = "2"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
thiserror = "2"
tokio = { version = "1", features = ["sync", "rt-multi-thread", "macros", "time"] }
tracing = "0.1"
uuid = { version = "1", features = ["serde", "v4", "v7"] }
db_kit = { package = "agentflare-db-kit", path = "../agentflare-db-kit" }
agentflare-workspace-hack = { version = "0.1", path = "../../agentflare-workspace-hack" }

[dev-dependencies]
tempfile = "3"

[lints.rust]
unsafe_code = "warn"
77 changes: 77 additions & 0 deletions crates/flare-workflow/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
# flare-workflow

Embedded **durable workflow engine** for agent orchestration — typed DAG
steps with journaled execution, durable waits, and human-in-the-loop events.
No external binary, no sidecar runtime: durability lives in agentflare's own
SQLite (via `agentflare-db-kit`).

This is the Cloudflare/Vercel-Workflows-style spine beneath the coordination
layer (items / claims / handoffs).

## Design lineage

| Source | Adopted | License |
|---|---|---|
| SMG `wfaas` | typed DAG engine, retry/backoff, `StateStore` trait, event bus | Apache-2.0 |
| OpenFang | StepMode semantics, `{{input}}`/`{{var}}` templating, ErrorMode, JSON schema | MIT / Apache-2.0 |
| Restate | journal durability (CompletableEntry), durable timers/promises | BSL — design only, no code |
| DBOS | step outputs checkpointed with user state ⇒ exactly-once | MIT |

## Usage

```rust
use flare_workflow::{WorkflowDefinition, StepDefinition, WorkflowEngine, InMemoryStore};
use flare_workflow::executor::FunctionStep;

let wf = WorkflowDefinition::new("wf", "wf")
.add_step(StepDefinition::new("a", "a", Arc::new(FunctionStep::new(|ctx| {
ctx.output = "hello".into();
Box::pin(async { Ok(StepResult::Success) })
}))))
.add_step(StepDefinition::new("b", "b", /* ... */).depends_on(&["a"]));

let engine = WorkflowEngine::<Ctx, InMemoryStore<Ctx>>::new();
engine.register_workflow(wf)?;
let run = engine.start_workflow(WorkflowId::new("wf"), ctx, "input".into()).await?;
engine.wait_for_completion(run, "wf", Duration::from_secs(300)).await?;
```
Comment on lines +22 to +37

@coderabbitai coderabbitai Bot Aug 13, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

# Check available code-intelligence and memory tools, then inspect the README example
# and the crate metadata without executing repository code.
set -u
printf '%s\n' '--- tool availability ---'
for tool in agentflare ctx_search ctx_read ctx_callgraph ctx_compose ctx_shell; do
  if command -v "$tool" >/dev/null 2>&1; then
    printf '%s: available\n' "$tool"
  else
    printf '%s: unavailable\n' "$tool"
  fi
done

printf '%s\n' '--- memory context ---'
if command -v agentflare >/dev/null 2>&1; then
  agentflare memory context 2>&1 | head -200
else
  printf '%s\n' 'agentflare unavailable'
fi

printf '%s\n' '--- candidate files ---'
git ls-files | grep -E '(^|/)(README\.md|Cargo\.toml)$' | grep -E 'crates/flare-workflow|^Cargo\.toml$' || true

printf '%s\n' '--- README excerpt ---'
cat -n crates/flare-workflow/README.md | sed -n '1,90p'

printf '%s\n' '--- crate manifest ---'
cat -n crates/flare-workflow/Cargo.toml | sed -n '1,180p'

printf '%s\n' '--- workspace manifest doctest settings ---'
rg -n -C 3 'doctest|documentation|flare-workflow|edition' Cargo.toml crates/flare-workflow/Cargo.toml

Repository: getappz/agentflare

Length of output: 7346

🏁 Script executed:

set -u
printf '%s\n' '--- workflow source files ---'
git ls-files crates/flare-workflow | sed -n '1,120p'

printf '%s\n' '--- public symbols and example identifiers ---'
rg -n -C 2 'pub (struct|enum|type|trait|fn)|WorkflowDefinition|StepDefinition|WorkflowEngine|InMemoryStore|FunctionStep|StepResult|WorkflowId|Ctx|start_workflow|wait_for_completion' crates/flare-workflow/src

printf '%s\n' '--- library outline ---'
if command -v ast-grep >/dev/null 2>&1; then
  ast-grep outline crates/flare-workflow/src
else
  printf '%s\n' 'ast-grep unavailable'
fi

Repository: getappz/agentflare

Length of output: 50374

🏁 Script executed:

set -u
printf '%s\n' '--- WorkflowContext and StepResult ---'
cat -n crates/flare-workflow/src/types.rs | sed -n '315,425p'

printf '%s\n' '--- StepDefinition constructor and dependency API ---'
cat -n crates/flare-workflow/src/definition.rs | sed -n '35,125p'
cat -n crates/flare-workflow/src/definition.rs | sed -n '145,160p'

printf '%s\n' '--- WorkflowEngine constructor and workflow methods ---'
rg -n -A 24 -B 8 'pub fn new|pub fn register_workflow|pub fn start_workflow|pub async fn wait_for_completion' crates/flare-workflow/src/engine.rs

printf '%s\n' '--- existing test setup for a compiling FunctionStep ---'
rg -n -A 35 -B 10 'FunctionStep::new|WorkflowContext|struct TestData|impl WorkflowData' crates/flare-workflow/tests crates/flare-workflow/src | head -260

Repository: getappz/agentflare

Length of output: 32063

🏁 Script executed:

python3 - <<'PY'
from pathlib import Path
import re

readme = Path("crates/flare-workflow/README.md").read_text()
match = re.search(r"```rust\n(.*?)\n```", readme, re.S)
assert match, "usage Rust block not found"
block = match.group(1)

ctx_source = Path("crates/flare-workflow/src/types.rs").read_text()
engine_source = Path("crates/flare-workflow/src/engine.rs").read_text()

checks = {
    "uses_top_level_await": ".await" in block,
    "uses_question_mark": "?" in block,
    "has_placeholder_argument": "/* ... */" in block,
    "declares_arc": bool(re.search(r"\b(?:use|let|type|struct|enum)\b.*\bArc\b", block)),
    "declares_duration": bool(re.search(r"\b(?:use|let|type|struct|enum)\b.*\bDuration\b", block)),
    "declares_step_result": bool(re.search(r"\b(?:use|let|type|struct|enum)\b.*\bStepResult\b", block)),
    "declares_workflow_id": bool(re.search(r"\b(?:use|let|type|struct|enum)\b.*\bWorkflowId\b", block)),
    "declares_ctx_type": bool(re.search(r"\b(?:struct|type|enum)\s+Ctx\b", block)),
    "declares_ctx_value": bool(re.search(r"\blet\s+ctx\b", block)),
    "workflow_context_has_output": bool(re.search(r"\bpub\s+output\s*:\s*String", ctx_source)),
    "start_workflow_is_async": bool(re.search(r"pub\s+async\s+fn\s+start_workflow", engine_source)),
    "wait_for_completion_is_async": bool(re.search(r"pub\s+async\s+fn\s+wait_for_completion", engine_source)),
}
for name, value in checks.items():
    print(f"{name}={value}")
PY

Repository: getappz/agentflare

Length of output: 477

Make the usage example compile as a doctest.

The block uses undeclared Arc, Duration, StepResult, WorkflowId, and Ctx. It also contains ?, top-level .await, and /* ... */ where an executor expression is required. Add valid hidden setup and an async Result-returning example function, or mark the block as non-Rust text.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@crates/flare-workflow/README.md` around lines 22 - 37, The README usage block
is not a compilable Rust doctest. Update the example around WorkflowDefinition
and WorkflowEngine with hidden imports and setup for Arc, Duration, StepResult,
WorkflowId, and Ctx, wrap the statements in an async Result-returning function,
and replace the placeholder executor argument with a valid expression;
alternatively mark the block as non-Rust text.


### Durable execution

- **Journal**: every terminal step result is appended (`StepRun` / `Sleep` /
`WaitEvent`); a step with a completed entry is **never re-executed**.
- **Recovery**: `engine.recover()` resumes `Running` runs from the SQLite
journal after a crash, skipping completed steps (exactly-once).
- **Durable waits**: `StepMode::Sleep { duration_secs }` (timer) and
`StepMode::WaitEvent { name, timeout_secs }` (promise) survive restart;
`engine.complete_event(run_id, name, result)` resolves them from anywhere
(journaled pre-delivery closes the notify-before-wait race).
- **Retries**: per-step `RetryPolicy` (fixed/exponential/linear backoff +
jitter), `is_retryable`, `RetryIndefinitely`, per-attempt timeout.

### JSON workflows (agent pipelines)

OpenFang-style JSON definitions route prompts to agents via a `SendMessage`
hook:

```rust
let json: JsonWorkflow = serde_json::from_str(WORKFLOW_JSON)?;
let wf = compile_workflow(&json, send_message)?;
engine.register_workflow(wf)?;
```

Step modes: `sequential` / `fan_out` / `collect` / `conditional` / `loop` /
`sleep` / `wait_event`; error modes `fail` / `skip` / `retry`; `{{input}}` and
`{{var}}` templating; per-step token accounting.

## Store backends

- `InMemoryStore<D>` — default, tests/dev.
- `SqliteStore<D>` — durable, on `agentflare-db-kit` (WAL, migrations);
tables `workflow_runs` (authoritative JSON state), `journal` (append-only),
`step_state`, `run_vars` (queryable projections).

## License

Apache-2.0. Restate ideas are adopted as design only (BSL 1.1); no Restate
code is copied.
Loading
Loading