diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index ebb59d239..3577ccf8e 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -257,6 +257,7 @@ impl_openapi_schema!( SubjectBodyRequestResolution, SubjectBodyResponse, SubjectBodyStartRequest, + SubjectBodyRecoverRequest, SubjectBodyStopRequest, SubjectBodyMutationResponse, SubjectBodyStopOutcomeResponse, @@ -1574,6 +1575,16 @@ pub trait ServerApi { #[body] request: SubjectBodyStartRequest, ) -> Result; + #[post("/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/body/recover", status = 200, error_status = 400, + additional_error_statuses = [401, 403, 404, 409, 500, 502], bearer_auth = true, browser_auth = true)] + async fn subjektiv_subject_body_recover( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[path] subject_id: String, + #[body] request: SubjectBodyRecoverRequest, + ) -> Result; + #[post("/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/body/stop", status = 200, error_status = 400, additional_error_statuses = [401, 403, 404, 409, 500], bearer_auth = true, browser_auth = true)] async fn subjektiv_subject_body_stop( diff --git a/crates/server-api/src/subject_body.rs b/crates/server-api/src/subject_body.rs index da20e2f9c..2284cceda 100644 --- a/crates/server-api/src/subject_body.rs +++ b/crates/server-api/src/subject_body.rs @@ -76,6 +76,8 @@ pub struct SubjectBodyRef { #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] pub enum SubjectBodyObservation { Absent, + /// Failed create reservation, not a published body. Runtime fencing is still required. + FailedCreate, Available { worker: WorkerSummary, }, @@ -125,6 +127,15 @@ pub struct SubjectBodyStopRequest { pub expected_observation_token: String, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(deny_unknown_fields)] +pub struct SubjectBodyRecoverRequest { + pub request_id: String, + /// Recovery releases only this failed reservation; it never starts a replacement. + pub expected_current: SubjectBodyRef, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(rename_all = "snake_case")] @@ -176,6 +187,7 @@ pub fn subject_body_api_typescript() -> String { SubjectBodyObservation::decl(&config), SubjectBodyResponse::decl(&config), SubjectBodyStartRequest::decl(&config), + SubjectBodyRecoverRequest::decl(&config), SubjectBodyStopRequest::decl(&config), SubjectBodyDisposition::decl(&config), SubjectBodyMutationResponse::decl(&config), diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index a1d99f0e5..710538fa1 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -6272,8 +6272,8 @@ impl WorkspaceApi { Ok(binding) => binding, Err(error) => { let mut diagnostics = cleanup_spawn_created_workdirs(self, &spawned_workdir_ids); - if let Err(cleanup) = self.config_store.fail_worker_create_reservation( - &self.config.workspace_id, + if let Err(cleanup) = terminalize_failed_worker_create( + self, runtime_id, worker_id, &reservation.create_fingerprint, @@ -6308,8 +6308,8 @@ impl WorkspaceApi { ) { let mut diagnostics = release_worker_workdir_attachment_reservations(self, &reserved_attachments); - if let Err(cleanup_error) = self.config_store.fail_worker_create_reservation( - &self.config.workspace_id, + if let Err(cleanup_error) = terminalize_failed_worker_create( + self, runtime_id, worker_id, &reservation_fingerprint, @@ -6394,8 +6394,8 @@ impl WorkspaceApi { // uncertain singleton generation forever. let mut diagnostics = release_worker_workdir_attachment_reservations(self, &reserved_attachments); - if let Err(cleanup_error) = self.config_store.fail_worker_create_reservation( - &self.config.workspace_id, + if let Err(cleanup_error) = terminalize_failed_worker_create( + self, runtime_id, worker_id, &reservation_fingerprint, @@ -7842,6 +7842,9 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou .merge(server_api::server_api_axum::subjektiv_subject_body_get( service.clone(), )) + .merge(server_api::server_api_axum::subjektiv_subject_body_recover( + service.clone(), + )) .merge(server_api::server_api_axum::subjektiv_subject_body_start( service.clone(), )) @@ -10276,6 +10279,27 @@ impl server_api::ServerApi for ServerApiContractService { }) } + async fn subjektiv_subject_body_recover( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + subject_id: String, + request: server_api::SubjectBodyRecoverRequest, + ) -> std::result::Result + { + subject_body::recover( + self.workspace_api()?.clone(), + context, + workspace_id.clone(), + subject_id.clone(), + request.clone(), + ) + .await + .map_err(|error| { + subject_body::api_error(error, workspace_id, subject_id, request.request_id) + }) + } + async fn subjektiv_subject_body_stop( &self, context: server_api::ServerRequestContext, @@ -35640,6 +35664,36 @@ fn release_worker_workdir_attachment_reservations( diagnostics } +// Complete the failed creation lifecycle, including Subject occupancy. A removed +// catalog entry alone is not a fence against delayed Runtime admission. +fn terminalize_failed_worker_create( + api: &WorkspaceApi, + runtime_id: &str, + worker_id: WorkerId, + create_fingerprint: &str, +) -> Result<()> { + api.config_store.fail_worker_create_reservation( + api.workspace_id(), + runtime_id, + worker_id, + create_fingerprint, + )?; + let worker = RuntimeWorkerRef::new(runtime_id, worker_id.to_string()); + if let Some(lease) = api + .config_store + .retained_worker_singleton_lease(api.workspace_id(), &worker)? + && let Some(subject) = lease.key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) + && api + .config_store + .current_worker_singleton_owner(api.workspace_id(), &lease.key)? + .as_ref() + == Some(&lease) + { + subjektiv_journal::recover_subject_create_reservation(api, subject, &lease)?; + } + Ok(()) +} + fn compensate_failed_workspace_worker_create( api: &WorkspaceApi, runtime_id: &str, @@ -35689,8 +35743,8 @@ fn compensate_failed_workspace_worker_create( attachment_reservations, )); if reserved_worker_absent { - if let Err(error) = api.config_store.fail_worker_create_reservation( - &api.config.workspace_id, + if let Err(error) = terminalize_failed_worker_create( + api, runtime_id, reservation_worker_id, create_fingerprint, diff --git a/crates/workspace-server/src/server/subject_body.rs b/crates/workspace-server/src/server/subject_body.rs index fb66fceee..1720c841e 100644 --- a/crates/workspace-server/src/server/subject_body.rs +++ b/crates/workspace-server/src/server/subject_body.rs @@ -92,6 +92,11 @@ pub(super) fn snapshot(api: &WorkspaceApi, subject: &str) -> ApiResult SubjectBodyObservation::Absent, + Some(lease) + if subjektiv_journal::failed_subject_create_is_unpublished(api, subject, lease)? => + { + SubjectBodyObservation::FailedCreate + } Some(lease) => { let observe = || -> ApiResult { let live = api @@ -351,6 +356,62 @@ pub(super) async fn stop( .await) } +pub(super) async fn recover( + api: WorkspaceApi, + context: server_api::ServerRequestContext, + workspace: String, + subject: String, + request: server_api::SubjectBodyRecoverRequest, +) -> ApiResult { + validate_request_id(&request.request_id)?; + account_context(&api, &context, &workspace).await?; + let recovery_api = api.clone(); + let recovery_subject = subject.clone(); + let expected = request.expected_current; + let result = api + .host_executors + .host()? + .run(move || -> ApiResult<()> { + let lease = require_expected(&recovery_api, &recovery_subject, Some(&expected))? + .ok_or(Error::SubjectBodySuperseded)?; + subjektiv_journal::recover_subject_create_reservation( + &recovery_api, + &recovery_subject, + &lease, + )?; + Ok(()) + }) + .await; + let (disposition, diagnostics) = match result { + Ok(()) => (SubjectBodyDisposition::Accepted, vec![]), + Err(error) + if matches!( + error.error, + Error::SubjectBodySuperseded + | Error::RepositoryConflict(_) + | Error::WorkspacePermissionDenied(_) + | Error::SubjektivSubjectNotFound(_) + ) => + { + return Err(error); + } + Err(error) => ( + SubjectBodyDisposition::Unknown, + failure_diagnostics("subject_body_recovery_unconfirmed", error), + ), + }; + Ok(finish( + api, + context, + workspace, + subject, + request.request_id, + disposition, + diagnostics, + ) + .await) +} + pub(super) async fn start( api: WorkspaceApi, context: server_api::ServerRequestContext, @@ -517,7 +578,24 @@ pub(super) async fn start( .await }) .await; + // A failed allocation is terminal even if its occupancy still awaits Runtime proof. + // Settle only this request; GET observations never settle a retained start intent. + let removed = if result.is_err() { + let store = api.config_store.clone(); + let workspace = workspace.clone(); + let operation = check_operation.clone(); + api.store + .db_executor() + .run(move || store.worker_create_allocation_removed(&workspace, &operation)) + .await? + } else { + false + }; let (disposition, diagnostics) = match result { + Err(error) if removed => ( + SubjectBodyDisposition::Rejected, + failure_diagnostics("subject_body_create_failed", error), + ), Ok(Json(result)) => (SubjectBodyDisposition::Accepted, result.diagnostics), Err(error) if matches!( diff --git a/crates/workspace-server/src/server/subjektiv_journal.rs b/crates/workspace-server/src/server/subjektiv_journal.rs index 3c987f33d..a8c90419d 100644 --- a/crates/workspace-server/src/server/subjektiv_journal.rs +++ b/crates/workspace-server/src/server/subjektiv_journal.rs @@ -545,9 +545,42 @@ pub(super) fn check_start_allowed(api: &WorkspaceApi, subject: &str) -> Result<( Ok(()) } +/// Read-only eligibility, not proof of Runtime absence or permission to replace a body. +pub(super) fn failed_subject_create_is_unpublished( + api: &WorkspaceApi, + subject: &str, + lease: &crate::store::WorkerSingletonLease, +) -> Result { + Ok(api + .config_store + .subject_create_recovery_candidate(api.workspace_id(), lease)? + .is_some() + && store(api)? + .journal_body_is_unpublished(subject, &lease.worker.runtime_id, &lease.worker.worker_id) + .map_err(domain)?) +} + /// Explicit creation can reconcile an earlier failed, never-published body. /// No missing catalog/status observation is treated as a Runtime fencing receipt. pub(super) fn recover_failed_subject_create(api: &WorkspaceApi, subject: &str) -> Result<()> { + let key = crate::subjektiv::subject_worker_singleton_key(subject) + .map_err(|e| Error::InvalidInput(e.to_string()))?; + if let Some(lease) = api + .config_store + .current_worker_singleton_owner(api.workspace_id(), &key)? + && failed_subject_create_is_unpublished(api, subject, &lease)? + { + recover_subject_create_reservation(api, subject, &lease)?; + } + Ok(()) +} + +/// Compensation and explicit recovery fence only the observed reservation, never a successor. +pub(super) fn recover_subject_create_reservation( + api: &WorkspaceApi, + subject: &str, + expected: &crate::store::WorkerSingletonLease, +) -> Result<()> { let continuation = lifecycle_lock(api, &format!("release-continuation:{subject}"))?; let _continuation = continuation .lock() @@ -560,25 +593,16 @@ pub(super) fn recover_failed_subject_create(api: &WorkspaceApi, subject: &str) - let _guard = gate .lock() .map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?; - let Some(lease) = api + let lease = api .config_store .current_worker_singleton_owner(api.workspace_id(), &subject_key)? - else { - return Ok(()); - }; - if api - .config_store - .subject_create_recovery_candidate(api.workspace_id(), &lease)? - .is_none() - { - return Ok(()); - } - if !store - .journal_body_is_unpublished(subject, &lease.worker.runtime_id, &lease.worker.worker_id) - .map_err(domain)? - { - // Published history must go through the ordinary continuity release. - return Ok(()); + .filter(|lease| lease == expected) + .ok_or(Error::SubjectBodySuperseded)?; + if !failed_subject_create_is_unpublished(api, subject, &lease)? { + return Err(Error::RepositoryConflict( + "Only a failed, unpublished Subject creation can be recovered; retain occupancy" + .into(), + )); } let operation_id = format!( "subject-create-recovery:{}:{}", diff --git a/crates/workspace-server/src/server/tests/subject_body_connected_tests.rs b/crates/workspace-server/src/server/tests/subject_body_connected_tests.rs index f71c471c3..c2c6c74e5 100644 --- a/crates/workspace-server/src/server/tests/subject_body_connected_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_body_connected_tests.rs @@ -848,3 +848,58 @@ async fn body_pending_lifecycle_conflicts_do_not_claim_stale_request_nonacceptan .await .unwrap(); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn denied_session_registration_releases_failed_body_and_allows_a_new_start() { + let f = ConnectedBody::new().await; + let subject = f.subject("failed-session-registration").await; + let store = open_subjektiv_store(&f.connected.api).unwrap(); + let initial = store.subject_permissions(&subject).unwrap().unwrap(); + let permissions = + format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"); + f.ok( + "POST", + &permissions, + json!({ + "operation_id":"deny-session-registration", "expected_change_id":initial.change_id, + "execution":[], "delegation":[], + }), + ) + .await; + let denied = store.subject_permissions(&subject).unwrap().unwrap(); + let request = absent_start("failed-body-start"); + let failed = f.mutate(&subject, "/start", request.clone()).await; + assert_eq!(failed["disposition"], "rejected", "{failed}"); + assert_eq!(failed["body"]["observation"]["kind"], "absent", "{failed}"); + assert_eq!(f.worker_count(), 0); + assert!( + store + .subject_session_attributions(&subject) + .unwrap() + .is_empty() + ); + assert_eq!( + store.subject_permissions(&subject).unwrap().unwrap(), + denied + ); + // A terminal allocation retry must settle, never allocate a second Worker. + let replay = f.mutate(&subject, "/start", request).await; + assert_eq!(replay["disposition"], "rejected", "{replay}"); + assert_eq!(f.worker_count(), 0); + f.ok( + "POST", + &permissions, + json!({ + "operation_id":"allow-session-registration", "expected_change_id":denied.change_id, + "execution":initial.execution, "delegation":initial.delegation, + }), + ) + .await; + let started = f + .mutate(&subject, "/start", absent_start("new-body-start")) + .await; + assert_eq!(started["disposition"], "accepted", "{started}"); + assert_eq!(f.read(&subject).await["current"]["generation"], "2"); + assert_eq!(f.worker_count(), 1); + f.stop(&subject, "stop-new-body").await; +} diff --git a/crates/workspace-server/src/server/tests/subject_body_tests.rs b/crates/workspace-server/src/server/tests/subject_body_tests.rs index 3f5fa2880..a1fa475ff 100644 --- a/crates/workspace-server/src/server/tests/subject_body_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_body_tests.rs @@ -27,6 +27,7 @@ pub(super) fn assert_failed_request(body: &Value, subject: &str, request: &str, struct BodyFixture { api: WorkspaceApi, + backend: Arc, owner: String, subject: String, _dir: tempfile::TempDir, @@ -35,7 +36,7 @@ struct BodyFixture { impl BodyFixture { async fn new() -> Self { let dir = tempfile::tempdir().unwrap(); - let api = test_api(dir.path()).await; + let (api, backend) = test_api_with_recording_backend(dir.path()).await; let owner = seed_test_workspace_owner_token(api.store.as_ref(), "body-owner").await; let subject = open_subjektiv_store(&api) .unwrap() @@ -44,6 +45,7 @@ impl BodyFixture { .id; Self { api, + backend, owner, subject, _dir: dir, @@ -64,6 +66,35 @@ impl BodyFixture { assert_eq!(status, StatusCode::OK, "{body}"); body } + fn failed_reservation(&self) -> crate::store::WorkerSingletonLease { + let key = crate::subjektiv::subject_worker_singleton_key(&self.subject).unwrap(); + let reservation = self + .api + .config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + EMBEDDED_RUNTIME_ID, + "failed-body-create", + &"f".repeat(64), + Some(&key), + &self + .api + .config_store + .get_workspace_memory_settings(TEST_WORKSPACE_ID) + .unwrap(), + ) + .unwrap(); + self.api + .config_store + .fail_worker_create_reservation( + TEST_WORKSPACE_ID, + EMBEDDED_RUNTIME_ID, + reservation.worker_id, + &reservation.create_fingerprint, + ) + .unwrap(); + reservation.singleton.unwrap() + } fn reserve(&self) -> crate::store::WorkerSingletonLease { let key = crate::subjektiv::subject_worker_singleton_key(&self.subject).unwrap(); self.api @@ -165,6 +196,10 @@ async fn browser_subject_and_conversation_reads_allow_missing_origin_but_mutatio format!("{}/start", f.path()), json!({"request_id":"must-not-start", "expected_current":null, "expected_observation_token":null}), ), + ( + format!("{}/recover", f.path()), + json!({"request_id":"must-not-recover", "expected_current":expected}), + ), ( format!("{}/stop", f.path()), json!({"request_id":"must-not-stop", "expected_current":expected, "expected_observation_token":"not-observed"}), @@ -233,6 +268,13 @@ async fn body_operations_reject_nonowners_and_cross_workspace_targets() { let f = BodyFixture::new().await; let reader = seed_test_api_token(f.api.store.as_ref(), "body-reader"); for (method, suffix, payload) in [ + ( + "POST", + "/recover", + Some( + json!({"request_id":"forbidden-recovery", "expected_current":{"runtime_id":"r", "worker_id":"w", "generation":"1"}}), + ), + ), ("GET", "", None), ( "POST", @@ -309,6 +351,10 @@ async fn body_mutations_reject_an_old_generation_without_releasing_its_owner() { let mut old = subject_body::reference(&lease); old.generation = (lease.generation + 1).to_string(); for (suffix, payload) in [ + ( + "/recover", + json!({"request_id":"old-recover", "expected_current":old}), + ), ( "/start", json!({"request_id":"old-start", "expected_current":old, "expected_observation_token":"old-token"}), @@ -364,6 +410,13 @@ async fn body_operations_reject_a_subject_with_deletion_already_admitted() { json!({"request_id":"start", "expected_current":null, "expected_observation_token":null}), ), ), + ( + "POST", + "/recover", + Some( + json!({"request_id":"recover-deleted", "expected_current":subject_body::reference(&lease)}), + ), + ), ("POST", "/stop", Some(stop.clone())), ("POST", "/stop/outcome", Some(stop)), ] { @@ -440,3 +493,124 @@ async fn body_action_payload_rejection_cannot_claim_request_resolution() { ); } } + +#[tokio::test] +async fn failed_body_reservation_can_be_recovered_without_starting_a_worker() { + let f = BodyFixture::new().await; + let lease = f.failed_reservation(); + let before = f.read().await; + assert_eq!(before["observation"]["kind"], "failed_create"); + assert!( + f.backend.recovery_requests.lock().unwrap().is_empty(), + "GET must not recover" + ); + let request = json!({"request_id":"recover-failed", "expected_current":before["current"]}); + *f.backend.recovery_failure.lock().unwrap() = Some("Runtime absence unconfirmed".into()); + let (status, unknown) = response( + &f.api, + human_request( + &f.owner, + "POST", + &format!("{}/recover", f.path()), + Some(request.clone()), + ), + ) + .await; + assert_eq!(status, StatusCode::OK, "{unknown}"); + assert_eq!(unknown["disposition"], "unknown"); + assert_eq!(f.read().await["current"], before["current"]); + let (status, recovered) = response( + &f.api, + human_request( + &f.owner, + "POST", + &format!("{}/recover", f.path()), + Some(request.clone()), + ), + ) + .await; + assert_eq!(status, StatusCode::OK, "{recovered}"); + assert_eq!(recovered["disposition"], "accepted"); + assert_eq!(recovered["body"]["observation"]["kind"], "absent"); + let requests = f.backend.recovery_requests.lock().unwrap().clone(); + assert_eq!(requests.len(), 2); + assert_eq!(requests[0], requests[1]); + assert!( + f.api + .runtime + .list_workers_for_runtime(EMBEDDED_RUNTIME_ID, 100) + .unwrap() + .items + .is_empty() + ); + let successor = f.reserve(); + assert_ne!(successor.worker, lease.worker); + let (status, stale) = response( + &f.api, + human_request( + &f.owner, + "POST", + &format!("{}/recover", f.path()), + Some(request), + ), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT, "{stale}"); + assert_failed_request(&stale, &f.subject, "recover-failed", "superseded"); + assert_eq!( + f.read().await["current"], + serde_json::to_value(subject_body::reference(&successor)).unwrap() + ); + assert_eq!(f.backend.recovery_requests.lock().unwrap().len(), 2); +} + +#[tokio::test] +async fn body_recovery_rejects_unfinished_or_published_creation() { + for publication in ["reserved", "writer", "session"] { + let f = BodyFixture::new().await; + let lease = if publication == "reserved" { + f.reserve() + } else { + f.failed_reservation() + }; + let store = open_subjektiv_store(&f.api).unwrap(); + if publication == "writer" { + store + .claim_journal_writer( + &f.subject, + &lease.worker.runtime_id, + &lease.worker.worker_id, + ) + .unwrap(); + } else if publication == "session" { + store + .record_session_attribution( + crate::subjektiv::SubjectSessionAttribution::new( + &f.subject, + &lease.worker.runtime_id, + &lease.worker.worker_id, + session_store::new_session_id().to_string(), + ) + .unwrap(), + ) + .unwrap(); + } + let before = f.read().await; + assert_eq!(before["observation"]["kind"], "unavailable"); + let request = + json!({"request_id":"must-not-recover", "expected_current":before["current"]}); + let (status, error) = response( + &f.api, + human_request( + &f.owner, + "POST", + &format!("{}/recover", f.path()), + Some(request), + ), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT, "{error}"); + assert_eq!(f.read().await["current"], before["current"]); + assert!(f.backend.recovery_requests.lock().unwrap().is_empty()); + } +} diff --git a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs index 4f0190e98..5d8537b53 100644 --- a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs @@ -404,7 +404,12 @@ async fn only_exact_authenticated_callbacks_can_reenter_deletion_drain() { #[tokio::test] async fn rejected_create_releases_key_but_retains_subject_occupancy_without_release_proof() { let workspace = tempfile::tempdir().unwrap(); - let api = test_api(workspace.path()).await; + let (api, backend) = test_api_with_recording_backend(workspace.path()).await; + let subject = open_subjektiv_store(&api) + .unwrap() + .create_subject(crate::subjektiv::SubjectRole::new("rejected-create").unwrap()) + .unwrap(); + let key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap(); let memory = api .config_store .get_workspace_memory_settings(TEST_WORKSPACE_ID) @@ -416,7 +421,7 @@ async fn rejected_create_releases_key_but_retains_subject_occupancy_without_rele EMBEDDED_RUNTIME_ID, "manual:rejected-create", &"f".repeat(64), - Some("subjektiv:rejected-create"), + Some(&key), &memory, ) .unwrap(); @@ -432,6 +437,7 @@ async fn rejected_create_releases_key_but_retains_subject_occupancy_without_rele spawned_workdir_ids: &[], }; for _ in 0..2 { + *backend.recovery_failure.lock().unwrap() = Some("Runtime proof unavailable".into()); let diagnostics = compensate_failed_workspace_worker_create( &api, EMBEDDED_RUNTIME_ID, @@ -442,7 +448,12 @@ async fn rejected_create_releases_key_but_retains_subject_occupancy_without_rele &context, &[], ); - assert!(diagnostics.is_empty(), "{diagnostics:?}"); + assert!( + diagnostics + .iter() + .any(|d| d.code == "worker_spawn_compensation_create_reservation_release_failed"), + "{diagnostics:?}" + ); assert!( !api.store .has_active_worker_create_reservation(TEST_WORKSPACE_ID, &worker) @@ -456,7 +467,7 @@ async fn rejected_create_releases_key_but_retains_subject_occupancy_without_rele ); assert_eq!( api.store - .current_worker_singleton_owner(TEST_WORKSPACE_ID, "subjektiv:rejected-create") + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) .unwrap() .expect("failed creation is not explicit Subject release authority") .worker, @@ -469,7 +480,7 @@ async fn rejected_create_releases_key_but_retains_subject_occupancy_without_rele EMBEDDED_RUNTIME_ID, "manual:retry-rejected-create", &"f".repeat(64), - Some("subjektiv:rejected-create"), + Some(&key), &memory, ); assert!( @@ -691,45 +702,66 @@ async fn unpublished_create_recovery_resumes_after_runtime_proof_before_host_com } #[tokio::test] -async fn failed_subject_spawn_can_recover_on_next_explicit_create() { - let root = tempfile::tempdir().unwrap(); - let (api, backend) = test_api_with_recording_backend(root.path()).await; - let store = open_subjektiv_store(&api).unwrap(); - let subject = store - .create_subject(crate::subjektiv::SubjectRole::new("failed-spawn").unwrap()) - .unwrap(); - let key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap(); - *backend.spawn_failure.lock().unwrap() = - Some("Session registration failed before publication".into()); - let rejected = api - .spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject.id)) - .unwrap(); - assert_eq!(rejected.state, InternalWorkerOperationState::Rejected); - let old = api - .config_store - .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) - .unwrap() - .unwrap(); - assert!( - api.config_store - .subject_create_recovery_candidate(TEST_WORKSPACE_ID, &old) +async fn failed_subject_spawn_releases_occupancy_only_after_runtime_proof() { + for uncertain in [false, true] { + let root = tempfile::tempdir().unwrap(); + let (api, backend) = test_api_with_recording_backend(root.path()).await; + let subject = open_subjektiv_store(&api) .unwrap() - .is_some() - ); - assert!( - backend.recovery_requests.lock().unwrap().is_empty(), - "failure alone must not release occupancy" - ); - let result = api - .spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject.id)) - .unwrap(); - assert_eq!(result.state, InternalWorkerOperationState::Accepted); - let current = api - .config_store - .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) - .unwrap() - .unwrap(); - assert_ne!(current.worker, old.worker); - assert_eq!(current.generation, old.generation + 1); - assert_eq!(backend.recovery_requests.lock().unwrap().len(), 1); + .create_subject(crate::subjektiv::SubjectRole::new("failed-spawn").unwrap()) + .unwrap(); + let key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap(); + *backend.spawn_failure.lock().unwrap() = + Some("Session registration failed before publication".into()); + if uncertain { + *backend.recovery_failure.lock().unwrap() = Some("Runtime absence unconfirmed".into()); + } + let rejected = api + .spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject.id)) + .unwrap(); + assert_eq!(rejected.state, InternalWorkerOperationState::Rejected); + let retained = api + .config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap(); + assert_eq!(retained.is_some(), uncertain); + assert_eq!(backend.recovery_requests.lock().unwrap().len(), 1); + if uncertain { + assert!( + rejected.diagnostics.iter().any( + |d| d.code == "worker_spawn_compensation_create_reservation_release_failed" + ) + ); + assert!( + api.config_store + .subject_create_recovery_candidate( + TEST_WORKSPACE_ID, + retained.as_ref().unwrap() + ) + .unwrap() + .is_some() + ); + } + let result = api + .spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject.id)) + .unwrap(); + assert_eq!(result.state, InternalWorkerOperationState::Accepted); + let current = api + .config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap() + .unwrap(); + let requests = backend.recovery_requests.lock().unwrap(); + assert_ne!(current.worker.worker_id, requests[0].worker_id.to_string()); + assert_eq!(current.generation, 2); + if uncertain { + assert_eq!(requests.len(), 2); + assert_eq!( + requests[0], requests[1], + "compensation and explicit recovery reuse one fence" + ); + } else { + assert_eq!(requests.len(), 1); + } + } } diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index e70d87403..8ef11a11f 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -3673,6 +3673,18 @@ impl SqliteWorkspaceStore { }) } + /// Terminal create evidence for a specific allocation, independent of retained Subject occupancy. + pub(crate) fn worker_create_allocation_removed( + &self, + workspace_id: &str, + allocation_key: &str, + ) -> Result { + self.with_conn(|conn| Ok(conn.query_row( + "SELECT EXISTS(SELECT 1 FROM worker_create_reservations WHERE workspace_id=?1 AND allocation_key=?2 AND state='removed')", + params![workspace_id, allocation_key], |row| row.get(0), + )?)) + } + pub(crate) fn reserved_worker_create_runtime( &self, workspace_id: &str, diff --git a/openapi/server-api.json b/openapi/server-api.json index 12673f7f0..63e3f6bbe 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -14914,6 +14914,20 @@ ], "type": "object" }, + { + "additionalProperties": false, + "description": "Failed create reservation, not a published body. Runtime fencing is still required.", + "properties": { + "kind": { + "const": "failed_create", + "type": "string" + } + }, + "required": [ + "kind" + ], + "type": "object" + }, { "additionalProperties": false, "properties": { @@ -14954,6 +14968,23 @@ } ] }, + "SubjectBodyRecoverRequest": { + "additionalProperties": false, + "properties": { + "expected_current": { + "$ref": "#/components/schemas/SubjectBodyRef", + "description": "Recovery releases only this failed reservation; it never starts a replacement." + }, + "request_id": { + "type": "string" + } + }, + "required": [ + "request_id", + "expected_current" + ], + "type": "object" + }, "SubjectBodyRef": { "additionalProperties": false, "properties": { @@ -23317,7 +23348,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:6a3d9c468ff204b44fc0b11d47e45f153b8131b9ec99c54ba2ebfdf2c0d14580" + "x-yoi-source-digest": "sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321" }, "openapi": "3.1.0", "paths": { @@ -42609,6 +42640,133 @@ ] } }, + "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/body/recover": { + "post": { + "operationId": "subjektiv_subject_body_recover", + "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/SubjectBodyRecoverRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyMutationResponse" + } + } + }, + "description": "Successful response" + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyApiError" + } + } + }, + "description": "Error response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyApiError" + } + } + }, + "description": "Error response" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyApiError" + } + } + }, + "description": "Error response" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyApiError" + } + } + }, + "description": "Error response" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyApiError" + } + } + }, + "description": "Error response" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyApiError" + } + } + }, + "description": "Error response" + }, + "502": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubjectBodyApiError" + } + } + }, + "description": "Error response" + } + }, + "security": [ + { + "bearerAuth": [] + }, + { + "browserSession": [] + } + ] + } + }, "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/body/start": { "post": { "operationId": "subjektiv_subject_body_start", diff --git a/tools/web-ux/browser-tests/subject_body_recovery_review.ts b/tools/web-ux/browser-tests/subject_body_recovery_review.ts new file mode 100644 index 000000000..3e4b11a5d --- /dev/null +++ b/tools/web-ux/browser-tests/subject_body_recovery_review.ts @@ -0,0 +1,212 @@ +#!/usr/bin/env -S deno run -A +/** 実Webと合成HTTPで失敗予約の回復導線を検証・撮影する。実Runtimeの証拠ではない。 */ +import { assert, assertEquals } from "@std/assert"; +import { dirname, fromFileUrl, join, resolve } from "@std/path"; +import { type Browser, chromium } from "playwright"; +import { sha256File } from "../src/artifacts.ts"; +import { startOwnedProcesses, stopOwnedProcesses } from "../src/processes.ts"; +import { assertNoPageWideOverflow, freePort } from "./memory_document_harness.ts"; + +const root = resolve(dirname(fromFileUrl(import.meta.url)), "../../.."); +const [buildArg, runId, mode] = Deno.args; +if ( + !buildArg || !runId || !/^[a-zA-Z0-9._-]+$/.test(runId) || !["before", "after"].includes(mode) +) { + throw new Error( + "usage: subject_body_recovery_review.ts ", + ); +} +const buildRoot = resolve(buildArg); +const output = join(root, "target/web-ux/body-recovery", runId); +await Deno.mkdir(dirname(output), { recursive: true }); +await Deno.mkdir(output); +const port = await freePort(); +const base = `http://127.0.0.1:${port}`; +const faults: string[] = []; +const captures: unknown[] = []; +const source = await new Deno.Command("git", { + args: ["rev-parse", "HEAD"], + cwd: root, + stdout: "piped", +}).output(); +const patch = await new Deno.Command("git", { args: ["diff", "HEAD"], cwd: root, stdout: "piped" }) + .output(); +if (!source.success || !patch.success) throw new Error("Cannot identify reviewed source"); +const sourceDigest = new Uint8Array(await crypto.subtle.digest("SHA-256", patch.stdout)); +const manifest = { + source: { + commit: new TextDecoder().decode(source.stdout).trim(), + diffSha256: Array.from(sourceDigest, (b) => b.toString(16).padStart(2, "0")).join(""), + runnerSha256: await sha256File(fromFileUrl(import.meta.url)), + }, + evidence: "actual-Web-with-synthetic-HTTP-NOT-real-Backend-or-Runtime", + mode, + buildRoot, + buildIndexSha256: await sha256File(join(buildRoot, "index.html")), + captures, + faults, + status: "running", + visualVerdict: "unreviewed", +}; +const owned = await startOwnedProcesses( + [{ + id: "body-recovery-fixture", + command: Deno.execPath(), + cwd: join(root, "tools/web-ux"), + args: [ + "run", + "--cached-only", + "-A", + join(root, "tools/web-ux/browser-tests/memory_document_fixture_server.ts"), + String(port), + buildRoot, + "3", + ], + readyUrl: `${base}/health`, + }], + join(output, "scenario.json"), + join(output, "logs"), + [], +); +let browser: Browser | undefined; +try { + browser = await chromium.launch({ headless: true }); + for (const width of [1440, 768, 390]) { + const context = await browser.newContext({ + viewport: { width, height: 1000 }, + colorScheme: "light", + reducedMotion: "reduce", + }); + const page = await context.newPage(); + page.setDefaultTimeout(15000); + page.on("pageerror", (e) => faults.push(e.message)); + page.on("console", (e) => { + if (e.type() === "error") faults.push(e.text()); + }); + page.on("response", (response) => { + if (response.status() >= 400) faults.push(`HTTP ${response.status()} ${response.url()}`); + }); + page.on("requestfailed", (request) => { + if (request.failure()?.errorText !== "net::ERR_ABORTED") { + faults.push(`${request.method()} ${request.url()}: ${request.failure()?.errorText}`); + } + }); + const current = { runtime_id: "runtime", worker_id: "failed-worker", generation: "2" }; + let recovered = false; + let authority = true; + const requests: unknown[] = []; + const body = () => ({ + workspace_id: "memory-review", + subject_id: "empty-subject", + current: recovered ? null : current, + observation: recovered ? { kind: "absent" } : mode === "before" + ? { + kind: "unavailable", + diagnostics: [{ + code: "subject_body_observation_unavailable", + severity: "warning", + message: "Unknown Worker", + }], + } + : { kind: "failed_create" }, + can_start: authority, + can_stop: authority, + }); + await context.route( + "**/api/w/memory-review/subjektiv/subjects/empty-subject/body**", + async (route) => { + const request = route.request(); + if (request.method() === "GET") return await route.fulfill({ json: body() }); + assert(request.url().endsWith("/recover"), "recovery must never dispatch start or stop"); + const input = request.postDataJSON(); + requests.push(input); + assertEquals(input.expected_current, current); + assert(!("expected_observation_token" in input)); + if (requests.length > 1) recovered = true; + await route.fulfill({ + json: { + request_id: input.request_id, + disposition: recovered ? "accepted" : "unknown", + diagnostics: [], + body: body(), + }, + }); + }, + ); + const conversation = { + id: "conversation", + subject_id: "empty-subject", + participant_account_id: "owner", + last_user_message_id: null, + }; + await context.route( + "**/api/w/memory-review/chat/conversations**", + (route) => + route.fulfill({ + json: route.request().url().includes("/messages") + ? { conversation, messages: [], has_more: false } + : conversation, + }), + ); + const capture = async (state: string) => { + await assertNoPageWideOverflow(page); + const image = join(output, `${width}-${state}.png`); + await page.screenshot({ path: image }); + captures.push({ + state, + viewport: page.viewportSize(), + route: new URL(page.url()).pathname, + image, + imageSha256: await sha256File(image), + accessibility: await page.locator("body").ariaSnapshot(), + visibleAlerts: await page.getByRole("alert").allTextContents(), + }); + }; + try { + const document = await page.goto(`${base}/w/memory-review/subjektiv/empty-subject/chat`); + assertEquals(document?.status(), 200); + await page.getByText("No public messages yet.").waitFor(); + if (mode === "before") { + await page.getByText("Body owner present; observation unavailable.", { exact: true }) + .waitFor(); + await capture("retained"); + continue; + } + const recover = page.getByRole("button", { name: "Release failed start reservation" }); + await recover.waitFor(); + await recover.focus(); + await capture("retained"); + await recover.click(); + const retry = page.getByRole("button", { name: "Retry same recover request" }); + await retry.waitFor(); + await capture("unknown"); + await page.reload(); + await page.getByRole("button", { name: "Retry same recover request" }).click(); + await page.getByRole("button", { name: "Start conversation body", exact: true }).waitFor(); + assertEquals(requests.length, 2); + assertEquals(requests[0], requests[1]); + await capture("released"); + authority = false; + recovered = false; + await page.getByRole("button", { name: "Refresh body state" }).click(); + await page.getByText("Body creation failed; reservation retained.", { exact: true }) + .waitFor(); + assertEquals(await recover.count(), 0); + await capture("no-authority"); + } finally { + await context.close(); + } + } + assertEquals(faults, []); + manifest.status = "browser-assertions-passed"; +} finally { + try { + await browser?.close(); + } finally { + await stopOwnedProcesses(owned); + await Deno.writeTextFile( + join(output, "review-context.json"), + JSON.stringify(manifest, null, 2), + ); + } +} diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index 972315416..382d570bf 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:6a3d9c468ff204b44fc0b11d47e45f153b8131b9ec99c54ba2ebfdf2c0d14580) +// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index 0847a3b4d..cfffdfe22 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:6a3d9c468ff204b44fc0b11d47e45f153b8131b9ec99c54ba2ebfdf2c0d14580) +// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index f36e4953d..ad88cbaf8 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:6a3d9c468ff204b44fc0b11d47e45f153b8131b9ec99c54ba2ebfdf2c0d14580) +// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) // 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 diff --git a/web/workspace/src/lib/generated/subject-body-api.ts b/web/workspace/src/lib/generated/subject-body-api.ts index 5654201fe..2dd55e191 100644 --- a/web/workspace/src/lib/generated/subject-body-api.ts +++ b/web/workspace/src/lib/generated/subject-body-api.ts @@ -31,10 +31,11 @@ export type SubjectBodyRef = { generation: string; }; -export type SubjectBodyObservation = { "kind": "absent" } | { - "kind": "available"; - worker: WorkerSummary; -} | { "kind": "unavailable"; diagnostics: Array }; +export type SubjectBodyObservation = + | { "kind": "absent" } + | { "kind": "failed_create" } + | { "kind": "available"; worker: WorkerSummary } + | { "kind": "unavailable"; diagnostics: Array }; export type SubjectBodyResponse = { workspace_id: string; @@ -57,6 +58,14 @@ export type SubjectBodyStartRequest = { expected_observation_token: string | null; }; +export type SubjectBodyRecoverRequest = { + request_id: string; + /** + * Recovery releases only this failed reservation; it never starts a replacement. + */ + expected_current: SubjectBodyRef; +}; + export type SubjectBodyStopRequest = { request_id: string; expected_current: SubjectBodyRef; diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index aec99f4fa..428cdc13a 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:6a3d9c468ff204b44fc0b11d47e45f153b8131b9ec99c54ba2ebfdf2c0d14580) +// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index d294ad745..722c0733b 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:6a3d9c468ff204b44fc0b11d47e45f153b8131b9ec99c54ba2ebfdf2c0d14580) +// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check diff --git a/web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts b/web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts index 42094f518..7da247bae 100644 --- a/web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts +++ b/web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts @@ -1563,3 +1563,87 @@ test("lost stop stays unknown until an identity-bound supersession from outcome JSON.parse(posts[0].body).request_id, ); }); + +test("failed creation offers fenced recovery, retaining exact input across reload and never starting a replacement", async () => { + let snapshot: Omit, "current"> & { current: typeof owner | null } = { + ...absent(), + current: owner, + observation: { kind: "failed_create" }, + }; + const posts: Array<{ path: string; body: unknown }> = []; + vi.stubGlobal( + "fetch", + vi.fn(async (url: string, init?: RequestInit) => { + if (init?.method === "POST") { + const request = JSON.parse(String(init.body)); + posts.push({ path: String(url), body: request }); + if (posts.length === 1) throw new Error("lost recovery response"); + snapshot = absent(); + return Response.json({ + request_id: request.request_id, + disposition: "accepted", + diagnostics: [], + body: snapshot, + }); + } + return Response.json(snapshot); + }), + ); + const view = render(SubjectBody, { props }); + await fireEvent.click( + await screen.findByRole("button", { + name: "Release failed start reservation", + }), + ); + await screen.findByRole("button", { name: "Retry same recover request" }); + expect(screen.queryByRole("button", { name: "Start conversation body" })) + .toBeNull(); + const saved = readSubjectBodyAttempt( + sessionStorage, + props.accountId, + props.workspaceId, + props.subjectId, + )!; + expect(saved.action).toBe("recover"); + expect(saved.request.expected_current).toEqual(owner); + expect(saved.request).not.toHaveProperty("expected_observation_token"); + view.unmount(); + render(SubjectBody, { props }); + const retry = await screen.findByRole("button", { + name: "Retry same recover request", + }); + await waitFor(() => expect(retry.hasAttribute("disabled")).toBe(false)); + await fireEvent.click(retry); + await screen.findByRole("button", { name: "Start conversation body" }); + expect(posts).toHaveLength(2); + expect(posts[1]).toEqual(posts[0]); + expect(posts[0].path).toMatch(/\/body\/recover$/); + expect( + readSubjectBodyAttempt( + sessionStorage, + props.accountId, + props.workspaceId, + props.subjectId, + ), + ).toBeNull(); +}); + +test("failed creation without authority exposes no recovery control", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async () => + Response.json({ + ...absent(), + current: owner, + observation: { kind: "failed_create" }, + can_start: false, + can_stop: false, + }) + ), + ); + render(SubjectBody, { props }); + await screen.findByText("Body creation failed; reservation retained."); + expect( + screen.queryByRole("button", { name: "Release failed start reservation" }), + ).toBeNull(); +}); diff --git a/web/workspace/src/lib/workspace/memory/SubjectBody.svelte b/web/workspace/src/lib/workspace/memory/SubjectBody.svelte index 61b6d355b..eac71dcea 100644 --- a/web/workspace/src/lib/workspace/memory/SubjectBody.svelte +++ b/web/workspace/src/lib/workspace/memory/SubjectBody.svelte @@ -5,7 +5,7 @@ import { clearSubjectBodyAttempt, createSubjectBodyAttempt, getSubjectBody, getSubjectBodyStopOutcome, readSubjectBodyAttempt, saveSubjectBodyAttempt, startSubjectBody, - stopSubjectBody, subjectBodyControls, SubjectBodyApiError, type SubjectBodyAttempt, + recoverSubjectBody, stopSubjectBody, subjectBodyControls, SubjectBodyApiError, type SubjectBodyAttempt, } from './subject-body.ts'; let { accountId, workspaceId, subjectId }: { accountId: string; workspaceId: string; subjectId: string } = $props(); @@ -29,8 +29,8 @@ let hasAccount = $derived(typeof accountId === 'string' && accountId.length > 0); let visible = $derived(hasAccount && activeRoute === identityKey(accountId, workspaceId, subjectId)); let snapshot = $derived(visible ? body : null); - let controls = $derived(snapshot ? subjectBodyControls(snapshot) : { start: false, stop: false }); - let retryAuthorized = $derived(snapshot && attempt && (attempt.action === 'start' ? snapshot.can_start : snapshot.can_stop)); + let controls = $derived(snapshot ? subjectBodyControls(snapshot) : { start: false, stop: false, recover: false }); + let retryAuthorized = $derived(snapshot && attempt && (attempt.action === 'stop' ? snapshot.can_stop : snapshot.can_start)); let worker = $derived(snapshot?.observation.kind === 'available' ? snapshot.observation.worker : null); let workerDiagnostics = $derived(worker?.diagnostics ?? []); @@ -108,7 +108,7 @@ + cause.message + '. Refresh the current body before choosing a new explicit action.'; } - async function submit(action?: 'start' | 'stop'): Promise { + async function submit(action?: 'start' | 'stop' | 'recover'): Promise { if (!visible || sending || checkingOutcome || loading || storageError || !fresh || !snapshot) return; if (attempt ? !retryAuthorized : (!action || !controls[action])) return; const generation = routeGeneration; @@ -135,6 +135,8 @@ try { const result = frozen.action === 'start' ? await startSubjectBody(fetch, workspace, subject, frozen.request) + : frozen.action === 'recover' + ? await recoverSubjectBody(fetch, workspace, subject, frozen.request) : await stopSubjectBody(fetch, workspace, subject, frozen.request); if (!current(generation, account, workspace, subject)) return; receipt = result; @@ -210,6 +212,8 @@ {#if !fresh}

{loading ? 'Updating body state…' : 'Last confirmed observation; refresh before acting.'}

{/if} {#if snapshot.observation.kind === 'absent'}

No current conversation body.

+ {:else if snapshot.observation.kind === 'failed_create'} +

Body creation failed; reservation retained. Confirm Runtime cleanup to release this reservation. This does not start a new body or change Subject permissions.

{:else if snapshot.observation.kind === 'unavailable'}

Body owner present; observation unavailable. This is not an absent or stopped body. A replacement will not be started.

{#each snapshot.observation.diagnostics as diagnostic} @@ -240,6 +244,9 @@ {#if controls.start} {/if} + {#if controls.recover} + + {/if} {#if controls.stop} {/if} diff --git a/web/workspace/src/lib/workspace/memory/subject-body.test.ts b/web/workspace/src/lib/workspace/memory/subject-body.test.ts index 0d31b2f88..dedbffc16 100644 --- a/web/workspace/src/lib/workspace/memory/subject-body.test.ts +++ b/web/workspace/src/lib/workspace/memory/subject-body.test.ts @@ -240,6 +240,7 @@ Deno.test("durable retry input is frozen complete account-and-route-scoped and n Object.isFrozen(restored) && Object.isFrozen(restored.request) && Object.isFrozen(restored.request.expected_current), ); + assert(restored.action === "start"); assert(restored.request.expected_observation_token === "observation-1"); assert( readSubjectBodyAttempt(saved, "account", "workspace", "other") === null, @@ -838,3 +839,30 @@ Deno.test("GET cannot supply a mutation resolution even when its error claims th assert(failure instanceof SubjectBodyApiError); assert(failure.request === null); }); + +Deno.test("failed create has a recovery-only control and a frozen token-free reservation intent", () => { + const body = parse({ + ...absent(), + current: owner, + observation: { kind: "failed_create" }, + }); + const controls = subjectBodyControls(body); + assert(controls.recover && !controls.start && !controls.stop); + const attempt = createSubjectBodyAttempt( + "account", + body, + "recover", + "recover-1", + ); + const saved = storage(); + saveSubjectBodyAttempt(saved, "account", attempt); + assert( + JSON.stringify( + readSubjectBodyAttempt(saved, "account", "workspace", "subject"), + ) === JSON.stringify(attempt), + ); + assert(!("expected_observation_token" in attempt.request)); + rejects(() => createSubjectBodyAttempt("account", body, "start", "no-replacement")); + rejects(() => parse({ ...absent(), observation: { kind: "failed_create" } })); + assert(!subjectBodyControls({ ...body, can_start: false }).recover); +}); diff --git a/web/workspace/src/lib/workspace/memory/subject-body.ts b/web/workspace/src/lib/workspace/memory/subject-body.ts index 5d531982f..26f45d96f 100644 --- a/web/workspace/src/lib/workspace/memory/subject-body.ts +++ b/web/workspace/src/lib/workspace/memory/subject-body.ts @@ -2,6 +2,7 @@ import type { SubjectBodyFailedRequest, SubjectBodyMutationResponse, SubjectBodyObservation, + SubjectBodyRecoverRequest, SubjectBodyRef, SubjectBodyResponse, SubjectBodyStartRequest, @@ -122,6 +123,13 @@ export function parseSubjectBodyResponse( if (current !== null) throw new Error("Absent body cannot have an owner"); observation = { kind: "absent" }; break; + case "failed_create": + record(observed, ["kind"]); + if (current === null) { + throw new Error("Failed creation must retain its reservation"); + } + observation = { kind: "failed_create" }; + break; case "unavailable": record(observed, ["kind", "diagnostics"]); if (current === null) { @@ -328,8 +336,11 @@ async function mutateSubjectBody( fetchFn: typeof fetch, workspaceId: string, subjectId: string, - action: "start" | "stop", - request: SubjectBodyStartRequest | SubjectBodyStopRequest, + action: "start" | "stop" | "recover", + request: + | SubjectBodyStartRequest + | SubjectBodyStopRequest + | SubjectBodyRecoverRequest, ): Promise { const requestId = parseRequestId(request.request_id); const response = await fetchFn( @@ -357,6 +368,15 @@ export function startSubjectBody( return mutateSubjectBody(fetchFn, workspaceId, subjectId, "start", request); } +export function recoverSubjectBody( + fetchFn: typeof fetch, + workspaceId: string, + subjectId: string, + request: SubjectBodyRecoverRequest, +): Promise { + return mutateSubjectBody(fetchFn, workspaceId, subjectId, "recover", request); +} + export function stopSubjectBody( fetchFn: typeof fetch, workspaceId: string, @@ -394,17 +414,21 @@ export async function getSubjectBodyStopOutcome( export function subjectBodyControls( body: SubjectBodyResponse, -): { start: boolean; stop: boolean } { +): { start: boolean; stop: boolean; recover: boolean } { + if (body.observation.kind === "failed_create") { + return { start: false, stop: false, recover: body.can_start }; + } if (body.observation.kind === "absent") { - return { start: body.can_start, stop: false }; + return { start: body.can_start, stop: false, recover: false }; } if (body.observation.kind !== "available") { - return { start: false, stop: false }; + return { start: false, stop: false, recover: false }; } const worker = body.observation.worker; const observed = worker.availability === "observed" && Boolean(worker.restore_observation_token); return { + recover: false, start: body.can_start && observed && worker.state === "stopped" && worker.worker_state == null, stop: body.can_stop && observed && worker.worker_state != null && @@ -422,6 +446,7 @@ export type SubjectBodyAttempt = & ( | { action: "start"; request: SubjectBodyStartRequest } | { action: "stop"; request: SubjectBodyStopRequest } + | { action: "recover"; request: SubjectBodyRecoverRequest } ); export function subjectBodyStorageKey( @@ -466,6 +491,16 @@ function parseAttempt( const expected_current = input.expected_current === null ? null : Object.freeze(bodyRef(input.expected_current)); + if (item.action === "recover" && expected_current !== null) { + record(input, ["request_id", "expected_current"]); + return Object.freeze({ + accountId, + workspaceId, + subjectId, + action: "recover", + request: Object.freeze({ request_id, expected_current }), + }); + } const expected_observation_token = input.expected_observation_token === null ? null : token(input.expected_observation_token); @@ -509,7 +544,7 @@ function parseAttempt( export function createSubjectBodyAttempt( accountId: string, body: SubjectBodyResponse, - action: "start" | "stop", + action: "start" | "stop" | "recover", requestId: string, ): SubjectBodyAttempt { if (!subjectBodyControls(body)[action]) { @@ -521,13 +556,15 @@ export function createSubjectBodyAttempt( workspaceId: body.workspace_id, subjectId: body.subject_id, action, - request: { - request_id: requestId, - expected_current: body.current, - expected_observation_token: body.observation.kind === "available" - ? body.observation.worker.restore_observation_token - : null, - }, + request: action === "recover" + ? { request_id: requestId, expected_current: body.current } + : { + request_id: requestId, + expected_current: body.current, + expected_observation_token: body.observation.kind === "available" + ? body.observation.worker.restore_observation_token + : null, + }, }, accountId, body.workspace_id,