fix: preserve system item identity in live events

This commit is contained in:
2026-10-03 21:44:36 +09:00
parent 00c7e223b1
commit 3875c8e675
9 changed files with 120 additions and 13 deletions
+16 -1
View File
@@ -1106,6 +1106,9 @@ pub enum Event {
/// One event per `LogEntry::AnnotatedSystemItem` commit. Disk-side and /// One event per `LogEntry::AnnotatedSystemItem` commit. Disk-side and
/// wire-side are 1:1. /// wire-side are 1:1.
SystemItem { SystemItem {
/// Stable durable history identity. Synthetic compatibility events omit it.
#[serde(default, skip_serializing_if = "Option::is_none")]
entry_id: Option<String>,
#[cfg_attr(feature = "typescript", ts(type = "unknown"))] #[cfg_attr(feature = "typescript", ts(type = "unknown"))]
item: serde_json::Value, item: serde_json::Value,
}, },
@@ -2615,17 +2618,29 @@ mod tests {
#[test] #[test]
fn event_system_item_roundtrip() { fn event_system_item_roundtrip() {
let event = Event::SystemItem { let event = Event::SystemItem {
entry_id: Some("entry-1".to_string()),
item: serde_json::json!({"kind": "notification", "message": "hello"}), item: serde_json::json!({"kind": "notification", "message": "hello"}),
}; };
let json = serde_json::to_string(&event).unwrap(); let json = serde_json::to_string(&event).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed["event"], "system_item"); assert_eq!(parsed["event"], "system_item");
assert_eq!(parsed["data"]["entry_id"], "entry-1");
assert_eq!(parsed["data"]["item"]["kind"], "notification"); assert_eq!(parsed["data"]["item"]["kind"], "notification");
let decoded: Event = serde_json::from_str(&json).unwrap(); let decoded: Event = serde_json::from_str(&json).unwrap();
match decoded { match decoded {
Event::SystemItem { item } => assert_eq!(item["kind"], "notification"), Event::SystemItem { entry_id, item } => {
assert_eq!(entry_id.as_deref(), Some("entry-1"));
assert_eq!(item["kind"], "notification");
}
other => panic!("expected SystemItem, got {other:?}"), other => panic!("expected SystemItem, got {other:?}"),
} }
let legacy: Event = serde_json::from_value(serde_json::json!({
"event": "system_item",
"data": { "item": { "kind": "notification", "message": "legacy" } }
}))
.unwrap();
assert!(matches!(legacy, Event::SystemItem { entry_id: None, .. }));
} }
#[test] #[test]
+14 -4
View File
@@ -1213,7 +1213,7 @@ impl App {
self.push_durable_run_error(message); self.push_durable_run_error(message);
} }
} }
Event::SystemItem { item } => { Event::SystemItem { item, .. } => {
self.apply_system_item(&item); self.apply_system_item(&item);
self.assistant_streaming = false; self.assistant_streaming = false;
} }
@@ -4006,7 +4006,10 @@ mod completion_flow_tests {
"slug": "build", "slug": "build",
"body": "[Workflow /build]\nRun the build", "body": "[Workflow /build]\nRun the build",
}); });
app.handle_worker_event(Event::SystemItem { item }); app.handle_worker_event(Event::SystemItem {
entry_id: None,
item,
});
assert!(app.blocks.is_empty()); assert!(app.blocks.is_empty());
} }
@@ -4443,7 +4446,10 @@ mod completion_flow_tests {
"message": "hi", "message": "hi",
"body": "[Notification] hi", "body": "[Notification] hi",
}); });
app.handle_worker_event(Event::SystemItem { item }); app.handle_worker_event(Event::SystemItem {
entry_id: None,
item,
});
assert!(matches!( assert!(matches!(
app.blocks.as_slice(), app.blocks.as_slice(),
[Block::Notify { message }] if message == "hi" [Block::Notify { message }] if message == "hi"
@@ -4458,7 +4464,10 @@ mod completion_flow_tests {
"event": { "kind": "turn_ended", "worker_name": "child" }, "event": { "kind": "turn_ended", "worker_name": "child" },
"body": "[Notification] worker `child` finished a turn", "body": "[Notification] worker `child` finished a turn",
}); });
app.handle_worker_event(Event::SystemItem { item }); app.handle_worker_event(Event::SystemItem {
entry_id: None,
item,
});
assert_eq!(app.blocks.len(), 1); assert_eq!(app.blocks.len(), 1);
match &app.blocks[0] { match &app.blocks[0] {
Block::WorkerEvent { Block::WorkerEvent {
@@ -4832,6 +4841,7 @@ mod completion_flow_tests {
// Snapshot text injected through an active system item kind; legacy // Snapshot text injected through an active system item kind; legacy
// workflow items are intentionally ignored and must not carry active state. // workflow items are intentionally ignored and must not carry active state.
app.handle_worker_event(Event::SystemItem { app.handle_worker_event(Event::SystemItem {
entry_id: None,
item: serde_json::json!({ item: serde_json::json!({
"kind": "task_reminder", "kind": "task_reminder",
"body": snapshot, "body": snapshot,
+5 -2
View File
@@ -5467,6 +5467,7 @@ fn input_protocol_event(input: &WorkerInput) -> Option<protocol::Event> {
WorkerInputKind::Compact WorkerInputKind::Compact
| WorkerInputKind::ListRewindTargets | WorkerInputKind::ListRewindTargets
| WorkerInputKind::RegisterPeer => Some(protocol::Event::SystemItem { | WorkerInputKind::RegisterPeer => Some(protocol::Event::SystemItem {
entry_id: None,
item: serde_json::json!({ item: serde_json::json!({
"kind": "embedded_worker_command_input", "kind": "embedded_worker_command_input",
"command": input.kind, "command": input.kind,
@@ -8618,6 +8619,7 @@ mod tests {
.observe_worker_event( .observe_worker_event(
&detail.worker_ref, &detail.worker_ref,
protocol::Event::SystemItem { protocol::Event::SystemItem {
entry_id: Some("entry-1".to_string()),
item: serde_json::json!({ item: serde_json::json!({
"kind": "notification", "kind": "notification",
"message": "note", "message": "note",
@@ -8631,14 +8633,15 @@ mod tests {
.read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero()) .read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero())
.unwrap(); .unwrap();
assert_eq!(observations.len(), 1); assert_eq!(observations.len(), 1);
let protocol::Event::SystemItem { item } = &observations[0].payload else { let protocol::Event::SystemItem { entry_id, item } = &observations[0].payload else {
panic!("committed notification observation must be a system item"); panic!("committed notification observation must be a system item");
}; };
assert_eq!(entry_id.as_deref(), Some("entry-1"));
assert_eq!(item["kind"], "notification"); assert_eq!(item["kind"], "notification");
assert!(observations.iter().all(|observation| { assert!(observations.iter().all(|observation| {
!matches!( !matches!(
&observation.payload, &observation.payload,
protocol::Event::SystemItem { item } protocol::Event::SystemItem { item, .. }
if item["kind"] == "embedded_worker_notification" if item["kind"] == "embedded_worker_notification"
) )
})); }));
+39 -1
View File
@@ -88,8 +88,12 @@ pub fn live_log_entry_event(entry: LogEntry) -> Option<Event> {
.map(|entry| Event::SessionEntryCommitted { entry }) .map(|entry| Event::SessionEntryCommitted { entry })
} }
LogEntry::AnnotatedSystemItem { entry, .. } => { LogEntry::AnnotatedSystemItem { entry, .. } => {
let entry_id = Some(entry.metadata.entry_id.0.clone());
let value = serde_json::to_value(&entry.item).expect("SystemItem is Serialize"); let value = serde_json::to_value(&entry.item).expect("SystemItem is Serialize");
Some(Event::SystemItem { item: value }) Some(Event::SystemItem {
entry_id,
item: value,
})
} }
LogEntry::Invoke { trigger, .. } => Some(Event::InvokeStart { kind: trigger }), LogEntry::Invoke { trigger, .. } => Some(Event::InvokeStart { kind: trigger }),
other => { other => {
@@ -303,6 +307,40 @@ mod tests {
} }
} }
#[test]
fn system_item_log_entry_keeps_identity_in_live_event() {
let entry_id = session_store::LoggedSessionHistoryEntryId::new();
let event = live_log_entry_event(LogEntry::AnnotatedSystemItem {
ts: session_store::segment_log::now_millis(),
entry: session_store::LoggedSystemHistoryEntry {
item: session_store::SystemItem::ResidentSummaryRefresh {
body: "refresh".to_string(),
prompt_provenance: None,
},
metadata: session_store::LoggedSessionHistoryMetadata {
entry_id: entry_id.clone(),
origin: session_store::LoggedSessionHistoryOrigin::BackendInstruction {
operation_id: None,
},
derivation: None,
},
},
extensions: Vec::new(),
})
.expect("SystemItem must be live-relevant");
match event {
Event::SystemItem {
entry_id: live_entry_id,
item,
} => {
assert_eq!(live_entry_id.as_deref(), Some(entry_id.0.as_str()));
assert_eq!(item["kind"], "resident_summary_refresh");
}
other => panic!("expected SystemItem, got {other:?}"),
}
}
#[test] #[test]
fn user_input_log_entry_maps_to_user_message_event() { fn user_input_log_entry_maps_to_user_message_event() {
let segments = vec![protocol::Segment::text("hello from log")]; let segments = vec![protocol::Segment::text("hello from log")];
+1 -1
View File
@@ -2717,7 +2717,7 @@ async fn socket_worker_event_turn_ended_while_idle_auto_starts_turn() {
tokio::select! { tokio::select! {
event = reader.next::<Event>() => { event = reader.next::<Event>() => {
match event { match event {
Ok(Some(Event::SystemItem { ref item })) Ok(Some(Event::SystemItem { ref item, .. }))
if item.get("kind").and_then(|k| k.as_str()) == Some("worker_event") if item.get("kind").and_then(|k| k.as_str()) == Some("worker_event")
&& item && item
.pointer("/event/worker_name") .pointer("/event/worker_name")
File diff suppressed because one or more lines are too long
+5 -1
View File
@@ -367,7 +367,11 @@ export type Event = { "event": "submission_accepted", "data": { submission_reque
/** /**
* Stable durable history identity. Older mixed-version producers omit it. * Stable durable history identity. Older mixed-version producers omit it.
*/ */
entry_id?: string | null, segments: Array<Segment>, } } | { "event": "session_entry_committed", "data": { entry: SessionSnapshotEntry, } } | { "event": "system_item", "data": { item: unknown, } } | { "event": "invoke_start", "data": { kind: InvokeKind, } } | { "event": "turn_start", "data": { turn: number, } } | { "event": "turn_end", "data": { turn: number, result: TurnResult, } } | { "event": "llm_call_start", "data": { llm_call: number, } } | { "event": "llm_call_end", "data": { llm_call: number, } } | { "event": "llm_retry", "data": { llm_call: number, entry_id?: string | null, segments: Array<Segment>, } } | { "event": "session_entry_committed", "data": { entry: SessionSnapshotEntry, } } | { "event": "system_item", "data": {
/**
* Stable durable history identity. Synthetic compatibility events omit it.
*/
entry_id?: string | null, item: unknown, } } | { "event": "invoke_start", "data": { kind: InvokeKind, } } | { "event": "turn_start", "data": { turn: number, } } | { "event": "turn_end", "data": { turn: number, result: TurnResult, } } | { "event": "llm_call_start", "data": { llm_call: number, } } | { "event": "llm_call_end", "data": { llm_call: number, } } | { "event": "llm_retry", "data": { llm_call: number,
/** /**
* The attempt that just failed. 1 origin. * The attempt that just failed. 1 origin.
*/ */
@@ -3781,6 +3781,37 @@ Deno.test("committed history reconciles overlapping Read groups before counting
} }
}); });
Deno.test("live system items reconcile with committed history by entry identity", () => {
const item = {
kind: "resident_summary_refresh",
body: "Resident memory refresh",
};
const event: Event = {
event: "system_item",
data: { entry_id: "resident-refresh-1", item },
};
const current = projectConsole([
{ eventId: "live-1", event },
{ eventId: "live-retry", event },
]).lines;
assertEquals(current.length, 1);
assertEquals(current[0].entryId, "resident-refresh-1");
const history = projectSessionHistoryEntries([{
kind: "system_item",
entry_id: "resident-refresh-1",
timestamp: 1,
provenance: "backend_instruction",
item_kind: "resident_summary_refresh",
content: item.body,
data: item,
}], null);
const merged = mergeCommittedHistoryLines(history, current);
assertEquals(merged.length, 1);
assertEquals(merged[0].entryId, "resident-refresh-1");
assertEquals(merged[0].body, item.body);
});
Deno.test("committed history keeps identical run failures with distinct entry identities", () => { Deno.test("committed history keeps identical run failures with distinct entry identities", () => {
const message = "the same failure happened again"; const message = "the same failure happened again";
const runError = (entryId: string): SessionSnapshotEntry => ({ const runError = (entryId: string): SessionSnapshotEntry => ({
@@ -1067,10 +1067,16 @@ export function applyProtocolEvent(
} }
break; break;
} }
case "system_item": case "system_item": {
next.lines.push(systemItemLine(envelope.eventId, event.data.item)); const entryId = event.data.entry_id ?? undefined;
if (entryId && next.lines.some((line) => line.entryId === entryId)) break;
next.lines.push({
...systemItemLine(envelope.eventId, event.data.item),
entryId,
});
applyTaskSystemItem(next, event.data.item); applyTaskSystemItem(next, event.data.item);
break; break;
}
case "text_delta": case "text_delta":
appendStreaming( appendStreaming(
next, next,