fix(workflow): replace NUL in webhook output and trigger fields

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

@@ -2949,14 +2949,8 @@ pub async fn workflow_webhook(
             .unwrap_or_default(),
         ..Default::default()
     };
-    if let Some(Value::Object(ref map)) = body_json {
-        for (k, v) in map {
-            let val_str = match v {
-                Value::String(s) => s.clone(),
-                other => other.to_string(),
-            };
-            trigger_ctx.webhook_fields.insert(k.clone(), val_str);
-        }
+    if let Some(Value::Object(map)) = body_json {
+        trigger_ctx.extend_webhook_fields(map);
     }
     let trigger_ctx_json = serde_json::to_value(&trigger_ctx).ok();
 

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

@@ -133,6 +133,16 @@ async fn post_webhook(
     workflow_id: Uuid,
     host: &str,
     secret: &str,
+) -> (StatusCode, Value) {
+    post_webhook_with_body(state, workflow_id, host, secret, Bytes::new()).await
+}
+
+async fn post_webhook_with_body(
+    state: &Arc<AppState>,
+    workflow_id: Uuid,
+    host: &str,
+    secret: &str,
+    body: Bytes,
 ) -> (StatusCode, Value) {
     let mut headers = HeaderMap::new();
     headers.insert(
@@ -149,7 +159,7 @@ async fn post_webhook(
         Path(workflow_id.to_string()),
         Query(WebhookQuery { secret: None }),
         headers,
-        Bytes::new(),
+        body,
     )
     .await
     {
@@ -355,3 +365,60 @@ async fn webhook_admission_enforces_owner_ban_tenant_scope_and_lookup_failure()
         .await
         .expect("drop restriction lookup failure schema");
 }
+
+/// PostgreSQL `jsonb` rejects `\u0000`; a webhook body field holding an
+/// escaped NUL must still create the run instead of failing the insert.
+#[tokio::test]
+#[ignore = "requires PostgreSQL and Redis"]
+async fn webhook_body_field_with_nul_creates_run() {
+    let pool = PgPool::connect(&database_url())
+        .await
+        .expect("connect to isolated PostgreSQL test database");
+    let db = buzz_db::Db::from_pool(pool.clone());
+
+    let community_id = Uuid::new_v4();
+    let host = format!("webhook-nul-{}.test", Uuid::new_v4().simple());
+    let owner_pubkey = [0x44; 32];
+    let workflow_id = Uuid::new_v4();
+    let secret = format!("webhook-secret-{}", Uuid::new_v4().simple());
+    seed_community(&pool, community_id, &host).await;
+    seed_webhook_workflow(
+        &pool,
+        WebhookWorkflowSeed {
+            community_id,
+            workflow_id,
+            channel_id: Uuid::new_v4(),
+            owner_pubkey: &owner_pubkey,
+            owner_role: "owner",
+            name: "nul-field-workflow",
+            secret: &secret,
+            steps: json!([{"id": "hold", "action": "delay", "duration": "1s"}]),
+        },
+    )
+    .await;
+
+    let (state, audit_shutdown) = app_state(db, &pool).await;
+    let (status, body) = post_webhook_with_body(
+        &state,
+        workflow_id,
+        &host,
+        &secret,
+        Bytes::from_static(br#"{"note": "a\u0000b"}"#),
+    )
+    .await;
+    assert_eq!(status, StatusCode::ACCEPTED, "response: {body}");
+
+    let trigger_context: Value = sqlx::query_scalar(
+        "SELECT trigger_context FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2",
+    )
+    .bind(community_id)
+    .bind(workflow_id)
+    .fetch_one(&pool)
+    .await
+    .expect("load run trigger context");
+    assert_eq!(trigger_context["webhook_fields"]["note"], "a\u{FFFD}b");
+
+    audit_shutdown
+        .drain(std::time::Duration::from_secs(1))
+        .await;
+}

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

@@ -914,13 +914,7 @@ async fn handle_workflow_trigger(
     };
     if !event.content.is_empty() {
         if let Ok(serde_json::Value::Object(map)) = serde_json::from_str(&event.content) {
-            for (k, v) in map {
-                let val_str = match v {
-                    serde_json::Value::String(s) => s,
-                    other => other.to_string(),
-                };
-                trigger_ctx.webhook_fields.insert(k, val_str);
-            }
+            trigger_ctx.extend_webhook_fields(map);
         }
     }
     let trigger_ctx_json = serde_json::to_value(&trigger_ctx).ok();

crates/buzz-workflow/src/executor.rs

@@ -61,6 +61,44 @@ impl TriggerContext {
             other => self.webhook_fields.get(other).map(|s| s.as_str()),
         }
     }
+
+    /// Flatten a JSON object (webhook body, command payload) into
+    /// `webhook_fields`: strings verbatim, other values as compact JSON.
+    ///
+    /// U+0000 in keys and values becomes U+FFFD: the context is persisted as
+    /// `workflow_runs.trigger_context`, and PostgreSQL `jsonb` rejects `\u0000`.
+    pub fn extend_webhook_fields(&mut self, map: serde_json::Map<String, JsonValue>) {
+        for (key, value) in map {
+            let value = match value {
+                JsonValue::String(s) => s,
+                other => other.to_string(),
+            };
+            self.webhook_fields.insert(storable(key), storable(value));
+        }
+    }
+}
+
+/// Replace U+0000 with U+FFFD so an untrusted string can be persisted in run
+/// state: PostgreSQL `jsonb` rejects `\u0000` ("unsupported Unicode escape
+/// sequence"), which would fail the run's status update and strand it.
+/// U+FFFD matches how `String::from_utf8_lossy` already treats invalid UTF-8.
+fn storable(s: String) -> String {
+    if s.contains('\0') {
+        s.replace('\0', "\u{FFFD}")
+    } else {
+        s
+    }
+}
+
+/// Step output of `call_webhook`: the status and the body decoded lossily
+/// into a [`storable`] string, because the output lands in the run's
+/// `execution_trace`.
+#[cfg(any(feature = "reqwest", test))]
+pub(crate) fn webhook_output(status: u16, body: &[u8]) -> JsonValue {
+    serde_json::json!({
+        "status": status,
+        "body": storable(String::from_utf8_lossy(body).into_owned()),
+    })
 }
 
 /// Resolve `{{trigger.X}}` and `{{steps.ID.output.X}}` placeholders in a string.
@@ -953,12 +991,7 @@ async fn call_webhook_impl(
         }
     }
 
-    let body_text = String::from_utf8_lossy(&body_bytes).into_owned();
-
-    Ok(serde_json::json!({
-        "status": status,
-        "body": body_text,
-    }))
+    Ok(webhook_output(status, &body_bytes))
 }
 
 /// Returns a shared `reqwest::Client` reused across all workflow HTTP calls.
