feat: page console history across compaction
This commit is contained in:
@@ -831,6 +831,34 @@ pub struct SessionSnapshot {
|
||||
pub entries: Vec<SessionSnapshotEntry>,
|
||||
}
|
||||
|
||||
/// One user-visible conversation turn. A turn starts at a canonical user entry
|
||||
/// and contains every public committed entry until the next user entry.
|
||||
/// `turn_id` is the stable `entry_id` of that first user entry.
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
|
||||
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
|
||||
pub struct SessionConversationTurn {
|
||||
pub turn_id: String,
|
||||
pub entries: Vec<SessionSnapshotEntry>,
|
||||
}
|
||||
|
||||
/// Bounded backward page over the adopted conversation lineage.
|
||||
///
|
||||
/// Turns remain chronological within a page. `next_cursor` is an opaque,
|
||||
/// identity-bound boundary for the page immediately preceding `turns`.
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
|
||||
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
|
||||
pub struct SessionHistoryPage {
|
||||
pub session_id: String,
|
||||
/// Stable identity of the adopted active lineage used to fence stale pages.
|
||||
pub lineage_id: String,
|
||||
pub turns: Vec<SessionConversationTurn>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub next_cursor: Option<String>,
|
||||
pub has_more: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
|
||||
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
|
||||
@@ -957,8 +985,17 @@ pub enum Event {
|
||||
/// activated for a turn. Broadcast to every subscribed client so TUI / GUI
|
||||
/// instances show the same user line that reconnect snapshots replay.
|
||||
UserMessage {
|
||||
/// Stable durable history identity. Older mixed-version producers omit it.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
entry_id: Option<String>,
|
||||
segments: Vec<Segment>,
|
||||
},
|
||||
/// A canonical assistant/tool entry has committed to durable history.
|
||||
/// Streaming events remain the in-flight authority; this event supplies the
|
||||
/// stable identity used to reconcile that live block with paged history.
|
||||
SessionEntryCommitted {
|
||||
entry: SessionSnapshotEntry,
|
||||
},
|
||||
/// One agent-injected system item committed to history.
|
||||
///
|
||||
/// Carries the JSON form of `session_store::SystemItem`. Covers
|
||||
@@ -2759,6 +2796,7 @@ mod tests {
|
||||
#[test]
|
||||
fn event_user_message_roundtrip() {
|
||||
let event = Event::UserMessage {
|
||||
entry_id: Some("entry-1".to_string()),
|
||||
segments: vec![Segment::text("hello 世界")],
|
||||
};
|
||||
let json = serde_json::to_string(&event).unwrap();
|
||||
@@ -2769,7 +2807,7 @@ mod tests {
|
||||
|
||||
let decoded: Event = serde_json::from_str(&json).unwrap();
|
||||
match decoded {
|
||||
Event::UserMessage { segments } => {
|
||||
Event::UserMessage { segments, .. } => {
|
||||
assert_eq!(segments.len(), 1);
|
||||
match &segments[0] {
|
||||
Segment::Text { content } => assert_eq!(content, "hello 世界"),
|
||||
|
||||
@@ -11,12 +11,13 @@ use crate::{
|
||||
InternalWorkerRef, InternalWorkerSnapshot, InvokeKind, MemoryWorkerEvent, Method,
|
||||
PasteArtifactAvailability, PasteArtifactMediaType, PasteArtifactRef, PendingSubmissionSummary,
|
||||
PendingSubmissionsSnapshot, Permission, RewindSummary, RewindTarget, RewindTargetId, RunResult,
|
||||
ScopeRule, Segment, SessionContentPart, SessionEntryProvenance, SessionMessageRole,
|
||||
SessionSnapshot, SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment,
|
||||
SubmissionDisposition, SymlinkPolicy, ToolResultDisposition, TurnResult,
|
||||
UploadedFileAvailability, UploadedFileRef, WorkerBusyState, WorkerCommandAcknowledgement,
|
||||
WorkerCommandDisposition, WorkerCommandEnvelope, WorkerCommandKind, WorkerEvent,
|
||||
WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStateSnapshot, WorkerStatus,
|
||||
ScopeRule, Segment, SessionContentPart, SessionConversationTurn, SessionEntryProvenance,
|
||||
SessionHistoryPage, SessionMessageRole, SessionSnapshot, SessionSnapshotEntry,
|
||||
SessionSnapshotEntryData, SessionToolAttachment, SubmissionDisposition, SymlinkPolicy,
|
||||
ToolResultDisposition, TurnResult, UploadedFileAvailability, UploadedFileRef, WorkerBusyState,
|
||||
WorkerCommandAcknowledgement, WorkerCommandDisposition, WorkerCommandEnvelope,
|
||||
WorkerCommandKind, WorkerEvent, WorkerMaintenanceState, WorkerRunState, WorkerState,
|
||||
WorkerStateSnapshot, WorkerStatus,
|
||||
subscription::{
|
||||
EventSubscriptionSelector, MAX_CORRELATION_ID_BYTES, MAX_REJECTION_MESSAGE_BYTES,
|
||||
MAX_RESOURCE_ID_BYTES, MAX_SUBSCRIPTION_COLLECTION_ITEMS,
|
||||
@@ -104,6 +105,8 @@ pub fn generated_protocol_types() -> String {
|
||||
push_decl::<SessionToolAttachment>(&cfg, &mut output);
|
||||
push_decl::<SessionSnapshotEntryData>(&cfg, &mut output);
|
||||
push_decl::<SessionSnapshotEntry>(&cfg, &mut output);
|
||||
push_decl::<SessionConversationTurn>(&cfg, &mut output);
|
||||
push_decl::<SessionHistoryPage>(&cfg, &mut output);
|
||||
push_decl::<PendingSubmissionSummary>(&cfg, &mut output);
|
||||
push_decl::<PendingSubmissionsSnapshot>(&cfg, &mut output);
|
||||
push_decl::<SubmissionDisposition>(&cfg, &mut output);
|
||||
|
||||
@@ -655,6 +655,41 @@ pub struct WorkerSessionRequest {
|
||||
pub workspace_id: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct WorkerSessionHistoryRequest {
|
||||
pub workspace_id: String,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub cursor: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub limit: Option<usize>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum WorkerSessionHistoryUnavailableReason {
|
||||
RetentionMissing,
|
||||
ActivePointerMissing,
|
||||
CorruptLog,
|
||||
MigrationRequired,
|
||||
RetentionExpired,
|
||||
StorageUnavailable,
|
||||
InvalidCursor,
|
||||
ResourceLimit,
|
||||
Unsupported,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
|
||||
#[serde(tag = "availability", rename_all = "snake_case")]
|
||||
pub enum WorkerSessionHistoryAvailability {
|
||||
Page {
|
||||
page: protocol::SessionHistoryPage,
|
||||
},
|
||||
Unavailable {
|
||||
reason: WorkerSessionHistoryUnavailableReason,
|
||||
message: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct RetainedSessionIdentity {
|
||||
pub session_id: String,
|
||||
@@ -718,6 +753,17 @@ pub trait RuntimeApi {
|
||||
#[query] request: WorkerSessionRequest,
|
||||
) -> Result<WorkerSessionAvailability, RuntimeApiError>;
|
||||
|
||||
#[get(
|
||||
"/v1/workers/{worker_id}/session/history",
|
||||
status = 200,
|
||||
error_status = 400
|
||||
)]
|
||||
async fn worker_session_history(
|
||||
&self,
|
||||
#[path] worker_id: String,
|
||||
#[query] request: WorkerSessionHistoryRequest,
|
||||
) -> Result<WorkerSessionHistoryAvailability, RuntimeApiError>;
|
||||
|
||||
#[post("/v1/workers", status = 200, error_status = 400)]
|
||||
async fn create_worker(
|
||||
&self,
|
||||
@@ -1053,6 +1099,17 @@ mod tests {
|
||||
Ok(WorkerSessionAvailability::LiveProtocol)
|
||||
}
|
||||
|
||||
async fn worker_session_history(
|
||||
&self,
|
||||
_worker_id: String,
|
||||
_request: WorkerSessionHistoryRequest,
|
||||
) -> Result<WorkerSessionHistoryAvailability, RuntimeApiError> {
|
||||
Ok(WorkerSessionHistoryAvailability::Unavailable {
|
||||
reason: WorkerSessionHistoryUnavailableReason::Unsupported,
|
||||
message: "unsupported".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
async fn create_worker(
|
||||
&self,
|
||||
_request: CreateWorkerRequest,
|
||||
@@ -1174,10 +1231,22 @@ mod tests {
|
||||
assert_eq!(value["message"], "snapshot exceeds limit");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn worker_session_history_availability_has_closed_tagged_wire_shape() {
|
||||
let value = serde_json::to_value(WorkerSessionHistoryAvailability::Unavailable {
|
||||
reason: WorkerSessionHistoryUnavailableReason::InvalidCursor,
|
||||
message: "cursor is stale".to_string(),
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(value["availability"], "unavailable");
|
||||
assert_eq!(value["reason"], "invalid_cursor");
|
||||
assert_eq!(value["message"], "cursor is stale");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn contract_inventory_is_complete_and_unique() {
|
||||
let operations = RuntimeApiMetadata::OPERATIONS;
|
||||
assert_eq!(operations.len(), 16);
|
||||
assert_eq!(operations.len(), 17);
|
||||
let mut routes = operations
|
||||
.iter()
|
||||
.map(|operation| (format!("{:?}", operation.method), operation.path))
|
||||
|
||||
@@ -389,6 +389,20 @@ pub struct WorkspaceWorkerSessionResponse {
|
||||
pub observation: runtime_api::WorkerSessionAvailability,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct WorkspaceWorkerSessionHistoryQuery {
|
||||
pub cursor: Option<String>,
|
||||
pub limit: Option<usize>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
|
||||
pub struct WorkspaceWorkerSessionHistoryResponse {
|
||||
pub subject: WorkspaceWorkerSubject,
|
||||
#[serde(flatten)]
|
||||
pub observation: runtime_api::WorkerSessionHistoryAvailability,
|
||||
}
|
||||
|
||||
#[api(reqwest, axum, openapi)]
|
||||
pub trait ServerApi {
|
||||
#[get("/health", status = 200, error_status = 400)]
|
||||
@@ -1141,6 +1155,20 @@ pub trait ServerApi {
|
||||
#[path] worker_id: String,
|
||||
) -> Result<WorkspaceWorkerSessionResponse, ServerApiError>;
|
||||
|
||||
#[get(
|
||||
"/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/session/history",
|
||||
status = 200,
|
||||
error_status = 400,
|
||||
openapi = false
|
||||
)]
|
||||
async fn worker_session_history(
|
||||
&self,
|
||||
#[path] workspace_id: String,
|
||||
#[path] runtime_id: String,
|
||||
#[path] worker_id: String,
|
||||
#[query] query: WorkspaceWorkerSessionHistoryQuery,
|
||||
) -> Result<WorkspaceWorkerSessionHistoryResponse, ServerApiError>;
|
||||
|
||||
#[get(
|
||||
"/api/workers",
|
||||
status = 200,
|
||||
@@ -11506,7 +11534,7 @@ mod openapi_artifact_tests {
|
||||
];
|
||||
// Session observation retains its existing explicit non-OpenAPI boundary independently of
|
||||
// the Runtime/Worker-source signed operation inventory above.
|
||||
const OTHER_OPENAPI_EXCLUDED: &[&str] = &["worker_session"];
|
||||
const OTHER_OPENAPI_EXCLUDED: &[&str] = &["worker_session", "worker_session_history"];
|
||||
const DOCUMENTED_SIGNED_INTERNAL: &[&str] = &["workspace_runtime_config"];
|
||||
const BROWSER_ONLY: &[&str] = &["merge_request_selector_repair"];
|
||||
|
||||
|
||||
@@ -37,23 +37,31 @@ use std::path::Path;
|
||||
/// Read an existing retained file without updating its access timestamp.
|
||||
/// Observation fails closed when the platform cannot provide that guarantee.
|
||||
pub(crate) fn read_without_atime(path: &Path) -> io::Result<Vec<u8>> {
|
||||
read_without_atime_bounded(path, u64::MAX)
|
||||
}
|
||||
|
||||
/// Read at most `max_bytes + 1` retained bytes without updating access time.
|
||||
/// The extra byte lets callers prove that the authoritative file exceeded a
|
||||
/// limit without first racing a separate metadata length observation.
|
||||
pub(crate) fn read_without_atime_bounded(path: &Path, max_bytes: u64) -> io::Result<Vec<u8>> {
|
||||
#[cfg(any(target_os = "linux", target_os = "android"))]
|
||||
{
|
||||
use std::fs::OpenOptions;
|
||||
use std::io::Read;
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
|
||||
let mut file = OpenOptions::new()
|
||||
let file = OpenOptions::new()
|
||||
.read(true)
|
||||
.custom_flags(libc::O_NOATIME)
|
||||
.open(path)?;
|
||||
let mut bytes = Vec::new();
|
||||
file.read_to_end(&mut bytes)?;
|
||||
file.take(max_bytes.saturating_add(1))
|
||||
.read_to_end(&mut bytes)?;
|
||||
Ok(bytes)
|
||||
}
|
||||
#[cfg(not(any(target_os = "linux", target_os = "android")))]
|
||||
{
|
||||
let _ = path;
|
||||
let _ = (path, max_bytes);
|
||||
Err(io::Error::new(
|
||||
io::ErrorKind::Unsupported,
|
||||
"read-only retained observation requires no-atime file reads",
|
||||
@@ -88,8 +96,12 @@ pub use history::{
|
||||
pub use logged_item::{LoggedContentPart, LoggedItem, LoggedRole, from_logged, to_logged};
|
||||
pub use paste_artifact::PasteArtifactLimits;
|
||||
pub use public_snapshot::{
|
||||
DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, RetainedSessionIdentity, RetainedSessionSnapshot,
|
||||
RetainedSnapshotReadError, project_session_snapshot, read_retained_session_snapshot,
|
||||
DEFAULT_RETAINED_HISTORY_MAX_ENTRIES, DEFAULT_RETAINED_HISTORY_MAX_RESPONSE_BYTES,
|
||||
DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES, DEFAULT_RETAINED_HISTORY_MAX_SEGMENTS,
|
||||
DEFAULT_RETAINED_HISTORY_PAGE_TURNS, DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
|
||||
MAX_RETAINED_HISTORY_PAGE_TURNS, RetainedHistoryReadError, RetainedHistoryReadLimits,
|
||||
RetainedSessionIdentity, RetainedSessionSnapshot, RetainedSnapshotReadError,
|
||||
project_session_snapshot, read_retained_session_history_page, read_retained_session_snapshot,
|
||||
};
|
||||
pub use segment::{
|
||||
SegmentStartState, append_entry, append_system_item, classify_logged_history_entry,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -33,6 +33,9 @@ pub enum StoreError {
|
||||
#[error("log corrupted at line {line}: {message}")]
|
||||
Corrupt { line: usize, message: String },
|
||||
|
||||
#[error("retained observation read limit exceeded")]
|
||||
ReadLimitExceeded,
|
||||
|
||||
#[error("paste artifact storage is unavailable")]
|
||||
PasteArtifactUnsupported,
|
||||
|
||||
|
||||
@@ -111,6 +111,19 @@ impl WorkerSessionStore {
|
||||
session_id: SessionId,
|
||||
segment_id: SegmentId,
|
||||
) -> Result<Vec<LogEntry>, StoreError> {
|
||||
self.read_all_read_only_bounded(session_id, segment_id, u64::MAX)
|
||||
.map(|(entries, _)| entries)
|
||||
}
|
||||
|
||||
/// Read one retained Segment under an actual byte bound without touching
|
||||
/// access time. The file descriptor read, rather than a preceding stat, is
|
||||
/// the limit authority so concurrent appends cannot evade accounting.
|
||||
pub fn read_all_read_only_bounded(
|
||||
&self,
|
||||
session_id: SessionId,
|
||||
segment_id: SegmentId,
|
||||
max_bytes: u64,
|
||||
) -> Result<(Vec<LogEntry>, u64), StoreError> {
|
||||
let retained_session_id = self.session_id.lock().map_err(|_| StoreError::Corrupt {
|
||||
line: 0,
|
||||
message: "Worker Session identity lock poisoned".to_string(),
|
||||
@@ -123,14 +136,18 @@ impl WorkerSessionStore {
|
||||
});
|
||||
}
|
||||
let path = self.log_path(segment_id);
|
||||
let bytes = crate::read_without_atime(&path).map_err(|error| {
|
||||
let bytes = crate::read_without_atime_bounded(&path, max_bytes).map_err(|error| {
|
||||
if error.kind() == std::io::ErrorKind::NotFound {
|
||||
StoreError::NotFound(segment_id)
|
||||
} else {
|
||||
StoreError::Io(error)
|
||||
}
|
||||
})?;
|
||||
parse_jsonl(&bytes)
|
||||
let byte_len = bytes.len() as u64;
|
||||
if byte_len > max_bytes {
|
||||
return Err(StoreError::ReadLimitExceeded);
|
||||
}
|
||||
Ok((parse_jsonl(&bytes)?, byte_len))
|
||||
}
|
||||
|
||||
pub fn segment_log_len(&self, segment_id: SegmentId) -> Result<u64, StoreError> {
|
||||
|
||||
@@ -117,7 +117,7 @@ async fn in_process_host_runs_text_and_read_tool_then_shuts_down() {
|
||||
.expect("protocol event")
|
||||
.expect("worker event")
|
||||
{
|
||||
Event::UserMessage { segments }
|
||||
Event::UserMessage { segments, .. }
|
||||
if format!("{segments:?}").contains("read the probe") =>
|
||||
{
|
||||
saw_user_message = true;
|
||||
|
||||
@@ -1156,7 +1156,7 @@ impl App {
|
||||
Event::PendingSubmissionsChanged { pending } => {
|
||||
self.pending_submissions = pending;
|
||||
}
|
||||
Event::UserMessage { segments } => {
|
||||
Event::UserMessage { segments, .. } => {
|
||||
self.turn_index += 1;
|
||||
self.blocks.push(Block::TurnHeader {
|
||||
turn: self.turn_index,
|
||||
@@ -1176,6 +1176,9 @@ impl App {
|
||||
}
|
||||
self.assistant_streaming = false;
|
||||
}
|
||||
// The TUI already renders assistant/tool streaming events. This commit
|
||||
// marker exists so identity-aware clients can reconcile durable history.
|
||||
Event::SessionEntryCommitted { .. } => {}
|
||||
Event::SystemItem { item } => {
|
||||
self.apply_system_item(&item);
|
||||
self.assistant_streaming = false;
|
||||
@@ -3399,6 +3402,7 @@ mod completion_flow_tests {
|
||||
]),
|
||||
});
|
||||
app.handle_worker_event(Event::UserMessage {
|
||||
entry_id: None,
|
||||
segments: vec![Segment::text("first persisted message")],
|
||||
});
|
||||
|
||||
@@ -3417,6 +3421,7 @@ mod completion_flow_tests {
|
||||
assert_eq!(input_text(&app), "");
|
||||
|
||||
app.handle_worker_event(Event::UserMessage {
|
||||
entry_id: None,
|
||||
segments: submitted,
|
||||
});
|
||||
// Simulate run-derived attachment display after the submitted user line.
|
||||
@@ -3457,6 +3462,7 @@ mod completion_flow_tests {
|
||||
let mut app = App::new("test".into());
|
||||
let submitted = submit_text(&mut app, "original submit");
|
||||
app.handle_worker_event(Event::UserMessage {
|
||||
entry_id: None,
|
||||
segments: submitted,
|
||||
});
|
||||
for c in "draft while running".chars() {
|
||||
@@ -3485,6 +3491,7 @@ mod completion_flow_tests {
|
||||
let mut app = App::new("test".into());
|
||||
let submitted = submit_text(&mut app, "normal run");
|
||||
app.handle_worker_event(Event::UserMessage {
|
||||
entry_id: None,
|
||||
segments: submitted,
|
||||
});
|
||||
app.handle_worker_event(Event::RunEnd { result });
|
||||
|
||||
@@ -1688,6 +1688,7 @@ mod tests {
|
||||
);
|
||||
|
||||
connection.observe_worker_event(&Event::UserMessage {
|
||||
entry_id: None,
|
||||
segments: vec![Segment::text("inspect"), Segment::UploadedFile { file }],
|
||||
});
|
||||
assert!(connection.pending_attachments.is_empty());
|
||||
@@ -1776,7 +1777,10 @@ mod tests {
|
||||
assert!(!reconcile_attachment_submission(
|
||||
&mut pending,
|
||||
&mut awaiting,
|
||||
&Event::UserMessage { segments },
|
||||
&Event::UserMessage {
|
||||
entry_id: None,
|
||||
segments,
|
||||
},
|
||||
));
|
||||
assert_eq!(awaiting, expected);
|
||||
assert!(pending.is_empty());
|
||||
|
||||
@@ -477,6 +477,13 @@ pub struct WorkerSessionObservationRequest {
|
||||
pub worker_ref: WorkerRef,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct WorkerSessionHistoryRequest {
|
||||
pub worker_ref: WorkerRef,
|
||||
pub cursor: Option<String>,
|
||||
pub limit: Option<usize>,
|
||||
}
|
||||
|
||||
pub trait WorkerExecutionBackend: Send + Sync + 'static {
|
||||
fn backend_id(&self) -> &str;
|
||||
|
||||
@@ -497,6 +504,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
|
||||
}
|
||||
}
|
||||
|
||||
fn worker_session_history(
|
||||
&self,
|
||||
_request: WorkerSessionHistoryRequest,
|
||||
) -> runtime_api::WorkerSessionHistoryAvailability {
|
||||
runtime_api::WorkerSessionHistoryAvailability::Unavailable {
|
||||
reason: runtime_api::WorkerSessionHistoryUnavailableReason::Unsupported,
|
||||
message: "session history paging is not supported by this Runtime".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult;
|
||||
|
||||
/// Read-only restore validation performed before any live controller,
|
||||
@@ -726,6 +743,13 @@ impl WorkerExecutionBackendRef {
|
||||
self.backend.worker_session(request)
|
||||
}
|
||||
|
||||
pub(crate) fn worker_session_history(
|
||||
&self,
|
||||
request: WorkerSessionHistoryRequest,
|
||||
) -> runtime_api::WorkerSessionHistoryAvailability {
|
||||
self.backend.worker_session_history(request)
|
||||
}
|
||||
|
||||
pub(crate) fn spawn_worker(
|
||||
&self,
|
||||
request: WorkerExecutionSpawnRequest,
|
||||
|
||||
@@ -260,6 +260,35 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
|
||||
.map_err(api_error)
|
||||
}
|
||||
|
||||
async fn worker_session_history(
|
||||
&self,
|
||||
worker_id: String,
|
||||
request: runtime_api::WorkerSessionHistoryRequest,
|
||||
) -> Result<runtime_api::WorkerSessionHistoryAvailability, runtime_api::RuntimeApiError> {
|
||||
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
|
||||
let scope = auth_workspace_scope(&self.state, auth_extension().as_ref())
|
||||
.map_err(api_error)?
|
||||
.ok_or_else(|| {
|
||||
runtime_api::RuntimeApiError::new(
|
||||
StatusCode::FORBIDDEN.as_u16(),
|
||||
"runtime_worker_session_scope_required",
|
||||
"Worker Session history requires a Workspace-scoped capability",
|
||||
)
|
||||
})?;
|
||||
if scope.workspace_id != request.workspace_id {
|
||||
return Err(runtime_api::RuntimeApiError::new(
|
||||
StatusCode::FORBIDDEN.as_u16(),
|
||||
"runtime_worker_session_workspace_scope_mismatch",
|
||||
"Worker Session Workspace scope does not match the authenticated capability",
|
||||
));
|
||||
}
|
||||
self.state
|
||||
.runtime
|
||||
.worker_session_history_scoped(&scope, &worker_ref, request.cursor, request.limit)
|
||||
.map_err(RuntimeHttpRestError::runtime)
|
||||
.map_err(api_error)
|
||||
}
|
||||
|
||||
async fn create_worker(
|
||||
&self,
|
||||
value: runtime_api::CreateWorkerRequest,
|
||||
|
||||
@@ -16,7 +16,8 @@ use crate::execution::{
|
||||
WorkerExecutionBackend, WorkerExecutionBackendRef, WorkerExecutionHandle,
|
||||
WorkerExecutionOperation, WorkerExecutionResult, WorkerExecutionSpawnRequest,
|
||||
WorkerExecutionSpawnResult, WorkerExecutionStopRequest, WorkerLifecycleOperationId,
|
||||
WorkerSessionObservationRequest, WorkspaceConfigFetchRequest, WorkspaceConfigFetchResult,
|
||||
WorkerSessionHistoryRequest, WorkerSessionObservationRequest, WorkspaceConfigFetchRequest,
|
||||
WorkspaceConfigFetchResult,
|
||||
};
|
||||
#[cfg(feature = "fs-store")]
|
||||
use crate::fs_store::{
|
||||
@@ -1613,6 +1614,38 @@ impl Runtime {
|
||||
})
|
||||
}
|
||||
|
||||
/// Read one bounded page from the adopted durable Session lineage without
|
||||
/// restoring the Worker or modifying model context.
|
||||
pub fn worker_session_history_scoped(
|
||||
&self,
|
||||
scope: &RuntimeWorkspaceScope,
|
||||
worker_ref: &WorkerRef,
|
||||
cursor: Option<String>,
|
||||
limit: Option<usize>,
|
||||
) -> Result<runtime_api::WorkerSessionHistoryAvailability, RuntimeError> {
|
||||
let operation_lock = self.worker_operation_lock(worker_ref.worker_id)?;
|
||||
let _operation_guard = operation_lock
|
||||
.lock()
|
||||
.map_err(|_| RuntimeError::StatePoisoned)?;
|
||||
self.ensure_worker_in_workspace(scope, worker_ref)?;
|
||||
let backend = {
|
||||
let state = self.lock()?;
|
||||
state.worker(worker_ref)?;
|
||||
state.execution_backend.clone()
|
||||
};
|
||||
Ok(match backend {
|
||||
Some(backend) => backend.worker_session_history(WorkerSessionHistoryRequest {
|
||||
worker_ref: worker_ref.clone(),
|
||||
cursor,
|
||||
limit,
|
||||
}),
|
||||
None => runtime_api::WorkerSessionHistoryAvailability::Unavailable {
|
||||
reason: runtime_api::WorkerSessionHistoryUnavailableReason::Unsupported,
|
||||
message: "session history paging is not supported by this Runtime".to_string(),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
/// Attach a live execution through a workspace-scoped Runtime authorization context.
|
||||
pub fn restore_worker_scoped(
|
||||
&self,
|
||||
|
||||
@@ -172,6 +172,15 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static {
|
||||
Err(session_store::RetainedSnapshotReadError::RetentionMissing)
|
||||
}
|
||||
|
||||
fn retained_session_history_page(
|
||||
&self,
|
||||
_worker_ref: &WorkerRef,
|
||||
_cursor: Option<&str>,
|
||||
_limit: Option<usize>,
|
||||
) -> Result<protocol::SessionHistoryPage, session_store::RetainedHistoryReadError> {
|
||||
Err(session_store::RetainedHistoryReadError::RetentionMissing)
|
||||
}
|
||||
|
||||
async fn spawn_controller(
|
||||
&self,
|
||||
request: WorkerExecutionSpawnRequest,
|
||||
@@ -1027,6 +1036,25 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
|
||||
)
|
||||
}
|
||||
|
||||
fn retained_session_history_page(
|
||||
&self,
|
||||
worker_ref: &WorkerRef,
|
||||
cursor: Option<&str>,
|
||||
limit: Option<usize>,
|
||||
) -> Result<protocol::SessionHistoryPage, session_store::RetainedHistoryReadError> {
|
||||
let aggregate_dir = self
|
||||
.worker_aggregate_dir(worker_ref)
|
||||
.map_err(|_| session_store::RetainedHistoryReadError::RetentionMissing)?;
|
||||
let worker_name = Self::runtime_worker_name_for_ref(worker_ref);
|
||||
session_store::read_retained_session_history_page(
|
||||
&aggregate_dir,
|
||||
&worker_name,
|
||||
cursor,
|
||||
limit,
|
||||
session_store::RetainedHistoryReadLimits::default(),
|
||||
)
|
||||
}
|
||||
|
||||
fn observe_workspace_prompt_projection(
|
||||
&self,
|
||||
projection: worker::WorkspacePromptProjection,
|
||||
@@ -2216,6 +2244,48 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
fn worker_session_history(
|
||||
&self,
|
||||
request: crate::execution::WorkerSessionHistoryRequest,
|
||||
) -> runtime_api::WorkerSessionHistoryAvailability {
|
||||
match self.factory.retained_session_history_page(
|
||||
&request.worker_ref,
|
||||
request.cursor.as_deref(),
|
||||
request.limit,
|
||||
) {
|
||||
Ok(page) => runtime_api::WorkerSessionHistoryAvailability::Page { page },
|
||||
Err(error) => {
|
||||
let reason = match error {
|
||||
session_store::RetainedHistoryReadError::RetentionMissing => {
|
||||
runtime_api::WorkerSessionHistoryUnavailableReason::RetentionMissing
|
||||
}
|
||||
session_store::RetainedHistoryReadError::ActivePointerMissing => {
|
||||
runtime_api::WorkerSessionHistoryUnavailableReason::ActivePointerMissing
|
||||
}
|
||||
session_store::RetainedHistoryReadError::MigrationRequired => {
|
||||
runtime_api::WorkerSessionHistoryUnavailableReason::MigrationRequired
|
||||
}
|
||||
session_store::RetainedHistoryReadError::CorruptLog => {
|
||||
runtime_api::WorkerSessionHistoryUnavailableReason::CorruptLog
|
||||
}
|
||||
session_store::RetainedHistoryReadError::StorageUnavailable => {
|
||||
runtime_api::WorkerSessionHistoryUnavailableReason::StorageUnavailable
|
||||
}
|
||||
session_store::RetainedHistoryReadError::InvalidCursor => {
|
||||
runtime_api::WorkerSessionHistoryUnavailableReason::InvalidCursor
|
||||
}
|
||||
session_store::RetainedHistoryReadError::ResourceLimit => {
|
||||
runtime_api::WorkerSessionHistoryUnavailableReason::ResourceLimit
|
||||
}
|
||||
};
|
||||
runtime_api::WorkerSessionHistoryAvailability::Unavailable {
|
||||
reason,
|
||||
message: error.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn observe_workspace_prompt_projection(
|
||||
&self,
|
||||
projection: worker::WorkspacePromptProjection,
|
||||
|
||||
@@ -619,8 +619,8 @@ mod tests {
|
||||
let (committed_tx, committed_rx) = mpsc::channel();
|
||||
let commit_thread = thread::spawn(move || {
|
||||
// This mirrors Worker::append_entry ordering: clear in-flight first,
|
||||
// then publish the finalized AssistantItem. AssistantItem entries
|
||||
// are mirror-only and are not delivered as live entry events.
|
||||
// then publish the finalized AssistantItem. Identity-aware clients
|
||||
// reconcile this durable commit with the preceding streamed block.
|
||||
in_flight_for_commit.clear_for_committed_item_then(&assistant_item, || {
|
||||
sink_for_commit.publish(assistant_entry);
|
||||
});
|
||||
@@ -639,7 +639,10 @@ mod tests {
|
||||
in_flight_snapshot.blocks.as_slice(),
|
||||
[InFlightBlock::Text { text, finished: true }] if text == "done"
|
||||
));
|
||||
assert!(entry_rx.try_recv().is_err());
|
||||
assert!(matches!(
|
||||
entry_rx.try_recv(),
|
||||
Ok(LogEntry::AnnotatedAssistantItem { .. })
|
||||
));
|
||||
let post_commit_guard = in_flight.snapshot_guard();
|
||||
assert!(snapshot_from_guard(&post_commit_guard).is_empty());
|
||||
}
|
||||
|
||||
@@ -34,7 +34,41 @@ pub fn live_log_entry_event(entry: LogEntry) -> Option<Event> {
|
||||
session_store::public_snapshot::project_current_session_snapshot(&[entry]);
|
||||
Some(Event::SegmentRotated { session })
|
||||
}
|
||||
LogEntry::AnnotatedUserInput { segments, .. } => Some(Event::UserMessage { segments }),
|
||||
LogEntry::AnnotatedUserInput {
|
||||
ts,
|
||||
segments,
|
||||
history,
|
||||
extensions,
|
||||
} => {
|
||||
let projected = session_store::public_snapshot::project_current_session_snapshot(&[
|
||||
LogEntry::AnnotatedUserInput {
|
||||
ts,
|
||||
segments: segments.clone(),
|
||||
history,
|
||||
extensions,
|
||||
},
|
||||
]);
|
||||
let entry_id = projected
|
||||
.entries
|
||||
.iter()
|
||||
.find(|entry| {
|
||||
matches!(
|
||||
entry.data,
|
||||
protocol::SessionSnapshotEntryData::UserInput { .. }
|
||||
)
|
||||
})
|
||||
.map(|entry| entry.entry_id.clone());
|
||||
Some(Event::UserMessage { entry_id, segments })
|
||||
}
|
||||
entry
|
||||
@ (LogEntry::AnnotatedAssistantItem { .. } | LogEntry::AnnotatedToolResult { .. }) => {
|
||||
let mut projected =
|
||||
session_store::public_snapshot::project_current_session_snapshot(&[entry]);
|
||||
projected
|
||||
.entries
|
||||
.pop()
|
||||
.map(|entry| Event::SessionEntryCommitted { entry })
|
||||
}
|
||||
LogEntry::AnnotatedSystemItem { entry, .. } => {
|
||||
let value = serde_json::to_value(&entry.item).expect("SystemItem is Serialize");
|
||||
Some(Event::SystemItem { item: value })
|
||||
@@ -94,7 +128,9 @@ mod tests {
|
||||
.expect("UserInput must be live-relevant");
|
||||
|
||||
match event {
|
||||
Event::UserMessage { segments: echoed } => assert_eq!(echoed, segments),
|
||||
Event::UserMessage {
|
||||
segments: echoed, ..
|
||||
} => assert_eq!(echoed, segments),
|
||||
other => panic!("expected UserMessage, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -92,14 +92,11 @@ impl SegmentLogSink {
|
||||
/// - `LogEntry::AnnotatedSegmentStart` → `Event::SegmentRotated` on the wire.
|
||||
/// - `LogEntry::AnnotatedUserInput` → `Event::UserMessage`.
|
||||
/// - `LogEntry::AnnotatedSystemItem` → `Event::SystemItem`.
|
||||
/// - assistant/tool commits → `Event::SessionEntryCommitted`.
|
||||
/// - `LogEntry::Invoke` → `Event::InvokeStart`.
|
||||
/// Everything else (AssistantItem, ToolResult, TurnEnd,
|
||||
/// RunCompleted, RunErrored, PausedTurnAbandoned, LlmUsage, Extension,
|
||||
/// ConfigChanged) is reflected in the mirror so reconnect snapshots stay accurate,
|
||||
/// but is not sent live — the streaming events (TextDelta /
|
||||
/// ToolCallStart / ToolResult / TurnEnd / etc.) already provide
|
||||
/// that data, and re-broadcasting it as a typed entry would just
|
||||
/// double-render every block on the client side.
|
||||
/// Operational checkpoints remain mirror-only. Assistant/tool streaming
|
||||
/// events are still the in-flight authority; the committed-entry event adds
|
||||
/// the durable stable identity needed to reconcile retained history.
|
||||
pub fn publish(&self, entry: LogEntry) {
|
||||
let mut mirror = self
|
||||
.inner
|
||||
@@ -122,6 +119,8 @@ impl SegmentLogSink {
|
||||
entry,
|
||||
LogEntry::AnnotatedSegmentStart { .. }
|
||||
| LogEntry::AnnotatedUserInput { .. }
|
||||
| LogEntry::AnnotatedAssistantItem { .. }
|
||||
| LogEntry::AnnotatedToolResult { .. }
|
||||
| LogEntry::AnnotatedSystemItem { .. }
|
||||
| LogEntry::Invoke { .. }
|
||||
)
|
||||
|
||||
@@ -1188,6 +1188,19 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
|
||||
})
|
||||
}
|
||||
|
||||
fn worker_session_history(
|
||||
&self,
|
||||
_workspace_id: &str,
|
||||
_worker_ref: &EmbeddedWorkerRef,
|
||||
_cursor: Option<String>,
|
||||
_limit: Option<usize>,
|
||||
) -> Result<runtime_api::WorkerSessionHistoryAvailability, String> {
|
||||
Ok(runtime_api::WorkerSessionHistoryAvailability::Unavailable {
|
||||
reason: runtime_api::WorkerSessionHistoryUnavailableReason::Unsupported,
|
||||
message: "session history paging is not supported by this Runtime".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
fn observation_source(
|
||||
&self,
|
||||
_worker_id: &str,
|
||||
@@ -2017,6 +2030,31 @@ impl RuntimeRegistry {
|
||||
})
|
||||
}
|
||||
|
||||
pub fn worker_session_history(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
worker: &RuntimeWorkerRef,
|
||||
cursor: Option<String>,
|
||||
limit: Option<usize>,
|
||||
) -> Result<runtime_api::WorkerSessionHistoryAvailability, RuntimeRegistryError> {
|
||||
validate_backend_identifier("runtime_id", &worker.runtime_id)?;
|
||||
validate_backend_identifier("worker_id", &worker.worker_id)?;
|
||||
let runtime = self.runtime(&worker.runtime_id)?;
|
||||
let worker_ref =
|
||||
EmbeddedWorkerRef::new(EmbeddedWorkerId::parse(&worker.worker_id).ok_or_else(
|
||||
|| RuntimeRegistryError::UnknownWorker {
|
||||
worker: worker.clone(),
|
||||
},
|
||||
)?);
|
||||
runtime
|
||||
.worker_session_history(workspace_id, &worker_ref, cursor, limit)
|
||||
.map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
|
||||
runtime_id: worker.runtime_id.clone(),
|
||||
code: "worker_session_history_failed".to_string(),
|
||||
message,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn observation_source(
|
||||
&self,
|
||||
worker: &RuntimeWorkerRef,
|
||||
@@ -2909,6 +2947,19 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
|
||||
fn worker_session_history(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
worker_ref: &EmbeddedWorkerRef,
|
||||
cursor: Option<String>,
|
||||
limit: Option<usize>,
|
||||
) -> Result<runtime_api::WorkerSessionHistoryAvailability, String> {
|
||||
let scope = RuntimeWorkspaceScope::new(workspace_id, "embedded-backend");
|
||||
self.runtime
|
||||
.worker_session_history_scoped(&scope, worker_ref, cursor, limit)
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
|
||||
fn observation_source(
|
||||
&self,
|
||||
worker_id: &str,
|
||||
@@ -4924,6 +4975,27 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
|
||||
.map_err(|diagnostic| diagnostic.message)
|
||||
}
|
||||
|
||||
fn worker_session_history(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
worker_ref: &EmbeddedWorkerRef,
|
||||
cursor: Option<String>,
|
||||
limit: Option<usize>,
|
||||
) -> Result<runtime_api::WorkerSessionHistoryAvailability, String> {
|
||||
let worker_id = worker_ref.worker_id.to_string();
|
||||
let request = runtime_api::WorkerSessionHistoryRequest {
|
||||
workspace_id: workspace_id.to_string(),
|
||||
cursor,
|
||||
limit,
|
||||
};
|
||||
self.run_runtime_api(
|
||||
self.request_timeout,
|
||||
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
|
||||
move |client| async move { client.worker_session_history(worker_id, request).await },
|
||||
)
|
||||
.map_err(|diagnostic| diagnostic.message)
|
||||
}
|
||||
|
||||
fn observation_source(
|
||||
&self,
|
||||
worker_id: &str,
|
||||
|
||||
@@ -4584,6 +4584,9 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
|
||||
.merge(server_api::server_api_axum::skill_get(service.clone()))
|
||||
.merge(server_api::server_api_axum::skill_activate(service.clone()))
|
||||
.merge(server_api::server_api_axum::worker_session(service.clone()))
|
||||
.merge(server_api::server_api_axum::worker_session_history(
|
||||
service.clone(),
|
||||
))
|
||||
.merge(server_api::server_api_axum::repository_list(
|
||||
service.clone(),
|
||||
))
|
||||
@@ -6248,6 +6251,50 @@ impl server_api::ServerApi for ServerApiContractService {
|
||||
})
|
||||
}
|
||||
|
||||
async fn worker_session_history(
|
||||
&self,
|
||||
workspace_id: String,
|
||||
runtime_id: String,
|
||||
worker_id: String,
|
||||
query: server_api::WorkspaceWorkerSessionHistoryQuery,
|
||||
) -> std::result::Result<
|
||||
server_api::WorkspaceWorkerSessionHistoryResponse,
|
||||
server_api::ServerApiError,
|
||||
> {
|
||||
let api = self.workspace_api().map_err(|_| {
|
||||
runtime_api::RuntimeApiError::new(
|
||||
StatusCode::NOT_FOUND.as_u16(),
|
||||
"workspace_not_found",
|
||||
"Workspace was not found",
|
||||
)
|
||||
})?;
|
||||
if workspace_id != api.config.workspace_id {
|
||||
return Err(runtime_api::RuntimeApiError::new(
|
||||
StatusCode::NOT_FOUND.as_u16(),
|
||||
"workspace_not_found",
|
||||
"Workspace was not found",
|
||||
));
|
||||
}
|
||||
let worker = RuntimeWorkerRef::new(runtime_id.clone(), worker_id.clone());
|
||||
let observation = api
|
||||
.runtime
|
||||
.worker_session_history(&workspace_id, &worker, query.cursor, query.limit)
|
||||
.map_err(|_error| {
|
||||
runtime_api::RuntimeApiError::new(
|
||||
StatusCode::BAD_GATEWAY.as_u16(),
|
||||
"worker_session_history_failed",
|
||||
"Worker Session history request failed",
|
||||
)
|
||||
})?;
|
||||
Ok(server_api::WorkspaceWorkerSessionHistoryResponse {
|
||||
subject: server_api::WorkspaceWorkerSubject::RuntimeWorker {
|
||||
runtime_id,
|
||||
worker_id,
|
||||
},
|
||||
observation,
|
||||
})
|
||||
}
|
||||
|
||||
async fn workspace_worker_list_alias(
|
||||
&self,
|
||||
) -> std::result::Result<server_api::WorkerListResponse, server_api::RepositoryApiError> {
|
||||
|
||||
Reference in New Issue
Block a user