diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 8460372abab..92714207274 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -211,6 +211,13 @@ pub struct AcpClient { /// deltas. Both goose and buzz-agent emit this notification; goose gates /// on client capability advertisement, buzz-agent emits unconditionally. goose_usage: UsageTracker, + /// Assistant text emitted during the current `session/prompt`. + /// + /// ACP streams assistant output as `agent_message_chunk` notifications, + /// while the terminal response only contains a stop reason. A bounded copy + /// lets the harness consume structured side output without asking the + /// agent to mutate the workspace or perform the network write itself. + agent_message_text: String, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -550,6 +557,7 @@ impl AcpClient { steering_supported: false, steer_rx: None, goose_usage: UsageTracker::default(), + agent_message_text: String::new(), }) } @@ -768,6 +776,7 @@ impl AcpClient { idle_timeout: std::time::Duration, max_duration: std::time::Duration, ) -> Result { + self.agent_message_text.clear(); let params = build_prompt_params(session_id, prompt_blocks); let hard_deadline = tokio::time::Instant::now() + max_duration; self.current_hard_deadline = Some(hard_deadline); @@ -824,6 +833,11 @@ impl AcpClient { self.parse_stop_reason(&result?) } + /// Take the assistant text streamed during the most recent prompt. + pub(crate) fn take_agent_message_text(&mut self) -> String { + std::mem::take(&mut self.agent_message_text) + } + /// Send a `session/cancel` **notification** (no `id` field, no response expected). /// /// After calling this, the agent will eventually respond to the in-flight @@ -1745,6 +1759,23 @@ impl AcpClient { match update_type { "agent_message_chunk" => { if let Some(text) = update["content"]["text"].as_str() { + // Match the ACP wire's 10 MiB line ceiling. Structured + // post-turn output is normally a few KiB, but this keeps a + // noisy adapter from growing the capture without bound. + const MAX_CAPTURED_AGENT_MESSAGE: usize = 10 * 1024 * 1024; + let remaining = + MAX_CAPTURED_AGENT_MESSAGE.saturating_sub(self.agent_message_text.len()); + if text.len() <= remaining { + self.agent_message_text.push_str(text); + } else if remaining > 0 { + let mut end = remaining.min(text.len()); + while end > 0 && !text.is_char_boundary(end) { + end -= 1; + } + if end > 0 { + self.agent_message_text.push_str(&text[..end]); + } + } tracing::info!(target: "acp::stream", "{text}"); } false diff --git a/crates/buzz-acp/src/dkg_memory.rs b/crates/buzz-acp/src/dkg_memory.rs new file mode 100644 index 00000000000..f1a38f5a5d3 --- /dev/null +++ b/crates/buzz-acp/src/dkg_memory.rs @@ -0,0 +1,529 @@ +//! Runtime-owned post-turn DKG memory finalization. +//! +//! The normal ACP turn remains responsible for the human-facing Buzz reply. +//! Once that succeeds, this module asks the same model for a structured, +//! evidence-bound semantic side output. The harness—not the model—signs and +//! submits the proposal. Signed proposals are persisted before the HTTP call so +//! a crash or transient network failure can be retried safely. + +use std::collections::HashSet; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use nostr::{Event, EventBuilder, Kind, Tag, Timestamp}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +use crate::acp::AcpError; +use crate::pool::{OwnedAgent, PromptContext}; +use crate::relay::{RelayError, RestClient}; + +const KIND_DKG_MEMORY_PROPOSAL: u16 = 40009; +const RESPONSE_QUERY_TIMEOUT: Duration = Duration::from_secs(3); +const MEMORY_IDLE_TIMEOUT: Duration = Duration::from_secs(45); +const MEMORY_HARD_TIMEOUT: Duration = Duration::from_secs(120); +const MEMORY_CANCEL_GRACE: Duration = Duration::from_secs(5); +const OUTBOX_RETRY_INTERVAL: Duration = Duration::from_secs(60); +const MAX_PROPOSAL_BYTES: usize = 64 * 1024; +const MAX_OUTBOX_DRAIN: usize = 64; + +/// Result of the post-turn memory phase. Failures never retract a response the +/// user has already received, but they remain observable and retryable. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum PostTurnMemoryOutcome { + Stored { proposal_event_id: String }, + SkippedNoResponse, + Failed(String), +} + +fn h_tag(event: &Event, channel_id: Uuid) -> bool { + let expected = channel_id.to_string(); + event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.first().map(String::as_str) == Some("h") + && parts.get(1).map(String::as_str) == Some(expected.as_str()) + }) +} + +fn response_kind(kind: Kind) -> bool { + response_kinds().contains(&kind) +} + +fn response_kinds() -> [Kind; 3] { + [Kind::Custom(9), Kind::Custom(45001), Kind::Custom(45003)] +} + +async fn query_response_events( + rest: &RestClient, + channel_id: Uuid, + turn_started_at: u64, +) -> Result, RelayError> { + use nostr::{Alphabet, SingleLetterTag}; + + let channel = channel_id.to_string(); + let h = SingleLetterTag::lowercase(Alphabet::H); + let filter = nostr::Filter::new() + .kinds(response_kinds()) + .author(rest.keys.public_key()) + .custom_tags(h, [channel]) + .since(Timestamp::from(turn_started_at.saturating_sub(1))) + .limit(32); + let raw = tokio::time::timeout(RESPONSE_QUERY_TIMEOUT, rest.query(&[filter])) + .await + .map_err(|_| RelayError::Timeout)??; + let mut events = raw + .as_array() + .into_iter() + .flatten() + .filter_map(|value| serde_json::from_value::(value.clone()).ok()) + .filter(|event| { + event.pubkey == rest.keys.public_key() + && event.created_at.as_secs() >= turn_started_at.saturating_sub(1) + && response_kind(event.kind) + && h_tag(event, channel_id) + }) + .collect::>(); + events.sort_by(|left, right| { + left.created_at + .cmp(&right.created_at) + .then_with(|| left.id.to_hex().cmp(&right.id.to_hex())) + }); + events.dedup_by_key(|event| event.id); + Ok(events) +} + +async fn discover_response_events( + rest: &RestClient, + channel_id: Uuid, + turn_started_at: u64, +) -> Result, RelayError> { + for delay in [ + Duration::ZERO, + Duration::from_millis(250), + Duration::from_millis(750), + Duration::from_millis(1_500), + ] { + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + let events = query_response_events(rest, channel_id, turn_started_at).await?; + if !events.is_empty() { + return Ok(events); + } + } + // Bound read-after-write retries: a missing response must never fabricate + // evidence or block the already-published human-facing response forever. + Ok(Vec::new()) +} + +fn extraction_prompt(schema: u8, channel_id: Uuid, sources: &[String]) -> String { + let sources = sources.join(", "); + match schema { + 2 => format!( + r#"[System: automatic DKG memory finalization] +The human-facing Buzz response for this turn was already published. Do not send another Buzz message and do not call any tool. Return exactly one JSON object, with no Markdown fence or surrounding prose, that captures the externally communicable semantics supported by the signed turn evidence. + +Channel: {channel_id} +Evidence event IDs: {sources} + +Use this schema: +{{"schemaVersion":2,"profiles":["dkg-memory@1"],"summary":"...","entities":[{{"id":"claim-1","type":"memory:Claim","name":"...","description":"..."}}],"relations":[],"model":"...","promptVersion":"agent-memory-post-turn-v1"}} + +Always include dkg-memory@1. Add dkg-software@1 only for code, repositories, commits, reviews, tests, builds, deployments, or software components. Use only the ontology terms and canonical locator rules from your standing DKG instructions. Record decisions, claims, tasks, questions, people, projects, software entities, and their useful relationships. Even a short conversational answer should produce one concise evidence-backed claim. Never include hidden reasoning, chain-of-thought, credentials, secrets, private keys, tool traces, or facts not supported by this turn."# + ), + _ => format!( + r#"[System: automatic DKG memory finalization] +The human-facing Buzz response for this turn was already published. Do not send another Buzz message and do not call any tool. Return exactly one JSON object, with no Markdown fence or surrounding prose, that captures the externally communicable semantics supported by the signed turn evidence. + +Channel: {channel_id} +Evidence event IDs: {sources} + +Use this schema: +{{"schemaVersion":1,"summary":"...","items":[{{"kind":"decision|claim|question|task|relationship","text":"..."}}],"model":"...","promptVersion":"agent-memory-post-turn-v1"}} + +Even a short conversational answer should produce one concise evidence-backed item. Never include hidden reasoning, chain-of-thought, credentials, secrets, private keys, tool traces, or facts not supported by this turn."# + ), + } +} + +fn json_object_slice(text: &str) -> Option<&str> { + let start = text.find('{')?; + let mut depth = 0usize; + let mut in_string = false; + let mut escaped = false; + for (offset, character) in text[start..].char_indices() { + if in_string { + if escaped { + escaped = false; + } else if character == '\\' { + escaped = true; + } else if character == '"' { + in_string = false; + } + continue; + } + match character { + '"' => in_string = true, + '{' => depth += 1, + '}' => { + depth = depth.checked_sub(1)?; + if depth == 0 { + return Some(&text[start..start + offset + character.len_utf8()]); + } + } + _ => {} + } + } + None +} + +fn parse_proposal_output(text: &str, schema: u8) -> Result { + let candidate = + json_object_slice(text).ok_or_else(|| "agent returned no JSON object".to_string())?; + if candidate.len() > MAX_PROPOSAL_BYTES { + return Err("agent memory proposal exceeds 64 KiB".into()); + } + let value: Value = serde_json::from_str(candidate) + .map_err(|error| format!("agent returned invalid proposal JSON: {error}"))?; + let object = value + .as_object() + .ok_or_else(|| "agent memory proposal is not an object".to_string())?; + if object.get("schemaVersion").and_then(Value::as_u64) != Some(u64::from(schema)) { + return Err(format!( + "agent memory proposal did not use schemaVersion {schema}" + )); + } + let summary_ok = object + .get("summary") + .and_then(Value::as_str) + .is_some_and(|summary| !summary.trim().is_empty() && summary.len() <= 1_000); + if !summary_ok { + return Err("agent memory proposal has an invalid summary".into()); + } + let shape_ok = if schema == 2 { + object + .get("profiles") + .and_then(Value::as_array) + .is_some_and(|profiles| profiles.iter().any(|value| value == "dkg-memory@1")) + && object + .get("entities") + .and_then(Value::as_array) + .is_some_and(|entities| !entities.is_empty()) + && object.get("relations").and_then(Value::as_array).is_some() + } else { + object + .get("items") + .and_then(Value::as_array) + .is_some_and(|items| !items.is_empty()) + }; + if !shape_ok { + return Err("agent memory proposal is missing required semantic fields".into()); + } + let lowered = candidate.to_ascii_lowercase(); + if ["nsec1", "private_key", "privatekey", "secret_key"] + .iter() + .any(|marker| lowered.contains(marker)) + { + return Err("agent memory proposal appears to contain private key material".into()); + } + serde_json::to_string(&value) + .map_err(|error| format!("could not normalize agent memory proposal: {error}")) +} + +fn outbox_dir(rest: &RestClient) -> PathBuf { + if let Some(path) = std::env::var_os("BUZZ_DKG_MEMORY_OUTBOX_DIR") { + return PathBuf::from(path); + } + let mut namespace = Sha256::new(); + namespace.update(rest.base_url.as_bytes()); + namespace.update(rest.keys.public_key().to_bytes()); + let namespace = hex::encode(namespace.finalize()); + platform_data_dir() + .join("buzz") + .join("dkg-memory-outbox") + .join(&namespace[..24]) +} + +fn platform_data_dir() -> PathBuf { + if let Some(path) = std::env::var_os("LOCALAPPDATA") { + return PathBuf::from(path); + } + if let Some(path) = std::env::var_os("XDG_DATA_HOME") { + return PathBuf::from(path); + } + if let Some(home) = std::env::var_os("HOME") { + let home = PathBuf::from(home); + if cfg!(target_os = "macos") { + return home.join("Library").join("Application Support"); + } + return home.join(".local").join("share"); + } + std::env::temp_dir() +} + +fn persist_event(path: &Path, event: &Event) -> Result<(), String> { + if path.exists() { + return Ok(()); + } + let parent = path + .parent() + .ok_or_else(|| "memory outbox path has no parent".to_string())?; + std::fs::create_dir_all(parent).map_err(|error| format!("create memory outbox: {error}"))?; + let temporary = parent.join(format!(".{}.tmp", Uuid::new_v4())); + let body = serde_json::to_vec(event) + .map_err(|error| format!("serialize memory outbox event: {error}"))?; + let mut options = std::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options + .open(&temporary) + .map_err(|error| format!("open memory outbox event: {error}"))?; + use std::io::Write; + if let Err(error) = file.write_all(&body).and_then(|_| file.sync_all()) { + let _ = std::fs::remove_file(&temporary); + return Err(format!("persist memory outbox event: {error}")); + } + if let Err(error) = std::fs::rename(&temporary, path) { + let _ = std::fs::remove_file(&temporary); + if !path.exists() { + return Err(format!("commit memory outbox event: {error}")); + } + } + Ok(()) +} + +async fn submit_persisted(rest: &RestClient, path: &Path, event: &Event) -> Result<(), String> { + rest.submit_dkg_memory(event) + .await + .map_err(|error| error.to_string())?; + match std::fs::remove_file(path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(format!("remove accepted memory outbox event: {error}")), + } +} + +/// Retry signed proposals left behind by a prior crash or transient outage. +pub(crate) async fn flush_outbox(rest: &RestClient) { + let directory = outbox_dir(rest); + let Ok(entries) = std::fs::read_dir(&directory) else { + return; + }; + let mut paths = entries + .filter_map(Result::ok) + .map(|entry| entry.path()) + .filter(|path| path.extension().and_then(|value| value.to_str()) == Some("json")) + .collect::>(); + paths.sort(); + for path in paths.into_iter().take(MAX_OUTBOX_DRAIN) { + let event = match std::fs::read(&path) + .map_err(|error| error.to_string()) + .and_then(|body| { + serde_json::from_slice::(&body).map_err(|error| error.to_string()) + }) { + Ok(event) if event.verify_id() && event.verify_signature() => event, + Ok(_) | Err(_) => { + tracing::error!(path = %path.display(), "invalid signed event in DKG memory outbox; leaving it for operator inspection"); + continue; + } + }; + match submit_persisted(rest, &path, &event).await { + Ok(()) => { + tracing::info!(proposal_event_id = %event.id, "retried DKG memory proposal from outbox") + } + Err(error) => { + tracing::warn!(proposal_event_id = %event.id, %error, "DKG memory outbox retry remains pending"); + break; + } + } + } +} + +/// Retry pending proposals at startup and on a bounded cadence. The same +/// already-signed event is reused, so retries cannot create new graph writes. +pub(crate) async fn run_outbox_retry(rest: RestClient) { + let mut interval = tokio::time::interval(OUTBOX_RETRY_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + flush_outbox(&rest).await; + } +} + +/// Finalize one successful channel response into signed semantic memory. +pub(crate) async fn finalize_turn( + agent: &mut OwnedAgent, + session_id: &str, + ctx: &PromptContext, + channel_id: Uuid, + trigger_event_ids: &[String], + turn_started_at: u64, + schema: u8, +) -> PostTurnMemoryOutcome { + if !matches!(schema, 1 | 2) { + return PostTurnMemoryOutcome::Failed(format!( + "relay advertised unsupported DKG memory schema {schema}" + )); + } + let responses = + match discover_response_events(&ctx.rest_client, channel_id, turn_started_at).await { + Ok(events) => events, + Err(error) => { + return PostTurnMemoryOutcome::Failed(format!( + "could not discover the published agent response: {error}" + )) + } + }; + if responses.is_empty() { + return PostTurnMemoryOutcome::SkippedNoResponse; + } + let mut seen = HashSet::new(); + let sources = trigger_event_ids + .iter() + .cloned() + .chain(responses.iter().map(|event| event.id.to_hex())) + .filter(|event_id| seen.insert(event_id.clone())) + .collect::>(); + let prompt = extraction_prompt(schema, channel_id, &sources); + let prompt_result = agent + .acp + .session_prompt_with_idle_timeout( + session_id, + &prompt, + MEMORY_IDLE_TIMEOUT, + MEMORY_HARD_TIMEOUT, + ) + .await; + if let Err(error) = prompt_result { + if matches!( + error, + AcpError::IdleTimeout(_) | AcpError::HardTimeout { .. } + ) { + let _ = agent + .acp + .cancel_with_cleanup_grace(session_id, MEMORY_CANCEL_GRACE) + .await; + agent.state.invalidate_channel(&channel_id); + } + return PostTurnMemoryOutcome::Failed(format!("semantic extraction failed: {error}")); + } + let output = agent.acp.take_agent_message_text(); + let content = match parse_proposal_output(&output, schema) { + Ok(content) => content, + Err(error) => return PostTurnMemoryOutcome::Failed(error), + }; + let mut tags = Vec::with_capacity(sources.len() + 2); + let channel = channel_id.to_string(); + let channel_tag = match Tag::parse(["h", channel.as_str()]) { + Ok(tag) => tag, + Err(error) => { + return PostTurnMemoryOutcome::Failed(format!("invalid channel tag: {error}")) + } + }; + tags.push(channel_tag); + let proposal_tag = match Tag::parse(["t", "dkg-memory-proposal"]) { + Ok(tag) => tag, + Err(error) => { + return PostTurnMemoryOutcome::Failed(format!("invalid proposal tag: {error}")) + } + }; + tags.push(proposal_tag); + for source in &sources { + let source_tag = match Tag::parse(["e", source, "", "source"]) { + Ok(tag) => tag, + Err(error) => { + return PostTurnMemoryOutcome::Failed(format!("invalid source tag: {error}")) + } + }; + tags.push(source_tag); + } + let event = match EventBuilder::new(Kind::Custom(KIND_DKG_MEMORY_PROPOSAL), content) + .tags(tags) + .sign_with_keys(&ctx.agent_keys) + { + Ok(event) => event, + Err(error) => { + return PostTurnMemoryOutcome::Failed(format!("sign memory proposal: {error}")) + } + }; + let path = outbox_dir(&ctx.rest_client).join(format!("{}.json", event.id.to_hex())); + if let Err(error) = persist_event(&path, &event) { + return PostTurnMemoryOutcome::Failed(error); + } + match submit_persisted(&ctx.rest_client, &path, &event).await { + Ok(()) => PostTurnMemoryOutcome::Stored { + proposal_event_id: event.id.to_hex(), + }, + Err(error) => PostTurnMemoryOutcome::Failed(format!( + "proposal remains queued in the durable outbox: {error}" + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extracts_one_json_object_from_plain_or_fenced_output() { + let plain = r#"{"schemaVersion":1,"summary":"x","items":[{"kind":"claim","text":"x"}]}"#; + let expected: Value = serde_json::from_str(plain).unwrap(); + assert_eq!( + serde_json::from_str::(&parse_proposal_output(plain, 1).unwrap()).unwrap(), + expected + ); + let fenced = format!("```json\n{plain}\n```\n"); + assert_eq!( + serde_json::from_str::(&parse_proposal_output(&fenced, 1).unwrap()).unwrap(), + expected + ); + } + + #[test] + fn response_query_uses_only_supported_buzz_message_kinds() { + let kinds = response_kinds() + .into_iter() + .map(|kind| kind.as_u16()) + .collect::>(); + assert_eq!(kinds, vec![9, 45001, 45003]); + assert!(kinds + .into_iter() + .all(|kind| response_kind(Kind::Custom(kind)))); + } + + #[test] + fn rejects_wrong_schema_missing_semantics_and_key_material() { + assert!(parse_proposal_output( + r#"{"schemaVersion":1,"summary":"x","items":[{"kind":"claim","text":"x"}]}"#, + 2, + ) + .is_err()); + assert!(parse_proposal_output( + r#"{"schemaVersion":2,"profiles":["dkg-memory@1"],"summary":"x","entities":[],"relations":[]}"#, + 2, + ) + .is_err()); + assert!(parse_proposal_output( + r#"{"schemaVersion":1,"summary":"private_key leaked","items":[{"kind":"claim","text":"x"}]}"#, + 1, + ) + .is_err()); + } + + #[test] + fn prompt_forbids_tools_and_second_chat_message() { + let prompt = extraction_prompt( + 2, + Uuid::parse_str("8e8cd542-e5d0-4f81-a060-e9980b20599d").unwrap(), + &["a".repeat(64), "b".repeat(64)], + ); + assert!(prompt.contains("Do not send another Buzz message")); + assert!(prompt.contains("do not call any tool")); + assert!(prompt.contains("schemaVersion\":2")); + } +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 55f727f62a6..b86ccee8a1c 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2,6 +2,7 @@ mod acp; mod config; +mod dkg_memory; mod dkg_queries; mod dkg_recall; mod engram_fetch; @@ -70,17 +71,21 @@ const AUTHENTICATE_TIMEOUT: Duration = Duration::from_secs(10 * 60); const DKG_MEMORY_AGENT_INSTRUCTIONS_V1: &str = r#"## DKG Channel Memory -This relay supports agent-authored DKG memory. After every normal channel turn in which you successfully publish a Buzz response, submit one signed memory proposal before ending the turn. Use the current channel UUID, the triggering message event ID(s), and your newly published response `event_id` as `--source` evidence: +This relay supports agent-authored DKG memory. After every successful normal channel turn, the Buzz harness automatically starts a separate memory-finalization phase. During the human-facing response phase, publish the Buzz response normally but do not submit a memory proposal yourself. During the memory-finalization phase, return only the requested JSON semantic side output; the harness binds the triggering and response event IDs, signs the proposal, persists it for retry, and submits it. + +For manual recovery or debugging, the equivalent explicit command is: ```sh printf '%s' '{"schemaVersion":1,"summary":"...","items":[{"kind":"decision|claim|question|task|relationship","text":"..."}],"model":"...","promptVersion":"agent-memory-v1"}' | buzz memory propose --channel --source --source --input - ``` -Extract concise outcomes, claims, open questions, tasks, and relationships from that turn. Relationship items also require `subject`, `predicate`, and `object`; `confidence` is optional from 0 to 1. Record only externally communicable semantics and evidence—never hidden reasoning, chain-of-thought, secrets, credentials, or tool traces. Do not send a second chat message about the memory operation. If proposal submission fails, keep the human response intact and surface the failure only when it affects the requested work."#; +Extract concise outcomes, claims, open questions, tasks, and relationships from that turn. Relationship items also require `subject`, `predicate`, and `object`; `confidence` is optional from 0 to 1. Record only externally communicable semantics and evidence—never hidden reasoning, chain-of-thought, secrets, credentials, or tool traces. Never send a second chat message during memory finalization. A memory failure must not retract or duplicate the human-facing response."#; const DKG_MEMORY_AGENT_INSTRUCTIONS_V2: &str = r#"## DKG Channel Memory -This relay supports versioned, agent-authored DKG memory. After every normal channel turn in which you successfully publish a Buzz response, submit exactly one signed memory proposal before ending the turn. Use the current channel UUID, the triggering message event ID(s), and your newly published response `event_id` as `--source` evidence: +This relay supports versioned, agent-authored DKG memory. After every successful normal channel turn, the Buzz harness automatically starts a separate memory-finalization phase. During the human-facing response phase, publish the Buzz response normally but do not submit a memory proposal yourself. During the memory-finalization phase, return only the requested JSON semantic side output; the harness binds the triggering and response event IDs, signs the proposal, persists it for retry, and submits it. + +For manual recovery or debugging, the equivalent explicit command is: ```sh printf '%s' '{"schemaVersion":2,"profiles":["dkg-memory@1"],"summary":"...","entities":[{"id":"decision-1","type":"decisions:Decision","name":"...","description":"...","attributes":[{"predicate":"decisions:status","value":"accepted"}]},{"id":"topic-1","type":"memory:Entity","name":"..."}],"relations":[{"subject":"decision-1","predicate":"memory:about","object":"topic-1"}],"model":"...","promptVersion":"agent-memory-v2"}' | buzz memory propose --channel --source --source --input - @@ -92,7 +97,7 @@ Every relation object uses exactly `subject`, `predicate`, and `object` (plus op Use compact local entity IDs. For stable software identity, use `locator`: GitHub resources use `{"kind":"github","repository":"owner/repo","resource":"commit|pull-request|issue|repository","id":"..."}`; every code package/file/symbol uses `{"kind":"code","repository":"https://github.com/owner/repo","package":"@scope/package","path":"src/file.ts","symbol":"qualified.name","symbolKind":"function|class|interface|type-alias|enum"}`. Omit path for packages and symbol fields for files, but never omit the canonical HTTPS repository URL. A `schema:Project` requires `{"kind":"uri","uri":"https://canonical.example/project"}`. Reuse exact canonical locators across turns and communities; names are labels, never identity. If evidence provides no trustworthy global locator, use `memory:Entity` and let it remain local rather than inventing an identifier. `schema:sameAs` may connect evidence-backed aliases. Useful literal attributes include `decisions:context|outcome|consequences|status`, `tasks:status|priority|dueDate`, `schema:dateCreated`, `code:language|startLine|endLine`, `github:state|mergedAt`, and `software:result|environment`. -Extract concise entities and queryable relationships supported by the signed turn. Record only externally communicable semantics and evidence—never hidden reasoning, chain-of-thought, secrets, credentials, or tool traces. Do not invent ontology terms. Do not send a second chat message about the memory operation. A proposal response with `state: "processing"` is durably accepted but not queryable yet; only `state: "stored"` confirms completion. If proposal submission fails or remains processing after the CLI wait, keep the human response intact and describe the memory status accurately rather than claiming it was stored."#; +Extract concise entities and queryable relationships supported by the signed turn. Record only externally communicable semantics and evidence—never hidden reasoning, chain-of-thought, secrets, credentials, or tool traces. Do not invent ontology terms. Never send a second chat message during memory finalization. A proposal response with `state: "processing"` is durably accepted but not queryable yet; only `state: "stored"` confirms completion. A memory failure must not retract or duplicate the human-facing response."#; const DKG_SEMANTIC_QUERY_AGENT_INSTRUCTIONS: &str = r#"## Query DKG Channel Memory @@ -232,13 +237,13 @@ async fn relay_dkg_capabilities(relay_url: &str) -> DkgCapabilities { for attempt in 0..DKG_CAPABILITY_ATTEMPTS { match fetch_relay_dkg_capabilities(relay_url).await { Ok(capabilities) => return capabilities, - Err(error) if attempt + 1 < DKG_CAPABILITY_ATTEMPTS => { - let delay = DKG_CAPABILITY_RETRY_DELAYS[attempt]; - tracing::warn!(attempt = attempt + 1, %error, ?delay, "relay capability discovery failed; retrying"); - tokio::time::sleep(delay).await; - } Err(error) => { - tracing::warn!(attempt = attempt + 1, %error, "relay capability discovery failed; DKG memory disabled for this agent session"); + if let Some(&delay) = DKG_CAPABILITY_RETRY_DELAYS.get(attempt) { + tracing::warn!(attempt = attempt + 1, %error, ?delay, "relay capability discovery failed; retrying"); + tokio::time::sleep(delay).await; + } else { + tracing::warn!(attempt = attempt + 1, %error, "relay capability discovery failed; DKG memory disabled for this agent session"); + } } } } @@ -2031,6 +2036,10 @@ async fn tokio_main() -> Result<()> { let system_prompt = append_dkg_memory_instructions(config.system_prompt.clone(), dkg_capabilities); + let _dkg_memory_outbox_retry = dkg_capabilities + .memory_schema + .map(|_| tokio::spawn(dkg_memory::run_outbox_retry(relay.rest_client()))); + let base_prompt_content = config.base_prompt_content.take(); let ctx = Arc::new(PromptContext { mcp_servers: build_mcp_servers(&config), @@ -2056,6 +2065,7 @@ async fn tokio_main() -> Result<()> { .to_string(), rest_client: relay.rest_client(), dkg_semantic_query: dkg_capabilities.semantic_query, + dkg_memory_schema: dkg_capabilities.memory_schema, channel_info: pool::ChannelInfoResolver::new(channel_info_map, relay.rest_client()), context_message_limit: config.context_message_limit, max_turns_per_session: config.max_turns_per_session, diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 93e284484e6..c8b910df0fe 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -583,6 +583,11 @@ pub struct PromptContext { /// When true, each substantive channel turn gets a bounded, fail-open /// relevant-memory lookup before the agent starts work. pub dkg_semantic_query: bool, + /// Agent-memory proposal schema advertised by the active relay. + /// + /// A value enables the runtime-owned post-turn semantic side-output and + /// signed proposal submission path. `None` keeps non-DKG relays unchanged. + pub dkg_memory_schema: Option, /// Shared channel metadata for startup-known and dynamically joined channels. pub channel_info: ChannelInfoResolver, /// Max messages to include in thread/DM context. 0 = disabled. @@ -1443,6 +1448,62 @@ fn send_prompt_result( }); } +async fn finalize_dkg_memory_after_success( + agent: &mut OwnedAgent, + source: &PromptSource, + session_id: &str, + ctx: &PromptContext, + triggering_event_ids: &[String], + turn_started_at: u64, +) { + let (Some(schema), PromptSource::Channel(channel_id)) = (ctx.dkg_memory_schema, source) else { + return; + }; + let outcome = crate::dkg_memory::finalize_turn( + agent, + session_id, + ctx, + *channel_id, + triggering_event_ids, + turn_started_at, + schema, + ) + .await; + match outcome { + crate::dkg_memory::PostTurnMemoryOutcome::Stored { proposal_event_id } => { + tracing::info!( + channel = %channel_id, + %proposal_event_id, + "automatic post-turn DKG memory accepted" + ); + agent.acp.observe( + "dkg_memory_finalized", + serde_json::json!({ + "status": "accepted", + "proposalEventId": proposal_event_id, + }), + ); + } + crate::dkg_memory::PostTurnMemoryOutcome::SkippedNoResponse => { + tracing::debug!( + channel = %channel_id, + "automatic post-turn DKG memory skipped because the turn published no response" + ); + } + crate::dkg_memory::PostTurnMemoryOutcome::Failed(error) => { + tracing::warn!( + channel = %channel_id, + %error, + "automatic post-turn DKG memory was not accepted" + ); + agent.acp.observe( + "dkg_memory_finalized", + serde_json::json!({ "status": "pending_or_failed", "error": error }), + ); + } + } +} + /// Core async function spawned for each prompt. /// /// Lifecycle: @@ -1473,6 +1534,7 @@ pub async fn run_prompt_task( PromptSource::Channel(channel_id) => Some(*channel_id), PromptSource::Heartbeat => None, }; + let turn_started_unix = nostr::Timestamp::now().as_secs(); let turn_started_at = chrono::Utc::now().to_rfc3339(); agent.acp.set_observer_context(observer::context_for_turn( observer_channel_id, @@ -1491,7 +1553,7 @@ pub async fn run_prompt_task( PromptSource::Channel(_) => "channel", PromptSource::Heartbeat => "heartbeat", }, - "triggeringEventIds": triggering_event_ids, + "triggeringEventIds": &triggering_event_ids, }), ); @@ -2283,6 +2345,15 @@ pub async fn run_prompt_task( &source, &control_signal, ); + finalize_dkg_memory_after_success( + &mut agent, + &source, + &session_id, + &ctx, + &triggering_event_ids, + turn_started_unix, + ) + .await; let usage = agent.acp.take_turn_usage(); publish_agent_turn_metric( &ctx, @@ -2323,6 +2394,16 @@ pub async fn run_prompt_task( agent.state.heartbeat_standing_context_sent = true; } + finalize_dkg_memory_after_success( + &mut agent, + &source, + &session_id, + &ctx, + &triggering_event_ids, + turn_started_unix, + ) + .await; + let should_rotate = matches!( stop_reason, StopReason::MaxTokens | StopReason::MaxTurnRequests @@ -7470,6 +7551,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" auth_tag_json: None, }, dkg_semantic_query: false, + dkg_memory_schema: None, channel_info: ChannelInfoResolver::new( std::collections::HashMap::new(), RestClient { diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 84a312cf8d4..acfc3590414 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -466,6 +466,21 @@ impl RestClient { .await .map_err(|e| RelayError::Http(format!("DKG query response error: {e}"))) } + + /// Submit a signed agent-memory proposal through the authenticated relay. + /// + /// The relay binds the signed event to channel evidence and forwards it to + /// the configured DKG provider. A successful 2xx response means the + /// provider durably accepted the proposal, even when graph materialization + /// is still processing. + pub async fn submit_dkg_memory(&self, event: &Event) -> Result { + let body_bytes = serde_json::to_vec(event) + .map_err(|e| RelayError::Http(format!("DKG memory serialize error: {e}")))?; + let resp = self.bridge_post("/api/dkg/memory", &body_bytes).await?; + resp.json() + .await + .map_err(|e| RelayError::Http(format!("DKG memory response error: {e}"))) + } } /// Events the harness cares about. diff --git a/docs/dkg-memory.md b/docs/dkg-memory.md index 295d108352a..a36e95f911b 100644 --- a/docs/dkg-memory.md +++ b/docs/dkg-memory.md @@ -87,10 +87,11 @@ node or authenticated community provider can later upgrade the same records. ## How it works -1. Humans and agents post ordinary Buzz messages. After a successful turn, a - participating agent may privately submit one signed semantic-memory - proposal citing its signed input and output events. This does not add a - second message to the conversation. +1. Humans and agents post ordinary Buzz messages. After a managed agent + successfully publishes its response, the Buzz runtime starts a separate, + tool-free semantic extraction phase. The runtime binds the signed input and + output event IDs, signs the resulting proposal with the agent identity, and + submits it without adding a second message to the conversation. 2. The relay authenticates the agent and channel access. The integration verifies signatures and evidence binding, then creates or reuses the channel's isolated Context Graph. Explicit `@dkg distill` remains a manual @@ -103,6 +104,11 @@ The desktop app never receives a DKG credential or accepts a caller-supplied Context Graph identifier. The relay derives graph scope from the authenticated community and channel. +The runtime writes each signed proposal to a local outbox before the network +request and retries the exact same event after transient failures or restarts. +The integration also deduplicates newly signed retries by channel and canonical +evidence-set digest, so retry recovery cannot create a second graph write. + ## Semantic profiles and canonical identities The relay advertises the proposal schema and ontology profiles it supports.