feat(agents): register external Hermes runtimes

crates/buzz-cli/src/commands/workflows.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use sha2::{Digest, Sha256};
 
 use crate::client::{
@@ -9,6 +11,16 @@ use crate::validate::{parse_uuid, read_or_stdin, sdk_err, validate_uuid};
 
 // TODO(phase-4): Replace raw nostr::EventBuilder usage with buzz-sdk builder functions
 
+fn lifecycle_precedence(kind: u64) -> u8 {
+    match kind {
+        46005..=46007 => 4,
+        46011..=46012 => 3,
+        46010 => 2,
+        46001 => 1,
+        _ => 0,
+    }
+}
+
 /// List workflows in a channel — query kind:30620 workflow definition events.
 pub async fn cmd_list_workflows(client: &BuzzClient, channel_id: &str) -> Result<(), CliError> {
     validate_uuid(channel_id)?;
@@ -57,12 +69,7 @@ pub async fn cmd_get_workflow(client: &BuzzClient, workflow_id: &str) -> Result<
     Ok(())
 }
 
-/// Get workflow run history — query kinds [46001, 46002, 46003].
-///
-/// NOTE: The relay does not currently emit workflow execution events (46001-46003).
-/// Run history is stored in the workflow_runs DB table, not as Nostr events.
-/// This command will return an empty array until the relay adds event emission
-/// or a dedicated REST endpoint for run history.
+/// Get workflow run history from durable lifecycle events.
 pub async fn cmd_get_workflow_runs(
     client: &BuzzClient,
     workflow_id: &str,
@@ -71,24 +78,51 @@ pub async fn cmd_get_workflow_runs(
     validate_uuid(workflow_id)?;
     let limit = limit.unwrap_or(20).min(100);
     let filter = serde_json::json!({
-        "kinds": [46001, 46002, 46003],
+        "kinds": [46001, 46005, 46006, 46007, 46010, 46011, 46012],
         "#d": [workflow_id],
-        "limit": limit
+        "limit": (limit * 10).min(500)
     });
     let resp = client.query(&filter).await?;
     let events: Vec<serde_json::Value> = serde_json::from_str(&resp).unwrap_or_default();
-    let normalized: Vec<serde_json::Value> = events
-        .iter()
-        .map(|e| {
-            serde_json::json!({
-                "event_id": e.get("id").and_then(|v| v.as_str()).unwrap_or(""),
-                "kind": e.get("kind").and_then(|v| v.as_u64()).unwrap_or(0),
-                "content": e.get("content").and_then(|v| v.as_str()).unwrap_or(""),
-                "created_at": e.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0),
-                "tags": e.get("tags").cloned().unwrap_or(serde_json::json!([])),
+    let mut latest: HashMap<String, (u64, u8, serde_json::Value)> = HashMap::new();
+    for event in events {
+        let Some(content) = event.get("content").and_then(|v| v.as_str()) else {
+            continue;
+        };
+        let Ok(payload) = serde_json::from_str::<serde_json::Value>(content) else {
+            continue;
+        };
+        let kind = event.get("kind").and_then(|v| v.as_u64()).unwrap_or(0);
+        let run = if (46010..=46012).contains(&kind) {
+            payload.get("run").cloned().unwrap_or(payload)
+        } else {
+            payload
+        };
+        if run.get("workflow_id").and_then(|v| v.as_str()) != Some(workflow_id) {
+            continue;
+        }
+        let Some(run_id) = run.get("id").and_then(|v| v.as_str()) else {
+            continue;
+        };
+        let created_at = event
+            .get("created_at")
+            .and_then(|v| v.as_u64())
+            .unwrap_or(0);
+        let precedence = lifecycle_precedence(kind);
+        if latest
+            .get(run_id)
+            .is_none_or(|(seen_at, seen_precedence, _)| {
+                (created_at, precedence) >= (*seen_at, *seen_precedence)
             })
-        })
-        .collect();
+        {
+            latest.insert(run_id.to_owned(), (created_at, precedence, run));
+        }
+    }
+    let mut normalized: Vec<(u64, u8, serde_json::Value)> = latest.into_values().collect();
+    normalized.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| b.1.cmp(&a.1)));
+    normalized.truncate(limit as usize);
+    let normalized: Vec<serde_json::Value> =
+        normalized.into_iter().map(|(_, _, run)| run).collect();
     let output = serde_json::to_string(&normalized).unwrap_or_default();
     println!("{output}");
     Ok(())
@@ -196,12 +230,18 @@ pub async fn cmd_approve_step(
     approved: bool,
     note: Option<&str>,
 ) -> Result<(), CliError> {
-    validate_uuid(approval_token)?;
-
     let content = note.unwrap_or("");
 
     // The relay expects d-tag = hex(SHA256(token)), not the raw token UUID.
-    let token_hash = hex::encode(Sha256::digest(approval_token.as_bytes()));
+    // Desktop approval cards already expose the stored hash, so accept either
+    // that 64-character hash or the original raw UUID for CLI compatibility.
+    let token_hash =
+        if approval_token.len() == 64 && approval_token.chars().all(|c| c.is_ascii_hexdigit()) {
+            approval_token.to_ascii_lowercase()
+        } else {
+            validate_uuid(approval_token)?;
+            hex::encode(Sha256::digest(approval_token.as_bytes()))
+        };
     let builder =
         buzz_sdk::build_workflow_approval(&token_hash, approved, content).map_err(sdk_err)?;
     let event = client.sign_event(builder)?;

crates/buzz-db/src/lib.rs

@@ -3825,6 +3825,16 @@ impl Db {
         workflow::create_approval(&self.pool, params).await
     }
 
+    /// Atomically suspend a workflow run and create its pending approval.
+    pub async fn suspend_workflow_run_for_approval(
+        &self,
+        params: workflow::CreateApprovalParams<'_>,
+        current_step: i32,
+        trace: &serde_json::Value,
+    ) -> Result<workflow::ApprovalRecord> {
+        workflow::suspend_workflow_run_for_approval(&self.pool, params, current_step, trace).await
+    }
+
     /// Fetch an approval by raw token.
     pub async fn get_approval(
         &self,

crates/buzz-db/src/workflow.rs

@@ -979,6 +979,88 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) ->
     Ok(())
 }
 
+/// Atomically suspend a run and create its pending approval record.
+///
+/// A waiting run without its approval row is unreachable, while an approval
+/// row without a waiting run is actionable garbage. Keep the two writes in the
+/// same transaction so the relay never commits either half of a human gate.
+pub async fn suspend_workflow_run_for_approval(
+    pool: &PgPool,
+    params: CreateApprovalParams<'_>,
+    current_step: i32,
+    trace: &serde_json::Value,
+) -> Result<ApprovalRecord> {
+    let CreateApprovalParams {
+        community_id,
+        token,
+        workflow_id,
+        run_id,
+        step_id,
+        step_index,
+        approver_spec,
+        expires_at,
+    } = params;
+    let token_hash = hash_approval_token(token);
+    let mut tx = pool.begin().await?;
+
+    let updated = sqlx::query(
+        r#"
+        UPDATE workflow_runs
+        SET status = 'waiting_approval',
+            current_step = $1,
+            execution_trace = $2,
+            error_message = NULL
+        WHERE community_id = $3 AND id = $4
+        "#,
+    )
+    .bind(current_step)
+    .bind(trace)
+    .bind(community_id.as_uuid())
+    .bind(run_id)
+    .execute(&mut *tx)
+    .await?
+    .rows_affected();
+
+    if updated == 0 {
+        return Err(DbError::NotFound(format!("workflow_run {run_id}")));
+    }
+
+    let created_at: DateTime<Utc> = sqlx::query_scalar(
+        r#"
+        INSERT INTO workflow_approvals
+            (community_id, token, workflow_id, run_id, step_id, step_index, approver_spec, status, expires_at)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, 'pending', $8)
+        RETURNING created_at
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(&token_hash)
+    .bind(workflow_id)
+    .bind(run_id)
+    .bind(step_id)
+    .bind(step_index)
+    .bind(approver_spec)
+    .bind(expires_at)
+    .fetch_one(&mut *tx)
+    .await?;
+
+    tx.commit().await?;
+
+    Ok(ApprovalRecord {
+        token: token_hash,
+        workflow_id,
+        run_id,
+        step_id: step_id.to_owned(),
+        step_index,
+        approver_spec: approver_spec.to_owned(),
+        status: ApprovalStatus::Pending,
+        approver_pubkey: None,
+        note: None,
+        expires_at,
+        created_at,
+    })
+}
+
 /// Fetch an approval record by raw token.
 ///
 /// The token is hashed before the DB lookup so plaintext tokens are never

crates/buzz-relay/src/api/bridge.rs

@@ -1918,6 +1918,11 @@ pub async fn workflow_webhook(
         .await
         .map_err(|e| super::internal_error(&format!("db error: {e}")))?;
 
+    state
+        .workflow_engine
+        .record_run_triggered(community_id, run_id)
+        .await;
+
     // Spawn workflow execution asynchronously.
     let engine = Arc::clone(&state.workflow_engine);
     let db = state.db.clone();

crates/buzz-relay/src/handlers/command_executor.rs

@@ -939,6 +939,11 @@ async fn handle_workflow_trigger(
         .await
         .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
 
+    state
+        .workflow_engine
+        .record_run_triggered(community_id, run_id)
+        .await;
+
     // 5. Spawn workflow execution
     let engine = Arc::clone(&state.workflow_engine);
     let db = state.db.clone();
@@ -1111,8 +1116,17 @@ async fn handle_approval_grant(
         .await
         .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
 
-    // 6. Resume workflow execution (post-commit, async)
     let community_id = tenant.community();
+    let mut resolved_approval = approval.clone();
+    resolved_approval.status = ApprovalStatus::Granted;
+    resolved_approval.approver_pubkey = Some(self_bytes.clone());
+    resolved_approval.note = note.map(str::to_owned);
+    state
+        .workflow_engine
+        .record_approval_resolution(community_id, &resolved_approval)
+        .await;
+
+    // 6. Resume workflow execution (post-commit, async)
     let run_id = approval.run_id;
     let workflow_id = approval.workflow_id;
     let resume_index = approval.step_index as usize + 1;
@@ -1222,44 +1236,48 @@ async fn handle_approval_deny(
         .await
         .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
 
-    // 6. Cancel the workflow run (post-commit, async)
+    // 6. Cancel the workflow run and persist its lifecycle receipt.
     let community_id = tenant.community();
     let run_id = approval.run_id;
     let pubkey_hex = self_hex.clone();
-    let db = state.db.clone();
-
-    tokio::spawn(async move {
-        let run = match db.get_workflow_run(community_id, run_id).await {
-            Ok(r) => r,
-            Err(e) => {
-                tracing::error!("approval_deny: failed to fetch run {run_id}: {e}");
-                return;
-            }
-        };
+    let run = state
+        .db
+        .get_workflow_run(community_id, run_id)
+        .await
+        .map_err(|e| IngestError::Internal(format!("error: db get_workflow_run: {e}")))?;
+    if run.status != RunStatus::WaitingApproval {
+        return Err(IngestError::Rejected(format!(
+            "invalid: workflow run is {} rather than waiting_approval",
+            run.status
+        )));
+    }
 
-        if run.status != RunStatus::WaitingApproval {
-            tracing::warn!(
-                "approval_deny: run {run_id} has status '{}', expected 'waiting_approval'",
-                run.status
-            );
-            return;
-        }
+    let cancel_msg = format!("workflow cancelled: approval denied by {pubkey_hex}");
+    state
+        .db
+        .update_workflow_run(
+            community_id,
+            run_id,
+            RunStatus::Cancelled,
+            run.current_step,
+            &run.execution_trace,
+            Some(&cancel_msg),
+        )
+        .await
+        .map_err(|e| IngestError::Internal(format!("error: db cancel workflow_run: {e}")))?;
 
-        let cancel_msg = format!("workflow cancelled: approval denied by {pubkey_hex}");
-        if let Err(e) = db
-            .update_workflow_run(
-                community_id,
-                run_id,
-                RunStatus::Cancelled,
-                run.current_step,
-                &run.execution_trace,
-                Some(&cancel_msg),
-            )
-            .await
-        {
-            tracing::error!("approval_deny: failed to cancel run {run_id}: {e}");
-        }
-    });
+    let mut resolved_approval = approval.clone();
+    resolved_approval.status = ApprovalStatus::Denied;
+    resolved_approval.approver_pubkey = Some(self_bytes);
+    resolved_approval.note = note.map(str::to_owned);
+    state
+        .workflow_engine
+        .record_approval_resolution(community_id, &resolved_approval)
+        .await;
+    state
+        .workflow_engine
+        .record_run_cancelled(community_id, run_id)
+        .await;
 
     // 7. Return response
     Ok(IngestResult {
@@ -1353,7 +1371,16 @@ async fn resume_workflow_after_approval(
         .unwrap_or_default();
 
     // Execute remaining steps
-    let existing_trace = run.execution_trace.as_array().cloned();
+    let mut existing_trace = run.execution_trace.as_array().cloned().unwrap_or_default();
+    if resume_index > 0 {
+        if let Some(entry) = existing_trace.get_mut(resume_index - 1) {
+            if entry.get("status").and_then(serde_json::Value::as_str) == Some("waiting_approval") {
+                entry["status"] = serde_json::json!("completed");
+                entry["output"] = serde_json::json!({"approval": "granted"});
+            }
+        }
+    }
+    let existing_trace = Some(existing_trace);
     let result = buzz_workflow::executor::execute_from_step(
         &engine,
         community_id,