External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Reduce repeated ACP session context (#5423)

Browse Source 
## Summary

- deliver legacy ACP standing context once per live session, committing
delivery state only after a successful turn
- send only new thread/DM event deltas on later turns, with fail-open
behavior for missing IDs and failed/cancelled prompts
- fence native steer delivery acknowledgements by ACP session identity
so stale acks cannot poison replacement sessions
- keep context hints truthful when a fetch contains only the triggering
event versus history delivered earlier

## Validation

The pre-push hook passed on exact pushed head
`6a768f1bc80fe63c686acf8d730f177fff8add3c`:

- `branch-skew`
- `desktop-check`
- `desktop-typecheck`
- `desktop-test`
- `rust-tests`
- `desktop-tauri-checks`

Focused regression tests were also run while iterating:

- `channel_prompt_commits_delivery_state_only_after_acp_success`
- `in_flight_stale_native_steer_ack_cannot_update_replacement_session`
- thread/DM trigger-only versus previously-delivered context hint tests

## Known limitations and follow-ups

A local Goose smoke timed out at `session/new`. This diff does not
change code that executes at or before `session/new`; its earliest
affected runtime behavior is delivery-state insertion after session
creation succeeds. The smoke failure is therefore bounded as
environmental or pre-existing, but no successful live-provider turn was
obtained. Scripted ACP wire/lifecycle tests carry the regression
coverage.

- #5421 — distinguish post-delta, already-delivered, and fetch-truncated
context counts
- #5422 — define a standing-context re-delivery policy if a legacy
provider compacts it away

Durable process-restart/session resume remains out of scope for this
slice of #5342. #5386 also remains separate pending upstream adapter
support.

---------

Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
2026-08-10 08:50:34 -07:00
committed by GitHub
co-authored by Carl
parent 5e4c05f90b
commit 563e4346da
4 changed files with 1585 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-acp/src/acp.rs
+12 -8
View File
@@ -1617,7 +1617,9 @@ impl AcpClient {
"steer accepted as {STEER_OUTCOME_STARTED_NEW_TURN}: \
awaited turn had ended — hard deadline not renewed"
);
crate::pool::SteerAck::Success
crate::pool::SteerAck::Success {
session_id: session_id.to_owned(),
}
}
Some(_) => {
let renew_now = Instant::now();
@@ -1629,7 +1631,9 @@ impl AcpClient {
"steer success: renewed hard deadline ({max_duration:?} from now)"
);
}
crate::pool::SteerAck::Success
crate::pool::SteerAck::Success {
session_id: session_id.to_owned(),
}
}
None => {
// Report the raw string when
@@ -3799,7 +3803,7 @@ mod tests {
.await
.expect("ack oneshot must have received a SteerAck");
match ack {
crate::pool::SteerAck::Success => {}
crate::pool::SteerAck::Success { .. } => {}
other => panic!("expected SteerAck::Success, got {other:?}"),
}
}
@@ -3860,7 +3864,7 @@ mod tests {
.await
.expect("ack oneshot must have received a SteerAck");
match ack {
crate::pool::SteerAck::Success => {}
crate::pool::SteerAck::Success { .. } => {}
other => panic!("expected SteerAck::Success, got {other:?}"),
}
}
@@ -4044,7 +4048,7 @@ mod tests {
"_session/steering must not carry expectedRunId; wrote: {written}"
);
assert!(
matches!(ack, crate::pool::SteerAck::Success),
matches!(ack, crate::pool::SteerAck::Success { .. }),
"injected outcome must ack Success, got {ack:?}"
);
}
@@ -4076,7 +4080,7 @@ mod tests {
// no `outcome`) — the OutcomeRejected guard applies only to
// `_session/steering`.
assert!(
matches!(ack, crate::pool::SteerAck::Success),
matches!(ack, crate::pool::SteerAck::Success { .. }),
"goose success result must ack Success, got {ack:?}"
);
}
@@ -4181,7 +4185,7 @@ mod tests {
assert_eq!(result.unwrap()["done"], serde_json::json!(true));
let ack = ack_rx.await.expect("ack must be received");
assert!(
matches!(ack, crate::pool::SteerAck::Success),
matches!(ack, crate::pool::SteerAck::Success { .. }),
"injected must ack Success, got {ack:?}"
);
}
@@ -4238,7 +4242,7 @@ mod tests {
// rather than released — hence Success, not an Err.
let ack = ack_rx.await.expect("ack must be received");
assert!(
matches!(ack, crate::pool::SteerAck::Success),
matches!(ack, crate::pool::SteerAck::Success { .. }),
"startedNewTurn is a delivery success, got {ack:?}"
);
}
crates/buzz-acp/src/lib.rs
+314 -5
View File
@@ -2821,7 +2821,7 @@ async fn tokio_main() -> Result<()> {
// treat as PromptCompletedNeutral to avoid leaking
// the withheld event in `withheld_native_steer`.
let (release_withheld, drop_withheld, signal_fallback) = match &ack {
Ok(pool::SteerAck::Success) => (false, true, false),
Ok(pool::SteerAck::Success { .. }) => (false, true, false),
// -32601 = method_not_found: agent does not implement the
// steer extension. Fire cancel+merge so the message still
// reaches the agent.
@@ -2854,8 +2854,19 @@ async fn tokio_main() -> Result<()> {
signal_fallback,
"non-cancelling steer ack received"
);
if matches!(ack, Ok(pool::SteerAck::Success)) {
if let Ok(pool::SteerAck::Success { session_id }) = &ack {
queue.extend_in_flight_deadline(channel_id, config.max_turn_duration_secs);
if !pool.record_successful_steer(
channel_id,
event_id.clone(),
session_id.clone(),
) {
tracing::warn!(
channel = %channel_id,
event_id = %event_id,
"successful steer lost its in-flight delivery ledger"
);
}
}
if drop_withheld {
queue.remove_event(channel_id, &event_id);
@@ -3318,6 +3329,7 @@ fn dispatch_pending(
recoverable_batch,
control_tx: Some(control_tx),
steer_tx,
successful_steer_deliveries: HashSet::new(),
},
);
dispatched_channels.push((channel_id, typing_scope));
@@ -3399,9 +3411,30 @@ fn handle_prompt_result(
) -> LoopAction {
let before = pool.task_map().len();
let agent_index = result.agent.index;
let successful_steer_deliveries = pool
.task_map()
.values()
.find(|meta| meta.agent_index == agent_index)
.map(|meta| meta.successful_steer_deliveries.clone())
.unwrap_or_default();
pool.task_map_mut()
.retain(|_, meta| meta.agent_index != agent_index);
debug_assert_eq!(before, pool.task_map().len() + 1);
if let PromptSource::Channel(channel_id) = &result.source {
// The task may have invalidated this session before returning. Never
// resurrect delivery state for a dead session; its replacement must
// receive fresh standing context and history.
if let Some(live_session_id) = result.agent.state.sessions.get(channel_id).cloned() {
let event_ids = successful_steer_deliveries
.into_iter()
.filter(|delivery| delivery.session_id == live_session_id)
.map(|delivery| delivery.event_id);
result
.agent
.state
.mark_channel_delivery_success(*channel_id, false, event_ids);
}
}
// The hard-timeout death_message (below) must describe the batch's
// *actual* fate, not just the `recently_active` eligibility flag — a
@@ -3932,6 +3965,7 @@ fn dispatch_heartbeat(
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
*heartbeat_in_flight = true;
@@ -4574,17 +4608,23 @@ mod heartbeat_base_prompt_tests {
// heartbeat user message, composed as `[Base]\n{bp}\n\n{prompt}`. This is
// the second half of the round-2 regression (the first being initial_message).
fn heartbeat_standing() -> queue::StandingContext<'static> {
queue::StandingContext {
base_prompt: Some("you are a helpful agent"),
..Default::default()
}
}
#[test]
fn test_heartbeat_legacy_agent_gets_base_prepended() {
// protocol_version 1 + Some(base_prompt): heartbeat prompt is prefixed
// with the [Base] section exactly as the legacy session/new path would.
let prompt = "[System: Heartbeat]\nrun feed get";
let composed = pool::prepend_base_for_legacy(1, Some("you are a helpful agent"), prompt);
let composed = pool::prepend_standing_for_legacy(1, &heartbeat_standing(), prompt);
assert_eq!(
composed,
"[Base]\nyou are a helpful agent\n\n[System: Heartbeat]\nrun feed get"
);
assert!(composed.starts_with("[Base]\nyou are a helpful agent\n\n"));
}
#[test]
@@ -4592,7 +4632,7 @@ mod heartbeat_base_prompt_tests {
// protocol_version 2 gets base_prompt via session/new; the heartbeat
// prompt is sent verbatim.
let prompt = "[System: Heartbeat]\nrun feed get";
let composed = pool::prepend_base_for_legacy(2, Some("you are a helpful agent"), prompt);
let composed = pool::prepend_standing_for_legacy(2, &heartbeat_standing(), prompt);
assert_eq!(composed, prompt);
}
}
@@ -4703,6 +4743,7 @@ mod owner_control_command_tests {
recoverable_batch: None,
control_tx: Some(control_tx),
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
@@ -6473,6 +6514,263 @@ mod error_outcome_emission_tests {
}
}
#[tokio::test]
async fn successful_native_steer_is_transferred_to_live_session_delivery_state() {
let channel_id = Uuid::new_v4();
let steer_event_id = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let mut agent = dummy_agent(0).await;
agent
.state
.sessions
.insert(channel_id, "live-session".into());
agent
.state
.deliveries
.insert(channel_id, Default::default());
let mut pool = AgentPool::from_slots(vec![None]);
let task_id = pool.join_set.spawn(async {}).id();
pool.task_map_mut().insert(
task_id,
crate::pool::TaskMeta {
agent_index: 0,
channel_id: Some(channel_id),
turn_id: "test-turn-id".into(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::from([
crate::pool::SuccessfulSteerDelivery {
event_id: steer_event_id.into(),
session_id: "live-session".into(),
},
]),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
let config = test_config();
let mut heartbeat_in_flight = false;
let removed_channels = HashSet::new();
let mut crash_history = vec![SlotCircuit {
crash_times: Vec::new(),
open_until: None,
respawn_in_flight: false,
}];
let (respawn_tx, _respawn_rx) = mpsc::channel(8);
let mut respawn_tasks = tokio::task::JoinSet::new();
let result = PromptResult {
agent,
source: PromptSource::Channel(channel_id),
turn_id: "test-turn-id".into(),
outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn),
batch: None,
};
handle_prompt_result(
&mut pool,
&mut queue,
&config,
result,
&mut heartbeat_in_flight,
&removed_channels,
&mut crash_history,
&respawn_tx,
&mut respawn_tasks,
None,
None,
);
let returned = pool.agents_mut()[0].as_ref().expect("returned agent");
assert!(returned.state.deliveries[&channel_id]
.delivered_event_ids
.contains(steer_event_id));
}
#[tokio::test]
async fn in_flight_stale_native_steer_ack_cannot_update_replacement_session() {
let channel_id = Uuid::new_v4();
let mut agent = dummy_agent(0).await;
agent
.state
.sessions
.insert(channel_id, "replacement-session".into());
agent
.state
.deliveries
.insert(channel_id, Default::default());
let mut pool = AgentPool::from_slots(vec![None]);
let task_id = pool.join_set.spawn(async {}).id();
pool.task_map_mut().insert(
task_id,
crate::pool::TaskMeta {
agent_index: 0,
channel_id: Some(channel_id),
turn_id: "test-turn-id".into(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::from([
crate::pool::SuccessfulSteerDelivery {
event_id: "stale-event".into(),
session_id: "old-session".into(),
},
]),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
let config = test_config();
let mut heartbeat_in_flight = false;
let removed_channels = HashSet::new();
let mut crash_history = vec![SlotCircuit {
crash_times: Vec::new(),
open_until: None,
respawn_in_flight: false,
}];
let (respawn_tx, _respawn_rx) = mpsc::channel(8);
let mut respawn_tasks = tokio::task::JoinSet::new();
let result = PromptResult {
agent,
source: PromptSource::Channel(channel_id),
turn_id: "test-turn-id".into(),
outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn),
batch: None,
};
handle_prompt_result(
&mut pool,
&mut queue,
&config,
result,
&mut heartbeat_in_flight,
&removed_channels,
&mut crash_history,
&respawn_tx,
&mut respawn_tasks,
None,
None,
);
let returned = pool.agents_mut()[0].as_ref().expect("returned agent");
assert!(returned.state.deliveries[&channel_id]
.delivered_event_ids
.is_empty());
}
#[tokio::test]
async fn successful_native_steer_ack_after_task_return_updates_matching_live_session() {
let channel_id = Uuid::new_v4();
let steer_event_id = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
let mut agent = dummy_agent(0).await;
agent
.state
.sessions
.insert(channel_id, "live-session".into());
agent
.state
.deliveries
.insert(channel_id, Default::default());
let mut pool = AgentPool::from_slots(vec![Some(agent)]);
assert!(pool.record_successful_steer(
channel_id,
steer_event_id.into(),
"live-session".into(),
));
let returned = pool.agents_mut()[0].as_ref().expect("idle returned agent");
assert!(returned.state.deliveries[&channel_id]
.delivered_event_ids
.contains(steer_event_id));
}
#[tokio::test]
async fn late_native_steer_ack_cannot_update_replacement_session() {
let channel_id = Uuid::new_v4();
let mut agent = dummy_agent(0).await;
agent
.state
.sessions
.insert(channel_id, "replacement-session".into());
agent
.state
.deliveries
.insert(channel_id, Default::default());
let mut pool = AgentPool::from_slots(vec![Some(agent)]);
assert!(!pool.record_successful_steer(
channel_id,
"stale-event".into(),
"old-session".into(),
));
let returned = pool.agents_mut()[0].as_ref().expect("replacement agent");
assert!(returned.state.deliveries[&channel_id]
.delivered_event_ids
.is_empty());
}
#[tokio::test]
async fn invalidated_session_does_not_resurrect_successful_steer_delivery_state() {
let channel_id = Uuid::new_v4();
let agent = dummy_agent(0).await;
// No live session: simulates the prompt task invalidating before return.
let mut pool = AgentPool::from_slots(vec![None]);
let task_id = pool.join_set.spawn(async {}).id();
pool.task_map_mut().insert(
task_id,
crate::pool::TaskMeta {
agent_index: 0,
channel_id: Some(channel_id),
turn_id: "test-turn-id".into(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::from([
crate::pool::SuccessfulSteerDelivery {
event_id: "stale-event".into(),
session_id: "invalidated-session".into(),
},
]),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
let config = test_config();
let mut heartbeat_in_flight = false;
let removed_channels = HashSet::new();
let mut crash_history = vec![SlotCircuit {
crash_times: Vec::new(),
open_until: None,
respawn_in_flight: false,
}];
let (respawn_tx, _respawn_rx) = mpsc::channel(8);
let mut respawn_tasks = tokio::task::JoinSet::new();
let result = PromptResult {
agent,
source: PromptSource::Channel(channel_id),
turn_id: "test-turn-id".into(),
outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn),
batch: None,
};
handle_prompt_result(
&mut pool,
&mut queue,
&config,
result,
&mut heartbeat_in_flight,
&removed_channels,
&mut crash_history,
&respawn_tx,
&mut respawn_tasks,
None,
None,
);
let returned = pool.agents_mut()[0].as_ref().expect("returned agent");
assert!(!returned.state.deliveries.contains_key(&channel_id));
}
/// Drive one error outcome through `handle_prompt_result` and return how
/// many `turn_error` events it emitted to the observer feed.
async fn turn_errors_emitted_for(outcome: PromptOutcome) -> usize {
@@ -6493,6 +6791,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
@@ -6569,6 +6868,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
started_rx.await.unwrap();
@@ -6661,6 +6961,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
@@ -6752,6 +7053,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
@@ -6857,6 +7159,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
@@ -6933,6 +7236,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
@@ -7027,6 +7331,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let config = test_config();
@@ -7143,6 +7448,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
@@ -7282,6 +7588,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
@@ -7470,6 +7777,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
@@ -7555,6 +7863,7 @@ mod error_outcome_emission_tests {
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(config::DedupMode::Queue);
crates/buzz-acp/src/pool.rs
+1019 -123
View File
File diff suppressed because it is too large Load Diff
crates/buzz-acp/src/queue.rs
+240 -40
View File
@@ -990,6 +990,9 @@ pub enum ConversationContext {
/// A single message in a conversation context section.
#[derive(Debug, Clone)]
pub struct ContextMessage {
/// Nostr event ID. Legacy REST fixtures may omit it, in which case it is
/// empty and cannot participate in delivery deduplication.
pub event_id: String,
pub pubkey: String,
pub timestamp: String,
pub content: String,
@@ -1241,6 +1244,7 @@ fn format_context_hints(
thread_tags: &ThreadTags,
is_dm: bool,
has_conversation_context: bool,
conversation_context_had_delivered_events: bool,
reply_anchor: Option<&str>,
) -> String {
let channel_display = match channel_info {
@@ -1258,6 +1262,10 @@ fn format_context_hints(
"Thread context included below. Use `buzz messages thread --channel <UUID> --event <ID>` for full history if truncated."
} else if has_conversation_context {
"Conversation context included below. Use `buzz messages get --channel <UUID>` for full history if truncated."
} else if conversation_context_had_delivered_events && is_reply {
"Earlier thread context was already delivered in this session. Use `buzz messages thread --channel <UUID> --event <ID>` to re-read the reply chain."
} else if conversation_context_had_delivered_events {
"Earlier conversation context was already delivered in this session. Use `buzz messages get --channel <UUID>` to re-read it."
} else if is_reply {
"Use `buzz messages thread --channel <UUID> --event <ID>` to fetch the reply chain."
} else {
@@ -1285,6 +1293,8 @@ fn format_context_hints(
} else if let Some(ref root) = thread_tags.root_event_id {
let ctx_hint = if has_conversation_context {
"Thread context included below. Use `buzz messages thread --channel <UUID> --event <ID>` for full history if truncated."
} else if conversation_context_had_delivered_events {
"Earlier thread context was already delivered in this session. Use `buzz messages thread --channel <UUID> --event <ID>` to re-read it."
} else {
"Use `buzz messages thread --channel <UUID> --event <ID>` to fetch thread context."
};
@@ -1359,6 +1369,9 @@ pub struct FormatPromptArgs<'a> {
pub agent_core: Option<&'a str>,
pub channel_info: Option<&'a PromptChannelInfo>,
pub conversation_context: Option<&'a ConversationContext>,
/// True when delivery-delta filtering removed at least one event that this
/// 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>,
/// When true, base_prompt and system_prompt are delivered via the system
/// role (session/new) and omitted from the user message. When false
@@ -1374,9 +1387,62 @@ pub struct FormatPromptArgs<'a> {
///
/// For modern agents (protocol_version >= 2) the section is delivered via
/// the system role in session/new; omit here to avoid duplication.
/// For legacy agents it rides in the user message on every turn of the
/// session, alongside `[Base]`/`[System]`/`[Agent Memory — core]`.
pub agent_canvas: Option<&'a str>,
/// Set once this session's standing context has already been delivered —
/// see [`StandingContext`]. Only meaningful for legacy agents; modern
/// agents are gated by `has_system_prompt_support` regardless.
///
/// Defaults to `false` so a caller that never sets it behaves as if this
/// were the session's first message.
pub standing_context_sent: bool,
}
/// The prompt sections that do not change for the life of a session: base
/// prompt, persona, team instructions, core memory, and channel canvas.
///
/// Protocol-v2 agents receive all of this through the system role at
/// `session/new`, once. Legacy agents (`protocol_version < 2`) have no system
/// role, so it has to ride in a user message — but only in the session's
/// *first* one. Re-sending it every turn makes the standing framing the newest
/// and most-repeated text in the window, outweighing the conversation it exists
/// to frame, and evicting real channel history that much sooner.
///
/// Both legacy dispatch paths (initial message, batch flush) render through
/// this one type so their section set and ordering cannot drift apart.
#[derive(Default)]
pub(crate) struct StandingContext<'a> {
pub base_prompt: Option<&'a str>,
pub system_prompt: Option<&'a str>,
pub team_instructions: Option<&'a str>,
pub agent_core: Option<&'a str>,
pub agent_canvas: Option<&'a str>,
}
impl StandingContext<'_> {
/// Render the sections in the order legacy agents have always seen them.
pub(crate) fn sections(&self) -> Vec<String> {
let mut sections = Vec::with_capacity(5);
if let Some(bp) = self.base_prompt {
sections.push(base_section(bp));
}
if let Some(sp) = self.system_prompt {
sections.push(format!("[System]\n{sp}"));
}
if let Some(team) = self
.team_instructions
.map(str::trim)
.filter(|value| !value.is_empty())
{
sections.push(format!("[Team Instructions]\n{team}"));
}
if let Some(core) = self.agent_core {
sections.push(core.to_string());
}
if let Some(canvas) = self.agent_canvas {
sections.push(canvas.to_string());
}
sections
}
}
/// Format the `[Base]` section for the base prompt.
@@ -1391,12 +1457,12 @@ pub(crate) fn base_section(base_prompt: &str) -> String {
/// Format a [`FlushBatch`] into the per-section prompt blocks for the agent.
///
/// Produces a stable prompt with these sections (in order):
/// 0. `[Base]` — base prompt (only for legacy agents without systemPrompt support)
/// 1. `[System]` — system prompt (only for legacy agents without systemPrompt support)
/// 2. `[Agent Memory — core]` — if agent core memory is set
/// 3. `[Context]` — scope, channel name, and contextual hints for the agent
/// 4. `[Thread Context]` or `[Conversation Context]` — if fetched
/// 5. `[Event]` / `[Buzz events]` — the triggering event(s)
/// 0. [`StandingContext`] — `[Base]`, `[System]`, `[Team Instructions]`,
/// `[Agent Memory — core]`, `[Channel Canvas]`. Legacy agents only, and only
/// on the session's first message (see `standing_context_sent`)
/// 1. `[Context]` — scope, channel name, and contextual hints for the agent
/// 2. `[Thread Context]` or `[Conversation Context]` — if fetched
/// 3. `[Event]` / `[Buzz events]` — the triggering event(s)
///
/// Each section is returned as its own block rather than one joined string so
/// the observer frame's size trimmer (`fit_observer_event_to_budget`) elides
@@ -1428,38 +1494,22 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
let mut sections: Vec<String> = Vec::with_capacity(7);
// For legacy agents (protocol_version < 2), inject base_prompt and
// system_prompt as user-message sections. Modern agents receive these
// via the system role in session/new.
if !args.has_system_prompt_support {
if let Some(bp) = args.base_prompt {
sections.push(base_section(bp));
}
if let Some(sp) = args.system_prompt {
sections.push(format!("[System]\n{sp}"));
}
if let Some(team) = args
.team_instructions
.map(str::trim)
.filter(|value| !value.is_empty())
{
sections.push(format!("[Team Instructions]\n{team}"));
}
}
// NIP-AE agent core memory (rendered by `engram_fetch::build_core_section`).
// For modern agents (protocol_version >= 2), core is delivered via the
// system role in session/new, so it is omitted here to avoid duplication.
// Legacy agents have no system role, so core rides in the user message
// alongside `[Base]`/`[System]`.
if !args.has_system_prompt_support {
if let Some(core) = args.agent_core {
sections.push(core.to_string());
}
// Channel canvas metadata — same delivery semantics as core for legacy agents.
if let Some(canvas) = args.agent_canvas {
sections.push(canvas.to_string());
}
// Standing context — base prompt, persona, team instructions, core memory
// and canvas. Modern agents received all of it via the system role in
// session/new. Legacy agents get it here, in the session's first message
// only; `standing_context_sent` means an earlier message in this session
// already carried it.
if !args.has_system_prompt_support && !args.standing_context_sent {
sections.extend(
StandingContext {
base_prompt: args.base_prompt,
system_prompt: args.system_prompt,
team_instructions: args.team_instructions,
agent_core: args.agent_core,
agent_canvas: args.agent_canvas,
}
.sections(),
);
}
// 2. Context hints (with a human-aware reply anchor).
@@ -1489,6 +1539,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
&thread_tags,
is_dm,
args.conversation_context.is_some(),
args.conversation_context_had_delivered_events,
reply_anchor.as_deref(),
));
@@ -2408,6 +2459,60 @@ mod tests {
);
}
#[test]
fn test_format_prompt_legacy_agent_omits_standing_after_first_message() {
// The defect this pins: standing context was re-sent on every turn of a
// legacy session, so the largest and least informative part of the
// prompt was also the most recent — crowding out the conversation and
// evicting real channel history sooner.
let ch = Uuid::new_v4();
let batch = FlushBatch {
channel_id: ch,
events: vec![BatchEvent {
event: make_event("hello"),
prompt_tag: "test".into(),
received_at: Instant::now(),
}],
cancelled_events: vec![],
cancel_reason: None,
};
let canvas = "[Channel Canvas]\ncanvas content";
let core = "[Agent Memory — core]\nremember this";
let args = |sent| FormatPromptArgs {
has_system_prompt_support: false,
base_prompt: Some("test base prompt"),
system_prompt: Some("test system prompt"),
team_instructions: Some("ship small"),
agent_core: Some(core),
agent_canvas: Some(canvas),
standing_context_sent: sent,
..Default::default()
};
let first = format_prompt(&batch, &args(false)).join("\n\n");
let later = format_prompt(&batch, &args(true)).join("\n\n");
for section in [
"[Base]",
"[System]",
"[Team Instructions]",
"[Agent Memory — core]",
"[Channel Canvas]",
] {
assert!(first.contains(section), "first message missing {section}");
assert!(!later.contains(section), "turn 2 repeated {section}");
}
// What the turn is actually about survives, and now leads.
assert!(later.starts_with("[Context]"), "got: {later}");
assert!(later.contains("hello"));
assert!(
later.len() < first.len(),
"later turns must be smaller: {} vs {}",
later.len(),
first.len()
);
}
#[test]
fn test_format_prompt_modern_agent_suppresses_base_and_system() {
let ch = Uuid::new_v4();
@@ -2464,6 +2569,7 @@ mod tests {
let ctx = ConversationContext::Thread {
messages: vec![ContextMessage {
event_id: String::new(),
pubkey: "npub1test".into(),
content: "prior message".into(),
timestamp: "2024-01-01T00:00:00Z".into(),
@@ -3078,11 +3184,13 @@ mod tests {
let ctx = ConversationContext::Thread {
messages: vec![
ContextMessage {
event_id: String::new(),
pubkey: "npub1xyz".into(),
timestamp: "2026-03-15T16:30:00Z".into(),
content: "Let's refactor auth".into(),
},
ContextMessage {
event_id: String::new(),
pubkey: "npub1def".into(),
timestamp: "2026-03-15T16:35:00Z".into(),
content: "yes go ahead".into(),
@@ -3125,6 +3233,7 @@ mod tests {
};
let ctx = ConversationContext::Dm {
messages: vec![ContextMessage {
event_id: String::new(),
pubkey: "npub1abc".into(),
timestamp: "2026-03-15T16:00:00Z".into(),
content: "Can you deploy?".into(),
@@ -3170,6 +3279,7 @@ mod tests {
};
let ctx = ConversationContext::Thread {
messages: vec![ContextMessage {
event_id: String::new(),
pubkey: author_hex.clone(),
timestamp: "2026-03-25T05:51:25Z".into(),
content: "follow up".into(),
@@ -3382,6 +3492,7 @@ mod tests {
// Thread context fetched (as the fetch path does for DM replies).
let ctx = ConversationContext::Thread {
messages: vec![ContextMessage {
event_id: String::new(),
pubkey: "npub1xyz".into(),
timestamp: "2026-03-15T16:30:00Z".into(),
content: "Should I deploy?".into(),
@@ -3418,6 +3529,95 @@ mod tests {
assert!(prompt.contains("Should I deploy?"));
}
#[test]
fn test_format_prompt_empty_thread_delta_distinguishes_trigger_only_from_delivered() {
let ch = Uuid::new_v4();
let event = make_event_with_tags(
"follow up",
vec![vec![
"e".into(),
"root123".into(),
"".into(),
"reply".into(),
]],
);
let batch = FlushBatch {
channel_id: ch,
events: vec![BatchEvent {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
}],
cancelled_events: vec![],
cancel_reason: None,
};
let trigger_only_prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n");
assert!(trigger_only_prompt.contains("fetch thread context"));
assert!(!trigger_only_prompt.contains("already delivered in this session"));
let prompt = format_prompt(
&batch,
&FormatPromptArgs {
conversation_context_had_delivered_events: true,
..Default::default()
},
)
.join("\n\n");
assert!(prompt.contains("Earlier thread context was already delivered in this session"));
assert!(prompt.contains("buzz messages thread"));
assert!(!prompt.contains("Thread context included below"));
assert!(!prompt.contains("[Thread Context"));
}
#[test]
fn test_format_prompt_empty_dm_delta_distinguishes_trigger_only_from_delivered() {
let ch = Uuid::new_v4();
let batch = FlushBatch {
channel_id: ch,
events: vec![BatchEvent {
event: make_event("follow up"),
prompt_tag: "dm".into(),
received_at: Instant::now(),
}],
cancelled_events: vec![],
cancel_reason: None,
};
let ci = PromptChannelInfo {
name: "DM".into(),
channel_type: "dm".into(),
};
let trigger_only_prompt = format_prompt(
&batch,
&FormatPromptArgs {
channel_info: Some(&ci),
..Default::default()
},
)
.join("\n\n");
assert!(trigger_only_prompt.contains("for conversation context"));
assert!(!trigger_only_prompt.contains("already delivered in this session"));
let prompt = format_prompt(
&batch,
&FormatPromptArgs {
channel_info: Some(&ci),
conversation_context_had_delivered_events: true,
..Default::default()
},
)
.join("\n\n");
assert!(
prompt.contains("Earlier conversation context was already delivered in this session")
);
assert!(prompt.contains("buzz messages get"));
assert!(!prompt.contains("Conversation context included below"));
assert!(!prompt.contains("[Conversation Context"));
}
#[test]
fn test_format_prompt_dm_non_reply_hints_get_messages() {
let ch = Uuid::new_v4();