feat: publish connected Subject body lifecycle candidate and validation fixes

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}"))

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());

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"

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 {

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<server_api::Diagnostic> {
+    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(

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

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"]

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<String>,
+    message: Option<String>,
+    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<Vec<runtime_api::WorkerLifecycleAck>>,
+    opened: Mutex<Vec<String>>,
+    closed: Mutex<Vec<String>>,
+    failures: Mutex<Vec<RuntimeHttpFailure>>,
+}
+
+impl StopTransport {
+    async fn forward(State(state): State<Arc<Self>>, request: Request<Body>) -> 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<String> {
+        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<StopTransport>,
+    _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(
+            &registry
+                .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:?}"
+    );
+}

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<String>);
+
+#[async_trait::async_trait]
+impl agen::llm_client::LlmClient for BodyModel {
+    fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
+        Box::new(self.clone())
+    }
+
+    async fn stream(
+        &self,
+        request: agen::llm_client::Request,
+    ) -> std::result::Result<agen::llm_client::client::ResponseStream, agen::llm_client::ClientError>
+    {
+        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<AtomicBool>,
+    inputs: tokio::sync::mpsc::Receiver<String>,
+}
+
+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, &current);
+        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", &current);
+    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::<u64>()
+            .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::<Vec<_>>();
+    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::<Vec<_>>();
+    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"
+    );
+}

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

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,

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",

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

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<Diagnostic>, current_worker?: | import("./worker-launch-api").WorkerLaunchWorkerSummary | null, };
 
 export type SubjektivSubjectListResponse = { limit: number, items: Array<SubjektivSubjectResponse>, next_cursor?: string | null, has_more: boolean, };
 

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

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<Diagnostic>; 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<Diagnostic>; runtime_id: string; state: WorkerOperationState; worker_id: string; };
+export type RuntimeWorkerLifecycleResult = { diagnostics?: Array<Diagnostic>; replayed?: boolean; runtime_id: string; state: WorkerOperationState; worker_id: string; };
 
 export type RuntimeWorkerListResponse = { diagnostics?: Array<Diagnostic>; items: Array<WorkerSummary>; 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<Diagnostic>; 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<string>; workdir_attachments?: Array<WorkerWorkdirAttachmentSummary>; worker_id: string; worker_state?: WorkerStateSnapshot | null; workspace: WorkerWorkspaceSummary; };
 
 export type WorkerWorkdirAttachmentSummary = { alias: string; effective_permissions: ExternalWorkdirPermissions; working_directory: WorkingDirectorySummary; };

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<Diagnostic> };
+
+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<Diagnostic>;
+  /**
+   * 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;
+};

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

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

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",
   );

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<void> {
+  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<T>() {
   let resolve!: (value: T) => void;
   const promise = new Promise<T>((accept) => {
@@ -80,6 +97,17 @@ function deferred<T>() {
   });
   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<Response>();
+  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<Response>();
+  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);
+});

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;

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);

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",
+  );
+});