Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

chore: merge develop into hare/develop

Browse Source
This commit is contained in:
Hare
2026-08-30 13:19:24 +09:00
parent 0cae4fd05c adb684a6bf
commit 74139aeb7e
44 changed files with 2391 additions and 979 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+18 -31
View File
@@ -84,10 +84,7 @@ impl WorkerHandle {
(entries, entry_rx, in_flight)
};
let event = Event::Snapshot {
entries: entries
.into_iter()
.map(|entry| serde_json::to_value(entry).expect("log entry serializes"))
.collect(),
session: session_store::public_snapshot::project_current_session_snapshot(&entries),
greeting: self.shared_state.greeting.clone(),
status: self.shared_state.get_status(),
in_flight,
@@ -634,8 +631,8 @@ fn protocol_command_status(status: WorkdirCommandStatus) -> ProtocolCommandStatu
///
/// `Worker::wire_history_persistence` is called separately to wire the
/// per-item history commit callback so every assistant / tool item
/// landing in `worker.history` becomes a singular `LogEntry::AssistantItem`
/// / `ToolResult` commit through the sync writer.
/// landing in `worker.history` becomes a singular `LogEntry::AnnotatedAssistantItem`
/// / `AnnotatedToolResult` commit through the sync writer.
pub(crate) fn wire_event_bridges_on_engine<C, St>(
worker: &mut Worker<C, St>,
event_tx: &broadcast::Sender<Event>,
@@ -1320,7 +1317,7 @@ async fn controller_loop<C, St>(
}
// Stage the run without a speculative user-message echo.
// `Worker::run` validates the input, commits
// `LogEntry::UserInput`, and the session-log sink turns that
// `LogEntry::AnnotatedUserInput`, and the session-log sink turns that
// committed entry into the live `Event::UserMessage`. That
// keeps every client ordered against `SegmentStart` replay and
// makes persisted history the single source of visible user
@@ -1346,7 +1343,7 @@ async fn controller_loop<C, St>(
Method::Notify { message, auto_run } => {
// Client-side live echo is delivered as `Event::SystemItem`
// once the interceptor commits the corresponding
// `LogEntry::SystemItem` entry — drained out of the
// `LogEntry::AnnotatedSystemItem` entry — drained out of the
// notify buffer + broadcast through the sink. No
// separate echo here.
worker.push_notify(message, auto_run);
@@ -1874,28 +1871,16 @@ where
St: Store,
{
match worker.rewind_to(target, expected_head_entries) {
Ok(applied) => match applied
.entries
.into_iter()
.map(serde_json::to_value)
.collect::<Result<Vec<_>, _>>()
{
Ok(entries) => {
let _ = event_tx.send(Event::RewindApplied {
entries,
input: applied.input,
summary: applied.summary,
});
true
}
Err(error) => {
let _ = event_tx.send(Event::Error {
code: ErrorCode::Internal,
message: format!("failed to encode rewind snapshot: {error}"),
});
false
}
},
Ok(applied) => {
let session =
session_store::public_snapshot::project_current_session_snapshot(&applied.entries);
let _ = event_tx.send(Event::RewindApplied {
session,
input: applied.input,
summary: applied.summary,
});
true
}
Err(err) => {
let _ = event_tx.send(Event::Error {
code: ErrorCode::InvalidRequest,
@@ -2101,7 +2086,9 @@ mod tests {
let mut writer = JsonLineWriter::new(w);
writer
.write(&Event::Snapshot {
entries: Vec::new(),
session: protocol::SessionSnapshot {
entries: Vec::new(),
},
greeting: protocol::Greeting {
worker_name: "parent".into(),
cwd: "/tmp".into(),
crates/worker/src/discovery.rs
+18 -6
View File
@@ -1481,7 +1481,9 @@ mod tests {
let mut writer = JsonLineWriter::new(stream);
writer
.write(&Event::Snapshot {
entries: Vec::new(),
session: protocol::SessionSnapshot {
entries: Vec::new(),
},
greeting: protocol::Greeting {
worker_name: "target".into(),
cwd: "/tmp".into(),
@@ -1514,7 +1516,9 @@ mod tests {
.unwrap();
writer
.write(&Event::Snapshot {
entries: Vec::new(),
session: protocol::SessionSnapshot {
entries: Vec::new(),
},
greeting: protocol::Greeting {
worker_name: "target".into(),
cwd: "/tmp".into(),
@@ -1603,7 +1607,9 @@ mod tests {
let mut writer = JsonLineWriter::new(stream);
writer
.write(&Event::Snapshot {
entries: Vec::new(),
session: protocol::SessionSnapshot {
entries: Vec::new(),
},
greeting: protocol::Greeting {
worker_name: "target".into(),
cwd: "/tmp".into(),
@@ -1627,7 +1633,9 @@ mod tests {
let mut writer = JsonLineWriter::new(writer_half);
writer
.write(&Event::Snapshot {
entries: Vec::new(),
session: protocol::SessionSnapshot {
entries: Vec::new(),
},
greeting: protocol::Greeting {
worker_name: "target".into(),
cwd: "/tmp".into(),
@@ -1729,7 +1737,9 @@ mod tests {
.unwrap();
writer
.write(&Event::Snapshot {
entries: Vec::new(),
session: protocol::SessionSnapshot {
entries: Vec::new(),
},
greeting: protocol::Greeting {
worker_name: "alerted".into(),
cwd: "/tmp".into(),
@@ -1779,7 +1789,9 @@ mod tests {
let mut writer = JsonLineWriter::new(stream);
let _ = writer
.write(&Event::Snapshot {
entries: Vec::new(),
session: protocol::SessionSnapshot {
entries: Vec::new(),
},
greeting: protocol::Greeting {
worker_name: "child-live".into(),
cwd: "/tmp".into(),
crates/worker/src/feature/builtin/flow_transition.rs
+4 -1
View File
@@ -963,9 +963,12 @@ permission = "read"
.append(
session_id,
segment_id,
&LogEntry::UserInput {
&LogEntry::AnnotatedUserInput {
ts: 1,
extensions: vec![],
history: vec![crate::session_history::test_logged_history_entry(
agen::Item::user_message("verify current Flow conditions"),
)],
segments: vec![Segment::Text {
content: "verify current Flow conditions".into(),
}],
crates/worker/src/feature/builtin/memory_extract.rs
+23 -54
View File
@@ -185,12 +185,9 @@ impl Tool for StageMemoryCandidateTool {
})?);
}
if matches!(params.kind, CandidateKind::Preference)
&& entries.iter().any(|entry| {
!matches!(
entry.origin,
crate::WorkerHistoryProvenance::HumanInput { .. }
)
})
&& entries
.iter()
.any(|entry| !matches!(entry.origin, protocol::SessionEntryProvenance::HumanInput))
{
return Err(ToolError::InvalidArgument(
"preference candidates require exclusively HumanInput evidence; model, Worker, Flow, backend, derived, and legacy-unknown origins are not preference authority"
@@ -324,10 +321,23 @@ fn evidence_kind(entry: &SessionEntryEvidence) -> EvidenceKind {
}
}
fn evidence_origin(origin: &crate::WorkerHistoryProvenance) -> EvidenceOrigin {
use crate::WorkerHistoryProvenance as Origin;
let mut evidence = EvidenceOrigin {
kind: EvidenceOriginKind::LegacyUnknown,
fn evidence_origin(origin: &protocol::SessionEntryProvenance) -> EvidenceOrigin {
use protocol::SessionEntryProvenance as Origin;
let kind = match origin {
Origin::HumanInput => EvidenceOriginKind::HumanInput,
Origin::WorkerInput => EvidenceOriginKind::WorkerInput,
Origin::FlowInstruction => EvidenceOriginKind::FlowInstruction,
Origin::BackendInstruction => EvidenceOriginKind::BackendInstruction,
Origin::ModelOutput => EvidenceOriginKind::ModelOutput,
Origin::ToolOutput => EvidenceOriginKind::ToolOutput,
Origin::DerivedSummary => EvidenceOriginKind::DerivedSummary,
Origin::LegacyUnknown => EvidenceOriginKind::LegacyUnknown,
};
EvidenceOrigin {
kind,
// The public SessionSnapshot intentionally excludes account, Worker,
// Runtime, and Flow internals. Preserve the authenticated origin class
// without inventing missing control-plane identity fields.
account_id: None,
workspace_id: None,
runtime_id: None,
@@ -335,46 +345,7 @@ fn evidence_origin(origin: &crate::WorkerHistoryProvenance) -> EvidenceOrigin {
flow_selector: None,
flow_definition_id: None,
flow_definition_revision: None,
};
match origin {
Origin::HumanInput { account_id } => {
evidence.kind = EvidenceOriginKind::HumanInput;
evidence.account_id = Some(account_id.clone());
}
Origin::WorkerInput { actor } => {
evidence.kind = EvidenceOriginKind::WorkerInput;
evidence.workspace_id = actor.workspace_id.clone();
evidence.runtime_id = actor.runtime_id.clone();
evidence.worker_id = Some(actor.worker_id.clone());
}
Origin::FlowInstruction {
selector,
definition_id,
definition_revision,
..
} => {
evidence.kind = EvidenceOriginKind::FlowInstruction;
evidence.flow_selector = Some(selector.clone());
evidence.flow_definition_id = Some(definition_id.clone());
evidence.flow_definition_revision = Some(*definition_revision);
}
Origin::BackendInstruction { .. } => evidence.kind = EvidenceOriginKind::BackendInstruction,
Origin::ModelOutput { worker } => {
evidence.kind = EvidenceOriginKind::ModelOutput;
evidence.workspace_id = worker.workspace_id.clone();
evidence.runtime_id = worker.runtime_id.clone();
evidence.worker_id = Some(worker.worker_id.clone());
}
Origin::ToolOutput { worker } => {
evidence.kind = EvidenceOriginKind::ToolOutput;
evidence.workspace_id = worker.workspace_id.clone();
evidence.runtime_id = worker.runtime_id.clone();
evidence.worker_id = Some(worker.worker_id.clone());
}
Origin::DerivedSummary => evidence.kind = EvidenceOriginKind::DerivedSummary,
Origin::LegacyUnknown => evidence.kind = EvidenceOriginKind::LegacyUnknown,
}
evidence
}
fn staging_evidence(entry: &SessionEntryEvidence) -> StagingEvidence {
@@ -502,12 +473,10 @@ mod tests {
}
#[test]
fn human_origin_projects_account_authority_into_evidence() {
let origin = evidence_origin(&crate::WorkerHistoryProvenance::HumanInput {
account_id: "account-1".into(),
});
fn public_human_origin_preserves_class_without_inventing_account_authority() {
let origin = evidence_origin(&protocol::SessionEntryProvenance::HumanInput);
assert_eq!(origin.kind, EvidenceOriginKind::HumanInput);
assert_eq!(origin.account_id.as_deref(), Some("account-1"));
assert_eq!(origin.account_id, None);
}
#[test]
crates/worker/src/feature/builtin/worker_observation.rs
+43 -48
View File
@@ -6,7 +6,7 @@ use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
use async_trait::async_trait;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use session_store::{LogEntry, collect_state};
use session_store::LogEntry;
use super::manage_worker::{WORKER_CONTROL_SERVICE_ID, WorkerControlService};
use crate::feature::{
@@ -61,7 +61,7 @@ pub struct WorkerObservationSubject {
#[derive(Debug, Clone)]
pub struct WorkerSessionCapture {
pub segment_id: String,
pub entries: Vec<agen::HistoryEntry<crate::SessionHistoryMetadata>>,
pub session: protocol::SessionSnapshot,
}
impl WorkerSessionCapture {
@@ -69,17 +69,9 @@ impl WorkerSessionCapture {
segment_id: impl Into<String>,
log_entries: &[LogEntry],
) -> Result<Self, String> {
let segment_id = segment_id.into();
let state = collect_state(log_entries);
let parsed_segment_id = segment_id.parse().unwrap_or_default();
let entries = crate::session_history::restore_history_entries(
state.session_id.unwrap_or_default(),
parsed_segment_id,
log_entries,
)?;
Ok(Self {
segment_id,
entries,
segment_id: segment_id.into(),
session: session_store::public_snapshot::project_current_session_snapshot(log_entries),
})
}
}
@@ -115,7 +107,7 @@ struct WorkspaceWorkerObservationListResponse {
#[derive(Debug, Deserialize)]
struct WorkspaceWorkerObservationCaptureResponse {
segment_id: String,
entries: Vec<serde_json::Value>,
session: protocol::SessionSnapshot,
}
pub struct WorkspaceClientWorkerObservationProvider {
@@ -173,26 +165,9 @@ impl WorkerObservationProvider for WorkspaceClientWorkerObservationProvider {
let body = workspace_response_body(response)?;
let response = serde_json::from_str::<WorkspaceWorkerObservationCaptureResponse>(&body)
.map_err(|error| WorkerObservationError::Unavailable(error.to_string()))?;
let entries = response
.entries
.into_iter()
.map(|entry| {
serde_json::from_value(entry)
.map_err(|error| WorkerObservationError::Unavailable(error.to_string()))
})
.collect::<Result<Vec<session_store::LogEntry>, _>>()?;
let state = collect_state(&entries);
let segment_id = response.segment_id;
let parsed_segment_id = segment_id.parse().unwrap_or_default();
let typed_entries = crate::session_history::restore_history_entries(
state.session_id.unwrap_or_default(),
parsed_segment_id,
&entries,
)
.map_err(WorkerObservationError::Unavailable)?;
Ok(WorkerSessionCapture {
segment_id,
entries: typed_entries,
segment_id: response.segment_id,
session: response.session,
})
}
}
@@ -420,16 +395,9 @@ impl WorkerObservationProvider for SpawnedSubWorkerObservationProvider {
.get_internal(name)
.ok_or(WorkerObservationError::NotFound)?;
let entries = record.session.entries();
let state = collect_state(&entries);
let typed_entries = crate::session_history::restore_history_entries(
state.session_id.unwrap_or_default(),
Default::default(),
&entries,
)
.map_err(WorkerObservationError::Unavailable)?;
Ok(WorkerSessionCapture {
segment_id: format!("subworker:{name}"),
entries: typed_entries,
session: session_store::public_snapshot::project_current_session_snapshot(&entries),
})
}
}
@@ -699,9 +667,9 @@ async fn latest_view(
.capture_worker_session(subject)
.await
.map_err(tool_error)?;
Ok(SessionCapture::from_history_entries(
Ok(SessionCapture::from_session_snapshot(
capture.segment_id,
capture.entries,
capture.session,
))
}
@@ -799,16 +767,43 @@ mod tests {
.clone()
.into_iter()
.enumerate()
.map(|(index, item)| {
let mut metadata = crate::SessionHistoryMetadata::legacy_unknown();
metadata.entry_id =
session_store::LoggedSessionHistoryEntryId(format!("fake-{index:08}"));
agen::HistoryEntry::new(item, metadata)
.filter_map(|(index, item)| {
let data = match item {
Item::Message { role, content, .. } => {
let role = match role {
Role::User => protocol::SessionMessageRole::User,
Role::Assistant => protocol::SessionMessageRole::Assistant,
Role::System => return None,
};
protocol::SessionSnapshotEntryData::Message {
role,
content: content
.into_iter()
.map(|part| match part {
agen::ContentPart::Text { text } => {
protocol::SessionContentPart::Text { text }
}
agen::ContentPart::Refusal { refusal } => {
protocol::SessionContentPart::Refusal { refusal }
}
})
.collect(),
}
}
_ => return None,
};
Some(protocol::SessionSnapshotEntry {
entry_id: format!("fake-{index:08}"),
timestamp: index as u64,
provenance: protocol::SessionEntryProvenance::LegacyUnknown,
derived_from: Vec::new(),
data,
})
})
.collect();
Ok(WorkerSessionCapture {
segment_id: "segment".to_string(),
entries,
session: protocol::SessionSnapshot { entries },
})
}
}
crates/worker/src/hook.rs
+1 -1
View File
@@ -161,7 +161,7 @@ impl From<HookTurnEndAction> for TurnEndAction {
///
/// Hook code can use this handle only when the Worker host includes it in an
/// event-specific context. The handle queues typed requests; the host drains the
/// queue, commits each entry through `LogEntry::SystemItem`, and only then makes
/// queue, commits each entry through `LogEntry::AnnotatedSystemItem`, and only then makes
/// the matching system message visible to the model. It deliberately exposes no
/// raw `agen::Item`, history writer, event sender, `Worker`, `Engine`, or
/// notification buffer.
crates/worker/src/in_flight.rs
+5 -5
View File
@@ -539,9 +539,9 @@ mod tests {
text: "done".into(),
}],
};
let assistant_entry = LogEntry::AssistantItem {
let assistant_entry = LogEntry::AnnotatedAssistantItem {
ts: 1,
item: assistant_item.clone(),
entry: crate::session_history::test_logged_history_entry(assistant_item.clone()),
};
let in_flight_guard = in_flight.snapshot_guard();
@@ -593,9 +593,9 @@ mod tests {
text: "done".into(),
}],
};
let assistant_entry = LogEntry::AssistantItem {
let assistant_entry = LogEntry::AnnotatedAssistantItem {
ts: 1,
item: assistant_item.clone(),
entry: crate::session_history::test_logged_history_entry(assistant_item.clone()),
};
in_flight.clear_for_committed_item_then(&assistant_item, || {
@@ -608,7 +608,7 @@ mod tests {
assert!(matches!(
entries_snapshot.as_slice(),
[LogEntry::AssistantItem { item, .. }] if item == &assistant_item
[LogEntry::AnnotatedAssistantItem { entry, .. }] if entry.item == assistant_item
));
assert!(in_flight_snapshot.is_empty());
}
crates/worker/src/internal_worker.rs
+2 -2
View File
@@ -335,7 +335,7 @@ enum InternalWorkerSessionCommand {
/// task; protocol access is consumed only by the owning parent registry.
#[derive(Debug, Clone)]
pub(crate) struct InternalWorkerSessionSnapshot {
pub entries: Vec<LogEntry>,
pub session: protocol::SessionSnapshot,
pub status: WorkerStatus,
pub error: Option<String>,
pub in_flight: InFlightSnapshot,
@@ -402,7 +402,7 @@ impl InternalWorkerSessionHandle {
(entries, snapshot_from_guard(&guard))
};
InternalWorkerSessionSnapshot {
entries,
session: session_store::public_snapshot::project_current_session_snapshot(&entries),
status: match self.status() {
InternalWorkerSessionStatus::Running => WorkerStatus::Running,
InternalWorkerSessionStatus::Paused => WorkerStatus::Paused,
crates/worker/src/ipc/interceptor.rs
+3 -4
View File
@@ -60,7 +60,7 @@ pub(crate) struct WorkerInterceptor {
pending_notifies: NotifyBuffer,
/// Submit-scoped stash of resolver-produced typed system items.
/// Drained inside `on_prompt_submit`, committed as
/// `LogEntry::SystemItem` entries through `log_writer`, and
/// `LogEntry::AnnotatedSystemItem` entries through `log_writer`, and
/// returned to the worker as `Item::system_message` via
/// `PromptAction::ContinueWith`. Populated by `Worker::run`
/// immediately before handing off to the worker.
@@ -71,7 +71,7 @@ pub(crate) struct WorkerInterceptor {
/// Workspace scope associated with Prompt projection provenance.
prompt_workspace_id: Option<String>,
/// Type-erased commit handle. The interceptor uses it to commit
/// `LogEntry::SystemItem` entries directly (sync) before
/// `LogEntry::AnnotatedSystemItem` entries directly (sync) before
/// returning the corresponding `Item::system_message`s up to the
/// worker. `None` in tests / `Worker::new` paths where no writer is
/// attached.
@@ -142,7 +142,7 @@ impl WorkerInterceptor {
self
}
/// Commit each `SystemItem` as its own `LogEntry::SystemItem`
/// Commit each `SystemItem` as its own `LogEntry::AnnotatedSystemItem`
/// entry through the attached writer (no-op when no writer is
/// wired). Sync — writes complete before the matching
/// `Item::system_message`s reach the worker via
@@ -540,7 +540,6 @@ mod tests {
entry: session_store::LogEntry,
) -> Result<(), session_store::StoreError> {
let item = match entry {
session_store::LogEntry::SystemItem { item, .. } => Some(item),
session_store::LogEntry::AnnotatedSystemItem { entry, .. } => Some(entry.item),
_ => None,
};
crates/worker/src/ipc/notify_buffer.rs
+1 -1
View File
@@ -5,7 +5,7 @@
//! `WorkerInterceptor::pending_history_appends`, which the Engine calls
//! at the head of each turn loop iteration. The drain renders each
//! pending entry into a typed `SystemItem` (with the `notify_wrapper`
//! prompt applied), commits a `LogEntry::SystemItem` per entry through
//! prompt applied), commits a `LogEntry::AnnotatedSystemItem` per entry through
//! the session-log sink, and returns the corresponding
//! `Item::system_message`s for the worker to append to its
//! persistent history.
crates/worker/src/ipc/protocol_session.rs
+9 -11
View File
@@ -29,17 +29,12 @@ pub fn subscribe_worker_protocol_session(handle: &WorkerHandle) -> WorkerProtoco
pub fn live_log_entry_event(entry: LogEntry) -> Option<Event> {
match entry {
entry @ (LogEntry::SegmentStart { .. } | LogEntry::AnnotatedSegmentStart { .. }) => {
let value = serde_json::to_value(&entry).expect("LogEntry is Serialize");
Some(Event::SegmentRotated { entry: value })
}
LogEntry::UserInput { segments, .. } | LogEntry::AnnotatedUserInput { segments, .. } => {
Some(Event::UserMessage { segments })
}
LogEntry::SystemItem { item, .. } => {
let value = serde_json::to_value(&item).expect("SystemItem is Serialize");
Some(Event::SystemItem { item: value })
entry @ LogEntry::AnnotatedSegmentStart { .. } => {
let session =
session_store::public_snapshot::project_current_session_snapshot(&[entry]);
Some(Event::SegmentRotated { session })
}
LogEntry::AnnotatedUserInput { segments, .. } => Some(Event::UserMessage { segments }),
LogEntry::AnnotatedSystemItem { entry, .. } => {
let value = serde_json::to_value(&entry.item).expect("SystemItem is Serialize");
Some(Event::SystemItem { item: value })
@@ -88,9 +83,12 @@ mod tests {
#[test]
fn user_input_log_entry_maps_to_user_message_event() {
let segments = vec![protocol::Segment::text("hello from log")];
let event = live_log_entry_event(LogEntry::UserInput {
let event = live_log_entry_event(LogEntry::AnnotatedUserInput {
ts: session_store::segment_log::now_millis(),
extensions: vec![],
history: vec![crate::session_history::test_logged_history_entry(
agen::Item::user_message("hello from log"),
)],
segments: segments.clone(),
})
.expect("UserInput must be live-relevant");
crates/worker/src/ipc/server.rs
+1 -1
View File
@@ -77,7 +77,7 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle)
let mut writer = JsonLineWriter::new(writer);
// Hold the in-flight stream lock while taking the session-log mirror
// snapshot. `LogEntry::AssistantItem` is mirror-only for live clients,
// snapshot. `LogEntry::AnnotatedAssistantItem` is mirror-only for live clients,
// so a finalized assistant block must be observed either as an already
// committed entry or as the still-present in-flight block. This lock
// order matches `append_entry` (in-flight clear before sink publish) and
crates/worker/src/segment_log_sink.rs
+31 -23
View File
@@ -50,7 +50,7 @@ struct SinkInner {
/// Broadcast channel for live entry updates. The same `Sender`
/// survives session swaps so existing subscribers keep their
/// receiver — they observe the swap as a freshly broadcast
/// `LogEntry::SegmentStart` and reset their view accordingly.
/// `LogEntry::AnnotatedSegmentStart` and reset their view accordingly.
broadcast_tx: broadcast::Sender<LogEntry>,
}
@@ -89,9 +89,9 @@ impl SegmentLogSink {
///
/// Live broadcast fires for committed session-log entries that
/// socket clients must see in log order:
/// - `LogEntry::SegmentStart` → `Event::SegmentRotated` on the wire.
/// - `LogEntry::UserInput` → `Event::UserMessage`.
/// - `LogEntry::SystemItem` → `Event::SystemItem`.
/// - `LogEntry::AnnotatedSegmentStart` → `Event::SegmentRotated` on the wire.
/// - `LogEntry::AnnotatedUserInput` → `Event::UserMessage`.
/// - `LogEntry::AnnotatedSystemItem` → `Event::SystemItem`.
/// - `LogEntry::Invoke` → `Event::InvokeStart`.
/// Everything else (AssistantItem, ToolResult, TurnEnd,
/// RunCompleted, RunErrored, PausedTurnAbandoned, LlmUsage, Extension,
@@ -120,11 +120,8 @@ impl SegmentLogSink {
fn is_live_relevant(entry: &LogEntry) -> bool {
matches!(
entry,
LogEntry::SegmentStart { .. }
| LogEntry::AnnotatedSegmentStart { .. }
| LogEntry::UserInput { .. }
LogEntry::AnnotatedSegmentStart { .. }
| LogEntry::AnnotatedUserInput { .. }
| LogEntry::SystemItem { .. }
| LogEntry::AnnotatedSystemItem { .. }
| LogEntry::Invoke { .. }
)
@@ -132,7 +129,7 @@ impl SegmentLogSink {
/// Atomically swap the mirror to `[initial]` and broadcast the new
/// session-start entry. Used during compaction / fork: the new
/// `LogEntry::SegmentStart` is the first entry of the replacement
/// `LogEntry::AnnotatedSegmentStart` is the first entry of the replacement
/// session, and existing subscribers transition by replaying it
/// like any other live entry.
///
@@ -234,7 +231,7 @@ mod tests {
use session_store::segment_log::now_millis;
fn session_start() -> LogEntry {
LogEntry::SegmentStart {
LogEntry::AnnotatedSegmentStart {
ts: now_millis(),
session_id: uuid::Uuid::nil(),
system_prompt: None,
@@ -253,9 +250,12 @@ mod tests {
}
fn user_input(text: &str) -> LogEntry {
LogEntry::UserInput {
LogEntry::AnnotatedUserInput {
ts: now_millis(),
extensions: vec![],
history: vec![crate::session_history::test_logged_history_entry(
agen::Item::user_message(text),
)],
segments: vec![protocol::Segment::Text {
content: text.to_owned(),
}],
@@ -270,7 +270,10 @@ mod tests {
let (snapshot, mut rx) = sink.subscribe_with_snapshot();
assert_eq!(snapshot.len(), 2);
assert!(matches!(snapshot[0], LogEntry::SegmentStart { .. }));
assert!(matches!(
snapshot[0],
LogEntry::AnnotatedSegmentStart { .. }
));
assert!(matches!(
snapshot[1],
LogEntry::TurnEnd { turn_count: 1, .. }
@@ -279,13 +282,15 @@ mod tests {
}
fn notification_entry(text: &str) -> LogEntry {
LogEntry::SystemItem {
LogEntry::AnnotatedSystemItem {
ts: now_millis(),
item: session_store::SystemItem::Notification {
message: text.to_owned(),
body: format!("[Notification] {text}"),
prompt_provenance: None,
},
entry: crate::session_history::test_logged_system_entry(
session_store::SystemItem::Notification {
message: text.to_owned(),
body: format!("[Notification] {text}"),
prompt_provenance: None,
},
),
}
}
@@ -305,7 +310,7 @@ mod tests {
// for Event::UserMessage.
sink.publish(user_input("hi from log"));
match rx.try_recv() {
Ok(LogEntry::UserInput { segments, .. }) => {
Ok(LogEntry::AnnotatedUserInput { segments, .. }) => {
assert_eq!(segments.len(), 1);
}
other => panic!("expected UserInput, got {other:?}"),
@@ -314,7 +319,7 @@ mod tests {
// SystemItem is live-relevant.
sink.publish(notification_entry("hi"));
match rx.try_recv() {
Ok(LogEntry::SystemItem { .. }) => {}
Ok(LogEntry::AnnotatedSystemItem { .. }) => {}
other => panic!("expected SystemItem, got {other:?}"),
}
@@ -332,7 +337,7 @@ mod tests {
assert_eq!(snapshot.len(), 1);
match rx.try_recv() {
Ok(LogEntry::SystemItem { .. }) => {}
Ok(LogEntry::AnnotatedSystemItem { .. }) => {}
other => panic!("unexpected: {other:?}"),
}
assert!(rx.try_recv().is_err());
@@ -348,13 +353,16 @@ mod tests {
sink.reset_with_initial(session_start());
match rx.try_recv() {
Ok(LogEntry::SegmentStart { .. }) => {}
Ok(LogEntry::AnnotatedSegmentStart { .. }) => {}
other => panic!("expected SegmentStart broadcast, got {other:?}"),
}
let (post_snapshot, _) = sink.subscribe_with_snapshot();
assert_eq!(post_snapshot.len(), 1);
assert!(matches!(post_snapshot[0], LogEntry::SegmentStart { .. }));
assert!(matches!(
post_snapshot[0],
LogEntry::AnnotatedSegmentStart { .. }
));
}
#[test]
crates/worker/src/session_capture.rs
+134 -33
View File
@@ -8,6 +8,10 @@ use std::sync::Arc;
use crate::session_history::{SessionHistoryMetadata, WorkerHistoryProvenance};
use agen::{HistoryEntry, Item, Role};
use protocol::{
SessionContentPart, SessionEntryProvenance, SessionMessageRole, SessionSnapshot,
SessionSnapshotEntryData,
};
use serde::{Deserialize, Serialize};
const DEFAULT_SEARCH_LIMIT: usize = 20;
@@ -105,7 +109,7 @@ impl ToolPart {
#[derive(Debug, Clone)]
pub(crate) struct OverviewItem {
pub id: SessionEntryRef,
pub origin: WorkerHistoryProvenance,
pub origin: SessionEntryProvenance,
pub entry_range: [u64; 2],
pub kind: ReferenceKind,
pub label: String,
@@ -116,7 +120,7 @@ pub(crate) struct OverviewItem {
#[derive(Debug, Clone)]
pub(crate) struct ReferenceEntry {
pub id: SessionEntryRef,
pub origin: WorkerHistoryProvenance,
pub origin: SessionEntryProvenance,
pub entry_range: [u64; 2],
pub kind: ReferenceKind,
pub tool_part: Option<ToolPart>,
@@ -142,7 +146,7 @@ pub(crate) struct SearchOptions {
#[derive(Debug, Clone)]
pub(crate) struct SearchHit {
pub id: SessionEntryRef,
pub origin: WorkerHistoryProvenance,
pub origin: SessionEntryProvenance,
pub kind: ReferenceKind,
pub tool_part: Option<ToolPart>,
pub tool_name: Option<String>,
@@ -188,7 +192,7 @@ impl Default for ReadOptions {
#[derive(Debug, Clone)]
pub(crate) struct ReadEntry {
pub id: SessionEntryRef,
pub origin: WorkerHistoryProvenance,
pub origin: SessionEntryProvenance,
pub kind: ReferenceKind,
pub tool_part: Option<ToolPart>,
pub tool_name: Option<String>,
@@ -207,7 +211,7 @@ pub(crate) struct ReadResult {
pub(crate) struct SessionEntryEvidence {
pub segment_id: String,
pub entry_ref: SessionEntryRef,
pub origin: WorkerHistoryProvenance,
pub origin: SessionEntryProvenance,
pub entry_range: [u64; 2],
pub kind: ReferenceKind,
pub tool_part: Option<ToolPart>,
@@ -216,35 +220,116 @@ pub(crate) struct SessionEntryEvidence {
pub excerpt: String,
}
#[derive(Debug, Clone)]
struct CapturedHistoryEntry {
item: Item,
entry_id: session_store::LoggedSessionHistoryEntryId,
origin: SessionEntryProvenance,
}
#[derive(Debug, Clone)]
pub(crate) struct SessionCapture {
segment_id: String,
entries: Arc<Vec<HistoryEntry<SessionHistoryMetadata>>>,
entries: Arc<Vec<CapturedHistoryEntry>>,
overview: Vec<OverviewItem>,
index: Vec<ReferenceEntry>,
}
impl SessionCapture {
pub(crate) fn from_session_snapshot(
segment_id: impl Into<String>,
snapshot: SessionSnapshot,
) -> Self {
let entries = snapshot
.entries
.into_iter()
.filter_map(|entry| {
let item = match entry.data {
SessionSnapshotEntryData::UserInput { segments } => {
Item::user_message(protocol::Segment::flatten_to_text(&segments))
}
SessionSnapshotEntryData::Message { role, content } => {
let role = match role {
SessionMessageRole::User => Role::User,
SessionMessageRole::Assistant => Role::Assistant,
};
Item::Message {
id: None,
role,
content: content
.into_iter()
.map(|part| match part {
SessionContentPart::Text { text } => {
agen::ContentPart::Text { text }
}
SessionContentPart::Refusal { refusal } => {
agen::ContentPart::Refusal { refusal }
}
})
.collect(),
status: None,
}
}
SessionSnapshotEntryData::ToolCall {
call_id,
name,
arguments,
} => Item::tool_call(call_id, name, arguments),
SessionSnapshotEntryData::ToolResult {
call_id,
summary,
content,
is_error,
attachments: _,
} => Item::tool_result_item(call_id, summary, content, is_error),
// Observation deliberately excludes system items and
// controller errors from model-visible session evidence.
SessionSnapshotEntryData::SystemItem { .. }
| SessionSnapshotEntryData::RunError { .. } => return None,
};
Some(CapturedHistoryEntry {
item,
entry_id: session_store::LoggedSessionHistoryEntryId(entry.entry_id),
origin: entry.provenance,
})
})
.collect();
Self::from_captured_entries(segment_id, entries)
}
pub(crate) fn new(segment_id: impl Into<String>, items: Vec<Item>) -> Self {
let entries = items
.into_iter()
.enumerate()
.map(|(index, item)| {
let mut metadata = SessionHistoryMetadata::legacy_unknown();
metadata.entry_id =
session_store::LoggedSessionHistoryEntryId(format!("{index:08}"));
HistoryEntry::new(item, metadata)
.map(|(index, item)| CapturedHistoryEntry {
item,
entry_id: session_store::LoggedSessionHistoryEntryId(format!("{index:08}")),
origin: SessionEntryProvenance::LegacyUnknown,
})
.collect();
Self::from_history_entries(segment_id, entries)
Self::from_captured_entries(segment_id, entries)
}
pub(crate) fn from_history_entries(
segment_id: impl Into<String>,
entries: Vec<HistoryEntry<SessionHistoryMetadata>>,
) -> Self {
let entries = entries
.into_iter()
.map(|entry| CapturedHistoryEntry {
item: entry.item,
entry_id: entry.annotation.entry_id,
origin: public_provenance(&entry.annotation.origin),
})
.collect();
Self::from_captured_entries(segment_id, entries)
}
fn from_captured_entries(
segment_id: impl Into<String>,
entries: Vec<CapturedHistoryEntry>,
) -> Self {
let segment_id = segment_id.into();
let entries = Arc::new(entries);
let mut overview = Vec::new();
let mut index = Vec::new();
@@ -253,7 +338,7 @@ impl SessionCapture {
let entry_range = [idx as u64, idx as u64];
match item {
Item::Message { role, content, .. } => {
let Some(kind) = message_reference_kind(&entry.annotation.origin, role) else {
let Some(kind) = message_reference_kind(&entry.origin, role) else {
continue;
};
let text = content
@@ -263,10 +348,10 @@ impl SessionCapture {
.join("");
let label = format!("{} message", kind.as_str());
let summary = truncate_chars(&text, 240);
let id = SessionEntryRef::from_history_entry_id(&entry.annotation.entry_id);
let id = SessionEntryRef::from_history_entry_id(&entry.entry_id);
index.push(ReferenceEntry {
id: id.clone(),
origin: entry.annotation.origin.clone(),
origin: entry.origin.clone(),
entry_range,
kind,
tool_part: None,
@@ -278,7 +363,7 @@ impl SessionCapture {
if matches!(kind, ReferenceKind::User | ReferenceKind::Assistant) {
overview.push(OverviewItem {
id: id.clone(),
origin: entry.annotation.origin.clone(),
origin: entry.origin.clone(),
entry_range,
kind,
label,
@@ -292,8 +377,8 @@ impl SessionCapture {
} => {
let text = format!("{name}\n{arguments}");
index.push(ReferenceEntry {
id: SessionEntryRef::from_history_entry_id(&entry.annotation.entry_id),
origin: entry.annotation.origin.clone(),
id: SessionEntryRef::from_history_entry_id(&entry.entry_id),
origin: entry.origin.clone(),
entry_range,
kind: ReferenceKind::Tool,
tool_part: Some(ToolPart::Input),
@@ -319,8 +404,8 @@ impl SessionCapture {
content.as_deref().unwrap_or_default(),
);
index.push(ReferenceEntry {
id: SessionEntryRef::from_history_entry_id(&entry.annotation.entry_id),
origin: entry.annotation.origin.clone(),
id: SessionEntryRef::from_history_entry_id(&entry.entry_id),
origin: entry.origin.clone(),
entry_range,
kind: ReferenceKind::Tool,
tool_part: Some(ToolPart::Output),
@@ -360,7 +445,7 @@ impl SessionCapture {
Self {
segment_id,
entries,
entries: Arc::new(entries),
overview,
index,
}
@@ -543,25 +628,41 @@ impl SessionCapture {
}
}
fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance {
match origin {
WorkerHistoryProvenance::HumanInput { .. } => SessionEntryProvenance::HumanInput,
WorkerHistoryProvenance::WorkerInput { .. } => SessionEntryProvenance::WorkerInput,
WorkerHistoryProvenance::FlowInstruction { .. } => SessionEntryProvenance::FlowInstruction,
WorkerHistoryProvenance::BackendInstruction { .. } => {
SessionEntryProvenance::BackendInstruction
}
WorkerHistoryProvenance::ModelOutput { .. } => SessionEntryProvenance::ModelOutput,
WorkerHistoryProvenance::ToolOutput { .. } => SessionEntryProvenance::ToolOutput,
WorkerHistoryProvenance::DerivedSummary => SessionEntryProvenance::DerivedSummary,
WorkerHistoryProvenance::LegacyUnknown => SessionEntryProvenance::LegacyUnknown,
}
}
fn message_reference_kind(
origin: &WorkerHistoryProvenance,
origin: &SessionEntryProvenance,
provider_role: &Role,
) -> Option<ReferenceKind> {
match origin {
WorkerHistoryProvenance::HumanInput { .. }
| WorkerHistoryProvenance::WorkerInput { .. } => Some(ReferenceKind::User),
WorkerHistoryProvenance::ModelOutput { .. } => Some(ReferenceKind::Assistant),
WorkerHistoryProvenance::ToolOutput { .. } => Some(ReferenceKind::Tool),
WorkerHistoryProvenance::LegacyUnknown => match provider_role {
SessionEntryProvenance::HumanInput | SessionEntryProvenance::WorkerInput => {
Some(ReferenceKind::User)
}
SessionEntryProvenance::ModelOutput => Some(ReferenceKind::Assistant),
SessionEntryProvenance::ToolOutput => Some(ReferenceKind::Tool),
SessionEntryProvenance::LegacyUnknown => match provider_role {
Role::User => Some(ReferenceKind::User),
Role::Assistant => Some(ReferenceKind::Assistant),
Role::System => None,
},
// Flow/backend/system content remains out of the observation surface
// even when represented with a provider user/system role.
WorkerHistoryProvenance::FlowInstruction { .. }
| WorkerHistoryProvenance::BackendInstruction { .. }
| WorkerHistoryProvenance::DerivedSummary => None,
SessionEntryProvenance::FlowInstruction
| SessionEntryProvenance::BackendInstruction
| SessionEntryProvenance::DerivedSummary => None,
}
}
@@ -658,13 +759,13 @@ mod tests {
assert_eq!(overview.len(), 1);
assert!(matches!(
overview[0].origin,
WorkerHistoryProvenance::HumanInput { .. }
SessionEntryProvenance::HumanInput
));
let evidence = capture.evidence_for(overview[0].id.as_str()).unwrap();
assert!(evidence.excerpt.ends_with("remember my preference"));
assert!(matches!(
evidence.origin,
WorkerHistoryProvenance::HumanInput { .. }
SessionEntryProvenance::HumanInput
));
}
crates/worker/src/session_history.rs
+27 -37
View File
@@ -5,7 +5,6 @@
//! retained only as explicit `LegacyUnknown` entries.
use agen::{HistoryEntry, Item};
use protocol::Segment;
use session_store::{
LogEntry, LoggedHistoryDerivation, LoggedHistoryEntry, LoggedSessionHistoryEntryId,
LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, LoggedWorkerSubject, SegmentId,
@@ -18,6 +17,32 @@ pub type WorkerHistoryProvenance = LoggedSessionHistoryOrigin;
pub type SessionHistoryDerivation = LoggedHistoryDerivation;
pub type WorkerSubjectSnapshot = LoggedWorkerSubject;
#[cfg(test)]
pub(crate) fn test_logged_history_entry(item: impl Into<Item>) -> LoggedHistoryEntry {
LoggedHistoryEntry {
item: session_store::LoggedItem::from(item.into()),
metadata: LoggedSessionHistoryMetadata {
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
},
}
}
#[cfg(test)]
pub(crate) fn test_logged_system_entry(
item: session_store::SystemItem,
) -> session_store::LoggedSystemHistoryEntry {
session_store::LoggedSystemHistoryEntry {
item,
metadata: LoggedSessionHistoryMetadata {
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
},
}
}
pub(crate) fn worker_subject(session_id: SessionId) -> WorkerSubjectSnapshot {
WorkerSubjectSnapshot {
workspace_id: None,
@@ -53,10 +78,6 @@ pub(crate) fn to_logged_history_entry(
}
}
fn legacy_entry(item: Item) -> HistoryEntry<SessionHistoryMetadata> {
HistoryEntry::new(item, SessionHistoryMetadata::legacy_unknown())
}
fn from_logged(entry: &LoggedHistoryEntry) -> HistoryEntry<SessionHistoryMetadata> {
HistoryEntry::new(Item::from(entry.item.clone()), entry.metadata.clone())
}
@@ -74,32 +95,15 @@ pub(crate) fn restore_history_entries(
LogEntry::AnnotatedSegmentStart { history: seed, .. } => {
history = seed.iter().map(from_logged).collect();
}
LogEntry::SegmentStart { history: seed, .. } => {
history = seed
.iter()
.cloned()
.map(Item::from)
.map(legacy_entry)
.collect();
}
LogEntry::AnnotatedUserInput { history: input, .. } => {
history.extend(input.iter().map(from_logged))
}
LogEntry::UserInput { segments, .. } => history.push(legacy_entry(Item::user_message(
Segment::flatten_to_text(segments),
))),
LogEntry::AnnotatedAssistantItem { entry, .. }
| LogEntry::AnnotatedToolResult { entry, .. } => history.push(from_logged(entry)),
LogEntry::AssistantItem { item, .. } | LogEntry::ToolResult { item, .. } => {
history.push(legacy_entry(Item::from(item.clone())));
}
LogEntry::AnnotatedSystemItem { entry, .. } => history.push(HistoryEntry::new(
entry.item.to_history_item(),
entry.metadata.clone(),
)),
LogEntry::SystemItem { item, .. } => {
history.push(legacy_entry(item.to_history_item()));
}
_ => {}
}
}
@@ -110,23 +114,9 @@ pub(crate) fn restore_history_entries(
mod tests {
use super::*;
use agen::llm_client::RequestConfig;
use protocol::Segment;
use session_store::LogEntry;
#[test]
fn legacy_user_role_is_not_inferred_as_human_authority() {
let entries = vec![LogEntry::UserInput {
ts: 1,
segments: vec![Segment::text("legacy")],
extensions: Vec::new(),
}];
let restored =
restore_history_entries(SessionId::now_v7(), SegmentId::now_v7(), &entries).unwrap();
assert!(matches!(
restored[0].annotation.origin,
WorkerHistoryProvenance::LegacyUnknown
));
}
#[test]
fn typed_flow_and_unknown_caller_input_round_trip_without_role_inference() {
let session_id = SessionId::now_v7();
crates/worker/src/spawn/comm_tools.rs
+15 -1
View File
@@ -278,7 +278,21 @@ mod tests {
fn snapshot(entries: Vec<serde_json::Value>) -> Event {
Event::Snapshot {
entries,
session: protocol::SessionSnapshot {
entries: entries
.into_iter()
.enumerate()
.map(|(index, value)| protocol::SessionSnapshotEntry {
entry_id: format!("test-{index}"),
timestamp: index as u64,
provenance: protocol::SessionEntryProvenance::LegacyUnknown,
derived_from: Vec::new(),
data: protocol::SessionSnapshotEntryData::RunError {
message: value.to_string(),
},
})
.collect(),
},
greeting: Greeting {
worker_name: "server".into(),
cwd: "/tmp".into(),
crates/worker/src/spawn/registry.rs
+30 -21
View File
@@ -112,8 +112,12 @@ impl InternalSpawnedWorkerRecord {
fn stop_summary(&self) -> SubWorkerStopSummary {
let mut counts = BTreeMap::<String, u64>::new();
for entry in self.session.entries() {
if let session_store::LogEntry::AssistantItem {
item: LoggedItem::ToolCall { name, .. },
if let session_store::LogEntry::AnnotatedAssistantItem {
entry:
session_store::LoggedHistoryEntry {
item: LoggedItem::ToolCall { name, .. },
..
},
..
} = entry
{
@@ -806,11 +810,7 @@ fn internal_worker_snapshot(
InternalWorkerSnapshot {
worker,
revision,
entries: snapshot
.entries
.into_iter()
.filter_map(|entry| serde_json::to_value(entry).ok())
.collect(),
session: snapshot.session,
status: snapshot.status,
error: snapshot.error,
in_flight: snapshot.in_flight,
@@ -1028,11 +1028,16 @@ mod tests {
&& worker.parent_session_id.as_deref() == Some("parent-session")
&& matches!(*event, Event::TextDone { ref text } if text == "answer")
));
record.session.publish_test_entry(LogEntry::UserInput {
ts: 1,
segments: vec![protocol::Segment::text("question")],
extensions: Vec::new(),
});
record
.session
.publish_test_entry(LogEntry::AnnotatedUserInput {
ts: 1,
segments: vec![protocol::Segment::text("question")],
history: vec![crate::session_history::test_logged_history_entry(
agen::Item::user_message("question"),
)],
extensions: Vec::new(),
});
let committed = tokio::time::timeout(Duration::from_secs(1), parent_rx.recv())
.await
.unwrap()
@@ -1045,7 +1050,7 @@ mod tests {
let snapshots = registry.internal_worker_snapshots();
assert_eq!(snapshots.len(), 1);
assert_eq!(snapshots[0].revision, 2);
assert_eq!(snapshots[0].entries.len(), 1);
assert_eq!(snapshots[0].session.entries.len(), 1);
record.session.emit_test_text_delta("partial");
let streamed = tokio::time::timeout(Duration::from_secs(1), parent_rx.recv())
@@ -1163,14 +1168,18 @@ mod tests {
let (mut record, _events) = record("child", InternalWorkerVisibility::ParentClient).await;
record.change_tracker = Some(tracker);
for (index, name) in ["Read", "Read", "Grep"].into_iter().enumerate() {
record.session.publish_test_entry(LogEntry::AssistantItem {
ts: index as u64,
item: LoggedItem::ToolCall {
call_id: format!("call-{index}"),
name: name.to_string(),
arguments: "{}".to_string(),
},
});
record
.session
.publish_test_entry(LogEntry::AnnotatedAssistantItem {
ts: index as u64,
entry: crate::session_history::test_logged_history_entry(
LoggedItem::ToolCall {
call_id: format!("call-{index}"),
name: name.to_string(),
arguments: "{}".to_string(),
},
),
});
}
registry.start_protocol_forwarding(record.clone());
install_record(&registry, record);
crates/worker/src/spawn/tool.rs
+6 -6
View File
@@ -960,9 +960,7 @@ mod tests {
use crate::WorkspaceId;
use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
use agen::llm_client::types::ContentPart;
use agen::llm_client::{ClientError, LlmClient, Request};
use agen::{Item, Role};
use async_trait::async_trait;
use futures::Stream;
use manifest::{AuthRef, ModelManifest, SchemeKind, WorkerManifest};
@@ -1252,9 +1250,11 @@ extract_threshold = 4000
)
.await
.unwrap();
assert!(first_capture.entries.iter().map(|entry| &entry.item).any(|item| {
matches!(item, Item::Message { role: Role::Assistant, content, .. } if content.iter().any(|part| matches!(part, ContentPart::Text { text } if text.contains("reviewed"))))
}));
assert!(
serde_json::to_string(&first_capture.session)
.unwrap()
.contains("reviewed")
);
let send = (crate::spawn::comm_tools::sub_worker_send_tool(registry.clone()))().1;
send.execute(
@@ -1274,7 +1274,7 @@ extract_threshold = 4000
)
.await
.unwrap();
assert!(latest_capture.entries.len() > first_capture.entries.len());
assert!(latest_capture.session.entries.len() > first_capture.session.entries.len());
fail_requests.store(true, Ordering::SeqCst);
send.execute(
crates/worker/src/worker.rs
+51 -42
View File
@@ -900,7 +900,6 @@ where
self.state.increment_entries();
if let Some(in_flight) = &self.in_flight {
let committed_item = match &entry {
LogEntry::AssistantItem { item, .. } => Some(item.clone()),
LogEntry::AnnotatedAssistantItem { entry, .. } => Some(entry.item.clone()),
_ => None,
};
@@ -1207,11 +1206,11 @@ pub struct Worker<C: LlmClient, St: Store> {
memory_task: Option<JoinHandle<()>>,
/// Typed user submissions in submit order. K-th entry corresponds to
/// the K-th `Item::user_message` in `worker.history()` (modulo seed
/// history loaded via `SegmentStart.history`, whose original segments
/// history loaded via `AnnotatedSegmentStart.history`, whose original segments
/// are not preserved). Populated from log on `restore_from_manifest`,
/// appended after `save_user_input` on each `run`. Pre-`Event::Snapshot`
/// this fed `WorkerSharedState.user_segments`; the new wire format
/// carries typed atoms via `LogEntry::UserInput { segments }` so
/// carries typed atoms via `LogEntry::AnnotatedUserInput { segments }` so
/// this remains purely an in-memory tracker for compact alignment.
user_segments: Vec<Vec<Segment>>,
/// Worker-side session-log mirror + broadcast sink. Populated alongside
@@ -1221,7 +1220,8 @@ pub struct Worker<C: LlmClient, St: Store> {
sink: SegmentLogSink,
/// `true` once `wire_history_persistence` has installed the
/// `Engine::on_history_append` callback that commits each appended
/// item as a singular `LogEntry::AssistantItem` / `ToolResult`
/// item as a singular `LogEntry::AnnotatedAssistantItem` /
/// `AnnotatedToolResult`
/// directly through the writer. Tests that drive `Worker::new` without
/// going through the controller leave this `false`; `persist_turn`
/// then walks the post-`history_before` slice inline so entries
@@ -1345,15 +1345,16 @@ impl<C: LlmClient + 'static, St: Store + Clone + 'static> Worker<C, St> {
}
/// Wire `Engine::on_history_append` to commit each appended item
/// directly as a singular `LogEntry::AssistantItem` / `ToolResult`
/// directly as a singular `LogEntry::AnnotatedAssistantItem` /
/// `AnnotatedToolResult`
/// through the writer. The controller calls this once per spawned
/// Worker after the worker is built; tests that drive `Worker::new` may
/// opt in to the same wiring or leave it off (in which case
/// `persist_turn`'s inline fallback writes entries at turn end).
///
/// `user_message` items are skipped because they are committed
/// up-front via `commit_entry(LogEntry::UserInput { segments })`.
/// `role:system` items are committed as typed `LogEntry::SystemItem`
/// up-front via `commit_entry(LogEntry::AnnotatedUserInput { segments })`.
/// `role:system` items are committed as typed `LogEntry::AnnotatedSystemItem`
/// entries by their producers (for example `WorkerInterceptor` and
/// interrupted-turn prep) before they reach the worker's history, so this
/// callback would otherwise double-write them.
@@ -1941,8 +1942,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
}
let input = match entries.get(target.user_input_entry_index) {
Some(LogEntry::UserInput { segments, .. })
| Some(LogEntry::AnnotatedUserInput { segments, .. }) => segments.clone(),
Some(LogEntry::AnnotatedUserInput { segments, .. }) => segments.clone(),
_ => {
return Err(RewindError::Invalid(
"rewind target is no longer a user message".into(),
@@ -2081,8 +2081,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
/// Cheap clone via `Option<Clone>`.
/// Snapshot of the typed user segments tracked alongside worker
/// history. The K-th entry corresponds to the K-th `Item::user_message`
/// derived from `LogEntry::UserInput` entries (post-compaction); seed
/// history loaded via `SegmentStart.history` does not contribute,
/// derived from `LogEntry::AnnotatedUserInput` entries (post-compaction); seed
/// history loaded via `AnnotatedSegmentStart.history` does not contribute,
/// which is acceptable because the original segments are unrecoverable.
pub fn user_segments(&self) -> &[Vec<Segment>] {
&self.user_segments
@@ -3533,12 +3533,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// slice from `history_before` inline so the test's
// `restore`-style assertions still see entries on disk.
if !self.history_persistence_wired {
let new_items: Vec<Item> = self.session.history().entries()[history_before..]
let new_entries: Vec<_> = self.session.history().entries()[history_before..]
.iter()
.map(|entry| entry.item.clone())
.map(to_logged_history_entry)
.collect();
let ts = segment_log::now_millis();
for item in &new_items {
for history_entry in new_entries {
let item = Item::from(history_entry.item.clone());
if item.is_user_message() {
continue;
}
@@ -3551,7 +3552,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
) {
continue;
}
let entry = session_store::classify_history_item(item, ts);
let entry = session_store::classify_logged_history_entry(history_entry, ts);
self.commit_entry(entry)?;
}
}
@@ -6257,8 +6258,7 @@ fn build_rewind_targets(segment_id: uuid::Uuid, entries: &[LogEntry]) -> Vec<Rew
let mut targets = Vec::new();
for (entry_index, entry) in entries.iter().enumerate() {
let (segments, ts) = match entry {
LogEntry::UserInput { segments, ts, .. }
| LogEntry::AnnotatedUserInput { segments, ts, .. } => (segments, ts),
LogEntry::AnnotatedUserInput { segments, ts, .. } => (segments, ts),
_ => continue,
};
turn_index += 1;
@@ -6300,8 +6300,7 @@ fn rewind_truncate_entries(entries: &[LogEntry], user_input_entry_index: usize)
fn suffix_has_tool_side_effects(entries: &[LogEntry]) -> bool {
entries.iter().any(|entry| match entry {
LogEntry::ToolResult { .. } | LogEntry::AnnotatedToolResult { .. } => true,
LogEntry::AssistantItem { item, .. } => logged_item_is_tool_call(item),
LogEntry::AnnotatedToolResult { .. } => true,
LogEntry::AnnotatedAssistantItem { entry, .. } => logged_item_is_tool_call(&entry.item),
_ => false,
})
@@ -7636,7 +7635,7 @@ mod build_summary_prompt_tests {
);
assert!(checkpoint.is_none());
let mut replacement_entries = vec![LogEntry::SegmentStart {
let mut replacement_entries = vec![LogEntry::AnnotatedSegmentStart {
ts: segment_log::now_millis(),
session_id: uuid::Uuid::nil(),
system_prompt: None,
@@ -7964,9 +7963,12 @@ mod build_summary_prompt_tests {
);
append_test_entry(
worker,
LogEntry::UserInput {
LogEntry::AnnotatedUserInput {
ts: ts + 1,
extensions: vec![],
history: vec![crate::session_history::test_logged_history_entry(
Item::user_message(text),
)],
segments: vec![text_segment(text)],
},
);
@@ -7986,16 +7988,18 @@ mod build_summary_prompt_tests {
append_user_turn(&worker, 20, "second message");
append_test_entry(
&worker,
LogEntry::ToolResult {
LogEntry::AnnotatedToolResult {
ts: 30,
item: session_store::LoggedItem::ToolResult {
call_id: "call-1".into(),
summary: "wrote a file".into(),
content: None,
attachments: Vec::new(),
disposition: Default::default(),
is_error: false,
},
entry: crate::session_history::test_logged_history_entry(
session_store::LoggedItem::ToolResult {
call_id: "call-1".into(),
summary: "wrote a file".into(),
content: None,
attachments: Vec::new(),
disposition: Default::default(),
is_error: false,
},
),
},
);
@@ -8029,16 +8033,18 @@ mod build_summary_prompt_tests {
append_user_turn(&worker, 20, "second message");
append_test_entry(
&worker,
LogEntry::ToolResult {
LogEntry::AnnotatedToolResult {
ts: 30,
item: session_store::LoggedItem::ToolResult {
call_id: "call-1".into(),
summary: "wrote a file".into(),
content: None,
attachments: Vec::new(),
disposition: Default::default(),
is_error: false,
},
entry: crate::session_history::test_logged_history_entry(
session_store::LoggedItem::ToolResult {
call_id: "call-1".into(),
summary: "wrote a file".into(),
content: None,
attachments: Vec::new(),
disposition: Default::default(),
is_error: false,
},
),
},
);
let (head_entries, targets) = worker.list_rewind_targets().unwrap();
@@ -8456,9 +8462,9 @@ mod build_summary_prompt_tests {
worker.wire_history_persistence();
let dangling_call = Item::tool_call("call-1", "SideEffect", "{}");
worker
.commit_entry(LogEntry::AssistantItem {
.commit_entry(LogEntry::AnnotatedAssistantItem {
ts: segment_log::now_millis(),
item: dangling_call.clone().into(),
entry: crate::session_history::test_logged_history_entry(dangling_call.clone()),
})
.unwrap();
worker.set_history_for_test(vec![dangling_call]);
@@ -8863,9 +8869,12 @@ mod build_summary_prompt_tests {
);
worker.set_history_for_test(vec![evidence.clone()]);
worker
.commit_entry(LogEntry::UserInput {
.commit_entry(LogEntry::AnnotatedUserInput {
ts: segment_log::now_millis(),
extensions: vec![],
history: vec![crate::session_history::test_logged_history_entry(
evidence.clone(),
)],
segments: vec![text_segment(
"The cancellation regression must leave this evidence available for retry.",
)],
crates/worker/tests/compact_events_test.rs
+20 -18
View File
@@ -25,6 +25,17 @@ use worker::{Worker, WorkerController};
type TestStore = CombinedStore<FsStore, FsWorkerStore>;
fn annotated(item: Item) -> session_store::LoggedHistoryEntry {
session_store::LoggedHistoryEntry {
item: session_store::LoggedItem::from(item),
metadata: session_store::LoggedSessionHistoryMetadata {
entry_id: session_store::LoggedSessionHistoryEntryId::new(),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
},
}
}
#[derive(Clone)]
struct MockClient {
responses: Arc<Vec<Vec<LlmEvent>>>,
@@ -210,7 +221,6 @@ fn system_texts_in_sink_session_start(
.into_iter()
.map(|entry| entry.item)
.collect::<Vec<_>>(),
session_store::LogEntry::SegmentStart { history, .. } => history,
_ => continue,
};
return history
@@ -310,17 +320,14 @@ permission = "write"
// Simulate a foreign writer appending to the same segment. This bumps
// the on-disk entry count past the Worker's own append tally without
// updating the Worker's `entries_written`.
store
.append(
session_id,
source_segment_id,
&LogEntry::UserInput {
ts: 9999,
segments: vec![protocol::Segment::text("interloper")],
extensions: vec![],
},
)
.unwrap();
session_store::save_user_input(
&store,
session_id,
source_segment_id,
vec![protocol::Segment::text("interloper")],
vec![annotated(Item::user_message("interloper"))],
)
.unwrap();
// Next run triggers ensure_segment_head, which sees the drift.
worker.run_text("second").await.unwrap();
@@ -348,11 +355,6 @@ permission = "write"
session_id: seg_session,
forked_from: Some(origin),
..
}
| LogEntry::SegmentStart {
session_id: seg_session,
forked_from: Some(origin),
..
} => {
assert_eq!(*seg_session, session_id);
assert_eq!(origin.segment_id, source_segment_id);
@@ -366,7 +368,7 @@ permission = "write"
assert_eq!(source_after.len(), source_len_before + 1);
assert!(matches!(
source_after.last(),
Some(LogEntry::UserInput { .. })
Some(LogEntry::AnnotatedUserInput { .. })
));
}
crates/worker/tests/controller_test.rs
+24 -43
View File
@@ -35,29 +35,16 @@ fn history_from_sink(handle: &WorkerHandle) -> Vec<Item> {
LogEntry::AnnotatedSegmentStart { history, .. } => {
items.extend(history.into_iter().map(|entry| Item::from(entry.item)));
}
LogEntry::SegmentStart { history, .. } => {
items.extend(history.into_iter().map(Item::from));
}
LogEntry::AnnotatedUserInput { history, .. } => {
items.extend(history.into_iter().map(|entry| Item::from(entry.item)));
}
LogEntry::UserInput { segments, .. } => {
let text = protocol::Segment::flatten_to_text(&segments);
items.push(Item::user_message(text));
}
LogEntry::AnnotatedAssistantItem { entry, .. }
| LogEntry::AnnotatedToolResult { entry, .. } => {
items.push(Item::from(entry.item));
}
LogEntry::AssistantItem { item, .. } | LogEntry::ToolResult { item, .. } => {
items.push(Item::from(item));
}
LogEntry::AnnotatedSystemItem { entry, .. } => {
items.push(entry.item.to_history_item());
}
LogEntry::SystemItem { item, .. } => {
items.push(item.to_history_item());
}
_ => {}
}
}
@@ -67,7 +54,6 @@ fn history_from_sink(handle: &WorkerHandle) -> Vec<Item> {
fn system_item(entry: &LogEntry) -> Option<&session_store::SystemItem> {
match entry {
LogEntry::AnnotatedSystemItem { entry, .. } => Some(&entry.item),
LogEntry::SystemItem { item, .. } => Some(item),
_ => None,
}
}
@@ -839,26 +825,26 @@ async fn snapshot_includes_user_input_for_in_flight_turn() {
loop {
let event = reader.next::<Event>().await.unwrap().unwrap();
match event {
Event::Snapshot { entries, .. } => {
// Walk the entries, find a `LogEntry::UserInput` and
// confirm its segments flatten to our submitted text.
let mut found = false;
for value in &entries {
let entry: session_store::LogEntry =
serde_json::from_value(value.clone()).expect("LogEntry deserialise");
if let session_store::LogEntry::UserInput { segments, .. }
| session_store::LogEntry::AnnotatedUserInput { segments, .. } = entry
{
let text = protocol::Segment::flatten_to_text(&segments);
if text == "hello in-flight" {
found = true;
break;
}
Event::Snapshot { session, .. } => {
let found = session.entries.iter().any(|entry| match &entry.data {
protocol::SessionSnapshotEntryData::UserInput { segments } => {
protocol::Segment::flatten_to_text(segments) == "hello in-flight"
}
}
protocol::SessionSnapshotEntryData::Message {
role: protocol::SessionMessageRole::User,
content,
} => content.iter().any(|part| {
matches!(
part,
protocol::SessionContentPart::Text { text }
if text == "hello in-flight"
)
}),
_ => false,
});
assert!(
found,
"snapshot must carry the in-flight UserInput entry: {entries:?}"
"snapshot must carry the in-flight UserInput entry: {session:?}"
);
return;
}
@@ -1095,7 +1081,7 @@ async fn run_with_paste_segment_inlines_content_and_emits_typed_user_message() {
// Mixed input: plain text + a paste chip + trailing text. Worker must
// flatten this into one user-message string (paste content inlined,
// no `[Clipboard ...]` label leaking to the LLM); the committed
// `LogEntry::UserInput` must carry the typed segments unchanged so
// `LogEntry::AnnotatedUserInput` must carry the typed segments unchanged so
// socket clients can derive `Event::UserMessage` and re-render the chip.
let segments = vec![
protocol::Segment::text("see "),
@@ -1130,7 +1116,7 @@ async fn run_with_paste_segment_inlines_content_and_emits_typed_user_message() {
_ => {}
},
entry = entry_rx.recv() => match entry {
Ok(session_store::LogEntry::UserInput { segments, .. } | session_store::LogEntry::AnnotatedUserInput { segments, .. }) => {
Ok(session_store::LogEntry::AnnotatedUserInput { segments, .. }) => {
user_input_segments = Some(segments);
if saw_turn_end {
break;
@@ -2410,17 +2396,12 @@ async fn snapshot_contains_user_input(handle: &WorkerHandle, needle: &str) -> bo
loop {
let event = reader.next::<Event>().await.unwrap().unwrap();
match event {
Event::Snapshot { entries, .. } => {
return entries.into_iter().any(|value| {
let entry: session_store::LogEntry =
serde_json::from_value(value).expect("LogEntry deserialise");
match entry {
session_store::LogEntry::UserInput { segments, .. }
| session_store::LogEntry::AnnotatedUserInput { segments, .. } => {
protocol::Segment::flatten_to_text(&segments).contains(needle)
}
_ => false,
Event::Snapshot { session, .. } => {
return session.entries.into_iter().any(|entry| match entry.data {
protocol::SessionSnapshotEntryData::UserInput { segments } => {
protocol::Segment::flatten_to_text(&segments).contains(needle)
}
_ => false,
});
}
Event::Alert(_) => continue,
crates/worker/tests/system_prompt_template_test.rs
+2 -2
View File
@@ -203,12 +203,12 @@ async fn session_start_state_captures_rendered_prompt() {
.unwrap();
let first = entries.first().expect("at least one entry");
match first {
LogEntry::SegmentStart { system_prompt, .. } => {
LogEntry::AnnotatedSegmentStart { system_prompt, .. } => {
let sp = system_prompt.as_deref().expect("system prompt set");
assert!(sp.starts_with("hello"));
assert!(sp.contains(&pwd.display().to_string()));
}
other => panic!("expected SegmentStart as first entry, got {other:?}"),
other => panic!("expected AnnotatedSegmentStart as first entry, got {other:?}"),
}
}