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

fix: release failed Subject creation reservations with Runtime proof

Browse Source
This commit is contained in:
Hare
2026-10-31 20:14:09 +09:00
parent f7ea2362dd
commit d08a21e38c
21 changed files with 1085 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/server-api/src/lib.rs
+11
View File
@@ -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<SubjectBodyMutationResponse, SubjectBodyApiError>;
#[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<SubjectBodyMutationResponse, SubjectBodyApiError>;
#[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(
crates/server-api/src/subject_body.rs
+12
View File
@@ -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),
crates/workspace-server/src/server.rs
+62 -8
View File
@@ -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<server_api::SubjectBodyMutationResponse, server_api::SubjectBodyApiError>
{
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,
crates/workspace-server/src/server/subject_body.rs
+78
View File
@@ -92,6 +92,11 @@ pub(super) fn snapshot(api: &WorkspaceApi, subject: &str) -> ApiResult<SubjectBo
let lease = current(api, subject)?;
let observation = match lease.as_ref() {
None => SubjectBodyObservation::Absent,
Some(lease)
if subjektiv_journal::failed_subject_create_is_unpublished(api, subject, lease)? =>
{
SubjectBodyObservation::FailedCreate
}
Some(lease) => {
let observe = || -> ApiResult<server_api::WorkerSummary> {
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<SubjectBodyMutationResponse> {
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!(
crates/workspace-server/src/server/subjektiv_journal.rs
+41 -17
View File
@@ -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<bool> {
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:{}:{}",
crates/workspace-server/src/server/tests/subject_body_connected_tests.rs
+55
View File
@@ -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;
}
crates/workspace-server/src/server/tests/subject_body_tests.rs
+175 -1
View File
@@ -27,6 +27,7 @@ pub(super) fn assert_failed_request(body: &Value, subject: &str, request: &str,
struct BodyFixture {
api: WorkspaceApi,
backend: Arc<DeterministicExecutionBackend>,
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());
}
}
crates/workspace-server/src/server/tests/subject_spawn_tests.rs
+77 -45
View File
@@ -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);
}
}
}
crates/workspace-server/src/store.rs
+12
View File
@@ -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<bool> {
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,
openapi/server-api.json
+159 -1
View File
@@ -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",
tools/web-ux/browser-tests/subject_body_recovery_review.ts
+212
View File
@@ -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 <actual-build-path> <unique-run-id> <before|after>",
);
}
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),
);
}
}
web/workspace/src/lib/generated/companion-api.ts
+1 -1
View File
@@ -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
web/workspace/src/lib/generated/repository-api.ts
+1 -1
View File
@@ -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
web/workspace/src/lib/generated/runtime-api.ts
+1 -1
View File
@@ -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
web/workspace/src/lib/generated/subject-body-api.ts
+13 -4
View File
@@ -31,10 +31,11 @@ export type SubjectBodyRef = {
generation: string;
};
export type SubjectBodyObservation = { "kind": "absent" } | {
"kind": "available";
worker: WorkerSummary;
} | { "kind": "unavailable"; diagnostics: Array<Diagnostic> };
export type SubjectBodyObservation =
| { "kind": "absent" }
| { "kind": "failed_create" }
| { "kind": "available"; worker: WorkerSummary }
| { "kind": "unavailable"; diagnostics: Array<Diagnostic> };
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;
web/workspace/src/lib/generated/ticket-api.ts
+1 -1
View File
@@ -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
web/workspace/src/lib/generated/worker-launch-api.ts
+1 -1
View File
@@ -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
web/workspace/src/lib/workspace/memory/SubjectBody.browser.spec.ts
+84
View File
@@ -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<ReturnType<typeof absent>, "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();
});
web/workspace/src/lib/workspace/memory/SubjectBody.svelte
+11 -4
View File
@@ -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<void> {
async function submit(action?: 'start' | 'stop' | 'recover'): Promise<void> {
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}<p class="muted">{loading ? 'Updating body state…' : 'Last confirmed observation; refresh before acting.'}</p>{/if}
{#if snapshot.observation.kind === 'absent'}
<p>No current conversation body.</p>
{:else if snapshot.observation.kind === 'failed_create'}
<p><strong>Body creation failed; reservation retained.</strong> Confirm Runtime cleanup to release this reservation. This does not start a new body or change Subject permissions.</p>
{:else if snapshot.observation.kind === 'unavailable'}
<p><strong>Body owner present; observation unavailable.</strong> This is not an absent or stopped body. A replacement will not be started.</p>
{#each snapshot.observation.diagnostics as diagnostic}
@@ -240,6 +244,9 @@
{#if controls.start}
<button type="button" class="workspace-secondary-button" disabled={!fresh || loading || sending || !!storageError} onclick={() => { void submit('start'); }}>{snapshot.current ? 'Restart conversation body' : 'Start conversation body'}</button>
{/if}
{#if controls.recover}
<button type="button" class="workspace-secondary-button" disabled={!fresh || loading || sending || !!storageError} onclick={() => { void submit('recover'); }}>Release failed start reservation</button>
{/if}
{#if controls.stop}
<button type="button" class="workspace-secondary-button" disabled={!fresh || loading || sending || !!storageError} onclick={() => { void submit('stop'); }}>Stop conversation body</button>
{/if}
web/workspace/src/lib/workspace/memory/subject-body.test.ts
+28
View File
@@ -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);
});
web/workspace/src/lib/workspace/memory/subject-body.ts
+50 -13
View File
@@ -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<SubjectBodyMutationResponse> {
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<SubjectBodyMutationResponse> {
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,