fix(runner): retain unlabeled prefixes and complete tool JSON

Retain the published message identity when a provider begins labeling deltas. Preserve the original full-prefix oracle, separate subsequent message boundaries, and keep unknown JSON result fields inspectable. Record native, server and browser proof and the pending release gates.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-09 05:19:40 -05:00
1 parent 2d9e483cb9
commit 334db53750
10 files changed
+239 -32

No files matched your search

+4 -1
View File
@@ -108,7 +108,10 @@ extend the task's execution deadline.
- Assistant message identity v2 includes the provider message and provider turn.
A final snapshot replaces only its own deltas, retaining preceding commentary
and steered messages. Providers without message IDs retain the original
identity; existing PRP v1 events replay unchanged. This is an identity change
identity. If a provider first supplies an ID after an unlabelled prefix,
that prefix and the continuation retain one item identity and one final
snapshot. A later distinct native message gets its own item. Existing PRP v1
events replay unchanged. This is an identity change
within the current event contract, not a new wire field or transcript rewrite.
- Native questions publish a 65,536-character limit for text and custom
answers. The form, canonical response validation and bridge count UTF-16
@@ -72,7 +72,7 @@ pub enum AcpxProviderStateEvent {
SemanticResult(AcpxSemanticResult),
AssistantMessage {
turn_id: String,
message_id: Option<String>,
provider_item_id: Option<String>,
text: String,
},
TurnTerminal {
@@ -110,6 +110,7 @@ pub struct AcpxProviderState {
plan_revision: u64,
assistant_text: String,
assistant_message_id: Option<String>,
assistant_item_id: Option<String>,
pi_assistant_message: Option<String>,
pi_assistant_message_ids: BTreeSet<String>,
pending_tools: BTreeMap<String, AcpxPendingTool>,
@@ -129,6 +130,7 @@ impl AcpxProviderState {
plan_revision: 0,
assistant_text: String::new(),
assistant_message_id: None,
assistant_item_id: None,
pi_assistant_message: None,
pi_assistant_message_ids: BTreeSet::new(),
pending_tools: BTreeMap::new(),
@@ -223,6 +225,7 @@ impl AcpxProviderState {
self.plan_revision = 0;
self.assistant_text.clear();
self.assistant_message_id = None;
self.assistant_item_id = None;
self.pi_assistant_message = None;
self.pi_assistant_message_ids.clear();
self.semantic_result = None;
@@ -389,13 +392,14 @@ impl AcpxProviderState {
if status == AcpxTurnStatus::Completed && !self.assistant_text.is_empty() {
events.push(AcpxProviderStateEvent::AssistantMessage {
turn_id: turn_id.clone(),
message_id: self.assistant_message_id.take(),
provider_item_id: self.assistant_item_id.take(),
text: std::mem::take(&mut self.assistant_text),
});
} else {
self.assistant_text.clear();
}
self.assistant_message_id = None;
self.assistant_item_id = None;
self.pi_assistant_message = None;
self.pi_assistant_message_ids.clear();
events.push(AcpxProviderStateEvent::TurnTerminal {
@@ -529,7 +533,7 @@ impl AcpxProviderState {
let starts_new_message = provider_message_id.as_ref().is_some_and(|message_id| {
self.assistant_message_id
.as_ref()
.is_none_or(|current| current != message_id)
.is_some_and(|current| current != message_id)
});
let raw_text = payload
.get("text")
@@ -594,6 +598,7 @@ impl AcpxProviderState {
// the task UI and can promote intermediate prose as final output.
if starts_new_message || native_boundary_clears {
self.assistant_text.clear();
self.assistant_item_id = None;
}
if self
.assistant_text
@@ -662,7 +667,7 @@ impl AcpxProviderState {
.as_deref()
.expect("a decoded runtime event has a turn binding");
let fallback_item_id = format!("acpx-event-{}", event.sequence);
let events = normalize_acpx_runtime_event(
let mut events = normalize_acpx_runtime_event(
kind,
&payload,
tool_operation,
@@ -670,6 +675,26 @@ impl AcpxProviderState {
turn_id,
self.provider_requests,
);
if kind == AcpxRuntimeEventKind::TextDelta
&& payload.get("piMessageHistory").and_then(Value::as_bool) != Some(true)
{
// A provider can begin identifying a message after its first delta.
// Adopt that native ID for boundary detection, but retain the item's
// already-published identity so its prefix and final snapshot agree.
for activity in &mut events {
let item = activity
.payload
.as_object_mut()
.expect("a normalized ACPX text delta is an object");
let item_id = item
.get("itemId")
.and_then(Value::as_str)
.expect("a normalized ACPX text delta has an item identity")
.to_owned();
let retained_id = self.assistant_item_id.get_or_insert(item_id);
item.insert("itemId".to_owned(), Value::String(retained_id.clone()));
}
}
if let Some(revision) = next_plan_revision {
self.plan_revision = revision;
}
@@ -378,13 +378,10 @@ pub fn project_acpx_state_event(
}
AcpxProviderStateEvent::AssistantMessage {
turn_id,
message_id,
provider_item_id,
text,
} => {
require_projected_turn(context, turn_id)?;
let provider_item_id = message_id
.as_deref()
.map(|message_id| acpx_message_item_id(message_id, turn_id, "assistant-message"));
one(
"item.completed",
EventPriority::P1,
@@ -151,7 +151,7 @@ fn keeps_durable_correlation_separate_from_the_active_provider_turn() {
&context,
&AcpxProviderStateEvent::AssistantMessage {
turn_id: "provider-turn-1".to_owned(),
message_id: None,
provider_item_id: None,
text: "Done".to_owned(),
},
)
@@ -339,7 +339,7 @@ fn projects_assistant_terminal_and_diagnostic_events_fail_closed() {
let assistant = project(AcpxProviderStateEvent::AssistantMessage {
turn_id: "turn-1".to_owned(),
message_id: Some("opaque-provider-message".to_owned()),
provider_item_id: Some("opaque-provider-message".to_owned()),
text: "Done".to_owned(),
});
assert_eq!(assistant[0].event_type, "item.completed");
@@ -629,7 +629,7 @@ fn runtime_request_projection_preserves_durable_identity_boundaries() {
fn recovery_preserves_the_preceding_provider_turn_answer() {
let first = project(AcpxProviderStateEvent::AssistantMessage {
turn_id: "turn-1".to_owned(),
message_id: None,
provider_item_id: None,
text: "Useful answer".to_owned(),
});
let mut recovery = context();
@@ -638,7 +638,7 @@ fn recovery_preserves_the_preceding_provider_turn_answer() {
&recovery,
&AcpxProviderStateEvent::AssistantMessage {
turn_id: "recovery-turn".to_owned(),
message_id: None,
provider_item_id: None,
text: "Recovery update".to_owned(),
},
)
@@ -673,6 +673,7 @@ fn assistant_message_identity_matches_its_snapshot_and_is_scoped_to_the_provider
1,
)
.remove(0);
let provider_item_id = activity.payload["itemId"].as_str().unwrap().to_owned();
let streamed = project(AcpxProviderStateEvent::Activity(activity.clone()));
assert_eq!(
streamed,
@@ -680,7 +681,7 @@ fn assistant_message_identity_matches_its_snapshot_and_is_scoped_to_the_provider
);
let final_message = project(AcpxProviderStateEvent::AssistantMessage {
turn_id: "turn-1".to_owned(),
message_id: message_id.map(str::to_owned),
provider_item_id: Some(provider_item_id.clone()),
text: "Done".to_owned(),
});
assert_eq!(
@@ -697,7 +698,7 @@ fn assistant_message_identity_matches_its_snapshot_and_is_scoped_to_the_provider
&recovery,
&AcpxProviderStateEvent::AssistantMessage {
turn_id: "recovery-turn".to_owned(),
message_id: message_id.map(str::to_owned),
provider_item_id: Some(provider_item_id.clone()),
text: "Recovered".to_owned(),
},
)
@@ -70,8 +70,9 @@ fn accumulates_assistant_text_preserves_reasoning_and_flushes_before_terminal()
&& event.payload["text"] == "Checking the result."
));
let mut assistant_item_id = None;
for (sequence, text) in [(3, "Hello "), (4, "world")] {
state
let emitted = state
.accept_event(&event(
sequence,
GeneratedAcpxSidecarEventType::RuntimeEvent,
@@ -79,6 +80,14 @@ fn accumulates_assistant_text_preserves_reasoning_and_flushes_before_terminal()
json!({"type":"text_delta","text":text}),
))
.unwrap();
let AcpxProviderStateEvent::Activity(activity) = &emitted[0] else {
panic!("an assistant delta must emit progress");
};
let item_id = activity.payload["itemId"].as_str().unwrap().to_owned();
if let Some(retained) = &assistant_item_id {
assert_eq!(retained, &item_id);
}
assistant_item_id = Some(item_id);
}
let terminal = state
.accept_event(&event(
@@ -90,8 +99,8 @@ fn accumulates_assistant_text_preserves_reasoning_and_flushes_before_terminal()
.unwrap();
assert!(matches!(
&terminal[0],
AcpxProviderStateEvent::AssistantMessage { turn_id, message_id, text }
if turn_id == "turn-1" && message_id.is_none() && text == "Hello world"
AcpxProviderStateEvent::AssistantMessage { turn_id, provider_item_id, text }
if turn_id == "turn-1" && provider_item_id == &assistant_item_id && text == "Hello world"
));
assert!(matches!(
&terminal[1],
@@ -183,8 +192,8 @@ fn promotes_only_the_latest_provider_message_as_the_terminal_reply() {
.unwrap();
assert!(matches!(
&terminal[0],
AcpxProviderStateEvent::AssistantMessage { message_id, text, .. }
if message_id.as_deref() == Some("message-2")
AcpxProviderStateEvent::AssistantMessage { provider_item_id, text, .. }
if provider_item_id.as_deref() == Some("message-2")
&& text == "Second paragraph.\n\nStill final."
));
let projected_final = project_acpx_state_event(&context, &terminal[0]).unwrap();
@@ -227,7 +236,7 @@ fn an_unlabelled_continuation_keeps_the_current_message_identity() {
project_acpx_state_event(&context, &emitted[0]).unwrap()[0].payload["itemId"].clone(),
);
}
assert_ne!(ids[0], ids[1]);
assert_eq!(ids[0], ids[1]);
assert_eq!(ids[1], ids[2]);
let terminal = state
.accept_event(&event(
@@ -239,7 +248,62 @@ fn an_unlabelled_continuation_keeps_the_current_message_identity() {
.unwrap();
let final_message = project_acpx_state_event(&context, &terminal[0]).unwrap();
assert_eq!(final_message[0].payload["itemId"], ids[2]);
assert_eq!(final_message[0].payload["text"], "Hello world");
assert_eq!(
final_message[0].payload["text"],
"Unlabelled progress.Hello world"
);
}
#[test]
fn an_identified_message_after_an_adopted_prefix_gets_its_own_item() {
let context = AcpxEventProjectionContext {
run_id: "run-1".to_owned(),
normalized_session_id: "session-1".to_owned(),
turn_id: "turn-1".to_owned(),
provider_turn_id: None,
item_id: "item-1".to_owned(),
};
let mut state = AcpxProviderState::new("run-1").unwrap();
state.begin_turn("turn-1").unwrap();
let mut ids = Vec::new();
for (sequence, payload) in [
(1, json!({"type":"text_delta","text":"Preface. "})),
(
2,
json!({"type":"text_delta","messageId":"first","text":"Progress."}),
),
(
3,
json!({"type":"text_delta","messageId":"second","text":"Final "}),
),
(4, json!({"type":"text_delta","text":"response."})),
] {
let emitted = state
.accept_event(&event(
sequence,
GeneratedAcpxSidecarEventType::RuntimeEvent,
Some("turn-1"),
payload,
))
.unwrap();
ids.push(
project_acpx_state_event(&context, &emitted[0]).unwrap()[0].payload["itemId"].clone(),
);
}
assert_eq!(ids[0], ids[1]);
assert_ne!(ids[1], ids[2]);
assert_eq!(ids[2], ids[3]);
let terminal = state
.accept_event(&event(
5,
GeneratedAcpxSidecarEventType::RuntimeTurnTerminal,
Some("turn-1"),
json!({"status":"completed"}),
))
.unwrap();
let final_message = project_acpx_state_event(&context, &terminal[0]).unwrap();
assert_eq!(final_message[0].payload["itemId"], ids[3]);
assert_eq!(final_message[0].payload["text"], "Final response.");
}
#[test]
@@ -333,9 +397,16 @@ fn pi_native_boundary_order_and_terminal_completeness_fail_closed() {
#[test]
fn preserves_an_idless_prefix_when_the_provider_begins_identifying_deltas() {
let context = AcpxEventProjectionContext {
run_id: "run-1".to_owned(),
normalized_session_id: "session-1".to_owned(),
turn_id: "turn-1".to_owned(),
provider_turn_id: None,
item_id: "item-1".to_owned(),
};
let mut state = AcpxProviderState::new("run-1").unwrap();
state.begin_turn("turn-1").unwrap();
state
let prefix = state
.accept_event(&event(
1,
GeneratedAcpxSidecarEventType::RuntimeEvent,
@@ -343,7 +414,7 @@ fn preserves_an_idless_prefix_when_the_provider_begins_identifying_deltas() {
json!({"type":"text_delta","text":"Preface.\n\n"}),
))
.unwrap();
state
let identified = state
.accept_event(&event(
2,
GeneratedAcpxSidecarEventType::RuntimeEvent,
@@ -364,6 +435,14 @@ fn preserves_an_idless_prefix_when_the_provider_begins_identifying_deltas() {
AcpxProviderStateEvent::AssistantMessage { text, .. }
if text == "Preface.\n\nFinal response."
));
let prefix = project_acpx_state_event(&context, &prefix[0]).unwrap();
let identified = project_acpx_state_event(&context, &identified[0]).unwrap();
let final_message = project_acpx_state_event(&context, &terminal[0]).unwrap();
assert_eq!(prefix[0].payload["itemId"], identified[0].payload["itemId"]);
assert_eq!(
prefix[0].payload["itemId"],
final_message[0].payload["itemId"]
);
}
#[test]
@@ -256,7 +256,7 @@ fn acpx_projection_matches_shared_plan_question_final_and_terminal_identity() {
},
&AcpxProviderStateEvent::AssistantMessage {
turn_id: context.turn_id.clone(),
message_id: None,
provider_item_id: None,
text: expected_final["text"]
.as_str()
.expect("final text")