@@ -1983,4 +2016,49 @@ mod tests {
                 .expect("override should be accepted");
         assert_eq!(resolved, override_channel_id.to_string());
     }
+
+    /// True if any string in `value` holds U+0000, which PostgreSQL `jsonb`
+    /// rejects as `\u0000`.
+    fn contains_nul(value: &JsonValue) -> bool {
+        match value {
+            JsonValue::String(s) => s.contains('\0'),
+            JsonValue::Array(items) => items.iter().any(contains_nul),
+            JsonValue::Object(map) => map.iter().any(|(k, v)| k.contains('\0') || contains_nul(v)),
+            _ => false,
+        }
+    }
+
+    #[test]
+    fn webhook_output_replaces_nul_like_invalid_utf8() {
+        let trace = json!([{
+            "step_id": "hook",
+            "status": "completed",
+            "output": webhook_output(200, b"ok\0tail\xff"),
+        }]);
+        assert!(
+            !contains_nul(&trace),
+            "trace is not jsonb-storable: {trace}"
+        );
+        assert_eq!(trace[0]["output"]["body"], "ok\u{FFFD}tail\u{FFFD}");
+        assert_eq!(trace[0]["output"]["status"], 200);
+    }
+
+    #[test]
+    fn extend_webhook_fields_replaces_nul_in_keys_and_values() {
+        let body: JsonValue = serde_json::from_str(
+            r#"{"note": "a\u0000b", "k\u0000": 1, "nested": {"x": "\u0000"}}"#,
+        )
+        .unwrap();
+        let mut ctx = TriggerContext::default();
+        ctx.extend_webhook_fields(body.as_object().cloned().unwrap());
+        let stored = serde_json::to_value(&ctx).unwrap();
+        assert!(
+            !contains_nul(&stored),
+            "context is not jsonb-storable: {stored}"
+        );
+        assert_eq!(ctx.get_field("note"), Some("a\u{FFFD}b"));
+        assert_eq!(ctx.get_field("k\u{FFFD}"), Some("1"));
+        // Non-string values are stored as JSON text, where NUL is escaped.
+        assert_eq!(ctx.get_field("nested"), Some(r#"{"x":"\u0000"}"#));
+    }
 }

crates/buzz-workflow/src/lib.rs

@@ -2618,4 +2618,80 @@ steps:
             run.error_message
         );
     }
+
+    /// PostgreSQL `jsonb` rejects `\u0000`. A `call_webhook` body carrying
+    /// U+0000 must still persist in the trace, or the final status update
+    /// fails and the run stays `running`.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn webhook_output_with_nul_completes_run() {
+        let db = setup_db().await;
+        let creator = nostr::Keys::generate().public_key().to_bytes().to_vec();
+        let member = nostr::Keys::generate().public_key().to_bytes().to_vec();
+        let (community, channel_id) = setup_channel(&db, &creator, &member).await;
+        let def = serde_json::json!({
+            "name": "nul-webhook",
+            "trigger": {"on": "webhook"},
+            "steps": [{"id": "hook", "action": "call_webhook", "url": "https://example.com/x"}],
+            "enabled": true,
+        })
+        .to_string();
+        let workflow_id = db
+            .create_workflow(
+                community,
+                Some(channel_id),
+                &creator,
+                "nul-webhook",
+                &def,
+                &[7u8; 32],
+            )
+            .await
+            .expect("create workflow");
+
+        let run_id = db
+            .create_workflow_run(community, workflow_id, None, None)
+            .await
+            .expect("create run");
+        db.update_workflow_run(
+            community,
+            run_id,
+            RunStatus::Running,
+            0,
+            &serde_json::json!([]),
+            None,
+        )
+        .await
+        .expect("mark run running");
+
+        let engine = WorkflowEngine::new(db.clone(), WorkflowConfig::default());
+        let trace = vec![serde_json::json!({
+            "step_id": "hook",
+            "status": "completed",
+            "output": executor::webhook_output(200, b"ok\0tail"),
+        })];
+        engine
+            .finalize_run(
+                community,
+                run_id,
+                Ok(ExecutionResult {
+                    approval_token: None,
+                    step_index: 1,
+                    step_outputs: HashMap::new(),
+                    trace,
+                }),
+                None,
+            )
+            .await;
+
+        let run = db
+            .get_workflow_run(community, run_id)
+            .await
+            .expect("get run");
+        assert_eq!(
+            run.status,
+            RunStatus::Completed,
+            "a NUL in the webhook body must not strand the run"
+        );
+        assert_eq!(run.execution_trace[0]["output"]["body"], "ok\u{FFFD}tail");
+    }
 }