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

Merge develop into T-693 and reconcile Subject behavior with Memory information

Browse Source 
Normal merge of develop@b0bce50b retains T-692 state hierarchy, diagnostics disclosures, ready-empty and candidate-only regressions alongside T-693 behavior/CAS/route fences. ResidentContext load and refresh retain the five-second timeout from 685b8aa6, with a seven-request regression.

Validation: root cargo check; full worker(serial), session-store, workspace-server and server-api(typescript) suites; frontend check with zero diagnostics; 464 Web tests and 171 component tests (19 Memory); production Memory browser-shell test/build; fmt/diff/OpenAPI checks.
This commit is contained in:
Hare
2026-10-05 17:31:17 +09:00
parent 06480b9c7c b0bce50bb4
commit a4af26e8fd
49 changed files with 2998 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/feature/builtin/memory.rs
+51 -9
View File
@@ -6,6 +6,7 @@
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
use arc_swap::ArcSwap;
@@ -157,9 +158,24 @@ impl dyn WorkspaceClient + '_ {
}
}
const RESIDENT_SUMMARY_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
async fn execute_subjektiv_memory_operation(
client: &dyn WorkspaceClient,
operation: server_api::SubjektivMemoryBackendOperation,
) -> Result<server_api::SubjektivMemoryBackendResponse, WorkspaceMemoryBackendError> {
execute_subjektiv_memory_operation_with(client, operation, |client, request| {
client.execute(request)
})
}
fn execute_subjektiv_memory_operation_with(
client: &dyn WorkspaceClient,
operation: server_api::SubjektivMemoryBackendOperation,
execute: impl FnOnce(
&dyn WorkspaceClient,
WorkspaceRequest,
) -> Result<crate::worker::WorkspaceResponse, WorkspaceClientError>,
) -> Result<server_api::SubjektivMemoryBackendResponse, WorkspaceMemoryBackendError> {
let workspace_id =
client
@@ -170,11 +186,14 @@ async fn execute_subjektiv_memory_operation(
client.kind()
),
})?;
let response = client.execute(WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("/api/w/{workspace_id}/subjektiv/memory"),
serde_json::to_string(&server_api::SubjektivMemoryBackendRequest { operation })?,
))?;
let response = execute(
client,
WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("/api/w/{workspace_id}/subjektiv/memory"),
serde_json::to_string(&server_api::SubjektivMemoryBackendRequest { operation })?,
),
)?;
let status = reqwest::StatusCode::from_u16(response.status)
.unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
if !response.is_success() {
@@ -712,12 +731,13 @@ impl WorkspaceSubjektivResidentSummarySource {
async fn fetch(
&self,
) -> Result<server_api::SubjektivResidentContextOutput, WorkspaceMemoryBackendError> {
match execute_subjektiv_memory_operation(
match execute_subjektiv_memory_operation_with(
self.client.as_ref(),
server_api::SubjektivMemoryBackendOperation::ResidentContext(Default::default()),
)
.await?
{
|client, request| {
client.execute_with_timeout(request, RESIDENT_SUMMARY_REQUEST_TIMEOUT)
},
)? {
server_api::SubjektivMemoryBackendResponse::ResidentContext(output) => Ok(output),
other => Err(WorkspaceMemoryBackendError::InvalidResponse(format!(
"unexpected subject resident context response: {other:?}"
@@ -1291,6 +1311,7 @@ mod tests {
scope_allowed: bool,
behavior: Mutex<(u64, String)>,
paths: Mutex<Vec<String>>,
timeouts: Mutex<Vec<Duration>>,
}
impl SubjectResidentClient {
@@ -1304,6 +1325,7 @@ mod tests {
scope_allowed: true,
behavior: Mutex::new((3, "Be deliberate.".to_string())),
paths: Mutex::new(Vec::new()),
timeouts: Mutex::new(Vec::new()),
}
}
@@ -1314,6 +1336,7 @@ mod tests {
scope_allowed: false,
behavior: Mutex::new((3, "must not be injected".to_string())),
paths: Mutex::new(Vec::new()),
timeouts: Mutex::new(Vec::new()),
}
}
}
@@ -1367,6 +1390,15 @@ mod tests {
.unwrap(),
})
}
fn execute_with_timeout(
&self,
request: WorkspaceRequest,
timeout: Duration,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.timeouts.lock().unwrap().push(timeout);
self.execute(request)
}
}
fn resident_client(content: &str) -> Arc<dyn WorkspaceClient> {
@@ -1681,6 +1713,10 @@ permission = "write"
ready_client.paths.lock().unwrap().clone(),
vec!["/api/w/workspace/subjektiv/memory".to_string()]
);
assert_eq!(
ready_client.timeouts.lock().unwrap().as_slice(),
&[RESIDENT_SUMMARY_REQUEST_TIMEOUT]
);
let ready_empty_client = Arc::new(SubjectResidentClient::new(
memory::backend::MemoryResidentSummaryAvailability::Ready,
@@ -1790,6 +1826,12 @@ permission = "write"
.body
.contains("No user-managed behavior is set")
);
// Initial load and every live update/clear/replay probe retain the
// target-side bounded resident request contract.
assert_eq!(
client.timeouts.lock().unwrap().as_slice(),
&[RESIDENT_SUMMARY_REQUEST_TIMEOUT; 7]
);
}
#[tokio::test]
crates/worker/src/image_recovery_tests.rs
+221
View File
@@ -0,0 +1,221 @@
//! Recovery tests use the normal Worker execution and append-log commit gate.
use super::*;
use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
use agen::llm_client::{ClientError, Request, ResponseStream};
use agen::tool::{Attachment, ImageAttachment};
use async_trait::async_trait;
#[derive(Clone, Default)]
struct RejectImages {
requests: Arc<Mutex<Vec<Request>>>,
cancel_on_retry: Arc<Mutex<Option<tokio::sync::mpsc::Sender<()>>>>,
}
#[async_trait]
impl LlmClient for RejectImages {
fn clone_boxed(&self) -> Box<dyn LlmClient> {
Box::new(self.clone())
}
async fn stream(&self, request: Request) -> Result<ResponseStream, ClientError> {
let has_image = request.items.iter().any(
|item| matches!(item, Item::ToolResult { attachments, .. } if !attachments.is_empty()),
);
self.requests.lock().unwrap().push(request);
if has_image {
return Err(ClientError::Api {
status: Some(400),
code: Some("invalid_value".into()),
message: "The image requires 52150 patches, exceeding the limit of 30000.".into(),
retry_after: None,
});
}
if let Some(cancel) = self.cancel_on_retry.lock().unwrap().take() {
cancel.try_send(()).unwrap();
return Ok(Box::pin(futures::stream::pending()));
}
Ok(Box::pin(futures::stream::iter(vec![
Ok(LlmEvent::text_block_start(0)),
Ok(LlmEvent::text_delta(0, "Resize the source image")),
Ok(LlmEvent::text_block_stop(0, None)),
Ok(LlmEvent::Status(StatusEvent {
status: ResponseStatus::Completed,
})),
])))
}
}
async fn image_worker() -> (
tempfile::TempDir,
Worker<RejectImages, session_store::FsStore>,
RejectImages,
) {
let dir = tempfile::tempdir().unwrap();
let store = session_store::FsStore::new(dir.path().join("session")).unwrap();
let manifest = WorkerManifest::from_toml(
r#"
[worker]
name = "image-recovery-test"
[model]
scheme = "anthropic"
model_id = "test"
[engine]
[scope]
allow = []
"#,
)
.unwrap();
let client = RejectImages::default();
let mut worker = Worker::new(
manifest,
Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(client.clone()),
store,
WorkerWorkspaceContext::local_filesystem(None),
WorkerFilesystemAuthority::None,
Scope::writable(dir.path()).unwrap(),
)
.await
.unwrap();
worker.disable_manifest_lifecycle_features();
worker.set_history_for_test(vec![
Item::user_message("inspect this image"),
Item::tool_call("image-call", "ViewImage", "{}"),
Item::tool_result_item_with_disposition_and_attachments(
"image-call",
"Attached image",
None,
agen::ToolResultDisposition::Success,
vec![Attachment::Image(ImageAttachment::new(
"image/png",
vec![1_u8, 2, 3],
))],
),
]);
worker.wire_history_persistence();
(dir, worker, client)
}
fn rejected(item: &Item) -> bool {
matches!(item, Item::ToolResult { call_id, is_error: true, attachments, .. } if call_id == "image-call" && attachments.is_empty())
}
#[tokio::test]
async fn image_correction_is_committed_before_retry_and_survives_restore_and_reseed() {
let (_dir, mut worker, client) = image_worker().await;
assert_eq!(
worker.run_text("continue").await.unwrap(),
WorkerRunResult::Finished
);
assert_eq!(client.requests.lock().unwrap().len(), 2);
let raw = worker
.store
.read_all(worker.session_id(), worker.segment_id())
.unwrap();
let correction = raw
.iter()
.find(|entry| matches!(entry, LogEntry::ToolResultCorrected { .. }))
.unwrap()
.clone();
let restored = restore_history_entries(worker.session_id(), worker.segment_id(), &raw).unwrap();
assert_eq!(
restored
.iter()
.filter(|entry| rejected(&entry.item))
.count(),
1
);
assert_eq!(
session_store::collect_state(&raw)
.history
.iter()
.filter(|item| rejected(item))
.count(),
1
);
let LogEntry::ToolResultCorrected {
entry: corrected, ..
} = &correction
else {
unreachable!()
};
let original = match &raw[0] {
LogEntry::AnnotatedSegmentStart { history, .. } => history
.iter()
.find(|entry| entry.metadata.entry_id == corrected.metadata.entry_id)
.unwrap(),
_ => panic!("missing seed"),
};
assert!(
matches!(&original.item, session_store::LoggedItem::ToolResult { attachments, is_error: false, .. } if attachments.len() == 1)
);
assert_eq!(
restored
.iter()
.find(|entry| rejected(&entry.item))
.unwrap()
.annotation,
original.metadata
);
let snapshot = session_store::public_snapshot::project_current_session_snapshot(&raw);
let projected = snapshot
.entries
.iter()
.find(|entry| entry.entry_id == corrected.metadata.entry_id.0)
.unwrap();
assert!(
matches!(&projected.data, protocol::SessionSnapshotEntryData::ToolResult { is_error: true, attachments, .. } if attachments.is_empty())
);
assert!(matches!(
crate::ipc::protocol_session::live_log_entry_event(correction),
Some(Event::SessionEntryCommitted { .. })
));
// Fork/compact seed writers serialize the effective annotated history.
let seed = LogEntry::AnnotatedSegmentStart {
ts: 1,
session_id: worker.session_id(),
system_prompt: None,
config: agen::llm_client::RequestConfig::default(),
history: restored.iter().map(to_logged_history_entry).collect(),
forked_from: None,
compacted_from: None,
};
let restored_again =
restore_history_entries(worker.session_id(), worker.segment_id(), &[seed]).unwrap();
assert_eq!(
restored_again
.iter()
.filter(|entry| rejected(&entry.item))
.count(),
1
);
let mut history = History::from_entries(restored_again);
let out = Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(client.clone())
.run_with_annotation(&mut history, "continue after restart", &mut |_| {
Ok(SessionHistoryMetadata::legacy_unknown())
})
.await;
assert!(matches!(out.result, EngineRunExit::Finished));
assert_eq!(client.requests.lock().unwrap().len(), 3);
}
#[tokio::test]
async fn cancellation_after_image_correction_does_not_rollback_the_correction() {
let (_dir, mut worker, client) = image_worker().await;
*client.cancel_on_retry.lock().unwrap() = Some(worker.engine().cancel_sender());
let result = worker.run_text("continue").await.unwrap();
assert_eq!(result, WorkerRunResult::Cancelled);
let raw = worker
.store
.read_all(worker.session_id(), worker.segment_id())
.unwrap();
assert!(
raw.iter()
.any(|entry| matches!(entry, LogEntry::ToolResultCorrected { .. }))
);
let restored = restore_history_entries(worker.session_id(), worker.segment_id(), &raw).unwrap();
assert_eq!(
restored
.iter()
.filter(|entry| rejected(&entry.item))
.count(),
1
);
}
crates/worker/src/ipc/protocol_session.rs
+1
View File
@@ -76,6 +76,7 @@ pub fn live_log_entry_event(entry: LogEntry) -> Option<Event> {
}
entry @ (LogEntry::AnnotatedAssistantItem { .. }
| LogEntry::AnnotatedToolResult { .. }
| LogEntry::ToolResultCorrected { .. }
| LogEntry::RunYielded { .. }
| LogEntry::RunResumed { .. }
| LogEntry::RunCancelled { .. }
crates/worker/src/segment_log_sink.rs
+1
View File
@@ -121,6 +121,7 @@ impl SegmentLogSink {
| LogEntry::AnnotatedUserInput { .. }
| LogEntry::AnnotatedAssistantItem { .. }
| LogEntry::AnnotatedToolResult { .. }
| LogEntry::ToolResultCorrected { .. }
| LogEntry::AnnotatedSystemItem { .. }
| LogEntry::Invoke { .. }
| LogEntry::RunYielded { .. }
crates/worker/src/session_history.rs
+11
View File
@@ -116,6 +116,17 @@ pub(crate) fn restore_history_entries(
}
LogEntry::AnnotatedAssistantItem { entry, .. }
| LogEntry::AnnotatedToolResult { entry, .. } => history.push(from_logged(entry)),
LogEntry::ToolResultCorrected { entry, .. } => {
let target = history
.iter_mut()
.find(|existing| existing.annotation.entry_id == entry.metadata.entry_id)
.ok_or_else(|| "tool result correction target is missing".to_string())?;
if !matches!((&target.item, &entry.item), (Item::ToolResult { call_id, .. }, session_store::LoggedItem::ToolResult { call_id: replacement, .. }) if call_id == replacement)
{
return Err("tool result correction changed call identity".to_string());
}
*target = from_logged(entry);
}
LogEntry::AnnotatedSystemItem { entry, .. } => history.push(HistoryEntry::new(
entry.item.to_history_item(),
entry.metadata.clone(),
crates/worker/src/worker.rs
+33
View File
@@ -1,3 +1,7 @@
#[cfg(test)]
#[path = "image_recovery_tests.rs"]
mod image_recovery_tests;
use std::collections::{HashMap, VecDeque};
#[cfg(test)]
use std::path::Path;
@@ -734,6 +738,17 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
fn execute(&self, request: WorkspaceRequest)
-> Result<WorkspaceResponse, WorkspaceClientError>;
/// Executes one bounded Workspace request when the caller is on a
/// latency-sensitive lifecycle boundary. Implementations without transport
/// timeout support retain their existing execution semantics.
fn execute_with_timeout(
&self,
request: WorkspaceRequest,
_timeout: Duration,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.execute(request)
}
fn execute_server_operation(
&self,
operation: WorkspaceServerOperation,
@@ -4061,6 +4076,24 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
St: Clone + 'static,
{
self.materialize_durable_session_head().await?;
let writer = self.log_writer_handle();
let activity = self.ai_activity_counter.clone();
self.engine_mut()
.set_image_rejection_handler(move |original, replacement| {
writer
.append_entry(LogEntry::ToolResultCorrected {
ts: segment_log::now_millis(),
entry: session_store::LoggedHistoryEntry {
item: replacement.clone().into(),
metadata: original.annotation.clone(),
},
})
.map_err(|error| error.to_string())?;
// A later Cancel must not roll back this committed correction while
// leaving the in-memory result changed (the original may predate Run).
activity.fetch_add(1, Ordering::SeqCst);
Ok(())
});
self.ensure_interceptor_installed();
if self.should_pre_run_compact() {
self.try_pre_run_compact().await?;