fix(acp): route edit-triggered replies through originals

Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz>
This commit is contained in:
Brother Darryl
2026-08-10 12:39:15 -04:00
committed by Larry
parent 7ad7d79a84
commit 914418de52
3 changed files with 258 additions and 23 deletions

View File

@@ -24,6 +24,7 @@ use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use nostr::EventId;
use tokio::sync::mpsc;
use tokio::task::{JoinHandle, JoinSet};
use tokio::time::timeout;
@@ -2002,8 +2003,28 @@ pub async fn run_prompt_task(
// Try startup cache first; lazy-fetch via REST for dynamic channels.
let channel_info = ctx.channel_info.resolve(b.channel_id).await;
let resolved_edit = match b.events.last() {
Some(event) => resolve_edit_routing(&event.event, &ctx.rest_client).await,
None => None,
};
let conversation_context = if ctx.context_message_limit > 0 {
fetch_conversation_context(b, &channel_info, &ctx).await
match resolved_edit
.as_ref()
.and_then(|edit| edit.target_thread_tags.root_event_id.as_deref())
{
Some(root_id) => {
fetch_thread_context(
b.channel_id,
root_id,
ctx.context_message_limit,
ctx.agent_keys.public_key(),
&ctx.rest_client,
)
.await
}
None => fetch_conversation_context(b, &channel_info, &ctx).await,
}
} else {
None
};
@@ -2060,6 +2081,7 @@ pub async fn run_prompt_task(
conversation_context: conversation_context.as_ref(),
conversation_context_had_delivered_events,
profile_lookup: profile_lookup.as_ref(),
resolved_edit: resolved_edit.as_ref(),
has_system_prompt_support: agent.has_system_prompt_support(),
base_prompt: standing.base_prompt,
system_prompt: standing.system_prompt,
@@ -2893,6 +2915,47 @@ fn conversation_context_delta(
}
}
/// Resolve a kind:40003 edit through its original event for reply routing.
/// Failure deliberately falls back to the edit target id in `format_prompt`.
async fn resolve_edit_routing(
event: &nostr::Event,
rest: &RestClient,
) -> Option<crate::queue::ResolvedEdit> {
let target_event_id = crate::queue::edit_target_id(event)?;
let target_id = EventId::from_hex(&target_event_id).ok()?;
let filter = nostr::Filter::new().id(target_id).limit(1);
let response = match timeout(Duration::from_millis(2_000), rest.query(&[filter])).await {
Ok(Ok(response)) => response,
Ok(Err(error)) => {
tracing::warn!(edit_event_id = %event.id, target_event_id, "edit routing: original event fetch failed: {error}");
return None;
}
Err(_) => {
tracing::warn!(edit_event_id = %event.id, target_event_id, "edit routing: original event fetch timed out");
return None;
}
};
let Some(raw) = response.as_array().and_then(|events| events.first()) else {
tracing::warn!(edit_event_id = %event.id, target_event_id, "edit routing: original event was not returned");
return None;
};
let original = match serde_json::from_value::<nostr::Event>(raw.clone()) {
Ok(original) if original.id == target_id && original.verify().is_ok() => original,
Ok(_) => {
tracing::warn!(edit_event_id = %event.id, target_event_id, "edit routing: original event failed id/signature verification");
return None;
}
Err(error) => {
tracing::warn!(edit_event_id = %event.id, target_event_id, "edit routing: malformed original event: {error}");
return None;
}
};
Some(crate::queue::ResolvedEdit {
target_event_id,
target_thread_tags: crate::queue::parse_thread_tags(&original),
})
}
/// Fetch conversation context (thread or DM) for a batch before prompting.
///
/// Returns `None` if:

View File

@@ -884,6 +884,23 @@ pub struct ThreadTags {
/// NOTE: Only handles NIP-10 marker-based format (preferred). The deprecated
/// positional format (no markers, `["e", id, relay_url]`) is not supported —
/// Buzz always generates marker-based tags (see relay messages.rs:762-783).
/// Return the original message id targeted by a kind:40003 edit event.
///
/// Edit events deliberately use a bare two-element `e` tag; this is not the
/// deprecated positional NIP-10 thread format and must not be accepted by
/// [`parse_thread_tags`] for other event kinds.
pub fn edit_target_id(event: &Event) -> Option<String> {
(event.kind.as_u16() == 40003)
.then(|| {
event.tags.iter().find_map(|tag| {
let parts = tag.as_slice();
(parts.len() == 2 && parts.first().map(String::as_str) == Some("e"))
.then(|| parts[1].clone())
})
})
.flatten()
}
pub fn parse_thread_tags(event: &Event) -> ThreadTags {
let mut root = None;
let mut reply = None;
@@ -1444,6 +1461,13 @@ fn format_conversation_context(
s
}
/// Original-message routing recovered for a kind:40003 edit event.
#[derive(Debug, Clone)]
pub struct ResolvedEdit {
pub target_event_id: String,
pub target_thread_tags: ThreadTags,
}
/// Arguments for [`format_prompt`] beyond the required [`FlushBatch`].
#[derive(Default)]
pub struct FormatPromptArgs<'a> {
@@ -1454,6 +1478,10 @@ pub struct FormatPromptArgs<'a> {
/// live session had already received. Trigger-only context does not set it.
pub conversation_context_had_delivered_events: bool,
pub profile_lookup: Option<&'a PromptProfileLookup>,
/// Routing derived by resolving a kind:40003 edit target. When present,
/// prompt scope and reply instructions use the original message rather
/// than the invisible edit auxiliary event.
pub resolved_edit: Option<&'a ResolvedEdit>,
/// When true, base_prompt and system_prompt are delivered via the system
/// role (session/new) and omitted from the user message. When false
/// (legacy agents), they are injected as `[Base]` and `[System]` sections.
@@ -1567,7 +1595,17 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
return Vec::new();
}
};
let thread_tags = parse_thread_tags(&last_event.event);
let edit_target = edit_target_id(&last_event.event);
let resolved_edit = args.resolved_edit.filter(|_| edit_target.is_some());
// An edit's own bare `e` tag is not a thread tag. Route via the original
// message when it was fetched, or its target id as the safe fallback.
let thread_tags = resolved_edit
.map(|edit| edit.target_thread_tags.clone())
.unwrap_or_else(|| parse_thread_tags(&last_event.event));
let routing_event_id = resolved_edit
.map(|edit| edit.target_event_id.clone())
.or(edit_target.clone())
.unwrap_or_else(|| last_event.event.id.to_hex());
let is_dm = args
.channel_info
.map(|ci| ci.channel_type == "dm")
@@ -1605,12 +1643,12 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
thread_tags
.root_event_id
.is_some()
.then(|| last_event.event.id.to_hex())
.then(|| routing_event_id.to_string())
} else {
resolve_reply_anchor(
&sender_pubkey,
&thread_tags,
&last_event.event.id.to_hex(),
&routing_event_id,
args.profile_lookup,
)
};
@@ -1623,6 +1661,16 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
args.conversation_context_had_delivered_events,
reply_anchor.as_deref(),
));
if let Some(edit) = resolved_edit {
sections.push(format!(
"[Edit routing]\nTrigger is a kind:40003 edit event; original message ID: {}",
edit.target_event_id
));
} else if let Some(target_id) = edit_target {
sections.push(format!(
"[Edit routing]\nTrigger is a kind:40003 edit event; original message ID: {target_id} (original event fetch failed; using target as reply anchor)"
));
}
// 3. Conversation context (thread or DM).
if let Some(ctx) = args.conversation_context {
@@ -5104,6 +5152,78 @@ mod tests {
);
}
fn edit_event(target: &str) -> Event {
let keys = Keys::generate();
EventBuilder::new(Kind::Custom(40003), "edited mention")
.tags([nostr::Tag::parse(["e", target]).unwrap()])
.sign_with_keys(&keys)
.unwrap()
}
fn one_event_batch(event: Event) -> FlushBatch {
FlushBatch {
channel_id: Uuid::new_v4(),
events: vec![BatchEvent {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
}],
cancelled_events: vec![],
cancel_reason: None,
}
}
#[test]
fn edit_of_top_level_anchors_original_message_in_rendered_prompt() {
let original_id = "11".repeat(32);
let edit = edit_event(&original_id);
let prompt = format_prompt(
&one_event_batch(edit),
&FormatPromptArgs {
resolved_edit: Some(&ResolvedEdit {
target_event_id: original_id.clone(),
target_thread_tags: ThreadTags::default(),
}),
..Default::default()
},
)
.join("\n");
assert!(prompt.contains(&format!("--reply-to {original_id}")));
assert!(prompt.contains(&format!("original message ID: {original_id}")));
}
#[test]
fn edit_of_threaded_reply_anchors_original_thread_root_in_rendered_prompt() {
let original_id = "22".repeat(32);
let root_id = "33".repeat(32);
let prompt = format_prompt(
&one_event_batch(edit_event(&original_id)),
&FormatPromptArgs {
resolved_edit: Some(&ResolvedEdit {
target_event_id: original_id,
target_thread_tags: ThreadTags {
root_event_id: Some(root_id.clone()),
parent_event_id: Some("44".repeat(32)),
mentioned_pubkeys: vec![],
},
}),
..Default::default()
},
)
.join("\n");
assert!(prompt.contains(&format!("--reply-to {root_id}")));
}
#[test]
fn edit_fetch_failure_anchors_target_never_auxiliary_edit_event() {
let original_id = "55".repeat(32);
let edit = edit_event(&original_id);
let edit_id = edit.id.to_hex();
let prompt = format_prompt(&one_event_batch(edit), &FormatPromptArgs::default()).join("\n");
assert!(prompt.contains(&format!("--reply-to {original_id}")));
assert!(!prompt.contains(&format!("--reply-to {edit_id}")));
}
// ── F2 case 2.5: steer renewal is monotonic across repeated steers ───────
/// Calling `extend_in_flight_deadline` twice with the same `max_turn_secs`

