public/galaxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Restore progressive agent output and reduce repeated narration

Browse Source 
Forward text and reasoning events as they arrive while retaining canonical history without replaying output. Tighten shared communication guidance and add streaming-order regression coverage. Include the staged dependency refresh.
This commit is contained in:
rkw6086
2026-10-04 19:12:58 -05:00
parent b29882d56c
commit 447113bdd0
5 changed files with 450 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/ai/runtime/provider_run_coordinator.rs
+38 -41
View File
@@ -753,6 +753,18 @@ impl ProviderRunCoordinator {
return Ok(());
}
buffer.text.push_str(&text);
if !self
.project_or_fail_acknowledged(
ProviderRunProjection::ModelEvent {
work_id: call.work_id.clone(),
event: AgentEvent::TextDelta { text },
},
project,
)
.await?
{
return Ok(());
}
}
AgentEvent::ReasoningDelta { text } => {
if !self
@@ -764,6 +776,18 @@ impl ProviderRunCoordinator {
return Ok(());
}
buffer.reasoning.push_str(&text);
if !self
.project_or_fail_acknowledged(
ProviderRunProjection::ModelEvent {
work_id: call.work_id.clone(),
event: AgentEvent::ReasoningDelta { text },
},
project,
)
.await?
{
return Ok(());
}
}
AgentEvent::ReasoningCompleted { text, signature }
if text.is_empty() && signature.is_none() =>
@@ -783,6 +807,18 @@ impl ProviderRunCoordinator {
}
buffer.reasoning.clone_from(&text);
buffer.reasoning_signature.clone_from(&signature);
if !self
.project_or_fail_acknowledged(
ProviderRunProjection::ModelEvent {
work_id: call.work_id.clone(),
event: AgentEvent::ReasoningCompleted { text, signature },
},
project,
)
.await?
{
return Ok(());
}
}
AgentEvent::Tool {
event: ToolEvent::Proposed { call: tool_call },
@@ -829,47 +865,8 @@ impl ProviderRunCoordinator {
{
return Ok(());
}
// A model can speak before its tool call arrives. Wait until the turn
// finishes so routine tool-turn narration never reaches the conversation.
// Keep it in the canonical turn for provider continuity.
if buffer.tool_calls.is_empty()
&& !matches!(
&reason,
StopReason::Cancelled | StopReason::ContextWindowExceeded
)
{
if !buffer.reasoning.is_empty()
&& !self
.project_or_fail_acknowledged(
ProviderRunProjection::ModelEvent {
work_id: call.work_id.clone(),
event: AgentEvent::ReasoningCompleted {
text: buffer.reasoning.clone(),
signature: buffer.reasoning_signature.clone(),
},
},
project,
)
.await?
{
return Ok(());
}
if !buffer.text.is_empty()
&& !self
.project_or_fail_acknowledged(
ProviderRunProjection::ModelEvent {
work_id: call.work_id.clone(),
event: AgentEvent::TextDelta {
text: buffer.text.clone(),
},
},
project,
)
.await?
{
return Ok(());
}
}
// Output is projected as it arrives. The buffer is only for canonical
// history; replaying it here would duplicate the streamed response.
if !self
.project_or_fail_acknowledged(
ProviderRunProjection::ModelTurnFinished {
app/src/ai/runtime/provider_run_coordinator_tests.rs
+92 -11
View File
@@ -20,6 +20,7 @@ type ScriptedTurn = Result<Vec<ScriptedEvent>, AgentError>;
struct ScriptedRuntime {
descriptor: RuntimeDescriptor,
turns: Mutex<VecDeque<ScriptedTurn>>,
events_consumed: Arc<AtomicUsize>,
requests: Mutex<Vec<TurnRequest>>,
}
@@ -37,6 +38,7 @@ impl ScriptedRuntime {
capabilities: RuntimeCapabilities::provider(),
},
turns: Mutex::new(turns.into()),
events_consumed: Arc::new(AtomicUsize::new(0)),
requests: Mutex::new(Vec::new()),
}
}
@@ -64,7 +66,10 @@ impl AgentRuntime for ScriptedRuntime {
.unwrap()
.pop_front()
.expect("scripted runtime ran out of turns")?;
Ok(Box::pin(futures::stream::iter(turn)))
let events_consumed = self.events_consumed.clone();
Ok(Box::pin(futures::stream::iter(turn).inspect(move |_| {
events_consumed.fetch_add(1, Ordering::Relaxed);
})))
}
}
@@ -302,7 +307,7 @@ fn answer_turn() -> ScriptedTurn {
}
#[tokio::test]
async fn tool_turn_narration_stays_in_history_but_out_of_visible_output() {
async fn tool_turn_text_is_streamed_and_retained_in_history() {
let runtime = Arc::new(ScriptedRuntime::new(vec![tool_turn()]));
let mut coordinator = coordinator(runtime);
let mut projections = Vec::new();
@@ -319,15 +324,91 @@ async fn tool_turn_narration_stays_in_history_but_out_of_visible_output() {
.transcript()
.iter()
.any(|message| format!("{:?}", message.content).contains("I will inspect it.")));
assert!(!projections.iter().any(|projection| matches!(
projection,
ProviderRunProjection::ModelEvent {
event: AgentEvent::TextDelta { .. }
| AgentEvent::ReasoningDelta { .. }
| AgentEvent::ReasoningCompleted { .. },
..
}
)));
let text: Vec<_> = projections
.iter()
.filter_map(|projection| match projection {
ProviderRunProjection::ModelEvent {
event: AgentEvent::TextDelta { text },
..
} => Some(text.as_str()),
_ => None,
})
.collect();
assert_eq!(text, ["I will inspect it."]);
}
#[tokio::test]
async fn text_and_reasoning_are_projected_before_reading_the_next_stream_event() {
let runtime = Arc::new(ScriptedRuntime::new(vec![Ok(vec![
started("request-streaming"),
Ok(AgentEvent::TextDelta {
text: "First ".to_string(),
}),
Ok(AgentEvent::ReasoningDelta {
text: "Thinking".to_string(),
}),
Ok(AgentEvent::ReasoningCompleted {
text: "Thinking".to_string(),
signature: Some("signature".to_string()),
}),
Ok(AgentEvent::TextDelta {
text: "second.".to_string(),
}),
usage(10, 4),
stopped(StopReason::Completed),
])]));
let events_consumed = runtime.events_consumed.clone();
let mut coordinator = coordinator(runtime);
let mut output = Vec::new();
let (_sender, control) = turn_control();
coordinator
.drive_until_blocked(control, |projection| {
if let ProviderRunProjection::ModelEvent { event, .. } = projection {
if matches!(
event,
AgentEvent::TextDelta { .. }
| AgentEvent::ReasoningDelta { .. }
| AgentEvent::ReasoningCompleted { .. }
) {
output.push((events_consumed.load(Ordering::Relaxed), event));
}
}
Ok(())
})
.await
.unwrap();
assert_eq!(
output,
vec![
(
2,
AgentEvent::TextDelta {
text: "First ".to_string()
}
),
(
3,
AgentEvent::ReasoningDelta {
text: "Thinking".to_string()
}
),
(
4,
AgentEvent::ReasoningCompleted {
text: "Thinking".to_string(),
signature: Some("signature".to_string()),
}
),
(
5,
AgentEvent::TextDelta {
text: "second.".to_string()
}
),
]
);
}
#[tokio::test]
app/src/ai/runtime/rig_request.rs
+1 -1
View File
@@ -832,7 +832,7 @@ fn build_system_prompt(
"You are Galaxy, a local-first software-engineering and terminal agent. Complete the user's task through inspection, implementation, and proportionate validation. Galaxy owns tool permissions and execution; use only the tools advertised in this request and treat every result as authoritative evidence.\n\n",
);
prompt.push_str(
"## Communication Style\nSpeak naturally, warmly, and directly, like a thoughtful collaborator working alongside the user. Work quietly by default while using tools; tool panes already show the activity. Do not send an introductory acknowledgement, plan, or generic progress message. Send prose during work only for a question needing the user's input, a consequential blocker, or a material finding the user needs to act on now. Tool results and compaction checkpoints continue the same request: resume work without another greeting or restatement. Supply the short purpose title requested by each tool schema. Do not narrate routine reads, repeat the plan, or end a turn with only a progress update when useful work can continue. In the final response, lead with the outcome and keep the handoff compact.\n\n",
"## Communication Style\nUse plain, direct language. During implementation, default to tool calls without accompanying prose; follow the Progress Updates contract below. Supply the short purpose title requested by each tool schema. Answer conversational questions naturally. When the task is complete, give one compact, self-contained result with the outcome, relevant validation, and any remaining blocker.\n\n",
);
prompt.push_str(
"## Execution Contract\nBefore another read or search, identify the missing fact it will resolve. Reuse findings already present in the conversation. Once enough evidence exists for the requested change, implement it and validate; additional inspection should resolve a specific uncertainty. Batch independent reads within the tool limits and wait for results before dependent work. Continue until the user's requested outcome is complete and validated. Do not stop at an intermediate analysis, plan, status update, or promise of future work, and do not ask the user to say \"continue\". After each tool result, choose and perform the next necessary step. Stop only when the request is fulfilled or a concrete blocker requires user input; identify that blocker explicitly.\n\n",