From 82616c2a41088d13d363b44ac05b2154acdbb87d Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 31 Oct 2026 14:38:17 +0900 Subject: [PATCH] feat: publish connected Subject body lifecycle candidate and validation fixes --- crates/server-api/src/subject_body.rs | 2 +- crates/worker-runtime/src/runtime.rs | 5 +- crates/workspace-server/src/hosts.rs | 24 +- crates/workspace-server/src/server.rs | 15 +- .../src/server/subject_body.rs | 27 +- .../src/server/tests/chat_tests.rs | 33 +- .../tests/subject_authorization_tests.rs | 4 + .../subject_body_cleanup_connected_tests.rs | 715 ++++++++++++++++ .../tests/subject_body_connected_tests.rs | 706 ++++++++++++++++ .../src/server/tests/subject_body_tests.rs | 24 +- .../tests/workdir_session_close_tests.rs | 48 +- openapi/server-api.json | 773 +++++++++++++++++- .../src/lib/generated/companion-api.ts | 2 +- web/workspace/src/lib/generated/memory-api.ts | 2 +- .../src/lib/generated/repository-api.ts | 2 +- .../src/lib/generated/runtime-api.ts | 8 +- .../src/lib/generated/subject-body-api.ts | 68 ++ web/workspace/src/lib/generated/ticket-api.ts | 2 +- .../src/lib/generated/worker-launch-api.ts | 2 +- .../src/lib/workspace/api/runtime-workers.ts | 2 +- .../memory/SubjectBody.browser.spec.ts | 282 ++++++- .../memory/memory-page.browser.spec.ts | 26 +- web/workspace/test/subject-body-review.ts | 491 +++++++++++ web/workspace/tests/runtime-workers.test.ts | 22 + 24 files changed, 3197 insertions(+), 88 deletions(-) create mode 100644 crates/workspace-server/src/server/tests/subject_body_cleanup_connected_tests.rs create mode 100644 crates/workspace-server/src/server/tests/subject_body_connected_tests.rs create mode 100644 web/workspace/src/lib/generated/subject-body-api.ts create mode 100644 web/workspace/test/subject-body-review.ts diff --git a/crates/server-api/src/subject_body.rs b/crates/server-api/src/subject_body.rs index fac571613..32f4b9561 100644 --- a/crates/server-api/src/subject_body.rs +++ b/crates/server-api/src/subject_body.rs @@ -123,7 +123,7 @@ pub fn subject_body_api_typescript() -> String { SubjectBodyStopOutcomeResponse::decl(&config), ]; format!( - "// Generated by `cargo run -p server-api --features typescript --example generate_subject_body_api_types`.\n// Do not edit manually.\n\nimport type {{ Diagnostic, WorkerSummary }} from './worker-launch-api';\n\n{}\n", + "// Generated by `cargo run -p server-api --features typescript --example generate_subject_body_api_types`.\n// Do not edit manually.\n\nimport type {{ Diagnostic, WorkerSummary }} from \"./worker-launch-api\";\n\n{}\n", declarations .into_iter() .map(|declaration| format!("export {declaration}")) diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index c27c76e27..493ecdf78 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -3567,7 +3567,10 @@ impl Runtime { if matches!(&method, Method::Shutdown { .. }) { self.stop_worker_with_scope( worker_ref, - Some("worker protocol shutdown".to_string()), + runtime_api::WorkerLifecycleRequest { + reason: Some("worker protocol shutdown".to_string()), + stop_intent: None, + }, scope, )?; return Ok(Vec::new()); diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index e95042a65..d95404c1d 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1208,7 +1208,7 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { Err(diagnostic( "worker_stop_admission_unsupported", HostDiagnosticSeverity::Error, - "Runtime does not support fenced stop admission".into(), + "Runtime does not support fenced stop admission", )) } @@ -1220,7 +1220,7 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { Err(diagnostic( "worker_stop_outcome_unsupported", HostDiagnosticSeverity::Error, - "Runtime does not support stop outcome queries".into(), + "Runtime does not support stop outcome queries", )) } @@ -3304,7 +3304,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { diagnostic( "embedded_worker_id_invalid", HostDiagnosticSeverity::Error, - "Worker ID could not be resolved".into(), + "Worker ID could not be resolved", ) })?; let outcome = self @@ -3329,7 +3329,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { diagnostic( "embedded_worker_id_invalid", HostDiagnosticSeverity::Error, - "Worker ID could not be resolved".into(), + "Worker ID could not be resolved", ) })?; let outcome = self @@ -4456,6 +4456,8 @@ fn workspace_runtime_operation(method: &str, path_and_query: &str) -> &'static s return "workers:input"; } if path.ends_with("/stop") + || path.ends_with("/stop/admit") + || path.ends_with("/stop/outcome") || path.ends_with("/cancel") || (path == "/v1/subject-release" && method == "POST") { @@ -6656,22 +6658,22 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti EmbeddedRuntimeError::RestoreAdmissionBusy { .. } => diagnostic( "restore_admission_busy", HostDiagnosticSeverity::Warning, - "Worker lifecycle admission is busy; retry the same Restore intent".into(), + "Worker lifecycle admission is busy; retry the same Restore intent", ), EmbeddedRuntimeError::StopAdmissionBusy { .. } => diagnostic( "stop_admission_busy", HostDiagnosticSeverity::Warning, - "Worker lifecycle admission is busy; retry the same stop intent".into(), + "Worker lifecycle admission is busy; retry the same stop intent", ), EmbeddedRuntimeError::StopObservationConflict { .. } => diagnostic( "stop_observation_conflict", HostDiagnosticSeverity::Warning, - "Stop observation changed; reload before a new intent".into(), + "Stop observation changed; reload before a new intent", ), EmbeddedRuntimeError::StopRequestConflict { .. } => diagnostic( "stop_request_conflict", HostDiagnosticSeverity::Warning, - "Stop request ID was already used with different input".into(), + "Stop request ID was already used with different input", ), EmbeddedRuntimeError::RestoreObservationConflict { .. } => diagnostic( "restore_observation_conflict", @@ -8810,6 +8812,12 @@ mod tests { ), "workers:read" ); + for suffix in ["stop", "stop/admit", "stop/outcome"] { + assert_eq!( + workspace_runtime_operation("POST", &format!("/v1/workers/{worker_id}/{suffix}")), + "workers:stop" + ); + } assert_eq!( workspace_runtime_operation("POST", "/v1/session-public/material/retention/erase"), "workers:delete" diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index fa2e123b2..42bf14a6e 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -43362,7 +43362,7 @@ mod tests { let cleanup_lock = current_worker_session_lock(&api, &identity); let held = Arc::new(Mutex::new(None)); let held_by_callback = held.clone(); - api.runtime.register_or_replace(WorkdirlessFixtureRuntime { + let runtime = WorkdirlessFixtureRuntime { stop_callback: Some(Arc::new(move || { // A genuine fresh-ACK path, not historical receipt replay. Capture // must release this fence before Runtime dispatch/callbacks. @@ -43371,7 +43371,18 @@ mod tests { notification.notify_one(); })), ..Default::default() - }); + }; + runtime + .workers + .lock() + .unwrap() + .push(worker_summary_from_registry( + &api.store + .get_worker_registry(TEST_WORKSPACE_ID, &identity) + .unwrap() + .unwrap(), + )); + api.runtime.register_or_replace(runtime); let worker = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap(); let waiter = tokio::spawn(async move { diff --git a/crates/workspace-server/src/server/subject_body.rs b/crates/workspace-server/src/server/subject_body.rs index fe28e3b8d..278ed3df7 100644 --- a/crates/workspace-server/src/server/subject_body.rs +++ b/crates/workspace-server/src/server/subject_body.rs @@ -171,6 +171,12 @@ fn operation_id(account: &str, workspace: &str, subject: &str, request: &str) -> format!("subject-body-start:{digest}") } +fn failure_diagnostics(code: &str, error: ApiError) -> Vec { + let mut diagnostics = vec![diagnostic(code, error.error.to_string())]; + diagnostics.extend(error.diagnostics.into_iter().map(Into::into)); + diagnostics +} + async fn finish( api: WorkspaceApi, context: server_api::ServerRequestContext, @@ -305,10 +311,7 @@ pub(super) async fn stop( // A post-dispatch failure may have followed a real stop. Keep the exact intent. Err(error) => ( SubjectBodyDisposition::Unknown, - vec![diagnostic( - "subject_body_stop_outcome_unknown", - error.error.to_string(), - )], + failure_diagnostics("subject_body_stop_outcome_unknown", error), ), }; Ok(finish( @@ -398,10 +401,7 @@ pub(super) async fn start( } Err(error) => ( SubjectBodyDisposition::Unknown, - vec![diagnostic( - "subject_body_restore_outcome_unknown", - error.error.to_string(), - )], + failure_diagnostics("subject_body_restore_outcome_unknown", error), ), }; return Ok(finish( @@ -470,7 +470,7 @@ pub(super) async fn start( runtime_id, display_name: subject_record.role.as_str().into(), singleton_key: None, - profile: Some("builtin:async-chat".into()), + profile: Some(server_api::authorization::SUBJECT_CONVERSATION_PROFILE.into()), ticket_assignment: None, initial_submit: vec![], workdir_attachments: vec![], @@ -496,17 +496,16 @@ pub(super) async fn start( Err(error) if matches!( error.error, - Error::RepositoryConflict(_) | Error::WorkspacePermissionDenied(_) + Error::RepositoryConflict(_) + | Error::WorkspacePermissionDenied(_) + | Error::SubjektivSubjectNotFound(_) ) => { return Err(error); } Err(error) => ( SubjectBodyDisposition::Unknown, - vec![diagnostic( - "subject_body_start_outcome_unknown", - error.error.to_string(), - )], + failure_diagnostics("subject_body_start_outcome_unknown", error), ), }; Ok(finish( diff --git a/crates/workspace-server/src/server/tests/chat_tests.rs b/crates/workspace-server/src/server/tests/chat_tests.rs index b975329dc..c0dd4e800 100644 --- a/crates/workspace-server/src/server/tests/chat_tests.rs +++ b/crates/workspace-server/src/server/tests/chat_tests.rs @@ -1274,7 +1274,22 @@ async fn chat_public_receive_busy_messages_commit_before_next_model_input_and_id let worker_ref = RuntimeWorkerRef::new("connected-chat", worker); let lifecycle_path = format!("/api/w/{TEST_WORKSPACE_ID}/runtimes/connected-chat/workers/{worker}"); - let stopped = f.post(&format!("{lifecycle_path}/stop"), json!({})).await; + let stop_observation = f + .api + .runtime + .worker(&worker_ref) + .unwrap() + .restore_observation_token + .unwrap(); + let stopped = f + .post( + &format!("{lifecycle_path}/stop"), + json!({"stop_intent": { + "request_id": "explicit-chat-stop", + "expected_observation_token": stop_observation, + }}), + ) + .await; assert_eq!(stopped["state"], "accepted"); let stopped_wire = json!({"operation_id":"connected-stopped","text":"CHAT_STOPPED_SAVED_NOT_NOTIFIED"}); @@ -1404,7 +1419,21 @@ async fn chat_public_receive_busy_messages_commit_before_next_model_input_and_id StatusCode::FORBIDDEN, "revocation cannot be repaired by profile or create replay: {body}" ); - f.post(&format!("{lifecycle_path}/stop"), json!({})).await; + let stop_observation = f + .api + .runtime + .worker(&worker_ref) + .unwrap() + .restore_observation_token + .unwrap(); + f.post( + &format!("{lifecycle_path}/stop"), + json!({"stop_intent": { + "request_id": "explicit-chat-stop-after-revocation", + "expected_observation_token": stop_observation, + }}), + ) + .await; let token = f .api .runtime diff --git a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs index ae8edb445..5d4a67061 100644 --- a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs @@ -9,6 +9,10 @@ use super::*; mod chat_delegation_connected; #[path = "chat_tests.rs"] mod chat_tests; +#[path = "subject_body_cleanup_connected_tests.rs"] +mod subject_body_cleanup_connected; +#[path = "subject_body_connected_tests.rs"] +mod subject_body_connected; #[path = "subject_body_tests.rs"] mod subject_body_tests; #[path = "subject_job_model_supply_tests.rs"] diff --git a/crates/workspace-server/src/server/tests/subject_body_cleanup_connected_tests.rs b/crates/workspace-server/src/server/tests/subject_body_cleanup_connected_tests.rs new file mode 100644 index 000000000..42cf89b3a --- /dev/null +++ b/crates/workspace-server/src/server/tests/subject_body_cleanup_connected_tests.rs @@ -0,0 +1,715 @@ +//! Connected stop/restore cleanup regression for the normal Server library target. +//! +//! The entry is authenticated Backend Worker control; Workspace capability signing, +//! Runtime HTTP, FsRuntimeStore, ProfileRuntimeWorkerFactory/Controller, canonical +//! Session restore, Git materialization and RemoteWorkdir sessions are production +//! implementations. Only the external model is scripted (no turn is submitted), +//! and a byte-preserving loopback proxy gates the first stop dispatch and observes +//! the actual Runtime ACK. Neither execution nor restore state is substituted. +//! +//! The test owns the existing Worker session fence after stop's resource snapshot +//! and before its ACK. While ACKed cleanup cannot reacquire that fence, it restores +//! the same ordinary Worker through signed Runtime HTTP, opens real provider +//! sessions/commands, and rebinds the same alias/Workdir through the SQLite lifetime +//! API. Registry insertion and rebind are fixture setup under that fence, not a +//! claim about the browser attachment-management endpoint or its authorization. +//! Provider calls still use the normal signed RemoteWorkdir HTTP boundary; there +//! is no unsigned router, fake ACK, synthetic restore result, or Subject-only +//! execution path. Successful Runtime DELETEs and subsequent provider operations +//! distinguish old resource release from survival of the new lifetimes. +//! +//! Run with the `subject_body_cleanup_connected` library filter. The launching +//! shell/normal runner must supply the isolated restore paths required by +//! tests/support/restore_environment.rs; this module never changes process env. +//! This is an in-process connection/race test, not an OS-crash or live-model test. +//! The command fixture requires the normal POSIX shell and `tail` on PATH. + +use super::*; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tokio::sync::{Notify, Semaphore}; + +const CLEANUP_TIMEOUT: Duration = Duration::from_secs(45); +const ALIAS: &str = "workdir"; + +#[derive(Debug)] +struct RuntimeHttpFailure { + method: Method, + path: String, + status: StatusCode, + code: Option, + message: Option, + closed_before_response: bool, + // Match the shared harness's nested-envelope view without discarding the + // flat Workdir envelope's actual code/message from our diagnostic evidence. + harness_message: String, +} + +impl RuntimeHttpFailure { + fn is_expected_creation_probe(&self, workdir: &str) -> bool { + self.method == Method::GET + && self.path == format!("/v1/working-directories/{workdir}") + && self.status == StatusCode::NOT_FOUND + && self.code.as_deref() == Some("working_directory_not_found") + && self.message.as_deref() + == Some("working_directory_not_found: Runtime working directory was not found") + } + + fn is_expected_closed_session(&self) -> bool { + self.method == Method::POST + && self.path.starts_with("/v1/workdir-sessions/") + && self.path.ends_with("/operations") + && self.status == StatusCode::NOT_FOUND + && self.code.as_deref() == Some("not_found") + && self.message.as_deref() == Some("Workdir session was not found") + && self.closed_before_response + } +} + +struct StopTransport { + upstream: String, + client: reqwest::Client, + hold_first_stop: AtomicBool, + dispatched: Notify, + release: Semaphore, + acked: Notify, + acks: Mutex>, + opened: Mutex>, + closed: Mutex>, + failures: Mutex>, +} + +impl StopTransport { + async fn forward(State(state): State>, request: Request) -> Response { + let method = request.method().clone(); + let path = request.uri().path().to_owned(); + let is_stop = request.method() == Method::POST + && path.starts_with("/v1/workers/") + && path.ends_with("/stop"); + if is_stop && state.hold_first_stop.swap(false, Ordering::SeqCst) { + // stop_impl has already released its pre-dispatch session fence. + state.dispatched.notify_one(); + state.release.acquire().await.unwrap().forget(); + } + let is_open = request.method() == Method::POST + && path.starts_with("/v1/working-directories/") + && path.ends_with("/sessions"); + let is_close = + request.method() == Method::DELETE && path.starts_with("/v1/workdir-sessions/"); + let (parts, body) = request.into_parts(); + let body = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); + // Keep method, URI, signed headers and exact body bytes. Runtime's + // production verifier, rather than this proxy, decides authorization. + let response = state + .client + .request(parts.method, format!("{}{}", state.upstream, parts.uri)) + .headers(parts.headers) + .body(body) + .send() + .await + .unwrap(); + let status = response.status(); + let headers = response.headers().clone(); + let bytes = response.bytes().await.unwrap(); + if status.is_client_error() || status.is_server_error() { + let body: Value = serde_json::from_slice(&bytes).unwrap(); + // REST uses {error:{code,message}}, whereas Workdir transport errors + // are {code,message}. Keep only sanitized diagnostic fields, never + // authorization headers, request bodies or raw response bodies. + let error = body.get("error").unwrap_or(&body); + let closed_before_response = path + .strip_suffix("/operations") + .filter(|session_path| session_path.starts_with("/v1/workdir-sessions/")) + .is_some_and(|session_path| { + state + .closed + .lock() + .unwrap() + .iter() + .any(|closed| closed == session_path) + }); + state.failures.lock().unwrap().push(RuntimeHttpFailure { + method, + path: path.clone(), + status, + code: error["code"].as_str().map(str::to_owned), + message: error["message"].as_str().map(str::to_owned), + closed_before_response, + harness_message: body["error"]["message"] + .as_str() + .unwrap_or("unknown fixture error") + .to_owned(), + }); + } + if is_stop && status.is_success() { + let response: runtime_api::WorkerLifecycleResponse = + serde_json::from_slice(&bytes).unwrap(); + state.acks.lock().unwrap().push(response.ack); + // This is a real completed Runtime stop, not merely proxy admission. + state.acked.notify_one(); + } + if is_open && status.is_success() { + let response: workdir::http::OpenWorkdirSessionResponse = + serde_json::from_slice(&bytes).unwrap(); + state + .opened + .lock() + .unwrap() + .push(response.session_id.as_str().to_owned()); + } + if is_close && status == StatusCode::NO_CONTENT { + state.closed.lock().unwrap().push(path); + } + let mut response = Response::new(Body::from(bytes)); + *response.status_mut() = status; + *response.headers_mut() = headers; + response + } + + fn closed_ids(&self) -> Vec { + let mut ids: Vec<_> = self + .closed + .lock() + .unwrap() + .iter() + .map(|path| { + path.strip_prefix("/v1/workdir-sessions/") + .unwrap() + .to_owned() + }) + .collect(); + ids.sort(); + ids + } +} + +struct StopProxy { + transport: Arc, + _server: ConnectedRuntimeServer, +} + +impl Drop for StopProxy { + fn drop(&mut self) { + // A failed assertion must not strand an admitted Host workflow or the + // proxy's graceful shutdown behind the test-only dispatch barrier. + self.transport.release.add_permits(1); + } +} + +impl StopProxy { + async fn install(harness: &ConnectedWorkerHarness) -> Self { + let api = &harness.api; + let binding = api + .store + .get_workspace_runtime_binding(TEST_WORKSPACE_ID, CONNECTED_RUNTIME) + .await + .unwrap() + .unwrap(); + let transport = Arc::new(StopTransport { + upstream: binding.base_url.clone(), + client: reqwest::Client::builder() + .no_proxy() + .timeout(CLEANUP_TIMEOUT) + .build() + .unwrap(), + hold_first_stop: AtomicBool::new(true), + dispatched: Notify::new(), + release: Semaphore::new(0), + acked: Notify::new(), + acks: Mutex::new(Vec::new()), + opened: Mutex::new(Vec::new()), + closed: Mutex::new(Vec::new()), + failures: Mutex::new(Vec::new()), + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let server = ConnectedRuntimeServer::start( + listener, + Router::new() + .fallback(axum::routing::any(StopTransport::forward)) + .with_state(transport.clone()), + ); + // Retain the real trusted binding and authorization issuer. Only the + // transport address changes, as in workdir_session_close_tests. + api.runtime.register_or_replace( + RemoteWorkerRuntime::new( + RemoteRuntimeConfig { + runtime_id: CONNECTED_RUNTIME.into(), + workspace_id: Some(TEST_WORKSPACE_ID.into()), + display_name: binding.display_name.clone(), + base_url: endpoint, + bearer_token: None, + workspace_authorization: Some( + crate::hosts::WorkspaceRuntimeAuthorization::new( + api.store.clone(), + api.signing_identities.clone(), + harness.backend_url.clone(), + binding, + ), + ), + strict_public_egress: false, + cached_worker_creation_available: true, + cached_os: "linux".into(), + cached_arch: "x86_64".into(), + cached_status: "active".into(), + timeout: CLEANUP_TIMEOUT, + }, + TEST_WORKSPACE_ID.into(), + harness.backend_url.clone(), + ) + .unwrap(), + ); + Self { + transport, + _server: server, + } + } + + async fn open( + &self, + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + workdir: &str, + ) -> (WorkdirSessionHandle, String) { + let before = self.transport.opened.lock().unwrap().len(); + let session = api + .runtime + .open_workdir_session(&worker.runtime_id, workdir, Some(&worker.worker_id)) + .await + .unwrap(); + let opened = self.transport.opened.lock().unwrap(); + assert_eq!(opened.len(), before + 1, "one actual Runtime session open"); + (session, opened[before].clone()) + } +} + +// The harness owns servers/temporary roots; this guard also stops an active +// restored Controller on assertion failure before those servers are dropped. +struct StopControllerOnDrop { + runtime: worker_runtime::Runtime, + worker: worker_runtime::identity::WorkerRef, +} +impl Drop for StopControllerOnDrop { + fn drop(&mut self) { + let _ = self.runtime.stop_worker( + &self.worker, + Some("cleanup connected stop/restore fixture".into()), + ); + } +} + +async fn start_held_command(session: &WorkdirSessionHandle) -> CommandHandle { + let command = session + .start_command(workdir::CommandRequest { + // A long-lived resource, not a timer used to infer progress. Session + // close must cancel and reap it; no external service is contacted. + command: "printf cleanup-command; exec tail -f /dev/null".into(), + timeout_secs: 120, + transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES, + cwd: workdir::WorkdirPath::root(), + tool_call_id: None, + }) + .await + .unwrap(); + assert_eq!( + session.command_status(command.clone()).await.unwrap(), + workdir::CommandStatus::Running, + "fixture command must really be live rather than a failed shell launch" + ); + command +} + +async fn assert_live(session: &WorkdirSessionHandle) { + session + .stat(workdir::StatRequest { + path: workdir::WorkdirPath::new("README.md").unwrap(), + }) + .await + .expect("new Runtime session must still serve the real materialized checkout"); +} + +async fn assert_closed(session: &WorkdirSessionHandle) { + session + .stat(workdir::StatRequest { + path: workdir::WorkdirPath::new("README.md").unwrap(), + }) + .await + .expect_err("observed Runtime session must no longer serve filesystem operations"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn stop_ack_cleanup_preserves_real_restored_sessions_and_rebound_connection() { + crate::restore_test_environment::require_restore_test_environment(); + let (completed, _completion) = tokio::sync::mpsc::channel(4); + let harness = ConnectedWorkerHarness::start(ConnectedAuthorizationModel { completed }).await; + let proxy = StopProxy::install(&harness).await; + let api = &harness.api; + let creation_probe_index = proxy.transport.failures.lock().unwrap().len(); + let (status, created) = harness + .human( + reqwest::Method::POST, + &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), + json!({"runtime_id":CONNECTED_RUNTIME, "repository_key":"test-repository", "selector":"HEAD", "operation_id":"cleanup-connected-workdir"}), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{created}"); + let workdir = created["item"]["working_directory_id"] + .as_str() + .unwrap() + .to_owned(); + { + // create_workspace_working_directory probes its reserved Runtime ID + // before materializing it. Restrict this expectation to the sole failure + // during that completed creation request and to the returned exact ID. + // Later misses of the same Workdir must not inherit this exemption. + let failures = proxy.transport.failures.lock().unwrap(); + assert_eq!( + failures.len(), + creation_probe_index + 1, + "creation must produce exactly one initial absence probe: {failures:?}" + ); + assert!( + failures[creation_probe_index].is_expected_creation_probe(&workdir), + "unexpected creation probe response: {failures:?}" + ); + } + let (status, created) = harness + .human( + reqwest::Method::POST, + &format!("/api/w/{TEST_WORKSPACE_ID}/workers"), + json!({ + "runtime_id":CONNECTED_RUNTIME, "display_name":"Connected stop cleanup", + "profile":"builtin:coder", "initial_submit":[], + "workdir_attachments":[{"alias":ALIAS,"working_directory_id":workdir}], + "feature_connections":{} + }), + ) + .await; + assert_eq!(status, StatusCode::OK, "{created}"); + let worker = RuntimeWorkerRef::new( + created["runtime_id"].as_str().unwrap(), + created["worker_id"].as_str().unwrap(), + ); + let local_worker = worker_runtime::identity::WorkerRef::try_from(&worker).unwrap(); + let _controller_cleanup = StopControllerOnDrop { + runtime: harness.runtime.clone(), + worker: local_worker.clone(), + }; + assert!( + api.store + .require_current_worker_singleton_owner(TEST_WORKSPACE_ID, &worker) + .unwrap() + .is_none(), + "this must remain the ordinary Worker stop/restore path" + ); + let old_link = api + .store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &worker) + .unwrap() + .into_iter() + .find(|link| link.unlinked_at.is_none() && link.alias == ALIAS) + .unwrap(); + + // Retain two independent pre-stop command sources. An attachment replacement + // normally closes the displaced session itself; do not manufacture an orphan + // by overwriting one and then demand that stop follow a removed registry row. + let (old_first_source, old_first_source_id) = proxy.open(api, &worker, &workdir).await; + let (old_command_source, old_command_id) = proxy.open(api, &worker, &workdir).await; + let old_first_provider_command = start_held_command(&old_first_source).await; + let old_provider_command = start_held_command(&old_command_source).await; + let (old_first_command, old_command) = { + let _guard = current_worker_session_lock(api, &worker).lock_owned().await; + let mut registry = api.workdir_sessions.lock().unwrap(); + assert!(!registry.attachments.contains_key(&worker)); + let first = registry.register_command( + worker.clone(), + ALIAS.into(), + old_first_source.clone(), + old_first_provider_command, + ); + let second = registry.register_command( + worker.clone(), + ALIAS.into(), + old_command_source.clone(), + old_provider_command, + ); + (first, second) + }; + let stop_path = format!( + "/api/w/{TEST_WORKSPACE_ID}/runtimes/{}/workers/{}/stop", + worker.runtime_id, worker.worker_id + ); + let stop_api = api.clone(); + let owner = harness.owner.clone(); + let first_stop_path = stop_path.clone(); + let stopping = tokio::spawn(async move { + response( + &stop_api, + human_request( + &owner, + "POST", + &first_stop_path, + Some(json!({"reason":"first real stop before restore"})), + ), + ) + .await + }); + tokio::time::timeout(CLEANUP_TIMEOUT, proxy.transport.dispatched.notified()) + .await + .expect("Backend stop must reach transport after its resource observation"); + let session_guard = tokio::time::timeout( + CLEANUP_TIMEOUT, + current_worker_session_lock(api, &worker).lock_owned(), + ) + .await + .expect("stop dispatch must not retain the pre-ACK session fence"); + proxy.transport.release.add_permits(1); + tokio::time::timeout(CLEANUP_TIMEOUT, proxy.transport.acked.notified()) + .await + .expect("real Runtime must acknowledge stop while Backend cleanup is fenced"); + { + let acks = proxy.transport.acks.lock().unwrap(); + assert_eq!(acks.len(), 1); + assert_eq!(acks[0].status, runtime_api::WorkerStatus::Stopped); + assert!( + !acks[0].replayed, + "must exercise fresh-ACK cleanup, not replay skip" + ); + } + assert!( + !stopping.is_finished(), + "ACK must not bypass the session fence" + ); + assert!(proxy.transport.closed_ids().is_empty()); + // Provider sessions are not Controller state: actual Runtime stop alone has + // not closed the Backend-observed command sources. + assert_live(&old_first_source).await; + assert_live(&old_command_source).await; + + let restore_api = api.clone(); + let restore_worker = worker.clone(); + let restored = tokio::time::timeout( + CLEANUP_TIMEOUT, + tokio::task::spawn_blocking(move || { + let stopped = restore_api.runtime.worker(&restore_worker).unwrap(); + assert_eq!(stopped.state, "stopped"); + restore_api + .runtime + .restore_worker( + &restore_worker, + runtime_api::WorkerRestoreRequest { + expected_observation_token: stopped.restore_observation_token.unwrap(), + request_id: "cleanup-connected-real-restore".into(), + preparation: Some(runtime_api::WorkerRestorePreparation::default()), + }, + ) + .unwrap() + }), + ) + .await + .expect("real restore must not need the cleanup-owned Worker control lane") + .unwrap(); + assert_eq!( + restored.state, + server_api::WorkerRestoreState::Accepted, + "{restored:?}" + ); + assert_eq!( + harness.runtime.worker_detail(&local_worker).unwrap().status, + worker_runtime::catalog::WorkerStatus::Idle, + "the actual Runtime catalog must expose the reconstructed Controller" + ); + + // The fixture owns the production session fence throughout rebind and + // resource publication. Keep worker/alias/workdir/capabilities identical; + // only the storage-issued connection lifetime is different. + let detached = api + .store + .detach_worker_workdir_connection( + TEST_WORKSPACE_ID, + &worker, + ALIAS, + &old_link.connection_id, + &now_registry_timestamp(), + ) + .unwrap(); + assert_eq!(detached.connection_id, old_link.connection_id); + assert!(detached.unlinked_at.is_some()); + let new_link = api.store.attach_worker_workdir(&old_link).unwrap(); + assert_ne!(new_link.connection_id, old_link.connection_id); + assert_eq!(new_link.worker, old_link.worker); + assert_eq!(new_link.alias, old_link.alias); + assert_eq!(new_link.workdir_id, old_link.workdir_id); + assert_eq!(new_link.capabilities, old_link.capabilities); + + let (new_attachment, new_attachment_id) = proxy.open(api, &worker, &workdir).await; + let (new_command_source, new_command_id) = proxy.open(api, &worker, &workdir).await; + let new_provider_command = start_held_command(&new_command_source).await; + let new_command = { + let mut registry = api.workdir_sessions.lock().unwrap(); + assert!( + registry + .insert_attachment(worker.clone(), new_attachment.clone()) + .is_none() + ); + registry.register_command( + worker.clone(), + ALIAS.into(), + new_command_source.clone(), + new_provider_command.clone(), + ) + }; + let mut record = api + .store + .get_workdir_registry(TEST_WORKSPACE_ID, &workdir) + .unwrap() + .unwrap(); + record.display_name = Some("new connection projection must survive old stop".into()); + api.store.upsert_workdir_registry(&record).unwrap(); + assert_live(&new_attachment).await; + assert_eq!( + new_command_source + .command_status(new_provider_command.clone()) + .await + .unwrap(), + workdir::CommandStatus::Running, + ); + assert!(!stopping.is_finished()); + drop(session_guard); + + let (status, stopped) = tokio::time::timeout(CLEANUP_TIMEOUT, stopping) + .await + .expect("old cleanup must finish after the session fence is released") + .unwrap(); + assert_eq!(status, StatusCode::OK, "{stopped}"); + assert_eq!(stopped["state"], "accepted", "{stopped}"); + let mut old_ids = vec![old_first_source_id, old_command_id]; + old_ids.sort(); + assert_eq!( + proxy.transport.closed_ids(), + old_ids, + "close only pre-dispatch sessions" + ); + assert_closed(&old_first_source).await; + assert_closed(&old_command_source).await; + assert_live(&new_attachment).await; + assert_live(&new_command_source).await; + assert_eq!( + new_command_source + .command_status(new_provider_command) + .await + .unwrap(), + workdir::CommandStatus::Running, + "old cleanup must not cancel a post-restore command" + ); + { + let registry = api.workdir_sessions.lock().unwrap(); + assert!(Arc::ptr_eq( + registry.attachments.get(&worker).unwrap(), + &new_attachment + )); + assert!( + registry + .command(&worker, ALIAS, &old_first_command) + .is_none() + ); + assert!(registry.command(&worker, ALIAS, &old_command).is_none()); + assert!(Arc::ptr_eq( + ®istry + .command(&worker, ALIAS, &new_command) + .unwrap() + .source, + &new_command_source, + )); + } + assert_eq!( + api.store + .get_workdir_registry(TEST_WORKSPACE_ID, &workdir) + .unwrap() + .unwrap() + .display_name, + record.display_name, + "old stop must not project a new connection even for identical worker/alias/workdir" + ); + assert_eq!( + api.store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &worker) + .unwrap() + .into_iter() + .find(|link| link.unlinked_at.is_none() && link.alias == ALIAS) + .unwrap() + .connection_id, + new_link.connection_id, + ); + + // Surviving resources are not leaked: a later explicit, non-replayed stop + // observes their own lifetimes and closes their actual Runtime sessions. + let (status, stopped) = harness + .human( + reqwest::Method::POST, + &stop_path, + json!({"reason":"explicit stop of restored Worker"}), + ) + .await; + assert_eq!(status, StatusCode::OK, "{stopped}"); + assert_eq!(stopped["state"], "accepted", "{stopped}"); + { + let acks = proxy.transport.acks.lock().unwrap(); + assert_eq!(acks.len(), 2); + assert_eq!(acks[1].status, runtime_api::WorkerStatus::Stopped); + assert!(!acks[1].replayed); + } + old_ids.extend([new_attachment_id, new_command_id]); + old_ids.sort(); + assert_eq!( + proxy.transport.closed_ids(), + old_ids, + "each actual session closes once" + ); + assert_closed(&new_attachment).await; + assert_closed(&new_command_source).await; + { + let registry = api.workdir_sessions.lock().unwrap(); + assert!(!registry.attachments.contains_key(&worker)); + assert!(registry.command(&worker, ALIAS, &new_command).is_none()); + } + assert_ne!( + api.store + .get_workdir_registry(TEST_WORKSPACE_ID, &workdir) + .unwrap() + .unwrap() + .display_name, + record.display_name, + "later stop must still project its current connection normally" + ); + let runtime_errors = harness.runtime_errors.lock().unwrap().clone(); + let failures = proxy.transport.failures.lock().unwrap(); + let unexpected: Vec<_> = failures + .iter() + .enumerate() + .filter(|(index, failure)| { + *index != creation_probe_index && !failure.is_expected_closed_session() + }) + .map(|(_, failure)| failure) + .collect(); + assert!( + unexpected.is_empty(), + "unexpected Runtime HTTP errors: {unexpected:?}; runtime_errors={runtime_errors:?}; all HTTP failures={failures:?}" + ); + // Do not silently lose failures from any Runtime request that bypassed our + // proxy. Reconcile even the shared harness's lossy flat-envelope messages. + // Apart from the single creation probe validated at its request boundary, + // a NotFound is expected only for operations on a session whose DELETE was + // already confirmed; a 401/403 or any other NotFound remains a failure. + let mut captured = runtime_errors.clone(); + let mut classified: Vec<_> = failures + .iter() + .map(|failure| failure.harness_message.clone()) + .collect(); + captured.sort(); + classified.sort(); + assert_eq!( + captured, classified, + "unclassified Runtime errors: runtime_errors={runtime_errors:?}; HTTP failures={failures:?}" + ); +} diff --git a/crates/workspace-server/src/server/tests/subject_body_connected_tests.rs b/crates/workspace-server/src/server/tests/subject_body_connected_tests.rs new file mode 100644 index 000000000..a0de8c448 --- /dev/null +++ b/crates/workspace-server/src/server/tests/subject_body_connected_tests.rs @@ -0,0 +1,706 @@ +//! Connected Subject Body API regressions in the normal Server lib target +//! (`subject_body_connected` filter). The authenticated human router, signed +//! Backend <-> RuntimeHTTP requests, ProfileRuntimeWorkerFactory/Controller, +//! singleton reservations, SQLite and canonical FS Session storage are real. +//! Only the external model stream is scripted. One transport fault discards an +//! already-produced stop response; it does not replace stop execution or receipts. +//! Existing Memory is fixture input through the domain store; Journal admission +//! uses real, authorized canonical Session capture, not invented source identity. +//! +//! Restore requires the runner-owned isolation checked by +//! `tests/support/restore_environment.rs` before any fixture starts. This module +//! never sets process environment, installs tools, or contacts a live model. +//! Resource races and Workdir rebinding belong to the separate lifecycle suite. +use super::*; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +const DEADLINE: Duration = Duration::from_secs(30); + +#[derive(Clone)] +struct BodyModel(tokio::sync::mpsc::Sender); + +#[async_trait::async_trait] +impl agen::llm_client::LlmClient for BodyModel { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + async fn stream( + &self, + request: agen::llm_client::Request, + ) -> std::result::Result + { + use agen::llm_client::event::{Event, StopReason}; + for name in [ + "ConversationRead", + "ConversationSend", + "SubjektivMemoryQuery", + ] { + assert!( + request.tools.iter().any(|tool| tool.name == name), + "the unchanged Subject conversation profile must expose {name}" + ); + } + self.0.send(format!("{:?}", request.items)).await.unwrap(); + Ok(Box::pin(futures::stream::iter([ + Ok(Event::text_block_start(0)), + Ok(Event::text_delta(0, "Retained internal body turn")), + Ok(Event::text_block_stop(0, Some(StopReason::EndTurn))), + ]))) + } +} + +struct ConnectedBody { + connected: ConnectedWorkerHarness, + lose_stop_ack: Arc, + inputs: tokio::sync::mpsc::Receiver, +} + +impl ConnectedBody { + async fn new() -> Self { + // Fail before starting either server or touching restore allocation paths. + crate::restore_test_environment::require_restore_test_environment(); + let (tx, inputs) = tokio::sync::mpsc::channel(8); + let lose_stop_ack = Arc::new(AtomicBool::new(false)); + let fault = lose_stop_ack.clone(); + let connected = ConnectedWorkerHarness::start_with_layer(BodyModel(tx), move |router| { + router.layer(axum::middleware::from_fn( + move |request: axum::extract::Request, next: axum::middleware::Next| { + let fault = fault.clone(); + async move { + let discard = request.uri().path().ends_with("/body/stop") + && fault.swap(false, Ordering::SeqCst); + let result = next.run(request).await; + if discard { + assert!( + result.status().is_success(), + "stop must execute before ACK loss" + ); + return Response::builder() + .status(StatusCode::BAD_GATEWAY) + .header(CONTENT_TYPE, "application/json") + .body(Body::from( + json!({"message":"injected lost stop response"}).to_string(), + )) + .unwrap(); + } + result + } + }, + )) + }) + .await; + // Body start intentionally chooses a Runtime itself and prefers embedded. + // Remove only the unused deterministic fixture Runtime via its real idle + // removal contract; never route around signed HTTP or replace execution. + assert!(matches!( + connected + .api + .runtime + .unregister_if_idle(EMBEDDED_WORKER_RUNTIME_ID, 100) + .unwrap(), + crate::hosts::RuntimeRegistryUnregisterResult::Removed + )); + let available = connected.api.runtime.list_runtimes(100); + assert_eq!(available.items.len(), 1); + assert_eq!(available.items[0].runtime_id, CONNECTED_RUNTIME); + assert!(available.items[0].worker_creation_available); + Self { + connected, + lose_stop_ack, + inputs, + } + } + + async fn request(&self, method: &str, path: &str, body: Value) -> (StatusCode, Value) { + self.connected + .human(method.parse().unwrap(), path, body) + .await + } + + async fn ok(&self, method: &str, path: &str, body: Value) -> Value { + let (status, result) = self.request(method, path, body).await; + assert_eq!(status, StatusCode::OK, "{method} {path}: {result}"); + result + } + + async fn subject(&self, operation: &str) -> String { + let (status, created) = self + .request( + "POST", + &format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects"), + json!({"operation_id":operation,"role":"connected-body","behavior_md":""}), + ) + .await; + assert_eq!( + status, + StatusCode::CREATED, + "normal Subject creation: {created}" + ); + created["id"].as_str().unwrap().to_owned() + } + + async fn read(&self, subject: &str) -> Value { + self.ok("GET", &body_path(subject), Value::Null).await + } + + async fn mutate(&self, subject: &str, suffix: &str, request: Value) -> Value { + self.ok("POST", &format!("{}{suffix}", body_path(subject)), request) + .await + } + + async fn stop(&self, subject: &str, request_id: &str) -> Value { + let current = self.read(subject).await; + let request = intent(request_id, ¤t); + let stopped = self.mutate(subject, "/stop", request.clone()).await; + assert_eq!(stopped["disposition"], "accepted", "{stopped}"); + let outcome = self.mutate(subject, "/stop/outcome", request.clone()).await; + assert_eq!(outcome["outcome"], "stopped", "{outcome}"); + request + } + + fn worker_count(&self) -> usize { + self.connected + .api + .runtime + .list_workers_for_runtime(CONNECTED_RUNTIME, 100) + .unwrap() + .items + .len() + } +} + +fn body_path(subject: &str) -> String { + format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/body") +} + +fn absent_start(request_id: &str) -> Value { + json!({"request_id":request_id,"expected_current":null,"expected_observation_token":null}) +} + +fn intent(request_id: &str, body: &Value) -> Value { + let token = body["observation"]["worker"]["restore_observation_token"] + .as_str() + .unwrap_or_else(|| { + panic!("available body must expose the Runtime observation fence: {body}") + }); + json!({"request_id":request_id,"expected_current":body["current"],"expected_observation_token":token}) +} + +async fn body_idle( + events: &mut tokio::sync::broadcast::Receiver< + worker_runtime::observation::WorkerObservationEvent, + >, + worker: &str, +) { + tokio::time::timeout(DEADLINE, async { + loop { + let event = events.recv().await.expect("real Runtime observation"); + if event.worker_ref.worker_id.to_string() != worker { + continue; + } + match event.payload { + protocol::Event::Error { code, message } => { + panic!("body turn failed: {code:?}: {message}") + } + protocol::Event::WorkerState { snapshot } + if snapshot.state == protocol::WorkerState::Idle => + { + break; + } + _ => {} + } + } + }) + .await + .expect("Controller must publish Idle after the scripted model turn"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn body_same_request_concurrent_start_stop_outcome_and_restore_preserve_subject_data() { + let mut f = ConnectedBody::new().await; + let subject = f.subject("body-new-subject").await; + let store = open_subjektiv_store(&f.connected.api).unwrap(); + let initial = store.subject_permissions(&subject).unwrap().unwrap(); + assert_eq!( + initial.execution, + server_api::authorization::subject_conversation_permissions(TEST_WORKSPACE_ID, &subject) + ); + assert!(initial.delegation.is_empty()); + assert!(f.read(&subject).await["current"].is_null()); + + // Retention inputs are deliberately independent of Worker execution and of + // Ticket conclusion. Direct Memory setup is not a claimed body mutation. + let memory = store + .create_memory( + &subject, + crate::subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Decision, + "Keep this decision", + "Retained Memory body", + "Survive body stop/restore", + "retention fixture", + ), + ) + .unwrap(); + let tickets = browser_ticket_backend(&f.connected.api).unwrap(); + let ticket = tickets + .create(NewTicket::new("Body stop does not conclude this case")) + .unwrap(); + let before_case = tickets.show(ticket.id.clone().into()).unwrap(); + let conversations = format!("/api/w/{TEST_WORKSPACE_ID}/chat/conversations"); + let conversation = f + .ok( + "POST", + &conversations, + json!({"operation_id":"body-conversation","subject_id":subject}), + ) + .await; + let messages = format!( + "{}/{}/messages", + conversations, + conversation["id"].as_str().unwrap() + ); + let saved = f + .ok( + "POST", + &messages, + json!({"operation_id":"body-before-start","text":"BEFORE_BODY_START_RETAINED"}), + ) + .await; + assert_ne!(saved["delivery"]["state"], "accepted"); + + let start = absent_start("body-concurrent-same"); + let path = format!("{}/start", body_path(&subject)); + let (left, right) = tokio::join!( + f.request("POST", &path, start.clone()), + f.request("POST", &path, start.clone()) + ); + for (status, result) in [&left, &right] { + assert_eq!(*status, StatusCode::OK, "exact concurrent retry: {result}"); + assert!( + matches!(result["disposition"].as_str(), Some("accepted" | "unknown")), + "{result}" + ); + } + // An in-flight response is not assumed to settle the other call. Reconcile + // with the same request identity, never with a second creation request. + let replay = f.mutate(&subject, "/start", start.clone()).await; + assert_eq!(replay["disposition"], "accepted", "{replay}"); + let running = f.read(&subject).await; + assert_eq!(running["current"]["runtime_id"], CONNECTED_RUNTIME); + assert_eq!(running["observation"]["kind"], "available"); + assert_eq!( + running["observation"]["worker"]["profile"], + server_api::authorization::SUBJECT_CONVERSATION_PROFILE + ); + assert_eq!(f.worker_count(), 1); + let current = running["current"].clone(); + let worker = current["worker_id"].as_str().unwrap().to_owned(); + let source = RuntimeWorkerRef::new(CONNECTED_RUNTIME, &worker); + let singleton = f + .connected + .api + .store + .require_current_worker_singleton_owner(TEST_WORKSPACE_ID, &source) + .unwrap() + .expect("normal body singleton owner"); + assert_eq!( + serde_json::to_value(subject_body::reference(&singleton)).unwrap(), + current + ); + let live_reuse = f + .mutate(&subject, "/start", intent("body-live-reuse", &running)) + .await; + assert_eq!(live_reuse["disposition"], "accepted", "{live_reuse}"); + assert_eq!(f.worker_count(), 1); + + let mut events = f.connected.runtime.subscribe_worker_observation().unwrap(); + let delivered = f + .ok( + "POST", + &messages, + json!({"operation_id":"body-live-message","text":"BODY_CANONICAL_SESSION_MARKER"}), + ) + .await; + assert_eq!(delivered["delivery"]["state"], "accepted"); + let input = tokio::time::timeout(DEADLINE, f.inputs.recv()) + .await + .unwrap() + .unwrap(); + assert!(input.contains("BODY_CANONICAL_SESSION_MARKER"), "{input}"); + body_idle(&mut events, &worker).await; + let root = f.connected.runtime_root.join("workers").join(&worker); + let history = session_store::read_retained_session_snapshot( + &root, + &format!("worker-runtime-{worker}"), + 2 * 1024 * 1024, + ) + .unwrap(); + assert!( + serde_json::to_string(&history.snapshot) + .unwrap() + .contains("BODY_CANONICAL_SESSION_MARKER") + ); + let attribution = store + .subject_session_attributions(&subject) + .unwrap() + .into_iter() + .find(|entry| entry.worker_id == worker) + .expect("signed startup registered real Session"); + assert_eq!(attribution.session_id, history.identity.session_id); + let host = + subjektiv::journal::JournalHostContext::validated_session(&store, attribution.clone()) + .unwrap(); + let capture = + subjektiv_journal::capture_input(&f.connected.api, &attribution, 0, 0, 128 * 1024).unwrap(); + let ranges = subjektiv::journal::recording::capture_ranges(&capture).unwrap(); + assert!(!ranges.is_empty(), "admit actual canonical Session bytes"); + let journal = store + .accept_journal_sources(&host, "Retained body Journal", &ranges) + .unwrap(); + let reader = + subjektiv::journal::JournalHostContext::validated_reader(&store, &subject).unwrap(); + let before_journal = + serde_json::to_value(store.read_journal(&reader, &journal).unwrap()).unwrap(); + let before_sources = serde_json::to_value( + store + .list_journal_sources( + &reader, + &journal, + None, + subjektiv::journal::MAX_JOURNAL_PAGE, + ) + .unwrap(), + ) + .unwrap(); + assert!( + before_sources["next"].is_null(), + "the small retention fixture must compare its complete Journal source ledger: {before_sources}" + ); + assert_eq!( + before_sources["items"].as_array().unwrap().len(), + ranges.len(), + "every admitted canonical source must be included in the retention comparison" + ); + let before_messages = f.ok("GET", &messages, Value::Null).await; + + let observed = f.read(&subject).await; + let stop = intent("body-stop-lost-ack", &observed); + f.lose_stop_ack.store(true, Ordering::SeqCst); + let (status, lost) = f + .request( + "POST", + &format!("{}/stop", body_path(&subject)), + stop.clone(), + ) + .await; + assert_eq!(status, StatusCode::BAD_GATEWAY, "{lost}"); + // The caller has an unknown result. A GET is only current observation: it + // cannot prove which request stopped it, nor manufacture a receipt. + let stopped = f.read(&subject).await; + assert_eq!(stopped["current"], current); + assert_eq!(stopped["observation"]["worker"]["state"], "stopped"); + let mut unrelated = stop.clone(); + unrelated["request_id"] = json!("body-stop-never-sent"); + let missing = f.mutate(&subject, "/stop/outcome", unrelated).await; + assert_eq!( + missing["outcome"], "not_found", + "GET must not invent stop acceptance: {missing}" + ); + let outcome = f.mutate(&subject, "/stop/outcome", stop.clone()).await; + assert_eq!( + outcome["outcome"], "stopped", + "only exact outcome resolves the lost ACK: {outcome}" + ); + let restore = intent("body-restore-same-generation", &stopped); + let restored = f.mutate(&subject, "/start", restore.clone()).await; + assert_eq!( + restored["disposition"], + "accepted", + "{restored}; Runtime errors: {:?}", + f.connected.runtime_errors.lock().unwrap() + ); + assert_eq!(restored["body"]["current"], current); + assert_ne!( + restored["body"]["observation"]["worker"]["state"], + "stopped" + ); + assert_eq!( + f.worker_count(), + 1, + "restore must not create a replacement Worker/generation" + ); + let replay_restore = f.mutate(&subject, "/start", restore).await; + assert_eq!( + replay_restore["disposition"], "accepted", + "{replay_restore}" + ); + let historical = f.mutate(&subject, "/stop/outcome", stop.clone()).await; + assert_eq!(historical["outcome"], "stopped"); + assert_ne!( + historical["body"]["observation"]["worker"]["state"], "stopped", + "historical stop receipt is not the restored body's current state" + ); + let stop_replay = f.mutate(&subject, "/stop", stop).await; + assert_eq!(stop_replay["disposition"], "accepted"); + assert_ne!( + f.read(&subject).await["observation"]["worker"]["state"], + "stopped", + "exact historical stop replay must not stop the restored lifetime" + ); + for suffix in ["/start", "/stop"] { + let stale = intent("body-old-observation", &observed); + let (status, result) = f + .request("POST", &format!("{}{suffix}", body_path(&subject)), stale) + .await; + assert_eq!( + status, + StatusCode::CONFLICT, + "old Runtime observation: {result}" + ); + } + + // Reopen authority rather than treating an earlier model callback as durable evidence. + let reopened = open_subjektiv_store(&f.connected.api).unwrap(); + assert_eq!( + reopened.subject_permissions(&subject).unwrap(), + Some(initial) + ); + assert_eq!( + serde_json::to_value(reopened.memory(&subject, &memory.id).unwrap().unwrap()).unwrap(), + serde_json::to_value(&memory).unwrap() + ); + assert_eq!( + serde_json::to_value(reopened.read_journal(&reader, &journal).unwrap()).unwrap(), + before_journal + ); + assert_eq!( + serde_json::to_value( + reopened + .list_journal_sources( + &reader, + &journal, + None, + subjektiv::journal::MAX_JOURNAL_PAGE + ) + .unwrap() + ) + .unwrap(), + before_sources + ); + assert_eq!(f.ok("GET", &messages, Value::Null).await, before_messages); + let after_case = tickets.show(ticket.id.into()).unwrap(); + assert_eq!( + after_case, before_case, + "body lifecycle must not conclude or edit the case" + ); + assert_eq!( + reopened.subject(&subject).unwrap().unwrap().state, + crate::subjektiv::SubjectState::Active + ); + let after_history = session_store::read_retained_session_snapshot( + &root, + &format!("worker-runtime-{worker}"), + 2 * 1024 * 1024, + ) + .unwrap(); + assert_eq!( + after_history.identity.session_id, + history.identity.session_id + ); + assert!( + serde_json::to_string(&after_history.snapshot) + .unwrap() + .contains("BODY_CANONICAL_SESSION_MARKER") + ); + f.stop(&subject, "body-final-stop").await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn body_distinct_concurrent_starts_keep_one_owner_and_reject_foreign_targets() { + let f = ConnectedBody::new().await; + let subject = f.subject("body-distinct-subject").await; + let other = f.subject("body-foreign-subject").await; + let path = format!("{}/start", body_path(&subject)); + let (a, b) = tokio::join!( + f.request("POST", &path, absent_start("body-click-a")), + f.request("POST", &path, absent_start("body-click-b")) + ); + assert!( + [&a, &b] + .iter() + .any(|(status, body)| *status == StatusCode::OK && body["disposition"] == "accepted"), + "one request must start the body: {a:?}, {b:?}" + ); + for (status, body) in [&a, &b] { + assert!( + matches!(*status, StatusCode::OK | StatusCode::CONFLICT), + "{status}: {body}" + ); + } + assert_eq!(f.worker_count(), 1); + let current = f.read(&subject).await; + let foreign = intent("body-foreign-intent", ¤t); + for suffix in ["/start", "/stop", "/stop/outcome"] { + let (status, result) = f + .request( + "POST", + &format!("{}{suffix}", body_path(&other)), + foreign.clone(), + ) + .await; + assert_eq!( + status, + StatusCode::CONFLICT, + "cross-Subject {suffix}: {result}" + ); + } + let mut old_generation = foreign.clone(); + old_generation["expected_current"]["generation"] = json!( + (current["current"]["generation"] + .as_str() + .unwrap() + .parse::() + .unwrap() + + 1) + .to_string() + ); + for suffix in ["/start", "/stop", "/stop/outcome"] { + let (status, result) = f + .request( + "POST", + &format!("{}{suffix}", body_path(&subject)), + old_generation.clone(), + ) + .await; + assert_eq!( + status, + StatusCode::CONFLICT, + "stale generation {suffix}: {result}" + ); + } + let reader = seed_test_api_token(f.connected.api.store.as_ref(), "connected-body-reader"); + for (method, suffix, body) in [ + ("GET", "", None), + ("POST", "/start", Some(foreign.clone())), + ("POST", "/stop", Some(foreign.clone())), + ("POST", "/stop/outcome", Some(foreign.clone())), + ] { + let (status, result) = response( + &f.connected.api, + human_request( + &reader, + method, + &format!("{}{suffix}", body_path(&subject)), + body, + ), + ) + .await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "non-owner {suffix}: {result}" + ); + } + let wrong_workspace = body_path(&subject).replace(TEST_WORKSPACE_ID, "not-this-workspace"); + let (status, result) = f + .request("POST", &format!("{wrong_workspace}/stop"), foreign) + .await; + assert_eq!(status, StatusCode::NOT_FOUND, "{result}"); + assert_eq!(f.read(&subject).await["current"], current["current"]); + assert!(f.read(&other).await["current"].is_null()); + assert_eq!(f.worker_count(), 1); + f.stop(&subject, "body-distinct-cleanup").await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn body_start_and_restore_do_not_repair_missing_session_record_permission() { + let f = ConnectedBody::new().await; + let subject = f.subject("body-revoked-subject").await; + let path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"); + let initial = open_subjektiv_store(&f.connected.api) + .unwrap() + .subject_permissions(&subject) + .unwrap() + .unwrap(); + let execution = initial + .execution + .iter() + .filter(|permission| permission.operation != "session.record") + .cloned() + .collect::>(); + let revoked = f.ok("POST", &path, json!({"operation_id":"body-remove-session-record", + "expected_change_id":initial.change_id,"execution":execution,"delegation":initial.delegation})).await; + let (status, denied) = f + .request( + "POST", + &format!("{}/start", body_path(&subject)), + absent_start("body-denied-start"), + ) + .await; + assert!( + status == StatusCode::FORBIDDEN + || (status == StatusCode::OK && denied["disposition"] != "accepted"), + "required startup authorization must refuse: {status}: {denied}" + ); + assert!( + denied.to_string().contains("session.record"), + "actionable missing-operation diagnostic: {denied}" + ); + assert_eq!( + f.ok("GET", &path, Value::Null).await["permissions"], + revoked["permissions"] + ); + assert!( + open_subjektiv_store(&f.connected.api) + .unwrap() + .subject_session_attributions(&subject) + .unwrap() + .is_empty(), + "refused startup must not register a protected Session" + ); + + // A separate Subject proves the same denial at restore, with normal defaults + // at creation and explicit owner revocation only after a successful start. + let subject = f.subject("body-restore-revoked-subject").await; + let started = f + .mutate(&subject, "/start", absent_start("body-authorized-start")) + .await; + assert_eq!(started["disposition"], "accepted", "{started}"); + f.stop(&subject, "body-before-revoked-restore").await; + let path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"); + let policy = open_subjektiv_store(&f.connected.api) + .unwrap() + .subject_permissions(&subject) + .unwrap() + .unwrap(); + let execution = policy + .execution + .iter() + .filter(|permission| permission.operation != "session.record") + .cloned() + .collect::>(); + let revoked = f.ok("POST", &path, json!({"operation_id":"body-revoke-before-restore", + "expected_change_id":policy.change_id,"execution":execution,"delegation":policy.delegation})).await; + let observed = f.read(&subject).await; + let request = intent("body-denied-restore", &observed); + let denied = f.mutate(&subject, "/start", request.clone()).await; + assert_ne!(denied["disposition"], "accepted", "{denied}"); + let replay = f.mutate(&subject, "/start", request).await; + assert_ne!(replay["disposition"], "accepted", "{replay}"); + assert_eq!( + f.ok("GET", &path, Value::Null).await["permissions"], + revoked["permissions"] + ); + assert_eq!( + f.read(&subject).await["current"], + observed["current"], + "denial must not replace singleton generation" + ); + assert_eq!( + f.read(&subject).await["observation"]["worker"]["state"], + "stopped" + ); +} diff --git a/crates/workspace-server/src/server/tests/subject_body_tests.rs b/crates/workspace-server/src/server/tests/subject_body_tests.rs index e086bdb06..e244dd9a6 100644 --- a/crates/workspace-server/src/server/tests/subject_body_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_body_tests.rs @@ -131,6 +131,10 @@ async fn body_operations_reject_nonowners_and_cross_workspace_targets() { async fn body_stop_and_outcome_reject_a_target_from_another_subject_before_runtime_dispatch() { let f = BodyFixture::new().await; let lease = f.reserve(); + let permissions = open_subjektiv_store(&f.api) + .unwrap() + .subject_permissions(&f.subject) + .unwrap(); let other = open_subjektiv_store(&f.api) .unwrap() .create_subject(crate::subjektiv::SubjectRole::new("other-companion").unwrap()) @@ -157,12 +161,12 @@ async fn body_stop_and_outcome_reject_a_target_from_another_subject_before_runti f.read().await["current"]["worker_id"], lease.worker.worker_id ); - assert!( + assert_eq!( open_subjektiv_store(&f.api) .unwrap() .subject_permissions(&f.subject) - .unwrap() - .is_none() + .unwrap(), + permissions ); } @@ -235,13 +239,13 @@ async fn body_operations_reject_a_subject_with_deletion_already_admitted() { "{status}: {body}" ); } - assert!( - open_subjektiv_store(&f.api) - .unwrap() - .subject(&f.subject) - .unwrap() - .is_some() - ); + let deletion = open_subjektiv_store(&f.api) + .unwrap() + .subject_deletion(&f.subject) + .unwrap() + .unwrap(); + assert_eq!(deletion.operation_id, "delete-body-fixture"); + assert!(!deletion.completed); assert_eq!( f.api .store diff --git a/crates/workspace-server/src/server/tests/workdir_session_close_tests.rs b/crates/workspace-server/src/server/tests/workdir_session_close_tests.rs index 76d1a0fac..9dc441a25 100644 --- a/crates/workspace-server/src/server/tests/workdir_session_close_tests.rs +++ b/crates/workspace-server/src/server/tests/workdir_session_close_tests.rs @@ -277,6 +277,29 @@ impl Fixture { RuntimeWorkerRef::new(&self.runtime_id, "registry-close-owner") } + async fn create_worker(&self) -> RuntimeWorkerRef { + self.source.publish_profile("builtin:companion"); + let Json(created) = create_workspace_worker( + State(self.api().clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: self.runtime_id.clone(), + display_name: "session-close Worker".into(), + singleton_key: None, + profile: Some("builtin:companion".into()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachments: Vec::new(), + feature_connections: Default::default(), + control_operation_id: None, + execution_grant: None, + }), + ) + .await + .unwrap(); + RuntimeWorkerRef::new(created.runtime_id, created.worker_id) + } + async fn open(&self) -> WorkdirSessionHandle { self.api() .runtime @@ -734,7 +757,7 @@ async fn stop_projection_does_not_follow_a_rebound_connection_with_identical_wor { let fixture = Fixture::new().await; let api = fixture.api(); - let worker = fixture.registry_owner(); + let worker = fixture.create_worker().await; let link = WorkerWorkdirLinkRecord { connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.into(), @@ -763,7 +786,7 @@ async fn stop_projection_does_not_follow_a_rebound_connection_with_identical_wor .get_workdir_registry(TEST_WORKSPACE_ID, &fixture.workdir_id) .unwrap() .unwrap(); - record.display_name = "new binding not owned by old stop".into(); + record.display_name = Some("new binding not owned by old stop".into()); api.store.upsert_workdir_registry(&record).unwrap(); sync_linked_workdir_after_worker_stop(api, &worker, &[old]).unwrap(); assert_eq!( @@ -790,26 +813,7 @@ async fn stop_projection_does_not_follow_a_rebound_connection_with_identical_wor async fn worker_removal_close_failure_retains_attachment_and_retry_releases_runtime_lease() { let fixture = Fixture::new().await; let api = fixture.api(); - fixture.source.publish_profile("builtin:companion"); - let Json(created) = create_workspace_worker( - State(api.clone()), - HeaderMap::new(), - Json(CreateWorkspaceWorkerRequest { - runtime_id: fixture.runtime_id.clone(), - display_name: "session-close removal".into(), - singleton_key: None, - profile: Some("builtin:companion".into()), - ticket_assignment: None, - initial_submit: Vec::new(), - workdir_attachments: Vec::new(), - feature_connections: Default::default(), - control_operation_id: None, - execution_grant: None, - }), - ) - .await - .unwrap(); - let worker = RuntimeWorkerRef::new(created.runtime_id, created.worker_id); + let worker = fixture.create_worker().await; api.runtime .stop_worker( &worker, diff --git a/openapi/server-api.json b/openapi/server-api.json index 1cb14c69c..b45c3dc02 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -13414,6 +13414,16 @@ "null" ] }, + "stop_intent": { + "anyOf": [ + { + "$ref": "#/components/schemas/WorkerStopIntent" + }, + { + "type": "null" + } + ] + }, "ticket_assignment": { "anyOf": [ { @@ -13437,6 +13447,11 @@ }, "type": "array" }, + "replayed": { + "default": false, + "description": "Historical receipt replay, not a new stop or evidence of current state.", + "type": "boolean" + }, "runtime_id": { "type": "string" }, @@ -14784,6 +14799,259 @@ ], "type": "object" }, + "SubjectBodyDisposition": { + "enum": [ + "accepted", + "rejected", + "unknown" + ], + "type": "string" + }, + "SubjectBodyMutationResponse": { + "additionalProperties": false, + "properties": { + "body": { + "anyOf": [ + { + "$ref": "#/components/schemas/SubjectBodyResponse" + }, + { + "type": "null" + } + ], + "description": "Fresh observation is independent of the historical operation receipt." + }, + "diagnostics": { + "items": { + "$ref": "#/components/schemas/Diagnostic" + }, + "type": "array" + }, + "disposition": { + "$ref": "#/components/schemas/SubjectBodyDisposition" + }, + "request_id": { + "type": "string" + } + }, + "required": [ + "request_id", + "disposition", + "diagnostics" + ], + "type": "object" + }, + "SubjectBodyObservation": { + "oneOf": [ + { + "additionalProperties": false, + "properties": { + "kind": { + "const": "absent", + "type": "string" + } + }, + "required": [ + "kind" + ], + "type": "object" + }, + { + "additionalProperties": false, + "properties": { + "kind": { + "const": "available", + "type": "string" + }, + "worker": { + "$ref": "#/components/schemas/WorkerSummary" + } + }, + "required": [ + "kind", + "worker" + ], + "type": "object" + }, + { + "additionalProperties": false, + "description": "The owner is still occupied; this must not invite a replacement launch.", + "properties": { + "diagnostics": { + "items": { + "$ref": "#/components/schemas/Diagnostic" + }, + "type": "array" + }, + "kind": { + "const": "unavailable", + "type": "string" + } + }, + "required": [ + "kind", + "diagnostics" + ], + "type": "object" + } + ] + }, + "SubjectBodyRef": { + "additionalProperties": false, + "properties": { + "generation": { + "description": "Decimal text preserves the full singleton generation through browsers.", + "type": "string" + }, + "runtime_id": { + "type": "string" + }, + "worker_id": { + "type": "string" + } + }, + "required": [ + "runtime_id", + "worker_id", + "generation" + ], + "type": "object" + }, + "SubjectBodyResponse": { + "additionalProperties": false, + "properties": { + "can_start": { + "description": "Account authority, not a claim that Profile/model/Runtime admission will succeed.", + "type": "boolean" + }, + "can_stop": { + "type": "boolean" + }, + "current": { + "anyOf": [ + { + "$ref": "#/components/schemas/SubjectBodyRef" + }, + { + "type": "null" + } + ] + }, + "observation": { + "$ref": "#/components/schemas/SubjectBodyObservation" + }, + "subject_id": { + "type": "string" + }, + "workspace_id": { + "type": "string" + } + }, + "required": [ + "workspace_id", + "subject_id", + "observation", + "can_start", + "can_stop" + ], + "type": "object" + }, + "SubjectBodyStartRequest": { + "additionalProperties": false, + "properties": { + "expected_current": { + "additionalProperties": false, + "description": "Null explicitly means the caller observed no owner, not an unspecified target.", + "properties": { + "generation": { + "description": "Decimal text preserves the full singleton generation through browsers.", + "type": "string" + }, + "runtime_id": { + "type": "string" + }, + "worker_id": { + "type": "string" + } + }, + "required": [ + "runtime_id", + "worker_id", + "generation" + ], + "type": "object" + }, + "expected_observation_token": { + "type": [ + "string", + "null" + ] + }, + "request_id": { + "type": "string" + } + }, + "required": [ + "request_id", + "expected_current" + ], + "type": "object" + }, + "SubjectBodyStopOutcome": { + "enum": [ + "not_found", + "pending", + "stopped" + ], + "type": "string" + }, + "SubjectBodyStopOutcomeResponse": { + "additionalProperties": false, + "properties": { + "body": { + "anyOf": [ + { + "$ref": "#/components/schemas/SubjectBodyResponse" + }, + { + "type": "null" + } + ] + }, + "outcome": { + "$ref": "#/components/schemas/SubjectBodyStopOutcome", + "description": "Historical operation evidence, never the current body's running state." + }, + "request_id": { + "type": "string" + } + }, + "required": [ + "request_id", + "outcome" + ], + "type": "object" + }, + "SubjectBodyStopRequest": { + "additionalProperties": false, + "properties": { + "expected_current": { + "$ref": "#/components/schemas/SubjectBodyRef" + }, + "expected_observation_token": { + "type": "string" + }, + "request_id": { + "type": "string" + } + }, + "required": [ + "request_id", + "expected_current", + "expected_observation_token" + ], + "type": "object" + }, "SubjectPermissionInterruption": { "additionalProperties": false, "properties": { @@ -15561,6 +15829,17 @@ "created_at": { "type": "string" }, + "current_body": { + "anyOf": [ + { + "$ref": "#/components/schemas/SubjectBodyRef" + }, + { + "type": "null" + } + ], + "description": "Current owner remains present even when its Runtime cannot be observed." + }, "current_worker": { "anyOf": [ { @@ -15570,7 +15849,13 @@ "type": "null" } ], - "description": "Current keyed-singleton owner, when one exists. This is a live Yoi\nprojection, not state duplicated in the subjektiv store." + "description": "Available launch metadata, not evidence of absence when observation failed." + }, + "current_worker_diagnostics": { + "items": { + "$ref": "#/components/schemas/Diagnostic" + }, + "type": "array" }, "id": { "type": "string" @@ -20361,6 +20646,22 @@ ], "type": "object" }, + "WorkerStopIntent": { + "additionalProperties": false, + "properties": { + "expected_observation_token": { + "type": "string" + }, + "request_id": { + "type": "string" + } + }, + "required": [ + "expected_observation_token", + "request_id" + ], + "type": "object" + }, "WorkerSummary": { "description": "Workspace-authoritative Worker projection.\n\n`resource_key` is required here even though Runtime-internal Worker summaries\ndo not carry one. The Workspace Server must resolve it from Workspace\nauthority before constructing this response.", "properties": { @@ -22725,7 +23026,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:85593527d939626a32b88d9b2760b531681b087d95fefa5119901fe86d60d76c" + "x-yoi-source-digest": "sha256:efcd8fd0911bdafcde6cebdb812649166e043377995a75d9cb9305af6294c28a" }, "openapi": "3.1.0", "paths": { @@ -41910,6 +42211,474 @@ ] } }, + "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/body": { + "get": { + "operationId": "subjektiv_subject_body_get", + "parameters": [ + { + "explode": false, + "in": "path", + "name": "workspace_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + }, + { + "explode": false, + "in": "path", + "name": "subject_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyResponse" + } + } + }, + "description": "Successful response" + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + } + }, + "security": [ + { + "bearerAuth": [] + }, + { + "browserSession": [] + } + ] + } + }, + "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/body/start": { + "post": { + "operationId": "subjektiv_subject_body_start", + "parameters": [ + { + "explode": false, + "in": "path", + "name": "workspace_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + }, + { + "explode": false, + "in": "path", + "name": "subject_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyStartRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyMutationResponse" + } + } + }, + "description": "Successful response" + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + } + }, + "security": [ + { + "bearerAuth": [] + }, + { + "browserSession": [] + } + ] + } + }, + "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/body/stop": { + "post": { + "operationId": "subjektiv_subject_body_stop", + "parameters": [ + { + "explode": false, + "in": "path", + "name": "workspace_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + }, + { + "explode": false, + "in": "path", + "name": "subject_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyStopRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyMutationResponse" + } + } + }, + "description": "Successful response" + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + } + }, + "security": [ + { + "bearerAuth": [] + }, + { + "browserSession": [] + } + ] + } + }, + "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/body/stop/outcome": { + "post": { + "operationId": "subjektiv_subject_body_stop_outcome", + "parameters": [ + { + "explode": false, + "in": "path", + "name": "workspace_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + }, + { + "explode": false, + "in": "path", + "name": "subject_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyStopRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyStopOutcomeResponse" + } + } + }, + "description": "Successful response" + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "502": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + } + }, + "security": [ + { + "bearerAuth": [] + }, + { + "browserSession": [] + } + ] + } + }, "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/memories": { "get": { "operationId": "subjektiv_memory_list", diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index 766492c8d..48e4042cf 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:85593527d939626a32b88d9b2760b531681b087d95fefa5119901fe86d60d76c) +// Input contract: 0.1.0 (sha256:efcd8fd0911bdafcde6cebdb812649166e043377995a75d9cb9305af6294c28a) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check diff --git a/web/workspace/src/lib/generated/memory-api.ts b/web/workspace/src/lib/generated/memory-api.ts index 3daa53463..bc8ef4d7f 100644 --- a/web/workspace/src/lib/generated/memory-api.ts +++ b/web/workspace/src/lib/generated/memory-api.ts @@ -76,7 +76,7 @@ export type SubjektivSubjectDeleteResponse = { subject_id: string, operation_id: export type SubjektivSubjectBehaviorUpdateRequest = { expected_behavior_md: string, behavior_md: string, }; -export type SubjektivSubjectResponse = { id: string, role: string, behavior_md: string, state: SubjektivSubjectState, memory_fingerprint: string, created_at: string, updated_at: string, current_worker?: | import("./worker-launch-api").WorkerLaunchWorkerSummary | null, }; +export type SubjektivSubjectResponse = { id: string, role: string, behavior_md: string, state: SubjektivSubjectState, memory_fingerprint: string, created_at: string, updated_at: string, current_body?: import("./subject-body-api").SubjectBodyRef | null, current_worker_diagnostics?: Array, current_worker?: | import("./worker-launch-api").WorkerLaunchWorkerSummary | null, }; export type SubjektivSubjectListResponse = { limit: number, items: Array, next_cursor?: string | null, has_more: boolean, }; diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index 81f19707e..b13d52985 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:85593527d939626a32b88d9b2760b531681b087d95fefa5119901fe86d60d76c) +// Input contract: 0.1.0 (sha256:efcd8fd0911bdafcde6cebdb812649166e043377995a75d9cb9305af6294c28a) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index 5d91a36bf..51405e46d 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:85593527d939626a32b88d9b2760b531681b087d95fefa5119901fe86d60d76c) +// Input contract: 0.1.0 (sha256:efcd8fd0911bdafcde6cebdb812649166e043377995a75d9cb9305af6294c28a) // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/runtime-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check @@ -127,9 +127,9 @@ export type RuntimeWorkerInputRequest = { content: string; kind?: string | null; export type RuntimeWorkerInputResult = { diagnostics?: Array; runtime_id: string; state: WorkerOperationState; worker_id: string; }; -export type RuntimeWorkerLifecycleRequest = { reason?: string | null; ticket_assignment?: RuntimeWorkerTicketAssignmentRequest | null; }; +export type RuntimeWorkerLifecycleRequest = { reason?: string | null; stop_intent?: WorkerStopIntent | null; ticket_assignment?: RuntimeWorkerTicketAssignmentRequest | null; }; -export type RuntimeWorkerLifecycleResult = { diagnostics?: Array; runtime_id: string; state: WorkerOperationState; worker_id: string; }; +export type RuntimeWorkerLifecycleResult = { diagnostics?: Array; replayed?: boolean; runtime_id: string; state: WorkerOperationState; worker_id: string; }; export type RuntimeWorkerListResponse = { diagnostics?: Array; items: Array; limit: number; source: string; workspace_id: string; }; @@ -191,6 +191,8 @@ export type WorkerState = { kind: "idle"; } | { kind: "busy"; state: WorkerBusyS export type WorkerStateSnapshot = { last_command_id: number; last_finished_submission_request_id?: string | null; state: WorkerState; }; +export type WorkerStopIntent = { expected_observation_token: string; request_id: string; }; + export type WorkerSummary = { availability: SubscriptionWorkerAvailability; diagnostics?: Array; display_name?: string; host_id: string; implementation: WorkerImplementationSummary; label: string; last_seen_at?: string | null; pinned?: boolean; profile?: string | null; resource_key: string; restore_observation_token?: string | null; retention_state?: string; runtime_id: string; singleton_key?: string | null; state: string; tags?: Array; workdir_attachments?: Array; worker_id: string; worker_state?: WorkerStateSnapshot | null; workspace: WorkerWorkspaceSummary; }; export type WorkerWorkdirAttachmentSummary = { alias: string; effective_permissions: ExternalWorkdirPermissions; working_directory: WorkingDirectorySummary; }; diff --git a/web/workspace/src/lib/generated/subject-body-api.ts b/web/workspace/src/lib/generated/subject-body-api.ts new file mode 100644 index 000000000..177cc0446 --- /dev/null +++ b/web/workspace/src/lib/generated/subject-body-api.ts @@ -0,0 +1,68 @@ +// Generated by `cargo run -p server-api --features typescript --example generate_subject_body_api_types`. +// Do not edit manually. + +import type { Diagnostic, WorkerSummary } from "./worker-launch-api"; + +export type SubjectBodyRef = { + runtime_id: string; + worker_id: string; + /** + * Decimal text preserves the full singleton generation through browsers. + */ + generation: string; +}; + +export type SubjectBodyObservation = { "kind": "absent" } | { + "kind": "available"; + worker: WorkerSummary; +} | { "kind": "unavailable"; diagnostics: Array }; + +export type SubjectBodyResponse = { + workspace_id: string; + subject_id: string; + current: SubjectBodyRef | null; + observation: SubjectBodyObservation; + /** + * Account authority, not a claim that Profile/model/Runtime admission will succeed. + */ + can_start: boolean; + can_stop: boolean; +}; + +export type SubjectBodyStartRequest = { + request_id: string; + /** + * Null explicitly means the caller observed no owner, not an unspecified target. + */ + expected_current: SubjectBodyRef | null; + expected_observation_token: string | null; +}; + +export type SubjectBodyStopRequest = { + request_id: string; + expected_current: SubjectBodyRef; + expected_observation_token: string; +}; + +export type SubjectBodyDisposition = "accepted" | "rejected" | "unknown"; + +export type SubjectBodyMutationResponse = { + request_id: string; + disposition: SubjectBodyDisposition; + diagnostics: Array; + /** + * Fresh observation is independent of the historical operation receipt. + */ + body: SubjectBodyResponse | null; +}; + +export type SubjectBodyStopOutcome = "not_found" | "pending" | "stopped"; + +export type SubjectBodyStopOutcomeResponse = { + request_id: string; + /** + * Historical operation evidence, never the current body's running state. + */ + outcome: SubjectBodyStopOutcome; + body: SubjectBodyResponse | null; +}; diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index e643dbc5b..58a6ef758 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:85593527d939626a32b88d9b2760b531681b087d95fefa5119901fe86d60d76c) +// Input contract: 0.1.0 (sha256:efcd8fd0911bdafcde6cebdb812649166e043377995a75d9cb9305af6294c28a) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index 7fc5de5f0..5c5805f3d 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:85593527d939626a32b88d9b2760b531681b087d95fefa5119901fe86d60d76c) +// Input contract: 0.1.0 (sha256:efcd8fd0911bdafcde6cebdb812649166e043377995a75d9cb9305af6294c28a) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check diff --git a/web/workspace/src/lib/workspace/api/runtime-workers.ts b/web/workspace/src/lib/workspace/api/runtime-workers.ts index 49c6861a1..b1875c8cf 100644 --- a/web/workspace/src/lib/workspace/api/runtime-workers.ts +++ b/web/workspace/src/lib/workspace/api/runtime-workers.ts @@ -319,7 +319,7 @@ export function parseRuntimeWorkerLifecycleResult( const item = object(value, "Runtime Worker lifecycle result"); exactKeys( item, - ["state", "runtime_id", "worker_id", "diagnostics", "replayed"], + ["state", "runtime_id", "worker_id", "diagnostics"], ["replayed"], "Runtime Worker lifecycle result", ); diff --git a/web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts b/web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts index 7bcb427d5..62e6a28a9 100644 --- a/web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts +++ b/web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts @@ -6,7 +6,7 @@ import { screen, waitFor, } from "@testing-library/svelte"; -import { afterEach, expect, test, vi } from "vitest"; +import { afterEach, beforeEach, expect, test, vi } from "vitest"; import { tick } from "svelte"; import SubjectBody from "./SubjectBody.svelte"; import { @@ -73,6 +73,23 @@ function savedStop(accountId = props.accountId) { }; } +// Happy DOM does not update bodyUsed when the production bounded reader consumes a stream. +// Observe releaseLock instead, so late-response assertions run after actual consumption. +function afterBodyRead(response: Response): Promise { + return new Promise((resolve) => { + const getReader = response.body!.getReader.bind(response.body); + vi.spyOn(response.body!, "getReader").mockImplementation(() => { + const reader = getReader(); + const release = reader.releaseLock.bind(reader); + vi.spyOn(reader, "releaseLock").mockImplementation(() => { + release(); + resolve(); + }); + return reader; + }); + }); +} + function deferred() { let resolve!: (value: T) => void; const promise = new Promise((accept) => { @@ -80,6 +97,17 @@ function deferred() { }); return { promise, resolve }; } +// Happy DOM Storage is a Proxy with bound own methods; spy on a plain adapter +// rather than its unrelated Storage.prototype or non-restorable proxy properties. +beforeEach(() => { + const storage = sessionStorage; + vi.stubGlobal("sessionStorage", { + getItem: storage.getItem.bind(storage), + setItem: storage.setItem.bind(storage), + removeItem: storage.removeItem.bind(storage), + clear: storage.clear.bind(storage), + }); +}); afterEach(() => { cleanup(); vi.restoreAllMocks(); @@ -337,7 +365,7 @@ test("storage write failure sends nothing and exposes a recovery diagnostic", as const start = await screen.findByRole("button", { name: "Start conversation body", }); - vi.spyOn(Storage.prototype, "setItem").mockImplementation(() => { + vi.spyOn(sessionStorage, "setItem").mockImplementation(() => { throw new Error("storage denied"); }); await fireEvent.click(start); @@ -366,8 +394,9 @@ test("late GET for an old route cannot overwrite the selected Subject", async () await view.rerender({ ...props, subjectId: "other" }); await screen.findByText(/You do not have permission to start or stop/); const response = Response.json(available()); + const consumed = afterBodyRead(response); old.resolve(response); - await waitFor(() => expect(response.bodyUsed).toBe(true)); + await consumed; await tick(); await waitFor(() => expect(screen.getByText("No current conversation body.")) @@ -414,8 +443,9 @@ test("late old-route mutation receipt cannot update new route or discard old rec diagnostics: [], body: available(), }); + const consumed = afterBodyRead(response); pending.resolve(response); - await waitFor(() => expect(response.bodyUsed).toBe(true)); + await consumed; await tick(); await waitFor(() => expect(screen.getByText("No current conversation body.")) @@ -753,8 +783,9 @@ test("late stop outcome cannot settle an old route request after navigation", as outcome: "stopped", body: available(), }); + const consumed = afterBodyRead(response); pending.resolve(response); - await waitFor(() => expect(response.bodyUsed).toBe(true)); + await consumed; await tick(); expect(screen.queryByText("Historical stop confirmed.")).toBeNull(); expect(screen.queryByRole("button", { name: "Check stop outcome" })) @@ -805,7 +836,7 @@ test("account switch hides old body and retry input while same-account reload pr const view = render(SubjectBody, props); await screen.findByText("Body stopped."); await screen.findByText(/An unresolved stop request is retained/); - const reads = vi.spyOn(Storage.prototype, "getItem"); + const reads = vi.spyOn(sessionStorage, "getItem"); reads.mockClear(); account = accountB.accountId; await view.rerender(accountB); @@ -875,8 +906,9 @@ test("late GET from another account cannot replace the same route's new observat await view.rerender({ ...props, accountId: account }); await screen.findByText(/You do not have permission to start or stop/); const oldResponse = Response.json(available()); + const consumed = afterBodyRead(oldResponse); pendingA.resolve(oldResponse); - await waitFor(() => expect(oldResponse.bodyUsed).toBe(true)); + await consumed; await tick(); expect(screen.getByText("No current conversation body.")).toBeTruthy(); expect(screen.queryByText("Body stopped.")).toBeNull(); @@ -962,7 +994,7 @@ test.each([false, true])( ).toBe(false) ); } - const removals = vi.spyOn(Storage.prototype, "removeItem"); + const removals = vi.spyOn(sessionStorage, "removeItem"); const getsBeforeResponse = gets; const oldResponse = Response.json({ request_id: requestA, @@ -970,8 +1002,9 @@ test.each([false, true])( diagnostics: [], body: available(), }); + const consumed = afterBodyRead(oldResponse); pendingA.resolve(oldResponse); - await waitFor(() => expect(oldResponse.bodyUsed).toBe(true)); + await consumed; await tick(); expect(screen.queryByText("Request accepted.")).toBeNull(); expect(screen.queryByText("Body stopped.")).toBeNull(); @@ -1024,14 +1057,15 @@ test("late stop outcome from another account cannot settle or clear either accou expect( screen.queryByText("Checking saved stop outcome… No stop request is sent."), ).toBeNull(); - const removals = vi.spyOn(Storage.prototype, "removeItem"); + const removals = vi.spyOn(sessionStorage, "removeItem"); const oldResponse = Response.json({ request_id: first.request.request_id, outcome: "stopped", body: available(), }); + const consumed = afterBodyRead(oldResponse); pending.resolve(oldResponse); - await waitFor(() => expect(oldResponse.bodyUsed).toBe(true)); + await consumed; await tick(); expect(screen.queryByText("Historical stop confirmed.")).toBeNull(); expect(screen.queryByText("Body stopped.")).toBeNull(); @@ -1057,7 +1091,7 @@ test("signout hides body and request details and never falls back to a dummy ide const view = render(SubjectBody, props); await screen.findByText("Body stopped."); expect(screen.getByText(first.request.request_id)).toBeTruthy(); - const reads = vi.spyOn(Storage.prototype, "getItem"); + const reads = vi.spyOn(sessionStorage, "getItem"); reads.mockClear(); await view.rerender({ ...props, accountId: "" }); await screen.findByText("Sign in to view this conversation body."); @@ -1071,3 +1105,227 @@ test("signout hides body and request details and never falls back to a dummy ide expect(fetch).toHaveBeenCalledTimes(1); expect(sessionStorage.getItem(key)).toBe(raw); }); + +test("Workspace switch fences late mutation even when the Subject ID is identical", async () => { + const pending = deferred(); + let requestId = ""; + const other = { ...props, workspaceId: "other-workspace" }; + vi.stubGlobal( + "fetch", + vi.fn(async (path: string, init?: RequestInit) => { + if (init?.method === "POST") { + requestId = JSON.parse(init.body as string).request_id; + return pending.promise; + } + return Response.json( + path.includes("/other-workspace/") + ? { + ...absent(), + workspace_id: other.workspaceId, + can_start: false, + can_stop: false, + } + : absent(), + ); + }), + ); + const view = render(SubjectBody, props); + await fireEvent.click( + await screen.findByRole("button", { name: "Start conversation body" }), + ); + const key = subjectBodyStorageKey( + props.accountId, + props.workspaceId, + props.subjectId, + ); + const retained = sessionStorage.getItem(key); + await view.rerender(other); + await screen.findByText(/You do not have permission to start or stop/); + const response = Response.json({ + request_id: requestId, + disposition: "accepted", + diagnostics: [], + body: available(), + }); + const consumed = afterBodyRead(response); + pending.resolve(response); + await consumed; + await tick(); + expect(screen.queryByText("Request accepted.")).toBeNull(); + expect(screen.queryByText("Body stopped.")).toBeNull(); + expect(sessionStorage.getItem(key)).toBe(retained); + expect( + readSubjectBodyAttempt( + sessionStorage, + props.accountId, + other.workspaceId, + props.subjectId, + ), + ).toBeNull(); + await view.rerender(props); + await screen.findByText(/An unresolved start request is retained/); +}); + +test("double click sends only one retained request, never a conversation ACK", async () => { + const pending = deferred(); + const posts: string[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (_path: string, init?: RequestInit) => { + if (init?.method !== "POST") return Response.json(absent()); + posts.push(init.body as string); + return pending.promise; + }), + ); + render(SubjectBody, props); + const start = await screen.findByRole("button", { + name: "Start conversation body", + }); + await fireEvent.click(start); + await fireEvent.click(start); + expect(posts).toHaveLength(1); + // A chat-submit acknowledgement is not a body mutation receipt. + const response = Response.json({ + command_id: "chat-command", + state: "accepted", + }); + const consumed = afterBodyRead(response); + pending.resolve(response); + await consumed; + await screen.findByText(/Request outcome unknown\. The request may/); + expect( + readSubjectBodyAttempt( + sessionStorage, + props.accountId, + props.workspaceId, + props.subjectId, + )?.request, + ) + .toEqual(JSON.parse(posts[0])); + expect(screen.queryByText("Request accepted.")).toBeNull(); +}); + +test("old generation rejection stays distinct from fresh body and requires a fresh explicit stop", async () => { + let observation = available("idle"); + const posts: { + expected_current: typeof owner; + expected_observation_token: string; + }[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (_path: string, init?: RequestInit) => { + if (init?.method !== "POST") return Response.json(observation); + const request = JSON.parse(init.body as string); + posts.push(request); + observation = available("idle", "generation-18-token"); + observation.current = { ...owner, generation: "18" }; + return Response.json({ + request_id: request.request_id, + disposition: "rejected", + diagnostics: [{ + code: "stale_owner", + severity: "warning", + message: + "The current body generation changed. Refresh before acting.", + }], + body: observation, + }); + }), + ); + render(SubjectBody, props); + await fireEvent.click( + await screen.findByRole("button", { name: "Stop conversation body" }), + ); + await screen.findByText("Request rejected."); + await screen.findByText( + "The current body generation changed. Refresh before acting.", + ); + await waitFor(() => + expect( + (screen.getByRole("button", { + name: "Stop conversation body", + }) as HTMLButtonElement).disabled, + ).toBe(false) + ); + expect(posts).toHaveLength(1); + expect(posts[0].expected_current.generation).toBe("17"); + expect(screen.queryByText("Body stopped.")).toBeNull(); + await fireEvent.click( + screen.getByRole("button", { name: "Stop conversation body" }), + ); + await waitFor(() => expect(posts).toHaveLength(2)); + expect(posts[1].expected_current.generation).toBe("18"); + expect(posts[1].expected_observation_token).toBe("generation-18-token"); +}); + +test("unknown stop cannot restart on GET alone; confirmed outcome permits restart with fresh token", async () => { + let observation = available("idle"); + let outcome = "pending"; + const posts: { path: string; body: string }[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (path: string, init?: RequestInit) => { + if (init?.method !== "POST") return Response.json(observation); + posts.push({ path, body: init.body as string }); + const request = JSON.parse(init.body as string); + if (path.endsWith("/outcome")) { + return Response.json({ + request_id: request.request_id, + outcome, + body: null, + }); + } + return Response.json({ + request_id: request.request_id, + disposition: "unknown", + diagnostics: [], + body: null, + }); + }), + ); + render(SubjectBody, props); + await fireEvent.click( + await screen.findByRole("button", { name: "Stop conversation body" }), + ); + await screen.findByText("Request outcome unknown."); + await waitFor(() => + expect( + (screen.getByRole("button", { + name: "Refresh body state", + }) as HTMLButtonElement).disabled, + ).toBe(false) + ); + observation = available("stopped", "after-confirmed-stop"); + await fireEvent.click( + screen.getByRole("button", { name: "Refresh body state" }), + ); + await screen.findByText("Body stopped."); + expect(screen.queryByRole("button", { name: "Restart conversation body" })) + .toBeNull(); + await fireEvent.click( + screen.getByRole("button", { name: "Check stop outcome" }), + ); + await screen.findByText("Stop outcome pending."); + expect(screen.queryByRole("button", { name: "Restart conversation body" })) + .toBeNull(); + outcome = "stopped"; + await fireEvent.click( + screen.getByRole("button", { name: "Check stop outcome" }), + ); + await screen.findByText("Historical stop confirmed."); + const restart = await screen.findByRole("button", { + name: "Restart conversation body", + }); + await waitFor(() => + expect((restart as HTMLButtonElement).disabled).toBe(false) + ); + await fireEvent.click(restart); + await waitFor(() => expect(posts).toHaveLength(4)); + expect(posts[1].body).toBe(posts[0].body); + expect(posts[2].body).toBe(posts[0].body); + const request = JSON.parse(posts[3].body); + expect(posts[3].path).toMatch(/\/start$/); + expect(request.expected_current).toEqual(owner); + expect(request.expected_observation_token).toBe("after-confirmed-stop"); + expect(request.request_id).not.toBe(JSON.parse(posts[0].body).request_id); +}); diff --git a/web/workspace/src/routes/w/[workspaceId]/memory/memory-page.browser.spec.ts b/web/workspace/src/routes/w/[workspaceId]/memory/memory-page.browser.spec.ts index e94b2342b..f0cc2bb9a 100644 --- a/web/workspace/src/routes/w/[workspaceId]/memory/memory-page.browser.spec.ts +++ b/web/workspace/src/routes/w/[workspaceId]/memory/memory-page.browser.spec.ts @@ -110,7 +110,7 @@ test("renders the subject index as the Memory product entry", () => { expect(screen.getByRole("link", { name: /Release coordinator/ })).not .toBeNull(); expect(screen.getByText("subject-1")).not.toBeNull(); - expect(screen.getByText("Not connected")).not.toBeNull(); + expect(screen.getByText("No current body")).not.toBeNull(); expect(screen.queryByText("Memory fingerprint", { exact: true })).toBeNull(); expect(screen.getByText("1 shown · more available")).not.toBeNull(); expect(screen.getByRole("link", { name: "First page" }).getAttribute("href")) @@ -346,8 +346,14 @@ test("renders a ready resident surface and committed Memory lifecycle states", a .toContain("Prefer explicit evidence."); expect(screen.getByRole("article", { name: "Resident context" })).not .toBeNull(); - expect(screen.getByText("Connected", { exact: true })).not.toBeNull(); - expect(screen.getByText("Release Worker", { exact: true })).not.toBeNull(); + // Legacy connection fields cannot authorize or infer the current singleton body. + expect(screen.queryByText("Connected", { exact: true })).toBeNull(); + expect(screen.queryByText("Release Worker", { exact: true })).toBeNull(); + expect( + screen.getByText( + "Confirming account identity before loading body controls…", + ), + ).toBeTruthy(); expect(screen.getByText("3 on this page", { exact: true })).not.toBeNull(); expect(screen.getByText("Memory fingerprint", { exact: true })).not .toBeNull(); @@ -791,7 +797,12 @@ test("candidate-only Subjects show no committed Memories, not an inferred candid expect(screen.getByText("None", { exact: true })).not.toBeNull(); expect(screen.getByText("No committed Memories yet.")).not.toBeNull(); - expect(screen.getByText("Not connected", { exact: true })).not.toBeNull(); + expect( + screen.getByText( + "Confirming account identity before loading body controls…", + { exact: true }, + ), + ).not.toBeNull(); expect(screen.getAllByText("Not generated", { exact: true })).toHaveLength(2); expect(screen.queryByRole("article", { name: "Resident context" })) .toBeNull(); @@ -834,7 +845,12 @@ test.each(["ready", "ungenerated", "stale", "failed"] as const)( expect(screen.getByText("Generated Memory context", { exact: true })).not .toBeNull(); expect(screen.getByText("No committed Memories yet.")).not.toBeNull(); - expect(screen.getByText("Not connected", { exact: true })).not.toBeNull(); + expect( + screen.getByText( + "Confirming account identity before loading body controls…", + { exact: true }, + ), + ).not.toBeNull(); const technical = view.container.querySelector( "details.subject-technical-details", ) as HTMLDetailsElement; diff --git a/web/workspace/test/subject-body-review.ts b/web/workspace/test/subject-body-review.ts new file mode 100644 index 000000000..7c5116bfc --- /dev/null +++ b/web/workspace/test/subject-body-review.ts @@ -0,0 +1,491 @@ +#!/usr/bin/env -S deno run -A +/** Actual Web build + synthetic HTTP only. No real Backend/Runtime is contacted. */ +import { assert, assertEquals } from "@std/assert"; +import { dirname, fromFileUrl, join, resolve } from "@std/path"; +import { chromium } from "playwright"; +import { sha256File } from "../../../tools/web-ux/src/artifacts.ts"; +import { + startOwnedProcesses, + stopOwnedProcesses, +} from "../../../tools/web-ux/src/processes.ts"; +import { + assertNoPageWideOverflow, + freePort, +} from "../../../tools/web-ux/browser-tests/memory_document_harness.ts"; + +const root = resolve(dirname(fromFileUrl(import.meta.url)), "../../.."); +const run = Deno.args[0] ?? "review"; +if (!/^[\w-]+$/.test(run)) throw new Error("safe unique run id required"); +const output = join(root, "target/t807-web", run); +await Deno.mkdir(output); // Never replace previous evidence. +const buildRoot = join(root, "web/workspace/build"); +await Deno.stat(join(buildRoot, "index.html")); +const git = async (args: string[]) => { + const result = await new Deno.Command("git", { + args, + cwd: root, + stdout: "piped", + }).output(); + assert(result.success); + return new TextDecoder().decode(result.stdout).trim(); +}; +const manifest = { + evidence: "actual-Web-with-synthetic-HTTP-NOT-real-Backend-or-Runtime", + source: { + commit: await git(["rev-parse", "HEAD"]), + status: await git(["status", "--short"]), + }, + buildIndexSha256: await sha256File(join(buildRoot, "index.html")), + browser: "", + status: "running", + visualVerdict: "unreviewed", + expectedFaults: [ + "structured session.record rejection", + "unknown start/stop receipts", + "stale generation rejection", + "503 body GET", + "navigation teardown aborts GET /workers refresh", + ], + captures: [] as unknown[], + assertions: [] as string[], + failures: [] as string[], +}; +const persist = () => + Deno.writeTextFile( + join(output, "review-context.json"), + JSON.stringify(manifest, null, 2), + ); +const port = await freePort(); +const base = `http://127.0.0.1:${port}`; +const owned = await startOwnedProcesses( + [{ + id: "synthetic-memory-http", + command: Deno.execPath(), + cwd: join(root, "tools/web-ux"), + args: [ + "run", + "--cached-only", + "-A", + "browser-tests/memory_document_fixture_server.ts", + String(port), + buildRoot, + "3", + ], + readyUrl: `${base}/health`, + }], + join(output, "scenario.json"), + join(output, "logs"), + [], +); +let browser; +try { + browser = await chromium.launch({ headless: true }); + manifest.browser = browser.version(); + for (const colorScheme of ["light", "dark"] as const) { + for (const width of [1440, 768, 390]) { + const prefix = `${colorScheme}-${width}`; + const context = await browser.newContext({ + viewport: { width, height: width === 390 ? 844 : 900 }, + colorScheme, + reducedMotion: "reduce", + }); + try { + let accountId = "owner"; + let observation: "absent" | "idle" | "stopped" | "unavailable" = + "absent"; + let permission = true; + let getFailure = false; + let action: "reject" | "unknown" | "accepted" | "stale" = "reject"; + let outcome: "pending" | "stopped" = "pending"; + let generation = "17"; + const posts: { path: string; body: string }[] = []; + const owner = () => ({ + runtime_id: "runtime", + worker_id: "worker", + generation, + }); + function body(workspace: string, subject: string) { + return { + workspace_id: workspace, + subject_id: subject, + can_start: permission, + can_stop: permission, + current: observation === "absent" ? null : owner(), + observation: observation === "absent" + ? { kind: "absent" } + : observation === "unavailable" + ? { + kind: "unavailable", + diagnostics: [{ + code: "runtime_unavailable", + severity: "warning", + message: + "Runtime observation unavailable; no replacement body will be started.", + }], + } + : { + kind: "available", + worker: { + runtime_id: "runtime", + worker_id: "worker", + resource_key: "W-1", + host_id: "host", + label: "Conversation body", + workspace: { + visibility: "workspace", + identity: workspace, + workspace_id: workspace, + }, + implementation: { kind: "embedded", display_hint: "" }, + availability: "observed", + state: observation, + worker_state: observation === "stopped" + ? null + : { last_command_id: 1, state: { kind: "idle" } }, + restore_observation_token: + `token-${generation}-${observation}`, + }, + }, + }; + } + await context.route("**/api/auth/whoami", async (route) => { + const response = await route.fetch(); + const value = await response.json(); + value.actor.account_id = accountId; + await route.fulfill({ response, json: value }); + }); + // The second Workspace uses a synthetic, silent protocol channel too. + await context.routeWebSocket( + "**/api/w/other-review/protocol/ws", + (socket) => { + socket.onMessage(() => {}); + }, + ); + // Reuse the existing isolated fixture for a second Workspace identity. + await context.route( + /\/api\/w\/other-review(?:\/.*)?$/, + async (route) => { + const response = await route.fetch({ + url: route.request().url().replace( + "/api/w/other-review/", + "/api/w/memory-review/", + ), + }); + await route.fulfill({ + response, + body: (await response.text()).replaceAll( + "memory-review", + "other-review", + ), + }); + }, + ); + await context.route( + /\/api\/w\/[^/]+\/subjektiv\/subjects\/[^/]+\/body(?:\/.*)?$/, + async (route) => { + const request = route.request(); + const path = new URL(request.url()).pathname; + const [, workspace, subject] = path.match( + /\/api\/w\/([^/]+)\/subjektiv\/subjects\/([^/]+)\/body/, + )!; + if (request.method() === "GET") { + await route.fulfill({ + status: getFailure ? 503 : 200, + json: getFailure + ? { error: "Synthetic body observation unavailable" } + : body(workspace, subject), + }); + return; + } + const raw = request.postData()!; + posts.push({ path, body: raw }); + const input = JSON.parse(raw); + assert(!("accountId" in input)); + if (path.endsWith("/outcome")) { + await route.fulfill({ + json: { request_id: input.request_id, outcome, body: null }, + }); + return; + } + if (action === "stale") generation = "18"; + if (action === "accepted" && path.endsWith("/start")) { + observation = "idle"; + } + await route.fulfill({ + json: { + request_id: input.request_id, + disposition: action === "reject" || action === "stale" + ? "rejected" + : action, + diagnostics: action === "reject" + ? [{ + code: "permission_missing", + severity: "error", + message: + "Required permission session.record is missing. Ask an administrator to review the default profile; no permissions were added.", + }] + : action === "stale" + ? [{ + code: "stale_owner", + severity: "warning", + message: + "The current body generation changed. Refresh before acting.", + }] + : [], + body: body(workspace, subject), + }, + }); + }, + ); + const page = await context.newPage(); + const errors: unknown[] = []; + let documentStatus: number | null = null; + page.on( + "pageerror", + (e) => errors.push({ kind: "page", message: e.message }), + ); + page.on("console", (e) => { + if (["warning", "error"].includes(e.type())) { + errors.push({ kind: "console", message: e.text() }); + } + }); + page.on( + "requestfailed", + (r) => + errors.push({ + kind: "request", + url: r.url(), + message: r.failure()?.errorText, + }), + ); + page.on("response", (r) => { + if (r.request().isNavigationRequest()) documentStatus = r.status(); + if (r.status() >= 400) { + errors.push({ kind: "http", status: r.status(), url: r.url() }); + } + }); + const panel = () => + page.getByRole("region", { name: "Conversation body", exact: true }); + const button = (name: string) => + panel().getByRole("button", { name, exact: true }); + async function ready() { + await page.locator(".subject-body header button:not(:disabled)") + .waitFor(); + } + async function refresh() { + await ready(); + await button("Refresh body state").click(); + await ready(); + } + async function capture(name: string, expectedErrors: string[] = []) { + await panel().scrollIntoViewIfNeeded(); + const id = `${prefix}-${name}`; + const image = join(output, `${id}.png`); + const aria = join(output, `${id}.aria.txt`); + const panelImage = join(output, `${id}.panel.png`); + await page.screenshot({ path: image }); + await panel().screenshot({ path: panelImage }); + await Deno.writeTextFile( + aria, + await page.locator("body").ariaSnapshot(), + ); + manifest.captures.push({ + id, + route: new URL(page.url()).pathname, + accountId, + viewport: page.viewportSize(), + theme: colorScheme, + documentStatus, + visibleAlerts: await page.getByRole("alert").allTextContents(), + bodyText: await panel().innerText(), + errors: [...errors], + expectedErrors, + image, + imageSha256: await sha256File(image), + panelImage, + panelImageSha256: await sha256File(panelImage), + accessibilitySnapshot: aria, + reviewed: false, + }); + await persist(); + await assertNoPageWideOverflow(page); + } + // Enter through the ordinary Subjects list, not a special-purpose shell. + await page.goto(`${base}/w/memory-review/memory`); + await page.locator('a.subject-row[href$="/empty-subject"]').click(); + await panel().getByText("No current conversation body.", { + exact: true, + }).waitFor(); + await ready(); + await button("Refresh body state").focus(); + await page.keyboard.press("Tab"); + assert( + await button("Start conversation body").evaluate((el) => + el === el.ownerDocument.activeElement + ), + ); + await capture("before-absent-focus"); + await page.keyboard.press("Enter"); + await panel().getByText("Request rejected.", { exact: true }).waitFor(); + await ready(); + await capture("permission-rejection"); + action = "unknown"; + await button("Start conversation body").click(); + await panel().getByText("Request outcome unknown.", { exact: true }) + .waitFor(); + await ready(); + const unknownInput = posts.at(-1)!.body; + observation = "idle"; + await refresh(); + assertEquals(await button("Start conversation body").count(), 0); + await capture("unknown-after-get"); + await page.reload(); + await panel().getByText(/An unresolved start request is retained/) + .waitFor(); + await ready(); + action = "accepted"; + await button("Retry same start request").click(); + await panel().getByText("Request accepted.", { exact: true }).waitFor(); + await ready(); + assertEquals(posts.at(-1)!.body, unknownInput); + await capture("running-reconciled"); + action = "unknown"; + await button("Stop conversation body").click(); + await panel().getByText("Request outcome unknown.", { exact: true }) + .waitFor(); + await ready(); + const stopInput = posts.at(-1)!.body; + await button("Check stop outcome").click(); + await panel().getByText("Stop outcome pending.", { exact: true }) + .waitFor(); + await ready(); + observation = "stopped"; + await refresh(); + assertEquals(await button("Restart conversation body").count(), 0); + await capture("stopped-but-unresolved"); + outcome = "stopped"; + await button("Check stop outcome").click(); + await panel().getByText("Historical stop confirmed.", { exact: true }) + .waitFor(); + await ready(); + assertEquals(posts.at(-1)!.body, stopInput); + await capture("stop-confirmed"); + action = "accepted"; + await button("Restart conversation body").click(); + await panel().getByText("Request accepted.", { exact: true }).waitFor(); + await ready(); + const restart = JSON.parse(posts.at(-1)!.body); + assertEquals(restart.expected_observation_token, "token-17-stopped"); + assertEquals(restart.expected_current.generation, "17"); + assert(restart.request_id !== JSON.parse(stopInput).request_id); + await capture("restarted-running"); + action = "stale"; + await button("Stop conversation body").click(); + await panel().getByText( + "The current body generation changed. Refresh before acting.", + { exact: true }, + ).waitFor(); + await ready(); + assertEquals( + JSON.parse(posts.at(-1)!.body).expected_current.generation, + "17", + ); + await capture("old-generation-rejected"); + permission = false; + await refresh(); + assertEquals(await button("Stop conversation body").count(), 0); + await capture("read-only"); + permission = true; + observation = "unavailable"; + await refresh(); + assertEquals(await button("Start conversation body").count(), 0); + await capture("unavailable-owner"); + // Account identity changes must not expose another account's retained retry. + observation = "absent"; + action = "unknown"; + await refresh(); + await button("Start conversation body").click(); + await panel().getByText("Request outcome unknown.", { exact: true }) + .waitFor(); + await ready(); + const beforeIdentitySwitch = posts.length; + accountId = "other-account"; + await page.reload(); + await panel().getByText("No current conversation body.", { + exact: true, + }).waitFor(); + await ready(); + assertEquals(await button("Retry same start request").count(), 0); + assertEquals(posts.length, beforeIdentitySwitch); + // Ordinary back-to-list and Subject navigation must isolate the old request too. + await page.getByRole("link", { name: "← All subjects", exact: true }) + .click(); + await page.locator('a.subject-row[href$="/stale-subject"]').click(); + await panel().getByText("No current conversation body.", { + exact: true, + }).waitFor(); + await ready(); + assertEquals(await button("Retry same start request").count(), 0); + await capture("identity-subject-switched"); + await button("Start conversation body").click(); + await panel().getByText("Request outcome unknown.", { exact: true }) + .waitFor(); + await ready(); + const beforeWorkspaceSwitch = posts.length; + await page.goto(`${base}/w/other-review/memory/stale-subject`); + await panel().getByText("No current conversation body.", { + exact: true, + }).waitFor(); + await ready(); + assertEquals(await button("Retry same start request").count(), 0); + assertEquals(posts.length, beforeWorkspaceSwitch); + await capture("workspace-switched"); + await page.goto(`${base}/w/memory-review/memory/stale-subject`); + await panel().getByText(/An unresolved start request is retained/) + .waitFor(); + await ready(); + assertEquals(posts.length, beforeWorkspaceSwitch); + getFailure = true; + await refresh(); + await panel().getByRole("alert").waitFor(); + assert(await button("Retry same start request").isDisabled()); + await capture("get-failure", [ + "intentional 503 body GET; browser console reports failed resource", + ]); + const unexpected = errors.filter((e: any) => { + if (e.kind === "request") { + return !(e.message === "net::ERR_ABORTED" && + /^\/api\/w\/(memory-review|other-review)\/workers$/.test( + new URL(e.url).pathname, + )); + } + if (e.kind === "http") { + return e.status !== 503 || !e.url.endsWith("/body"); + } + if (e.kind === "console") { + return !e.message.includes("503 (Service Unavailable)"); + } + return true; + }); + assertEquals(unexpected, [], "unexpected browser diagnostics"); + manifest.assertions.push( + `${prefix}: normal entry, keyboard focus/activation, missing session.record, unknown GET/reload/exact retry, pending stop, confirmed stop/restart, old generation, read-only, unavailable owner, account, Workspace and Subject isolation, GET failure`, + ); + } finally { + await context.close(); + } + } + } + manifest.status = "pass"; +} catch (error) { + manifest.status = "fail"; + manifest.failures.push( + error instanceof Error ? error.stack ?? error.message : String(error), + ); + throw error; +} finally { + await persist(); + await browser?.close(); + await stopOwnedProcesses(owned); +} +console.log(output); diff --git a/web/workspace/tests/runtime-workers.test.ts b/web/workspace/tests/runtime-workers.test.ts index a868cf508..3743e1fb7 100644 --- a/web/workspace/tests/runtime-workers.test.ts +++ b/web/workspace/tests/runtime-workers.test.ts @@ -297,3 +297,25 @@ Deno.test("Restore parser bounds diagnostics, validates wrapper/summary identity "wire contract", ); }); + +Deno.test("lifecycle replay metadata is optional but validated when present", () => { + const receipt = { + state: "accepted", + runtime_id: "runtime-a", + worker_id: "worker-1", + diagnostics: [], + }; + assert( + parseRuntimeWorkerLifecycleResult(receipt).replayed === false, + "missing replay is not a replay", + ); + assert( + parseRuntimeWorkerLifecycleResult({ ...receipt, replayed: true }) + .replayed === true, + "replay flag lost", + ); + assertThrows( + () => parseRuntimeWorkerLifecycleResult({ ...receipt, replayed: "true" }), + "must be a boolean", + ); +});