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

fix(workflow): complete managed owner updates

Browse Source 
Co-authored-by: Larry <larry@buzz.local>

Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-08-21 21:22:14 +00:00
committed by Larry
parent 9f58b64041
commit 9ca9d8ccbc
10 changed files with 1640 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
crates/buzz-acp/src/lib.rs
View File

@@ -23,6 +23,7 @@ use anyhow::{ensure, Context, Result};
use buzz_core::kind::{
KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_STREAM_MESSAGE,
KIND_STREAM_REMINDER, KIND_WORKFLOW_AGENT_WAKE, KIND_WORKFLOW_APPROVAL_REQUESTED,
KIND_WORKFLOW_DEF,
};
use buzz_core::observer::{
decrypt_observer_payload, encrypt_observer_payload, OBSERVER_FRAME_TELEMETRY,
@@ -314,6 +315,139 @@ struct ClaimedWorkflowDelivery {
trigger_context: Option<serde_json::Value>,
}
#[derive(Clone, Debug, serde::Deserialize)]
struct PendingWorkflowOwnerCommand {
command_id: Uuid,
event_id: String,
owner_pubkey: String,
agent_pubkey: String,
workflow_id: Uuid,
channel_id: Uuid,
expected_revision: String,
proposed_yaml: String,
}
#[derive(Clone, Debug, serde::Serialize)]
struct PendingWorkflowOwnerCompletion {
replacement_event: Option<nostr::Event>,
rejection_reason: Option<String>,
}
async fn list_pending_workflow_owner_commands(
rest_client: &relay::RestClient,
) -> Vec<PendingWorkflowOwnerCommand> {
rest_client
.post_json("/workflows/owner-commands/pending", &serde_json::json!({}))
.await
.ok()
.and_then(|response| serde_json::from_value(response.get("commands")?.clone()).ok())
.unwrap_or_default()
}
fn one_exact_tag(event: &nostr::Event, name: &str, expected: &str) -> bool {
let values = exact_tags(event, name);
values.len() == 1 && values[0].as_slice().len() == 2 && values[0].as_slice()[1] == expected
}
async fn prepare_workflow_owner_completion(
command: &PendingWorkflowOwnerCommand,
rest_client: &relay::RestClient,
immutable_owner: &str,
) -> PendingWorkflowOwnerCompletion {
let reject = |reason: &str| PendingWorkflowOwnerCompletion {
replacement_event: None,
rejection_reason: Some(reason.to_owned()),
};
let Ok(command_event_id) = nostr::EventId::from_hex(&command.event_id) else {
return reject("invalid_command_event_id");
};
let Some(command_event) = query_exact_event(command_event_id, rest_client).await else {
return reject("command_event_unavailable");
};
let Ok(parsed) = buzz_core::workflow_owner_command::parse_owner_command(&command_event) else {
return reject("invalid_command_event");
};
let agent = rest_client.keys.public_key();
let owner_matches = nostr::PublicKey::from_hex(immutable_owner)
.is_ok_and(|owner| owner == command_event.pubkey);
if !owner_matches
|| command.owner_pubkey != command_event.pubkey.to_hex()
|| command.agent_pubkey != agent.to_hex()
|| parsed.command_id != command.command_id
|| parsed.agent_pubkey != agent
|| parsed.recipient != agent
|| parsed.workflow_id != command.workflow_id
|| parsed.expected_revision.to_hex() != command.expected_revision
|| parsed.yaml_definition.as_deref() != Some(command.proposed_yaml.as_str())
{
return reject("command_binding_mismatch");
}
let Ok(revision) = nostr::EventId::from_hex(&command.expected_revision) else {
return reject("invalid_expected_revision");
};
let Some(definition) = query_exact_event(revision, rest_client).await else {
return reject("stale_workflow_revision");
};
if definition.kind.as_u16() as u32 != KIND_WORKFLOW_DEF
|| definition.pubkey != agent
|| !one_exact_tag(&definition, "d", &command.workflow_id.to_string())
|| !one_exact_tag(&definition, "h", &command.channel_id.to_string())
{
return reject("workflow_binding_mismatch");
}
if buzz_workflow::WorkflowEngine::parse_yaml(&command.proposed_yaml).is_err() {
return reject("invalid_workflow_yaml");
}
let tags = [
nostr::Tag::parse(["d", &command.workflow_id.to_string()]),
nostr::Tag::parse(["h", &command.channel_id.to_string()]),
nostr::Tag::parse(["expected-revision", &command.expected_revision]),
nostr::Tag::parse(["command", &command.command_id.to_string()]),
];
let Ok(tags) = tags.into_iter().collect::<Result<Vec<_>, _>>() else {
return reject("invalid_replacement_binding");
};
let Ok(event) = nostr::EventBuilder::new(
nostr::Kind::Custom(KIND_WORKFLOW_DEF as u16),
command.proposed_yaml.clone(),
)
.tags(tags)
.sign_with_keys(&rest_client.keys) else {
return reject("replacement_signing_failed");
};
PendingWorkflowOwnerCompletion {
replacement_event: Some(event),
rejection_reason: None,
}
}
async fn reconcile_workflow_owner_completions(
rest_client: &relay::RestClient,
pending: &mut HashMap<Uuid, PendingWorkflowOwnerCompletion>,
) {
let attempts = pending
.iter()
.map(|(command_id, completion)| (*command_id, completion.clone()))
.collect::<Vec<_>>();
for (command_id, completion) in attempts {
let path = format!("/workflows/owner-commands/{command_id}/complete");
match rest_client
.post_json(
&path,
&serde_json::to_value(&completion).unwrap_or_default(),
)
.await
{
Ok(_) => {
pending.remove(&command_id);
}
Err(error) => {
tracing::warn!(%command_id, %error, "workflow owner command completion pending reconciliation");
}
}
}
}
async fn claim_workflow_delivery(
rest_client: &relay::RestClient,
delivery_id: Option<Uuid>,
@@ -3340,6 +3474,10 @@ async fn tokio_main() -> Result<()> {
// turning successful work into a future redelivery.
let mut pending_workflow_finalizations: HashMap<Uuid, PendingWorkflowFinalization> =
HashMap::new();
// Signed owner-command terminal intents survive lost HTTP responses and are
// replayed verbatim until the relay returns the stored terminal result.
let mut pending_workflow_owner_completions: HashMap<Uuid, PendingWorkflowOwnerCompletion> =
HashMap::new();
// Native steers retain durable delivery ownership on the queued event until
// the agent acknowledges success. Results that race ahead of that ack are
// retained so success can finish against the exact fenced turn outcome.
@@ -3643,6 +3781,27 @@ async fn tokio_main() -> Result<()> {
// Remaining branches don't touch pool — evaluated when pool is idle.
_ = workflow_delivery_poll.tick() => {
let _ = result_rx;
reconcile_workflow_owner_completions(
&ctx.rest_client,
&mut pending_workflow_owner_completions,
).await;
if let Some(immutable_owner) = owner_cache.get() {
for command in list_pending_workflow_owner_commands(&ctx.rest_client).await {
if pending_workflow_owner_completions.contains_key(&command.command_id) {
continue;
}
let completion = prepare_workflow_owner_completion(
&command,
&ctx.rest_client,
immutable_owner,
).await;
pending_workflow_owner_completions.insert(command.command_id, completion);
}
}
reconcile_workflow_owner_completions(
&ctx.rest_client,
&mut pending_workflow_owner_completions,
).await;
if let Some(delivery) = claim_workflow_delivery(
&ctx.rest_client,
None,

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

@@ -3882,6 +3882,69 @@ impl Db {
.await
}
/// Insert or update a workflow within a caller-owned transaction.
#[datastore_span(name = "upsert_workflow_in_transaction", system = "postgresql")]
#[allow(clippy::too_many_arguments)]
pub async fn upsert_workflow_in_transaction(
&self,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
community_id: CommunityId,
id: Uuid,
channel_id: Option<Uuid>,
owner_pubkey: &[u8],
name: &str,
definition_json: &str,
definition_hash: &[u8],
definition_event_id: &[u8],
enabled: bool,
) -> Result<()> {
workflow::upsert_workflow_in_transaction(
tx,
community_id,
id,
channel_id,
owner_pubkey,
name,
definition_json,
definition_hash,
definition_event_id,
enabled,
)
.await
}
/// Persist a signed workflow definition and executable projection in one transaction.
#[allow(clippy::too_many_arguments)]
pub async fn persist_workflow_definition_in_transaction(
&self,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
community_id: CommunityId,
workflow_id: Uuid,
channel_id: Uuid,
owner_pubkey: &[u8],
event: &nostr::Event,
event_persistence: workflow::WorkflowDefinitionEventPersistence<'_>,
name: &str,
definition_json: &str,
definition_hash: &[u8],
enabled: bool,
) -> Result<()> {
workflow::persist_workflow_definition_in_transaction(
tx,
community_id,
workflow_id,
channel_id,
owner_pubkey,
event,
event_persistence,
name,
definition_json,
definition_hash,
enabled,
)
.await
}
/// Fetch a single workflow by ID, scoped to its community.
#[datastore_span(name = "get_workflow", system = "postgresql")]
pub async fn get_workflow(
@@ -4073,6 +4136,35 @@ impl Db {
workflow::find_by_owner_and_name(&self.pool, community_id, owner_pubkey, name).await
}
/// List pending owner update commands addressed to one managed agent.
#[datastore_span(name = "list_pending_workflow_owner_commands", system = "postgresql")]
pub async fn list_pending_workflow_owner_commands(
&self,
community_id: CommunityId,
agent_pubkey: &[u8],
limit: i64,
) -> Result<Vec<workflow::PendingWorkflowOwnerCommand>> {
workflow::list_pending_workflow_owner_commands(
&self.pool,
community_id,
agent_pubkey,
limit,
)
.await
}
/// Fetch one owner command for its authenticated target agent.
#[datastore_span(name = "get_workflow_owner_command", system = "postgresql")]
pub async fn get_workflow_owner_command(
&self,
community_id: CommunityId,
command_id: Uuid,
agent_pubkey: &[u8],
) -> Result<workflow::PendingWorkflowOwnerCommand> {
workflow::get_workflow_owner_command(&self.pool, community_id, command_id, agent_pubkey)
.await
}
/// Verify and record one owner-authored workflow command in the caller transaction.
#[allow(clippy::too_many_arguments)]
pub async fn admit_workflow_owner_command(
@@ -4087,6 +4179,7 @@ impl Db {
expected_revision: &[u8],
operation: &str,
proposed_yaml: Option<&str>,
receipt_event: Option<&nostr::Event>,
) -> Result<workflow::WorkflowOwnerCommandAdmission> {
workflow::admit_workflow_owner_command(
tx,
@@ -4099,6 +4192,43 @@ impl Db {
expected_revision,
operation,
proposed_yaml,
receipt_event,
)
.await
}
/// Atomically complete one pending owner-authored workflow update.
#[datastore_span(name = "complete_workflow_owner_command", system = "postgresql")]
#[allow(clippy::too_many_arguments)]
pub async fn complete_workflow_owner_command(
&self,
community_id: CommunityId,
command_id: Uuid,
agent_pubkey: &[u8],
replacement_event: Option<&nostr::Event>,
definition_name: Option<&str>,
definition_json: Option<&str>,
definition_hash: Option<&[u8]>,
definition_enabled: Option<bool>,
rejection_reason: Option<&str>,
receipt_event: &nostr::Event,
coordinate_conflict_receipt: &nostr::Event,
revision_conflict_receipt: &nostr::Event,
) -> Result<workflow::WorkflowOwnerCommandResult> {
workflow::complete_workflow_owner_command(
&self.pool,
community_id,
command_id,
agent_pubkey,
replacement_event,
definition_name,
definition_json,
definition_hash,
definition_enabled,
rejection_reason,
receipt_event,
coordinate_conflict_receipt,
revision_conflict_receipt,
)
.await
}

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

File diff suppressed because it is too large Load Diff

389
crates/buzz-relay/src/api/workflows.rs
View File

@@ -13,9 +13,11 @@ use axum::{
use chrono::{DateTime, Utc};
use serde::Deserialize;
use serde_json::Value;
use sha2::{Digest, Sha256};
use uuid::Uuid;
use buzz_core::TenantContext;
use buzz_core::kind::{KIND_WORKFLOW_DEF, KIND_WORKFLOW_OWNER_RECEIPT};
use buzz_core::{StoredEvent, TenantContext};
use crate::{
api::{api_error, bridge, internal_error},
@@ -280,6 +282,356 @@ async fn authorize_delivery_write(
Ok((tenant, pubkey))
}
/// List pending owner update commands for the authenticated target agent.
pub async fn pending_owner_commands(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
body: axum::body::Bytes,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let path = "/workflows/owner-commands/pending";
let (tenant, agent) = authorize_delivery_write(&state, &headers, path, &body).await?;
let commands = state
.db
.list_pending_workflow_owner_commands(tenant.community(), agent.as_bytes(), 100)
.await
.map_err(|error| internal_error(&format!("list workflow owner commands: {error}")))?;
Ok(Json(serde_json::json!({
"commands": commands.iter().map(|command| serde_json::json!({
"command_id": command.command_id,
"event_id": hex::encode(&command.event_id),
"owner_pubkey": hex::encode(&command.owner_pubkey),
"agent_pubkey": hex::encode(&command.agent_pubkey),
"workflow_id": command.workflow_id,
"expected_revision": hex::encode(&command.expected_revision),
"proposed_yaml": command.proposed_yaml,
"channel_id": command.channel_id,
})).collect::<Vec<_>>(),
})))
}
/// Agent-authenticated terminal intent for one owner command.
#[derive(Debug, Deserialize)]
pub struct CompleteOwnerCommandRequest {
replacement_event: Option<nostr::Event>,
rejection_reason: Option<String>,
}
pub(crate) struct OwnerCommandReceiptInput<'a> {
pub command_id: Uuid,
pub owner_pubkey: &'a [u8],
pub agent_pubkey: &'a nostr::PublicKey,
pub workflow_id: Uuid,
pub expected_revision: &'a [u8],
pub status: &'a str,
pub resulting_revision: Option<&'a str>,
pub reason: Option<&'a str>,
}
pub(crate) fn build_owner_command_receipt(
state: &AppState,
input: OwnerCommandReceiptInput<'_>,
) -> Result<nostr::Event, String> {
let OwnerCommandReceiptInput {
command_id,
owner_pubkey,
agent_pubkey,
workflow_id,
expected_revision,
status,
resulting_revision,
reason,
} = input;
nostr::EventBuilder::new(
nostr::Kind::Custom(KIND_WORKFLOW_OWNER_RECEIPT as u16),
serde_json::json!({
"status": status,
"resulting_revision": resulting_revision,
"reason": reason,
})
.to_string(),
)
.tags([
nostr::Tag::parse(["d", &command_id.to_string()]).map_err(|e| e.to_string())?,
nostr::Tag::parse(["p", &hex::encode(owner_pubkey)]).map_err(|e| e.to_string())?,
nostr::Tag::parse(["p", &agent_pubkey.to_hex()]).map_err(|e| e.to_string())?,
nostr::Tag::parse([
"a",
&format!(
"{KIND_WORKFLOW_DEF}:{}:{workflow_id}",
agent_pubkey.to_hex()
),
])
.map_err(|e| e.to_string())?,
nostr::Tag::parse(["revision", &hex::encode(expected_revision)])
.map_err(|e| e.to_string())?,
nostr::Tag::parse(["status", status]).map_err(|e| e.to_string())?,
])
.custom_created_at(nostr::Timestamp::from(0))
.sign_with_keys(&state.relay_keypair)
.map_err(|e| e.to_string())
}
fn prepare_managed_definition(
name: String,
enabled: bool,
definition_json: &str,
existing: &serde_json::Value,
is_webhook: bool,
) -> Result<(String, String, Vec<u8>, bool), String> {
let mut definition: serde_json::Value = serde_json::from_str(definition_json)
.map_err(|error| format!("replacement JSON: {error}"))?;
crate::webhook_secret::prepare_definition(&mut definition, Some(existing), is_webhook, false)
.map_err(str::to_string)?;
let json =
serde_json::to_string(&definition).map_err(|error| format!("replacement JSON: {error}"))?;
let hash = Sha256::digest(json.as_bytes()).to_vec();
Ok((name, json, hash, enabled))
}
/// Complete one pending owner update after agent-side verification/signing.
pub async fn complete_owner_command(
State(state): State<Arc<AppState>>,
Path(command_id): Path<Uuid>,
headers: HeaderMap,
body: axum::body::Bytes,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let path = format!("/workflows/owner-commands/{command_id}/complete");
let (tenant, agent) = authorize_delivery_write(&state, &headers, &path, &body).await?;
let request = match serde_json::from_slice::<CompleteOwnerCommandRequest>(&body) {
Ok(request)
if request.replacement_event.is_some() != request.rejection_reason.is_some() =>
{
request
}
Ok(_) => CompleteOwnerCommandRequest {
replacement_event: None,
rejection_reason: Some("invalid_completion_shape".into()),
},
Err(_) => CompleteOwnerCommandRequest {
replacement_event: None,
rejection_reason: Some("invalid_completion_json".into()),
},
};
let pending = state
.db
.get_workflow_owner_command(tenant.community(), command_id, agent.as_bytes())
.await
.map_err(|error| match error {
buzz_db::DbError::NotFound(_) => api_error(
StatusCode::NOT_FOUND,
"owner command was not found for this agent",
),
other => internal_error(&format!("read owner command: {other}")),
})?;
let workflow = state
.db
.get_workflow(tenant.community(), pending.workflow_id)
.await
.map_err(|error| internal_error(&format!("read owner command workflow: {error}")))?;
let channel_id = workflow
.channel_id
.ok_or_else(|| api_error(StatusCode::CONFLICT, "workflow is not channel-scoped"))?;
let mut replacement_event = request.replacement_event;
let mut rejection_reason = request.rejection_reason;
let mut definition = None;
if let Some(event) = replacement_event.as_ref() {
let bindings_match = event.verify().is_ok()
&& event.kind.as_u16() as u32 == KIND_WORKFLOW_DEF
&& event.pubkey == agent
&& event.content == pending.proposed_yaml
&& event
.tags
.iter()
.filter(|tag| tag.as_slice().first().map(String::as_str) == Some("d"))
.filter_map(|tag| tag.content())
.collect::<Vec<_>>()
== [pending.workflow_id.to_string()]
&& event
.tags
.iter()
.filter(|tag| tag.as_slice().first().map(String::as_str) == Some("h"))
.filter_map(|tag| tag.content())
.collect::<Vec<_>>()
== [channel_id.to_string()]
&& event
.tags
.iter()
.filter(|tag| {
tag.as_slice().first().map(String::as_str) == Some("expected-revision")
})
.filter_map(|tag| tag.content())
.collect::<Vec<_>>()
== [hex::encode(&pending.expected_revision)]
&& event
.tags
.iter()
.filter(|tag| tag.as_slice().first().map(String::as_str) == Some("command"))
.filter_map(|tag| tag.content())
.collect::<Vec<_>>()
== [command_id.to_string()];
if !bindings_match {
replacement_event = None;
rejection_reason = Some("replacement_binding_mismatch".into());
} else {
match buzz_workflow::WorkflowEngine::parse_yaml(&event.content) {
Ok((def, json)) => {
let authority_ok = if def.requires_elevated_authority() {
let role = state
.db
.get_member_role(tenant.community(), channel_id, agent.as_bytes())
.await
.map_err(|error| {
internal_error(&format!("replacement authority lookup: {error}"))
})?;
matches!(role.as_deref(), Some("owner") | Some("admin"))
} else {
true
};
if authority_ok {
match prepare_managed_definition(
def.name.clone(),
def.enabled,
&json,
&workflow.definition,
matches!(def.trigger, buzz_workflow::TriggerDef::Webhook),
) {
Ok(prepared) => definition = Some(prepared),
Err(reason) => {
replacement_event = None;
rejection_reason = Some(reason);
}
}
} else {
replacement_event = None;
rejection_reason = Some("insufficient_workflow_authority".into());
}
}
Err(_) => {
replacement_event = None;
rejection_reason = Some("invalid_workflow_yaml".into());
}
}
}
}
let status = if replacement_event.is_some() {
"applied"
} else {
"rejected"
};
let resulting_revision = replacement_event.as_ref().map(|event| event.id.to_hex());
let receipt = build_owner_command_receipt(
&state,
OwnerCommandReceiptInput {
command_id,
owner_pubkey: &pending.owner_pubkey,
agent_pubkey: &agent,
workflow_id: pending.workflow_id,
expected_revision: &pending.expected_revision,
status,
resulting_revision: resulting_revision.as_deref(),
reason: rejection_reason.as_deref(),
},
)
.map_err(|error| internal_error(&format!("sign owner command receipt: {error}")))?;
let conflict_receipt = |reason| {
build_owner_command_receipt(
&state,
OwnerCommandReceiptInput {
command_id,
owner_pubkey: &pending.owner_pubkey,
agent_pubkey: &agent,
workflow_id: pending.workflow_id,
expected_revision: &pending.expected_revision,
status: "rejected",
resulting_revision: None,
reason: Some(reason),
},
)
.map_err(|error| internal_error(&format!("sign owner command conflict receipt: {error}")))
};
let coordinate_conflict_receipt = conflict_receipt("workflow_coordinate_changed")?;
let revision_conflict_receipt = conflict_receipt("workflow_revision_changed")?;
let result = state
.db
.complete_workflow_owner_command(
tenant.community(),
command_id,
agent.as_bytes(),
replacement_event.as_ref(),
definition.as_ref().map(|value| value.0.as_str()),
definition.as_ref().map(|value| value.1.as_str()),
definition.as_ref().map(|value| value.2.as_slice()),
definition.as_ref().map(|value| value.3),
rejection_reason.as_deref(),
&receipt,
&coordinate_conflict_receipt,
&revision_conflict_receipt,
)
.await
.map_err(|error| match error {
buzz_db::DbError::AccessDenied(_) => api_error(
StatusCode::FORBIDDEN,
"owner command targets a different agent",
),
buzz_db::DbError::InvalidData(_) => {
api_error(StatusCode::CONFLICT, "owner command completion conflict")
}
other => internal_error(&format!("complete owner command: {other}")),
})?;
if result.transitioned {
if let Some(event) = replacement_event.as_ref() {
state
.workflow_engine
.invalidate_channel_workflows(tenant.community(), channel_id);
publish_internal_event(&state, &tenant, event, Some(channel_id)).await;
}
let published_receipt = if result.receipt_event_id == receipt.id.as_bytes() {
&receipt
} else if result.receipt_event_id == coordinate_conflict_receipt.id.as_bytes() {
&coordinate_conflict_receipt
} else {
&revision_conflict_receipt
};
publish_internal_event(&state, &tenant, published_receipt, Some(channel_id)).await;
}
Ok(Json(serde_json::json!({
"status": result.status,
"resulting_revision": result.resulting_revision.as_deref().map(hex::encode),
"reason": result.terminal_reason,
"receipt_event_id": hex::encode(result.receipt_event_id),
"transitioned": result.transitioned,
})))
}
pub(crate) async fn publish_internal_event(
state: &Arc<AppState>,
tenant: &TenantContext,
event: &nostr::Event,
channel_id: Option<Uuid>,
) {
let topic = channel_id
.map(buzz_pubsub::EventTopic::Channel)
.unwrap_or(buzz_pubsub::EventTopic::Global);
state.mark_local_event(tenant.community(), &event.id);
if state
.pubsub
.publish_event(tenant, topic, event)
.await
.is_err()
{
state
.local_event_ids
.invalidate(&(tenant.community(), event.id.to_bytes()));
}
let stored = StoredEvent::new(event.clone(), channel_id);
crate::handlers::event::fan_out_event_to_local_subscribers(state, tenant.community(), &stored)
.await;
}
/// Claim the oldest due workflow delivery for the authenticated managed agent.
pub async fn claim_agent_delivery(
State(state): State<Arc<AppState>>,
@@ -494,6 +846,41 @@ mod tests {
);
}
#[test]
fn managed_webhook_update_preserves_existing_secret() {
let existing = serde_json::json!({
"name": "existing",
"_webhook_secret": "keep-me",
});
let (_, json, hash, _) = prepare_managed_definition(
"replacement".into(),
true,
r#"{"name":"replacement"}"#,
&existing,
true,
)
.unwrap();
let prepared: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(prepared["_webhook_secret"], "keep-me");
assert_eq!(hash, Sha256::digest(json.as_bytes()).to_vec());
}
#[test]
fn managed_transition_to_webhook_rejects_undisclosable_secret() {
let existing = serde_json::json!({"name": "existing"});
assert_eq!(
prepare_managed_definition(
"replacement".into(),
true,
r#"{"name":"replacement"}"#,
&existing,
true,
),
Err("webhook_secret_unavailable".into())
);
assert!(existing.get("_webhook_secret").is_none());
}
#[test]
fn approval_wire_does_not_expose_hash_as_token() {
let approval = buzz_db::workflow::ApprovalRecord {

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

@@ -795,26 +795,17 @@ async fn handle_workflow_def(
}
};
// Preserve the existing webhook secret across updates. A new secret is
// returned only when the workflow first gains a webhook trigger.
let webhook_secret = if matches!(def.trigger, buzz_workflow::TriggerDef::Webhook) {
let existing_secret = existing_workflow
// Apply the shared secret projection before hashing. This synchronous path
// can securely return a newly generated webhook secret to its caller.
let webhook_secret = webhook_secret::prepare_definition(
&mut definition_json,
existing_workflow
.as_ref()
.and_then(|workflow| webhook_secret::extract_secret(&workflow.definition));
let secret = existing_secret.unwrap_or_else(webhook_secret::generate_webhook_secret);
webhook_secret::inject_secret(&mut definition_json, &secret);
if existing_workflow
.as_ref()
.and_then(|workflow| webhook_secret::extract_secret(&workflow.definition))
.is_none()
{
Some(secret)
} else {
None
}
} else {
None
};
.map(|workflow| &workflow.definition),
matches!(def.trigger, buzz_workflow::TriggerDef::Webhook),
true,
)
.map_err(|error| IngestError::Internal(format!("error: prepare webhook secret: {error}")))?;
// Compute hash AFTER secret injection
let definition_json_final = serde_json::to_string(&definition_json)
@@ -822,7 +813,7 @@ async fn handle_workflow_def(
let hash = compute_definition_hash(&definition_json_final);
// Persist the command event — returns open transaction
let tx = match persist_command_event(state, tenant, event, None).await? {
let mut tx = match persist_command_event(state, tenant, event, None).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),
@@ -833,7 +824,8 @@ async fn handle_workflow_def(
PersistResult::Inserted(tx) => tx,
};
// 4. Execute: upsert by the NIP-33 d-tag UUID. A retry updates the same
// 4. Execute in the same locked transaction that persists the NIP-33
// definition. A retry updates the same
// row instead of creating another enabled workflow that would fan out on
// every matching event. The workflow's community is the request's
// server-bound tenant — never re-derived from the (client-supplied) channel
@@ -853,15 +845,17 @@ async fn handle_workflow_def(
state
.db
.upsert_workflow(
.persist_workflow_definition_in_transaction(
&mut tx,
community_id,
workflow_id,
Some(channel_id),
channel_id,
&self_bytes,
event,
buzz_db::workflow::WorkflowDefinitionEventPersistence::AlreadyStored,
&workflow_name,
&definition_json_final,
&hash,
event.id.as_bytes(),
def.enabled,
)
.await
@@ -872,17 +866,16 @@ async fn handle_workflow_def(
other => IngestError::Internal(format!("error: db upsert_workflow: {other}")),
})?;
// Drop the trigger-path cache entry so the new/updated definition fires on
// the next matching event instead of after the cache TTL.
state
.workflow_engine
.invalidate_channel_workflows(community_id, channel_id);
// Commit the event transaction after the idempotent workflow upsert succeeds.
// Commit the definition event and executable workflow projection together.
tx.commit()
.await
.map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
// Drop the trigger-path cache only after both representations commit.
state
.workflow_engine
.invalidate_channel_workflows(community_id, channel_id);
// 5. Return response
let mut resp = serde_json::json!({
"workflow_id": workflow_id.to_string(),
@@ -928,27 +921,30 @@ async fn handle_workflow_owner_command(
let channel_id = workflow
.channel_id
.ok_or_else(|| IngestError::Rejected("invalid: workflow is not channel-scoped".into()))?;
if workflow.owner_pubkey != command.agent_pubkey.to_bytes()
|| workflow.definition_event_id.as_deref() != Some(command.expected_revision.as_bytes())
{
return Err(IngestError::Rejected(
"conflict: workflow coordinate or revision changed".into(),
));
}
if !state
.db
.is_agent_owner(
tenant.community(),
command.agent_pubkey.as_bytes(),
event.pubkey.as_bytes(),
)
.await
.map_err(|error| IngestError::Internal(format!("error: owner lookup: {error}")))?
{
return Err(IngestError::Rejected(
"forbidden: command author is not the immutable agent owner".into(),
));
}
// The workflow lookup supplies only the channel binding needed to persist and
// publish the command event. Immutable owner, coordinate author, and revision
// authority are rechecked under the workflow row lock during admission.
let control_receipt = (operation != "update")
.then(|| {
let resulting_revision = command.expected_revision.to_hex();
crate::api::workflows::build_owner_command_receipt(
state,
crate::api::workflows::OwnerCommandReceiptInput {
command_id: command.command_id,
owner_pubkey: event.pubkey.as_bytes(),
agent_pubkey: &command.agent_pubkey,
workflow_id: command.workflow_id,
expected_revision: command.expected_revision.as_bytes(),
status: "applied",
resulting_revision: Some(&resulting_revision),
reason: None,
},
)
})
.transpose()
.map_err(|error| {
IngestError::Internal(format!("error: sign owner command receipt: {error}"))
})?;
let mut tx = match persist_command_event(state, tenant, event, Some(channel_id)).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
@@ -972,6 +968,7 @@ async fn handle_workflow_owner_command(
command.expected_revision.as_bytes(),
operation,
command.yaml_definition.as_deref(),
control_receipt.as_ref(),
)
.await
.map_err(|error| match error {
@@ -989,6 +986,10 @@ async fn handle_workflow_owner_command(
state
.workflow_engine
.invalidate_channel_workflows(tenant.community(), channel_id);
if let Some(receipt) = control_receipt.as_ref() {
crate::api::workflows::publish_internal_event(state, tenant, receipt, Some(channel_id))
.await;
}
let status = match admission {
WorkflowOwnerCommandAdmission::PendingAgent => "pending_agent",
WorkflowOwnerCommandAdmission::Applied => "applied",

4
crates/buzz-relay/src/main.rs
View File

@@ -657,8 +657,8 @@ async fn main() -> anyhow::Result<()> {
let wf_cron = Arc::clone(&workflow_engine);
tokio::spawn(async move { wf_cron.run().await });
// Durable managed-agent delivery reaper. Expired leases become retryable by
// the claim query; exhausted or TTL-expired rows become terminal here and
// Durable managed-agent delivery reaper. Expired leases become terminal failed
// rows and are never reassigned; TTL-expired rows are terminalized here and
// emit one owner-visible channel signal from the rows returned by the
// guarded UPDATE.
{

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/owner-commands/pending",
post(api::workflows::pending_owner_commands),
)
.route(
"/workflows/owner-commands/{command_id}/complete",
post(api::workflows::complete_owner_command),
)
.route(
"/workflows/agent-deliveries/claim",
post(api::workflows::claim_agent_delivery),

59
crates/buzz-relay/src/webhook_secret.rs
View File

@@ -68,6 +68,32 @@ pub fn strip_secret(def: &serde_json::Value) -> serde_json::Value {
}
}
/// Apply webhook-secret projection semantics before hashing and persistence.
///
/// Existing webhook workflows retain their secret. Callers that can securely
/// return a newly minted secret may set `allow_generate`; background owner
/// command completion must not mint an undiscoverable credential.
pub fn prepare_definition(
definition: &mut serde_json::Value,
existing: Option<&serde_json::Value>,
is_webhook: bool,
allow_generate: bool,
) -> Result<Option<String>, &'static str> {
if !is_webhook {
return Ok(None);
}
if let Some(secret) = existing.and_then(extract_secret) {
inject_secret(definition, &secret);
return Ok(None);
}
if !allow_generate {
return Err("webhook_secret_unavailable");
}
let secret = generate_webhook_secret();
inject_secret(definition, &secret);
Ok(Some(secret))
}
/// Compare `provided` against `stored` in constant time.
///
/// Returns `true` only when the two strings are identical. The XOR-fold
@@ -139,6 +165,39 @@ mod tests {
assert_eq!(stripped.get("b").and_then(|v| v.as_i64()), Some(2));
}
#[test]
fn prepare_definition_preserves_existing_secret() {
let existing = serde_json::json!({"_webhook_secret": "keep-me"});
let mut replacement = serde_json::json!({"name": "replacement"});
assert_eq!(
prepare_definition(&mut replacement, Some(&existing), true, false),
Ok(None)
);
assert_eq!(extract_secret(&replacement).as_deref(), Some("keep-me"));
}
#[test]
fn prepare_definition_rejects_undeliverable_new_secret() {
let mut replacement = serde_json::json!({"name": "replacement"});
assert_eq!(
prepare_definition(&mut replacement, None, true, false),
Err("webhook_secret_unavailable")
);
assert!(extract_secret(&replacement).is_none());
}
#[test]
fn prepare_definition_generates_when_caller_can_return_secret() {
let mut replacement = serde_json::json!({"name": "replacement"});
let generated = prepare_definition(&mut replacement, None, true, true)
.unwrap()
.expect("new secret");
assert_eq!(
extract_secret(&replacement).as_deref(),
Some(generated.as_str())
);
}
#[test]
fn verify_secret_matches() {
assert!(verify_secret("hello-world", "hello-world"));

1
migrations/0035_workflow_owner_commands.sql
View File

@@ -13,6 +13,7 @@ CREATE TABLE workflow_owner_commands (
proposed_yaml TEXT,
status workflow_owner_command_status NOT NULL,
resulting_revision BYTEA,
receipt_event_id BYTEA,
terminal_reason TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

1
schema/schema.sql
View File

@@ -1776,6 +1776,7 @@ CREATE TABLE workflow_owner_commands (
proposed_yaml TEXT,
status workflow_owner_command_status NOT NULL,
resulting_revision BYTEA,
receipt_event_id BYTEA,
terminal_reason TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),