View File

@@ -464,6 +464,7 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) ->
if let Err(e) = publish_setup_nudge(
&publisher,
&config.keys,
&rest_client,
buzz_event.channel_id,
&buzz_event.event,
&payload,
@@ -595,6 +596,48 @@ async fn handle_setup_membership(
}
}
/// Fetch and verify the original event targeted by a kind:40003 edit.
async fn fetch_edit_original(
target_event_id: &str,
rest_client: &crate::relay::RestClient,
) -> Option<nostr::Event> {
use std::time::Duration;
let target_id = nostr::EventId::from_hex(target_event_id).ok()?;
let filter = nostr::Filter::new().id(target_id).limit(1);
let response = match tokio::time::timeout(
Duration::from_millis(2_000),
rest_client.query(&[filter]),
)
.await
{
Ok(Ok(response)) => response,
Ok(Err(error)) => {
tracing::warn!(
target_event_id,
"setup-mode: edit original fetch failed: {error}"
);
return None;
}
Err(_) => return None,
};
let raw = response.as_array()?.first()?;
match serde_json::from_value::<nostr::Event>(raw.clone()) {
Ok(event) if event.id == target_id && event.verify().is_ok() => Some(event),
_ => None,
}
}
/// Construct the flat thread reference used by a setup nudge.
fn setup_nudge_thread_ref(target_event_id: &str) -> Result<buzz_sdk::ThreadRef> {
let target_id = nostr::EventId::from_hex(target_event_id)
.map_err(|e| anyhow::anyhow!("invalid nudge anchor event id: {e}"))?;
Ok(buzz_sdk::ThreadRef {
root_event_id: target_id,
parent_event_id: target_id,
})
}
/// Build and publish a setup nudge reply to the triggering event.
///
/// Threading: flat reply to the thread root if one exists; otherwise reply
@@ -602,30 +645,31 @@ async fn handle_setup_membership(
async fn publish_setup_nudge(
publisher: &RelayEventPublisher,
keys: &nostr::Keys,
rest_client: &crate::relay::RestClient,
channel_id: Uuid,
triggering_event: &nostr::Event,
payload: &SetupPayload,
) -> Result<()> {
use buzz_sdk::ThreadRef;
// Parse NIP-10 thread tags to determine reply target.
let thread_tags = crate::queue::parse_thread_tags(triggering_event);
let thread_ref = if let Some(root_str) = &thread_tags.root_event_id {
// Threaded event: reply flat to the root.
let root_id = nostr::EventId::from_hex(root_str)
.map_err(|e| anyhow::anyhow!("invalid root event id: {e}"))?;
Some(ThreadRef {
root_event_id: root_id,
parent_event_id: root_id,
})
} else {
// Top-level event: reply to the triggering event.
Some(ThreadRef {
root_event_id: triggering_event.id,
parent_event_id: triggering_event.id,
})
let edit_target = crate::queue::edit_target_id(triggering_event);
let target_event_id = match edit_target.as_deref() {
Some(target) => match fetch_edit_original(target, rest_client).await {
Some(original) => {
let thread_tags = crate::queue::parse_thread_tags(&original);
thread_tags
.root_event_id
.unwrap_or_else(|| target.to_string())
}
None => {
tracing::warn!(edit_event_id = %triggering_event.id, target_event_id = target,
"setup-mode: original edit target unavailable; using target as nudge anchor");
target.to_string()
}
},
None => crate::queue::parse_thread_tags(triggering_event)
.root_event_id
.unwrap_or_else(|| triggering_event.id.to_hex()),
};
let thread_ref = Some(setup_nudge_thread_ref(&target_event_id)?);
let body = payload.nudge_body();
let author_hex = triggering_event.pubkey.to_hex();
@@ -1012,6 +1056,14 @@ mod tests {
// nudge. They use the extracted `should_nudge_for_event` helper, which is
// the exact code the live loop calls.
#[test]
fn setup_nudge_edit_fallback_anchors_edit_target_not_edit_event() {
let target = "66".repeat(32);
let thread_ref = setup_nudge_thread_ref(&target).unwrap();
assert_eq!(thread_ref.root_event_id.to_hex(), target);
assert_eq!(thread_ref.parent_event_id.to_hex(), target);
}
fn fake_event_id(byte: u8) -> EventId {
EventId::from_byte_array([byte; 32])
}