docs(flare-workflow): land metrics observability design spec with its implementation #530
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,128 @@ | ||
| # Design-spec: workflow observability (instance & step metrics) | ||
|
|
||
| Scope: the ask — "add observability similar to Cloudflare Workflows' | ||
| metrics-analytics" (`cloudflare-docs/.../workflows/observability/metrics-analytics.mdx`) | ||
| — for `flare-workflow`. Recommendation only; no code changed by this doc. | ||
|
|
||
| ## What Cloudflare's feature actually is | ||
|
|
||
| Two access paths over the same underlying data: the Cloudflare dashboard | ||
| (charts, per-Workflow and account-wide) and the GraphQL Analytics API | ||
| (`workflowsMetricsAdaptiveGroups`-style dataset, filtered/grouped by workflow | ||
| name, status, time bucket). Metrics exposed: **instance counts by status** | ||
| (queued/running/errored/terminated/complete/paused), **CPU time** and | ||
| **wall-clock duration** per instance, **step-level** duration/CPU/outcome, | ||
| and error rates. | ||
|
|
||
| The important structural fact: this is a **read-only aggregate query layer | ||
| over data the execution engine already records for every run** — not a | ||
| bolt-on instrumentation layer. Cloudflare isn't adding new tracing to | ||
| Workflows to build this; they're exposing what the durable-execution engine | ||
| already durably writes (state transitions, timing, step outcomes) via | ||
| structured queries. | ||
|
|
||
| ## Current state in `flare-workflow` (verified directly against source) | ||
|
|
||
| - `crates/flare-workflow` is a standalone library crate — checked the | ||
| workspace `Cargo.toml` dependency graph and no other crate currently | ||
| depends on it (only worktree copies of itself do). There is no existing | ||
| CLI/API/MCP surface exposing workflow runs today; whatever this spec | ||
| proposes ships as a library capability first. | ||
| - Per-run/per-step telemetry **is already recorded**, matching Cloudflare's | ||
| "expose what's already there" framing: | ||
| - `WorkflowState` (`types.rs:411-426`): `status`, `created_at`/`updated_at`, | ||
| `step_states: HashMap<StepId, StepState>`. | ||
| - `StepState` (`types.rs:198-211`): `status`, `attempt`, | ||
| `started_at`/`completed_at`, `input_tokens`, `output_tokens`, | ||
| `duration_ms` — this is already the Cloudflare "CPU time / duration / | ||
| step outcome" shape, per step, per attempt. | ||
| - `JournalEntry` (`types.rs:257-308`, append-only, `journal` table in | ||
| `sqlite_store.rs`): typed entries (`StepRun`, `Sleep`, `WaitEvent`, | ||
| `Rollback`, `LoopIteration`, `Output`). No timestamp column on the | ||
| journal table itself (`run_id, seq, entry_type, payload` only) — it's | ||
| ordered by `seq`, not wall-clock, so it's not a time-series source on | ||
| its own. `StepState`'s own timestamps are the right source for | ||
| duration-style metrics, not the journal. | ||
| - **No aggregate query capability exists on the store actually used in | ||
| production.** `count_by_status`/`count` (`store.rs:82-93`) are **inherent | ||
| methods on `InMemoryStore` only** — verified by reading `SqliteStore`'s | ||
| full `StateStore` impl in `sqlite_store.rs`: it has no such methods. The | ||
| `StateStore` trait itself declares no aggregate method. `list_all()`/ | ||
| `list_active()` return fully deserialized `WorkflowState` structs; today, | ||
| "count by status" against SQLite means loading and deserializing every | ||
| run's full `state_json` blob client-side, not a SQL `GROUP BY`. | ||
| - **Real schema gap for step-level metrics:** the SQLite `step_state` | ||
| projection table (`sqlite_store.rs` migrations) has columns `run_id, | ||
| step_id, status, attempt, last_error, started_at, completed_at` — it does | ||
| **not** have `duration_ms`, `input_tokens`, or `output_tokens` as columns, | ||
| even though those fields exist on the `StepState` struct and are written | ||
| into the `workflow_runs.state_json` blob. So step-level duration/token | ||
| aggregation can't be done in SQL today without a migration; it would | ||
| require deserializing every run's JSON. | ||
|
Comment on lines
+46
to
+61
There was a problem hiding this comment. 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win Update the current-state and recommendation sections to match the shipped implementation. The document says that Lines 73-99 also describe the migration, SQL queries, engine passthrough, and tests as future work. Rewrite this as a historical baseline plus shipped implementation, or mark it explicitly as a pre-implementation proposal. The documented query plan and test location must also match the implementation. Also applies to: 73-99 🤖 Prompt for AI Agents |
||
| - No metrics/tracing/analytics/GraphQL/dashboard infrastructure exists | ||
| anywhere else in agentflare to reuse — repo-wide search for | ||
| `metrics|analytics|tracing|instrumentation|prometheus|otel|graphql| | ||
| grafana` returned zero hits outside this investigation. This is new | ||
| infra, not a wire-up of something that already exists elsewhere. | ||
|
|
||
| ## Recommended shape | ||
|
|
||
| Cloudflare's feature reduces to "SQL-level aggregate queries over | ||
| already-durable execution data." The equivalent here: | ||
|
|
||
| 1. **Migration**: add `duration_ms INTEGER`, `input_tokens INTEGER`, | ||
| `output_tokens INTEGER` columns to `step_state` (new `M::up(...)` entry — | ||
| the crate already uses `rusqlite_migration` incrementally for its 4 | ||
| existing migrations, no new dependency). `write_state`'s existing | ||
| `step_state` UPSERT gets the 3 extra columns. | ||
| 2. **New `StateStore` trait method**: `async fn workflow_metrics(&self, | ||
| filter: MetricsFilter) -> WorkflowResult<WorkflowMetrics>` (`store.rs`). | ||
| `MetricsFilter`: optional `workflow_id`, optional status, optional time | ||
| range. `WorkflowMetrics`: counts-by-status, `avg`/`sum` of | ||
| `duration_ms` and tokens, and a per-`step_id` breakdown (status counts + | ||
| avg duration) — directly mirrors Cloudflare's instance-count / | ||
| CPU-time / step-outcome shape. | ||
| 3. **`SqliteStore` impl**: one indexed `GROUP BY status` query against | ||
| `workflow_runs` for instance counts, one `GROUP BY step_id, status` | ||
| query against `step_state` for step-level aggregates. Both tables | ||
| already have the needed indexed columns after the migration — no | ||
| full-row JSON deserialization needed for the aggregate path. | ||
| 4. **`InMemoryStore` impl**: mirror with the same signature, iterating the | ||
| in-memory map (trivial, same pattern as its existing `count_by_status`). | ||
| 5. **`WorkflowEngine::metrics(filter)`**: thin passthrough to the store. | ||
| This is where the spec deliberately stops — no dashboard, GraphQL API, | ||
| or CLI command. No consumer crate wires `flare-workflow` in yet (see | ||
| above), so building a query surface beyond the library boundary would | ||
| be speculative UI ahead of an actual integration point. | ||
| 6. **Test**: seed N runs with mixed statuses/step outcomes in a temp | ||
| `SqliteStore`, assert `workflow_metrics()` aggregate counts match, | ||
| alongside existing coverage in `tests/engine_test.rs`. | ||
|
|
||
| ## Cost estimate | ||
|
|
||
| - Migration + `write_state` column additions: ~20 LOC. | ||
| - `MetricsFilter`/`WorkflowMetrics` types (new `metrics.rs` or appended to | ||
| `types.rs`): ~30-40 LOC. | ||
| - Trait method + `SqliteStore` aggregate SQL + `InMemoryStore` mirror: | ||
| ~70-90 LOC. | ||
| - `WorkflowEngine::metrics()` passthrough: ~10 LOC. | ||
| - Test: ~40-60 LOC. | ||
|
|
||
| Total: roughly 170-210 LOC plus one migration, no new external | ||
| dependencies. No schema migration risk beyond the usual additive | ||
| `ALTER TABLE ADD COLUMN` (existing rows get `NULL`, aggregates already | ||
| need to tolerate that via SQL `COALESCE`/Rust `Option`). | ||
|
Comment on lines
+111
to
+114
There was a problem hiding this comment. 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
# Verify whether the migration backfills existing state_json metric values.
rg -n -C 6 'duration_ms|input_tokens|output_tokens|state_json|ALTER TABLE|M::up' \
crates/flare-workflow/src crates/flare-workflow/testsRepository: getappz/agentflare Length of output: 46005 🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '%s\n' '--- design document ---'
sed -n '80,130p' crates/flare-workflow/OBSERVABILITY_METRICS_DESIGN.md
printf '%s\n' '--- migration and projection code ---'
sed -n '55,125p' crates/flare-workflow/src/sqlite_store.rs
sed -n '575,680p' crates/flare-workflow/src/sqlite_store.rs
printf '%s\n' '--- focused parent diff for metric fields and migration ---'
git diff --unified=5 HEAD^ -- \
crates/flare-workflow/src/types.rs \
crates/flare-workflow/src/sqlite_store.rs \
crates/flare-workflow/tests/metrics_test.rs \
crates/flare-workflow/OBSERVABILITY_METRICS_DESIGN.mdRepository: getappz/agentflare Length of output: 18603 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import json
import sqlite3
# Model the documented migration and the existing authoritative state_json
# representation without executing repository code.
old_state = {
"step_states": {
"s1": {
"status": "succeeded",
"attempt": 1,
"input_tokens": 10,
"output_tokens": 20,
"duration_ms": 100,
}
}
}
old_state_without_metrics = {
"step_states": {
"s1": {
"status": "succeeded",
"attempt": 1,
}
}
}
db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE workflow_runs (
id TEXT PRIMARY KEY,
state_json TEXT NOT NULL
);
CREATE TABLE step_state (
run_id TEXT NOT NULL,
step_id TEXT NOT NULL,
status TEXT NOT NULL,
PRIMARY KEY (run_id, step_id)
);
INSERT INTO workflow_runs VALUES ('with-data', ?);
INSERT INTO workflow_runs VALUES ('without-data', ?);
INSERT INTO step_state VALUES ('with-data', 's1', 'succeeded');
INSERT INTO step_state VALUES ('without-data', 's1', 'succeeded');
ALTER TABLE step_state ADD COLUMN duration_ms INTEGER;
ALTER TABLE step_state ADD COLUMN input_tokens INTEGER;
ALTER TABLE step_state ADD COLUMN output_tokens INTEGER;
""", (json.dumps(old_state), json.dumps(old_state_without_metrics)))
print("projected rows after additive migration:")
for row in db.execute("""
SELECT run_id, duration_ms, input_tokens, output_tokens
FROM step_state ORDER BY run_id
"""):
print(row)
print("aggregates after additive migration:")
print(db.execute("""
SELECT
COALESCE(SUM(input_tokens), 0),
COALESCE(SUM(output_tokens), 0),
AVG(duration_ms)
FROM step_state
""").fetchone())
print("authoritative state_json values:")
for row in db.execute("SELECT id, state_json FROM workflow_runs ORDER BY id"):
print(row[0], json.loads(row[1])["step_states"]["s1"])
PYRepository: getappz/agentflare Length of output: 285 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import json
import sqlite3
db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE workflow_runs (
id TEXT PRIMARY KEY,
state_json TEXT NOT NULL
);
CREATE TABLE step_state (
run_id TEXT NOT NULL,
step_id TEXT NOT NULL,
status TEXT NOT NULL,
PRIMARY KEY (run_id, step_id)
);
""")
states = {
"with-data": {
"step_states": {
"s1": {
"status": "succeeded",
"attempt": 1,
"input_tokens": 10,
"output_tokens": 20,
"duration_ms": 100,
}
}
},
"without-data": {
"step_states": {
"s1": {"status": "succeeded", "attempt": 1}
}
},
}
for run_id, state in states.items():
db.execute(
"INSERT INTO workflow_runs VALUES (?, ?)",
(run_id, json.dumps(state)),
)
db.execute(
"INSERT INTO step_state VALUES (?, ?, ?)",
(run_id, "s1", "succeeded"),
)
db.executescript("""
ALTER TABLE step_state ADD COLUMN duration_ms INTEGER;
ALTER TABLE step_state ADD COLUMN input_tokens INTEGER;
ALTER TABLE step_state ADD COLUMN output_tokens INTEGER;
""")
print("projected rows after additive migration:")
for row in db.execute("""
SELECT run_id, duration_ms, input_tokens, output_tokens
FROM step_state ORDER BY run_id
"""):
print(row)
print("aggregates after additive migration:")
print(db.execute("""
SELECT
COALESCE(SUM(input_tokens), 0),
COALESCE(SUM(output_tokens), 0),
AVG(duration_ms)
FROM step_state
""").fetchone())
print("authoritative state_json values:")
for row in db.execute("SELECT id, state_json FROM workflow_runs ORDER BY id"):
print(row[0], json.loads(row[1])["step_states"]["s1"])
PY
printf '%s\n' '--- migration-related tests and all writes to metric columns ---'
rg -n -C 4 'migrations|ALTER TABLE step_state|duration_ms|input_tokens|output_tokens|workflow_metrics' \
crates/flare-workflow/tests crates/flare-workflow/src/sqlite_store.rsRepository: getappz/agentflare Length of output: 19606 Define historical metric completeness after migration. The migration leaves new columns Add a backfill from 🤖 Prompt for AI Agents |
||
|
|
||
| ## Recommendation | ||
|
|
||
| Worth doing, scoped as above: it's a genuine SQL-level aggregate layer over | ||
| data the engine already durably records, which is what Cloudflare's feature | ||
| actually is once you look past the dashboard chrome. The one real gap to | ||
| close is schema-level (`step_state` missing duration/token columns), not | ||
| architectural — no new instrumentation, tracing, or execution-path changes | ||
| are needed. Recommend explicitly **not** building a dashboard, GraphQL API, | ||
| or CLI surface in this pass: `flare-workflow` has no consumer crate today, | ||
| and a query surface with no caller would be built ahead of need. Land the | ||
| library-level `workflow_metrics()` capability now; revisit an exposed query | ||
| surface (CLI subcommand, MCP tool, or HTTP endpoint) once a consumer | ||
| actually wires the engine in and needs one. | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
Repository: getappz/agentflare
Length of output: 50377
🏁 Script executed:
Repository: getappz/agentflare
Length of output: 19845
🌐 Web query:
Cloudflare Workflows observability metrics analytics workflowsAdaptiveGroups workflowName instanceId stepName wallTime💡 Result:
Cloudflare Workflows provides observability through metrics and analytics that can be viewed in the Cloudflare dashboard or queried programmatically via the GraphQL Analytics API [1][2]. The primary dataset used for these analytics is
workflowsAdaptiveGroups[1][2]. This dataset allows you to filter and group execution metrics using several dimensions [1][2]: - workflowName: The name of the specific workflow [1][2]. - instanceId: The unique identifier for a workflow instance [1][2]. - stepName: The name assigned to a step within the workflow [1][2]. - eventType: The type of event (e.g., execution start, completion, error) [1][2]. - stepCount: The step number within a given instance [1][2]. - Time-based dimensions: Includes date, datetimeFiveMinutes, datetimeHour, and datetimeMinute [1][2]. While wallTime is a common metric concept in distributed systems, the official documentation specifically highlights that these metrics allow you to measure execution, error rates, steps, and total duration [1][2]. Metrics are retained for 31 days and require an accountTag filter (your Cloudflare account ID) when queried via the GraphQL API [1][2]. For more granular debugging, you can also retrieve detailed execution logs and status for specific instances, including step-level output, through the Cloudflare API [3][4][5].Citations:
🏁 Script executed:
Repository: getappz/agentflare
Length of output: 11889
🏁 Script executed:
Repository: getappz/agentflare
Length of output: 946
🏁 Script executed:
Repository: getappz/agentflare
Length of output: 621
Align the metrics design with the implemented contract.
Use Cloudflare’s
workflowsAdaptiveGroupsdataset name. The localWorkflowMetricsAPI exposes status counts, average duration, token totals, and step outcomes only. It does not expose CPU time, error rates, time buckets, or duration sums.duration_msrecords elapsed duration, and token totals are not CPU metrics. Update lines 9–15, 35–38, and 78–84 to describe a library-level subset, or extend the API before claiming parity.🤖 Prompt for AI Agents
Source: MCP tools