jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make workflow run history authoritative in Desktop (#5780)

Browse Source 
## Summary

- persist stable workflow run `error_code` values separately from human
diagnostics
- expose NIP-98 authenticated, channel-authorized run history and
approval reads with stable keyset pagination
- connect Desktop to those authoritative reads and return the
relay-created run ID on trigger
- show truthful loading, failure, and pending-trace states, and do not
render approval actions from non-actionable stored hashes

## Validation

- pre-push `branch-skew`, `desktop-typecheck`, `desktop-test`,
`rust-tests`, `desktop-tauri-checks`, and `desktop-check` all passed on
`a097dbe5f`
- Desktop tests: 4,761 passed, 0 failed
- `cargo check -p buzz-relay`
- `git diff --check`

## Remaining gate

This does not claim a relay-backed Playwright workflow journey. The
browser relay bridge still routes workflow invokes through in-memory
handlers; that production-shaped acceptance gate remains follow-up work
before Workflows can leave preview.

---------

Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
Co-authored-by: Mongo <5c25403eab7271f9f94ddd4f2b270e8cac2c92e2c830c51877cca6ec974ffb3f@buzz.block.builderlab.xyz>
Co-authored-by: Princess Donut <68157ebd23b3897c1991015c3038658ea916200c67d3a54620b0754d1b92f6e0@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
2026-08-13 13:56:37 -06:00
committed by GitHub
parent 45f4b91a36
commit 2693e0db1f
21 changed files with 726 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
crates/buzz-db/src/lib.rs
View File

@@ -4051,6 +4051,27 @@ impl Db {
workflow::list_workflow_runs(&self.pool, community_id, workflow_id, limit).await
}
/// List one keyset-paginated page of workflow runs.
#[datastore_span(name = "list_workflow_runs_page", system = "postgresql")]
pub async fn list_workflow_runs_page(
&self,
community_id: CommunityId,
workflow_id: Uuid,
before: Option<chrono::DateTime<chrono::Utc>>,
before_id: Option<Uuid>,
limit: i64,
) -> Result<Vec<workflow::WorkflowRunRecord>> {
workflow::list_workflow_runs_page(
&self.pool,
community_id,
workflow_id,
before,
before_id,
limit,
)
.await
}
/// Update a workflow run's status.
#[datastore_span(name = "update_workflow_run", system = "postgresql")]
pub async fn update_workflow_run(
@@ -4060,7 +4081,7 @@ impl Db {
status: workflow::RunStatus,
current_step: i32,
trace: &serde_json::Value,
error: Option<&str>,
failure: Option<workflow::WorkflowRunFailure<'_>>,
) -> Result<()> {
workflow::update_workflow_run(
&self.pool,
@@ -4069,7 +4090,7 @@ impl Db {
status,
current_step,
trace,
error,
failure,
)
.await
}

23
crates/buzz-db/src/migration.rs
View File

@@ -625,7 +625,7 @@ mod tests {
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
migrations.sort_by_key(|migration| migration.version);
assert_eq!(migrations.len(), 30);
assert_eq!(migrations.len(), 31);
assert_eq!(migrations[0].version, 1);
assert_eq!(&*migrations[0].description, "initial schema");
assert!(migrations[0]
@@ -1038,6 +1038,27 @@ mod tests {
assert!(deletion_recovery.contains("SET LOCAL lock_timeout = '5s'"));
}
#[test]
fn workflow_run_error_codes_are_additive_and_backfilled_without_parsing_diagnostics() {
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
migrations.sort_by_key(|migration| migration.version);
assert_eq!(migrations[30].version, 31);
let sql = migrations[30].sql.as_str();
assert!(sql.contains("ALTER TABLE workflow_runs ADD COLUMN error_code TEXT"));
assert!(sql.contains("SET error_code = 'legacy_unclassified'"));
assert!(sql.contains("status IN ('failed', 'cancelled')"));
assert!(!sql.contains("error_message LIKE"));
assert!(!MIGRATOR
.iter()
.find(|migration| migration.version == 1)
.expect("initial migration")
.sql
.as_str()
.contains("error_code"));
assert!(include_str!("../../../schema/schema.sql").contains("error_code TEXT"));
}
#[test]
fn migration_lint_detects_tables_missing_community_id_by_default() {
let sql = r#"

75
crates/buzz-db/src/workflow.rs
View File

@@ -216,8 +216,11 @@ pub struct WorkflowRunRecord {
pub started_at: Option<DateTime<Utc>>,
/// When execution finished (success or failure).
pub completed_at: Option<DateTime<Utc>>,
/// Error message if the run failed.
/// Redacted human-readable diagnostic for failed or cancelled runs.
pub error_message: Option<String>,
/// Stable machine-readable failure or cancellation classification.
/// Kept separate from `error_message` so callers never parse diagnostics.
pub error_code: Option<String>,
/// When the run record was created.
pub created_at: DateTime<Utc>,
}
@@ -831,7 +834,7 @@ pub async fn get_workflow_run(
let row = sqlx::query(
r#"
SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step,
execution_trace, trigger_context, started_at, completed_at, error_message, created_at
execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at
FROM workflow_runs
WHERE community_id = $1 AND id = $2
"#,
@@ -845,26 +848,40 @@ pub async fn get_workflow_run(
row_to_run_record(row)
}
/// List runs for a workflow, newest first, up to `limit` rows.
pub async fn list_workflow_runs(
/// List runs for a workflow using a stable newest-first keyset.
///
/// Rows are ordered by `(created_at DESC, id DESC)`. A cursor is valid only
/// when both `before` and `before_id` are supplied; callers should pass the
/// final row from the previous page. `limit` is clamped to the shared list
/// bounds.
pub async fn list_workflow_runs_page(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
before: Option<DateTime<Utc>>,
before_id: Option<Uuid>,
limit: i64,
) -> Result<Vec<WorkflowRunRecord>> {
let limit = limit.min(1000);
let limit = limit.clamp(1, LIST_MAX_LIMIT);
let rows = sqlx::query(
r#"
SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step,
execution_trace, trigger_context, started_at, completed_at, error_message, created_at
execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at
FROM workflow_runs
WHERE community_id = $1 AND workflow_id = $2
ORDER BY created_at DESC
LIMIT $3
AND (
$3::timestamptz IS NULL
OR $4::uuid IS NULL
OR (created_at, id) < ($3, $4)
)
ORDER BY created_at DESC, id DESC
LIMIT $5
"#,
)
.bind(community_id.as_uuid())
.bind(workflow_id)
.bind(before)
.bind(before_id)
.bind(limit)
.fetch_all(pool)
.await?;
@@ -872,7 +889,26 @@ pub async fn list_workflow_runs(
rows.into_iter().map(row_to_run_record).collect()
}
/// Update run status, current step, execution trace, and optional error message.
/// List runs for a workflow, newest first, up to `limit` rows.
pub async fn list_workflow_runs(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
limit: i64,
) -> Result<Vec<WorkflowRunRecord>> {
list_workflow_runs_page(pool, community_id, workflow_id, None, None, limit).await
}
/// Structured failure persisted for a workflow run.
#[derive(Debug, Clone, Copy)]
pub struct WorkflowRunFailure<'a> {
/// Stable machine-readable failure code.
pub code: &'a str,
/// Human-readable failure detail.
pub message: &'a str,
}
/// Update run status, current step, execution trace, and optional failure.
///
/// Fix C3: `started_at` is set when the NEW status is 'running' and `started_at`
/// has not yet been stamped (IS NULL). The original code read `status` from the
@@ -885,26 +921,31 @@ pub async fn update_workflow_run(
status: RunStatus,
current_step: i32,
trace: &serde_json::Value,
error: Option<&str>,
failure: Option<WorkflowRunFailure<'_>>,
) -> Result<()> {
let status_str = status.to_string();
let (error_code, error) = failure
.map(|failure| (Some(failure.code), Some(failure.message)))
.unwrap_or((None, None));
let affected = sqlx::query(
r#"
UPDATE workflow_runs
SET status = $1::run_status,
current_step = $2,
execution_trace = $3,
error_message = $4,
started_at = CASE WHEN $5 = 'running' AND started_at IS NULL
error_code = $4,
error_message = $5,
started_at = CASE WHEN $6 = 'running' AND started_at IS NULL
THEN NOW() ELSE started_at END,
completed_at = CASE WHEN $6 IN ('completed','failed','cancelled')
completed_at = CASE WHEN $7 IN ('completed','failed','cancelled')
THEN NOW() ELSE completed_at END
WHERE community_id = $7 AND id = $8
WHERE community_id = $8 AND id = $9
"#,
)
.bind(&status_str)
.bind(current_step)
.bind(trace)
.bind(error_code)
.bind(error)
.bind(&status_str) // for started_at CASE
.bind(&status_str) // for completed_at CASE
@@ -1169,6 +1210,7 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRunRecord> {
started_at: row.try_get("started_at")?,
completed_at: row.try_get("completed_at")?,
error_message: row.try_get("error_message")?,
error_code: row.try_get("error_code")?,
created_at: row.try_get("created_at")?,
})
}
@@ -1473,6 +1515,7 @@ mod tests {
started_at: Some(now),
completed_at: None,
error_message: None,
error_code: None,
created_at: now,
};
@@ -1501,6 +1544,7 @@ mod tests {
started_at: None,
completed_at: None,
error_message: None,
error_code: None,
created_at: now,
};
@@ -1524,6 +1568,7 @@ mod tests {
started_at: Some(now),
completed_at: Some(now),
error_message: Some("step timeout exceeded".to_owned()),
error_code: Some("step_timeout".to_owned()),
created_at: now,
};
@@ -1555,6 +1600,7 @@ mod tests {
started_at: Some(now),
completed_at: Some(now),
error_message: None,
error_code: None,
created_at: now,
};
@@ -1577,6 +1623,7 @@ mod tests {
started_at: None,
completed_at: None,
error_message: None,
error_code: None,
created_at: now,
};

7
crates/buzz-relay/src/api/bridge.rs
View File

@@ -21,7 +21,7 @@ use crate::state::AppState;
use super::{api_error, internal_error, not_found};
async fn enforce_http_admission(
pub(crate) async fn enforce_http_admission(
state: &AppState,
tenant: &TenantContext,
pubkey: &nostr::PublicKey,
@@ -1938,7 +1938,10 @@ pub async fn workflow_webhook(
buzz_db::workflow::RunStatus::Failed,
0,
&serde_json::json!([]),
Some(&format!("definition parse error: {e}")),
Some(buzz_db::workflow::WorkflowRunFailure {
code: "invalid_definition",
message: &format!("definition parse error: {e}"),
}),
)
.await
{

1
crates/buzz-relay/src/api/mod.rs
View File

@@ -9,6 +9,7 @@ pub mod media;
pub mod mesh_demo;
pub mod nip05;
pub mod operator;
pub mod workflows;
// Re-export imeta helpers used by ingest pipeline.
pub use crate::handlers::imeta::{validate_imeta_tags, verify_imeta_blobs};

264
crates/buzz-relay/src/api/workflows.rs Normal file
View File

@@ -0,0 +1,264 @@
//! Authorized structured reads for workflow execution state.
//!
//! Runs and approvals are relay-owned database rows, not Nostr events. These
//! endpoints expose those read models without inventing synthetic events.
use std::sync::Arc;
use axum::{
extract::{Path, Query, RawQuery, State},
http::{HeaderMap, StatusCode},
response::Json,
};
use chrono::{DateTime, Utc};
use serde::Deserialize;
use serde_json::Value;
use uuid::Uuid;
use buzz_core::TenantContext;
use crate::{
api::{api_error, bridge, internal_error},
state::AppState,
};
const DEFAULT_RUN_LIMIT: i64 = 20;
const MAX_RUN_LIMIT: i64 = 100;
/// Pagination query for workflow run history.
#[derive(Debug, Deserialize, Default)]
pub struct RunsQuery {
before: Option<DateTime<Utc>>,
before_id: Option<Uuid>,
limit: Option<i64>,
}
fn request_path(path: &str, raw_query: Option<&str>) -> String {
match raw_query {
Some(query) if !query.is_empty() => format!("{path}?{query}"),
_ => path.to_string(),
}
}
async fn authorize_workflow_read(
state: &Arc<AppState>,
headers: &HeaderMap,
path: &str,
raw_query: Option<&str>,
workflow_id: Uuid,
) -> Result<TenantContext, (StatusCode, Json<Value>)> {
let raw_host = headers
.get(axum::http::header::HOST)
.and_then(|value| value.to_str().ok())
.unwrap_or("");
let tenant = crate::tenant::bind_community(&state.db, raw_host)
.await
.map_err(|_| {
api_error(
StatusCode::NOT_FOUND,
"relay: no community is configured for this host",
)
})?;
let path_with_query = request_path(path, raw_query);
let url = bridge::nip98_expected_url(&state.config.relay_url, &tenant, &path_with_query);
let (pubkey, event_id_bytes) =
bridge::verify_bridge_auth(headers, "GET", &url, None, state.config.require_auth_token)?;
bridge::enforce_http_admission(state, &tenant, &pubkey).await?;
bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?;
let pubkey_bytes = pubkey.to_bytes().to_vec();
let auth_tag = headers
.get("x-auth-tag")
.and_then(|value| value.to_str().ok());
super::relay_members::enforce_relay_membership(
state,
tenant.community(),
&pubkey_bytes,
auth_tag,
)
.await?;
let workflow = state
.db
.get_workflow(tenant.community(), workflow_id)
.await
.map_err(|error| match error {
buzz_db::error::DbError::NotFound(_) => {
api_error(StatusCode::NOT_FOUND, "workflow not found")
}
other => internal_error(&format!("get workflow for run read: {other}")),
})?;
let channel_id = workflow
.channel_id
.ok_or_else(|| api_error(StatusCode::FORBIDDEN, "workflow is not channel-scoped"))?;
let accessible = state
.get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes)
.await
.map_err(|error| internal_error(&format!("workflow channel access lookup: {error}")))?;
if !accessible.contains(&channel_id) {
return Err(api_error(
StatusCode::FORBIDDEN,
"workflow is not accessible",
));
}
Ok(tenant)
}
/// `GET /workflows/{workflow_id}/runs` — one authorized, keyset-paginated page.
pub async fn workflow_runs(
State(state): State<Arc<AppState>>,
Path(workflow_id): Path<Uuid>,
headers: HeaderMap,
RawQuery(raw_query): RawQuery,
Query(query): Query<RunsQuery>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
if query.before.is_some() != query.before_id.is_some() {
return Err(api_error(
StatusCode::BAD_REQUEST,
"before and before_id must be supplied together",
));
}
let limit = query.limit.unwrap_or(DEFAULT_RUN_LIMIT);
if !(1..=MAX_RUN_LIMIT).contains(&limit) {
return Err(api_error(
StatusCode::BAD_REQUEST,
"limit must be between 1 and 100",
));
}
let path = format!("/workflows/{workflow_id}/runs");
let tenant =
authorize_workflow_read(&state, &headers, &path, raw_query.as_deref(), workflow_id).await?;
let mut rows = state
.db
.list_workflow_runs_page(
tenant.community(),
workflow_id,
query.before,
query.before_id,
limit + 1,
)
.await
.map_err(|error| internal_error(&format!("list workflow runs: {error}")))?;
let has_more = rows.len() > limit as usize;
rows.truncate(limit as usize);
let next = if has_more {
rows.last().map(|last| {
serde_json::json!({
"before": last.created_at,
"before_id": last.id,
})
})
} else {
None
};
Ok(Json(serde_json::json!({
"runs": rows.iter().map(run_json).collect::<Vec<_>>(),
"next": next,
})))
}
/// `GET /workflows/{workflow_id}/runs/{run_id}/approvals` — approvals for a run.
pub async fn run_approvals(
State(state): State<Arc<AppState>>,
Path((workflow_id, run_id)): Path<(Uuid, Uuid)>,
headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let path = format!("/workflows/{workflow_id}/runs/{run_id}/approvals");
let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id).await?;
let run = state
.db
.get_workflow_run(tenant.community(), run_id)
.await
.map_err(|error| match error {
buzz_db::error::DbError::NotFound(_) => {
api_error(StatusCode::NOT_FOUND, "workflow run not found")
}
other => internal_error(&format!("get workflow run for approval read: {other}")),
})?;
if run.workflow_id != workflow_id {
return Err(api_error(StatusCode::NOT_FOUND, "workflow run not found"));
}
let approvals = state
.db
.get_run_approvals(tenant.community(), workflow_id, run_id)
.await
.map_err(|error| internal_error(&format!("list run approvals: {error}")))?;
Ok(Json(serde_json::json!({
"approvals": approvals.iter().map(approval_json).collect::<Vec<_>>(),
})))
}
fn run_json(run: &buzz_db::workflow::WorkflowRunRecord) -> Value {
serde_json::json!({
"id": run.id,
"workflow_id": run.workflow_id,
"status": run.status,
"current_step": run.current_step,
"execution_trace": run.execution_trace,
"started_at": run.started_at.map(|value| value.timestamp()),
"completed_at": run.completed_at.map(|value| value.timestamp()),
"error_code": run.error_code,
"error_message": run.error_message,
"created_at": run.created_at.timestamp(),
})
}
fn approval_json(approval: &buzz_db::workflow::ApprovalRecord) -> Value {
serde_json::json!({
"approval_ref": hex::encode(&approval.token),
"workflow_id": approval.workflow_id,
"run_id": approval.run_id,
"step_id": approval.step_id,
"step_index": approval.step_index,
"approver_spec": approval.approver_spec,
"status": approval.status,
"approver_pubkey": approval.approver_pubkey.as_ref().map(hex::encode),
"note": approval.note,
"expires_at": approval.expires_at,
"created_at": approval.created_at.timestamp(),
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn request_path_preserves_signed_query_verbatim() {
assert_eq!(
request_path("/workflows/id/runs", Some("limit=20&before_id=abc")),
"/workflows/id/runs?limit=20&before_id=abc"
);
assert_eq!(
request_path("/workflows/id/runs", None),
"/workflows/id/runs"
);
}
#[test]
fn approval_wire_does_not_expose_hash_as_token() {
let approval = buzz_db::workflow::ApprovalRecord {
token: vec![0xab; 32],
workflow_id: Uuid::new_v4(),
run_id: Uuid::new_v4(),
step_id: "review".to_string(),
step_index: 1,
approver_spec: "any".to_string(),
status: buzz_db::workflow::ApprovalStatus::Pending,
approver_pubkey: None,
note: None,
expires_at: Utc::now(),
created_at: Utc::now(),
};
let wire = approval_json(&approval);
assert!(wire.get("token").is_none());
assert_eq!(wire["approval_ref"], hex::encode([0xab; 32]));
}
}

15
crates/buzz-relay/src/handlers/command_executor.rs
View File

@@ -964,7 +964,10 @@ async fn handle_workflow_trigger(
RunStatus::Failed,
0,
&serde_json::json!([]),
Some(&format!("definition parse error: {e}")),
Some(buzz_db::workflow::WorkflowRunFailure {
code: "invalid_definition",
message: &format!("definition parse error: {e}"),
}),
)
.await
{
@@ -1261,7 +1264,10 @@ async fn handle_approval_deny(
RunStatus::Cancelled,
run.current_step,
&run.execution_trace,
Some(&cancel_msg),
Some(buzz_db::workflow::WorkflowRunFailure {
code: "approval_denied",
message: &cancel_msg,
}),
)
.await
{
@@ -1329,7 +1335,10 @@ async fn resume_workflow_after_approval(
RunStatus::Failed,
run.current_step,
&run.execution_trace,
Some(&format!("definition parse error: {e}")),
Some(buzz_db::workflow::WorkflowRunFailure {
code: "invalid_definition",
message: &format!("definition parse error: {e}"),
}),
)
.await
{

8
crates/buzz-relay/src/router.rs
View File

@@ -72,6 +72,14 @@ pub fn build_router(state: Arc<AppState>) -> Router {
.route("/events", post(api::bridge::submit_event))
.route("/query", post(api::bridge::query_events))
.route("/count", post(api::bridge::count_events))
.route(
"/workflows/{workflow_id}/runs",
get(api::workflows::workflow_runs),
)
.route(
"/workflows/{workflow_id}/runs/{run_id}/approvals",
get(api::workflows::run_approvals),
)
.route(
"/operator/communities",
get(api::operator::list_owned_communities).post(api::operator::provision_community),

42
crates/buzz-workflow/src/error.rs
View File

@@ -65,8 +65,50 @@ pub enum WorkflowError {
NotImplemented(String),
}
impl WorkflowError {
/// Stable run-level classification. Diagnostics remain in `Display` output.
pub const fn code(&self) -> &'static str {
match self {
Self::InvalidYaml(_) => "invalid_yaml",
Self::InvalidDefinition(_) => "invalid_definition",
Self::ConditionError(_) => "condition_evaluation_failed",
Self::TemplateError(_) => "template_resolution_failed",
Self::StepTimeout { .. } => "step_timeout",
Self::WebhookError(_) => "webhook_failed",
Self::CapacityExceeded => "capacity_exceeded",
Self::Database(_) => "database_error",
Self::Unauthorized(_) => "owner_unauthorized",
Self::NotImplemented(_) => "action_not_implemented",
}
}
}
impl From<buzz_db::error::DbError> for WorkflowError {
fn from(e: buzz_db::error::DbError) -> Self {
WorkflowError::Database(e.to_string())
}
}
#[cfg(test)]
mod tests {
use super::WorkflowError;
#[test]
fn workflow_error_codes_are_stable_and_separate_from_diagnostics() {
let timeout = WorkflowError::StepTimeout {
step_id: "notify".to_owned(),
timeout_secs: 30,
};
assert_eq!(timeout.code(), "step_timeout");
assert!(timeout.to_string().contains("notify"));
let webhook = WorkflowError::WebhookError("secret-bearing detail".to_owned());
assert_eq!(webhook.code(), "webhook_failed");
assert!(!webhook.code().contains("secret-bearing detail"));
assert_eq!(
WorkflowError::NotImplemented("SendDm".to_owned()).code(),
"action_not_implemented"
);
}
}

10
crates/buzz-workflow/src/lib.rs
View File

@@ -242,7 +242,10 @@ impl WorkflowEngine {
RunStatus::Failed,
step_count,
&trace_json,
Some("approval gates not yet implemented — see WF-08"),
Some(buzz_db::workflow::WorkflowRunFailure {
code: "approval_not_supported",
message: "approval gates not yet implemented — see WF-08",
}),
)
.await
{
@@ -285,7 +288,10 @@ impl WorkflowEngine {
RunStatus::Failed,
progress.step_index as i32,
&trace_json,
Some(&e.to_string()),
Some(buzz_db::workflow::WorkflowRunFailure {
code: e.code(),
message: &e.to_string(),
}),
)
.await
{