diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 6adafbc90..6ae183d6a 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -803,6 +803,20 @@ pub struct SessionMaterialDescriptor { pub total_bytes: u64, } +/// Explicit Host erasure, distinct from ordinary retention policy or pin release. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SessionMaterialEraseRequest { + pub workspace_id: String, + pub source: SessionPublicSource, + pub expected_session_id: String, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct SessionMaterialEraseResponse { + pub session_id: String, +} + #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct SessionMaterialPinRequest { @@ -860,6 +874,11 @@ pub struct SessionMaterialReadPage { #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(tag = "availability", rename_all = "snake_case")] pub enum SessionMaterialCatalogAvailability { + /// The saved active pointer is pending and storage proves no Session material + /// has ever been published. Missing/corrupt materialized sources are not empty. + NotMaterialized { + session_id: String, + }, Page { page: SessionMaterialCatalogPage, }, @@ -1123,6 +1142,16 @@ pub trait RuntimeApi { #[query] request: WorkerSessionHistoryRequest, ) -> Result; + #[post( + "/v1/session-public/material/retention/erase", + status = 200, + error_status = 400 + )] + async fn session_material_erase( + &self, + #[body] request: SessionMaterialEraseRequest, + ) -> Result; + #[post( "/v1/session-public/material/retention/pin", status = 200, @@ -1588,6 +1617,13 @@ mod tests { }) } + async fn session_material_erase( + &self, + _request: SessionMaterialEraseRequest, + ) -> Result { + Err(test_error(501)) + } + async fn session_material_pin( &self, _request: SessionMaterialPinRequest, diff --git a/crates/session-store/src/fs_store.rs b/crates/session-store/src/fs_store.rs index 29897eb4e..cbde85674 100644 --- a/crates/session-store/src/fs_store.rs +++ b/crates/session-store/src/fs_store.rs @@ -329,6 +329,14 @@ impl FsStore { } impl Store for FsStore { + fn retain_session_before_publication( + &self, + session_id: SessionId, + pin: crate::retention::SessionRetentionPin, + ) -> Result<(), StoreError> { + self.retention().retain_before_publication(session_id, pin) + } + fn append( &self, session_id: SessionId, diff --git a/crates/session-store/src/retention.rs b/crates/session-store/src/retention.rs index d155d602e..4acecbc96 100644 --- a/crates/session-store/src/retention.rs +++ b/crates/session-store/src/retention.rs @@ -109,6 +109,23 @@ impl SessionRetentionStore { Ok(guard) } + /// Protect a Host-owned Session before its first source publication. This is + /// retention intent, not proof that any original material exists. The owner must + /// establish it before accepting input and use the same control store for writes + /// and deletion. Source-serving APIs must still verify actual material separately. + pub fn retain_before_publication( + &self, + session: SessionId, + pin: SessionRetentionPin, + ) -> Result<(), StoreError> { + let _guard = self.lock_live(session)?; + let mut state = self.read(session)?; + if state.pins.insert(pin) { + self.save(session, &state)?; + } + Ok(()) + } + /// Call `source_exists` under the same lock used by deletion. Replaying the /// same pin is idempotent; missing source never becomes a successful lease. pub fn pin( @@ -184,6 +201,43 @@ pub(crate) fn fs_retention(root: &Path) -> SessionRetentionStore { mod tests { use super::*; + #[test] + fn publication_intent_survives_reopen_without_attesting_missing_originals() { + let temp = tempfile::tempdir().unwrap(); + let source = temp.path().join("source"); + let controls = temp.path().join("controls"); + let session = crate::new_session_id(); + let pin = SessionRetentionPin(uuid::Uuid::now_v7()); + let retention = SessionRetentionStore::new(&controls); + retention.retain_before_publication(session, pin).unwrap(); + let reopened = SessionRetentionStore::new(&controls); + assert_eq!(reopened.pins(session).unwrap(), [pin]); + // An early owner is not a successful source-availability receipt. + assert!( + matches!(reopened.pin(session, pin, || Ok(source.is_file())), Err(StoreError::SessionSourceMissing(id)) if id == session) + ); + { + let _publication = reopened.lock_live(session).unwrap(); + fs::write(&source, "committed before notification").unwrap(); + } + assert!( + matches!(reopened.delete(session, SessionDeletionIntent::Ordinary, || { fs::remove_file(&source)?; Ok(()) }), Err(StoreError::SessionPinned(id)) if id == session) + ); + assert_eq!( + fs::read_to_string(&source).unwrap(), + "committed before notification" + ); + reopened + .delete(session, SessionDeletionIntent::ErasePermanently, || { + fs::remove_file(&source)?; + Ok(()) + }) + .unwrap(); + assert!( + matches!(retention.retain_before_publication(session, pin), Err(StoreError::SessionErased(id)) if id == session) + ); + } + #[test] fn pin_survives_reopen_blocks_delete_and_releases_idempotently() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/session-store/src/store.rs b/crates/session-store/src/store.rs index 750c0730c..a876b7429 100644 --- a/crates/session-store/src/store.rs +++ b/crates/session-store/src/store.rs @@ -42,6 +42,9 @@ pub enum StoreError { #[error("session was permanently erased: {0}")] SessionErased(SessionId), + #[error("session source retention is unsupported by this store")] + SessionRetentionUnsupported, + #[error("retained observation read limit exceeded")] ReadLimitExceeded, @@ -89,6 +92,17 @@ pub enum StoreError { /// take `(SessionId, SegmentId)` so segments can be physically grouped /// per Session on disk (or per session_id in a DB). pub trait Store: Send + Sync { + /// Establish durable retention intent before the Host accepts public input. + /// This does not attest source availability; it must use the same control store + /// as publishers and physical deletion. Unsupported stores fail closed. + fn retain_session_before_publication( + &self, + _session_id: SessionId, + _pin: crate::retention::SessionRetentionPin, + ) -> Result<(), StoreError> { + Err(StoreError::SessionRetentionUnsupported) + } + /// Append a single log entry to the segment log. /// /// One committed line per successful call. Implementations must not expose diff --git a/crates/session-store/src/worker_metadata.rs b/crates/session-store/src/worker_metadata.rs index 40a9f3895..e2a8ffe72 100644 --- a/crates/session-store/src/worker_metadata.rs +++ b/crates/session-store/src/worker_metadata.rs @@ -806,6 +806,15 @@ where S: crate::Store, W: Send + Sync, { + fn retain_session_before_publication( + &self, + session_id: SessionId, + pin: crate::retention::SessionRetentionPin, + ) -> Result<(), crate::StoreError> { + self.session_store + .retain_session_before_publication(session_id, pin) + } + fn append( &self, session_id: SessionId, diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index be85b1fca..62c6024cd 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -716,6 +716,17 @@ impl WorkerSessionStore { } impl Store for WorkerSessionStore { + fn retain_session_before_publication( + &self, + session_id: SessionId, + pin: crate::retention::SessionRetentionPin, + ) -> Result<(), StoreError> { + self.retention + .as_ref() + .ok_or(StoreError::SessionRetentionUnsupported)? + .retain_before_publication(session_id, pin) + } + fn append( &self, session_id: SessionId, diff --git a/crates/standalone/src/subjektiv/journal/tests.rs b/crates/standalone/src/subjektiv/journal/tests.rs index b8194db42..5cdf832d3 100644 --- a/crates/standalone/src/subjektiv/journal/tests.rs +++ b/crates/standalone/src/subjektiv/journal/tests.rs @@ -659,6 +659,253 @@ impl agen::llm_client::LlmClient for JournalScript { } } +#[derive(Clone, Default)] +struct DialogAndJournal { + journal: JournalScript, + dialog_calls: Arc, +} +#[async_trait::async_trait] +impl agen::llm_client::LlmClient for DialogAndJournal { + async fn stream( + &self, + request: agen::llm_client::Request, + ) -> Result< + std::pin::Pin< + Box< + dyn futures::Stream< + Item = Result< + agen::llm_client::event::Event, + agen::llm_client::error::ClientError, + >, + > + Send, + >, + >, + agen::llm_client::error::ClientError, + > { + if request + .tools + .iter() + .any(|tool| tool.name == "ReadJournalState") + { + return agen::llm_client::LlmClient::stream(&self.journal, request).await; + } + use agen::llm_client::{ + event::{Event, StopReason}, + types::Item, + }; + let surface = request + .tools + .iter() + .any(|tool| tool.name == "SubmitMemorySurface"); + if surface + || request + .tools + .iter() + .any(|tool| tool.name == "SubmitJobResult") + { + let events = if request.items.iter().any(|item| matches!(item, Item::ToolResult { call_id, .. } if call_id == "empty-memory")) { + vec![Event::text_block_start(0), Event::text_delta(0, "empty Memory acknowledged"), Event::text_block_stop(0, Some(StopReason::EndTurn))] + } else { + let (tool, value) = if surface { + ("SubmitMemorySurface", serde_json::json!({"points":[]})) + } else { + let input: serde_json::Value = request.items.iter().filter_map(Item::as_text).find_map(|text| text.lines().find_map(|line| line.strip_prefix("input_json: "))).map(|json| serde_json::from_str(json).unwrap()).expect("Memory Job input"); + assert_eq!(input["candidate_ids"], serde_json::json!([]), "fixture has no staged Memory; do not synthesize decisions"); + ("SubmitJobResult", serde_json::json!({"result":input})) + }; + vec![Event::tool_use_start(0, "empty-memory", tool), Event::tool_input_delta(0, value.to_string()), Event::tool_use_stop(0)] + }; + return Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok)))); + } + self.dialog_calls.fetch_add(1, Ordering::SeqCst); + Ok(Box::pin(futures::stream::iter( + [ + Event::text_block_start(0), + Event::text_delta(0, "public scripted response"), + Event::text_block_stop(0, Some(StopReason::EndTurn)), + ] + .into_iter() + .map(Ok), + ))) + } + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +} + +async fn submit_dialog(host: &crate::StandaloneHost, input: &str) { + let mut client = host.connect(); + client + .send(&protocol::Method::submit_text( + protocol::new_submission_request_id(), + input, + )) + .await + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(20), async { + let mut answered = false; + loop { + match client.next_event().await.unwrap().unwrap() { + protocol::Event::TextDelta { text } + if text.contains("public scripted response") => + { + answered = true + } + protocol::Event::RunEnd { .. } => { + assert!(answered); + return; + } + protocol::Event::Error { code, message } => panic!("{code:?}: {message}"), + _ => {} + } + } + }) + .await + .expect("bounded main dialog completion"); +} + +/// Public standalone launch/protocol -> real controller/Worker Feature -> LocalSubjectHost -> +/// original FsStore -> generic Job runner/tools -> Journal read API. Only the two model clients +/// are scripted; no manual admission, generation reservation, source append or domain mutation. +#[tokio::test] +async fn public_host_dialog_auto_records_while_generation_waits_and_drains_without_another_input() { + let temp = tempfile::tempdir().unwrap(); + let state = temp.path().join("state"); + let catalog = StandaloneSubjects::open(&state).unwrap(); + let subject = catalog.create("public-dialog-owner", None).unwrap(); + let model = DialogAndJournal { + journal: JournalScript { + gate_first: true, + ..Default::default() + }, + ..Default::default() + }; + let launch = crate::StandaloneLaunchConfig::new( + temp.path(), + &state, + manifest::ProfileSelector::source_named( + manifest::ProfileRegistrySource::Builtin, + "standalone-subjektiv", + ), + "public-journal-dialog", + ) + .with_subject(subject.id.clone()) + .resolve() + .unwrap(); + let host = crate::StandaloneHost::start_with_model_client(launch, model.clone()) + .await + .unwrap(); + let session = host.record().active_session_id; + assert_eq!( + crate::store::session_retention(&state) + .pins(session) + .unwrap(), + [session_store::retention::SessionRetentionPin(session)] + ); + submit_dialog(&host, "first small public turn").await; + tokio::time::timeout( + std::time::Duration::from_secs(20), + model.journal.entered.notified(), + ) + .await + .unwrap(); + let reader = JournalHostContext::validated_reader(&catalog.store, &subject.id).unwrap(); + let journal = catalog + .store + .list_journals(&reader, None, 1) + .unwrap() + .items + .pop() + .unwrap(); + let attr = catalog + .store + .session_attribution(&session.to_string()) + .unwrap() + .unwrap(); + let context = JournalHostContext::validated_session(&catalog.store, attr).unwrap(); + let first = catalog + .store + .active_journal_generation(&context, &journal.id) + .unwrap() + .unwrap(); + submit_dialog(&host, "second small public turn").await; + submit_dialog(&host, "last input; no further turn follows").await; + assert_eq!(model.dialog_calls.load(Ordering::SeqCst), 3); + assert_eq!( + catalog + .store + .active_journal_generation(&context, &journal.id) + .unwrap() + .unwrap(), + first, + "running generation/input remain frozen while dialog proceeds" + ); + model.journal.release.notify_one(); + let jobs = host.jobs(); + tokio::time::timeout( + std::time::Duration::from_secs(20), + jobs.wait(&first.job.job_id), + ) + .await + .unwrap() + .unwrap(); + tokio::time::timeout( + std::time::Duration::from_secs(20), + model.journal.entered.notified(), + ) + .await + .unwrap(); + let second = catalog + .store + .active_journal_generation(&context, &journal.id) + .unwrap() + .unwrap(); + assert_ne!(first.id, second.id); + model.journal.release.notify_one(); + tokio::time::timeout( + std::time::Duration::from_secs(20), + jobs.wait(&second.job.job_id), + ) + .await + .unwrap() + .unwrap(); + assert!(!catalog.store.journal_retention_required(&context).unwrap()); + let saved = catalog.store.read_journal(&reader, &journal.id).unwrap(); + assert!(saved.current_position_bytes > 0); + let position = catalog + .store + .read_journal_text( + &reader, + &journal.id, + None, + JournalTextField::CurrentPosition, + None, + 0, + MAX_JOURNAL_READ_BYTES, + ) + .unwrap(); + assert!(!position.text.is_empty()); + let (_, open) = catalog + .store + .journal_resume_state(&context, &journal.id) + .unwrap(); + let open = open.unwrap(); + let body = catalog + .store + .read_journal_text( + &reader, + &journal.id, + Some(&open.id), + JournalTextField::Body, + None, + 0, + MAX_JOURNAL_READ_BYTES, + ) + .unwrap(); + assert!(body.text.contains("last input")); + host.shutdown().await.unwrap(); +} + #[tokio::test] async fn real_job_tools_save_and_drain_last_input_without_another_turn() { let fixture = Fixture::new("Proposed A, corrected to B; partly performed; interrupted"); diff --git a/crates/subjektiv/src/journal/mod.rs b/crates/subjektiv/src/journal/mod.rs index 6f5e3ecb0..c8d25c7cf 100644 --- a/crates/subjektiv/src/journal/mod.rs +++ b/crates/subjektiv/src/journal/mod.rs @@ -62,6 +62,25 @@ pub const MAX_JOURNAL_READ_BYTES: usize = 16 * 1024; /// Whole pretty-serialized public page budget, including escaped text and metadata. pub const MAX_JOURNAL_RESPONSE_BYTES: usize = 64 * 1024; +/// Stable owner of Backend source retention, shared by attribution-time protection +/// and later source admission/release. This identifier confers no read authority. +/// Local standalone storage uses its scope-exclusive Session UUID as the owner. +pub fn source_retention_pin(workspace: &str, subject: &str, session: &str) -> Uuid { + use sha2::{Digest, Sha256}; + let digest = Sha256::digest( + serde_json::to_vec(&( + "subjektiv-journal-retention-v1", + workspace, + subject, + session, + )) + .expect("string tuple serialization"), + ); + let mut bytes = [0; 16]; + bytes.copy_from_slice(&digest[..16]); + Uuid::from_bytes(bytes) +} + #[derive(Debug, thiserror::Error)] pub enum JournalError { #[error("journal permission denied: {0}")] diff --git a/crates/worker-runtime/src/http_server/runtime_management_api.rs b/crates/worker-runtime/src/http_server/runtime_management_api.rs index 210288535..6f17337d8 100644 --- a/crates/worker-runtime/src/http_server/runtime_management_api.rs +++ b/crates/worker-runtime/src/http_server/runtime_management_api.rs @@ -369,6 +369,43 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { .map_err(api_error) } + async fn session_material_erase( + &self, + request: runtime_api::SessionMaterialEraseRequest, + ) -> Result { + let scope = material_scope(&self.state, &request.workspace_id)?; + #[cfg(feature = "fs-store")] + { + self.state + .runtime + .delete_session_public_source_scoped( + &scope, + &request.source, + &request.expected_session_id, + session_store::retention::SessionDeletionIntent::ErasePermanently, + ) + .map_err(|_| { + runtime_api::RuntimeApiError::new( + 409, + "session_erasure_incomplete", + "Session erasure could not complete; retain erasure intent and retry", + ) + })?; + Ok(runtime_api::SessionMaterialEraseResponse { + session_id: request.expected_session_id, + }) + } + #[cfg(not(feature = "fs-store"))] + { + let _ = scope; + Err(runtime_api::RuntimeApiError::new( + 503, + "session_retention_unavailable", + "Session erasure requires filesystem storage", + )) + } + } + async fn session_material_pin( &self, request: runtime_api::SessionMaterialPinRequest, diff --git a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs index e70c80d00..a5d06c6ba 100644 --- a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs @@ -194,6 +194,101 @@ impl Fixture { } } +#[test] +fn pending_session_proof_is_not_missing_material_and_pin_release_allows_empty_archive() { + let f = Fixture::new(); + std::fs::remove_dir_all(f.root()).unwrap(); + let aggregate = f.root().parent().unwrap().to_path_buf(); + let metadata = + WorkerAggregateStore::new(&aggregate, format!("worker-runtime-{}", f.worker.worker_id)) + .unwrap(); + metadata + .write( + &session_store::WorkerMetadata::new( + format!("worker-runtime-{}", f.worker.worker_id), + Some(session_store::WorkerActiveSegmentRef::pending_segment( + f.session, + )), + ) + .with_workspace_id(&f.scope.workspace_id), + ) + .unwrap(); + let pin = SessionRetentionPin(uuid::Uuid::now_v7()); + let retention = runtime_session_retention(f.temp.path()); + retention.retain_before_publication(f.session, pin).unwrap(); + assert!( + matches!(f.runtime.session_material_catalog_scoped(&f.scope, &f.request()), Catalog::NotMaterialized { session_id } if session_id == f.session.to_string()) + ); + assert!( + f.runtime + .pin_session_public_source_scoped(&f.scope, &f.source(), &f.session.to_string(), pin) + .is_err(), + "retention intent is not source availability" + ); + // A missing committed source is never the same fact as a pending pointer. + metadata + .write( + &session_store::WorkerMetadata::new( + format!("worker-runtime-{}", f.worker.worker_id), + Some(session_store::WorkerActiveSegmentRef::active_segment( + f.session, f.segment, + )), + ) + .with_workspace_id(&f.scope.workspace_id), + ) + .unwrap(); + assert!(matches!( + f.runtime + .session_material_catalog_scoped(&f.scope, &f.request()), + Catalog::Unavailable { .. } + )); + metadata + .write( + &session_store::WorkerMetadata::new( + format!("worker-runtime-{}", f.worker.worker_id), + Some(session_store::WorkerActiveSegmentRef::pending_segment( + f.session, + )), + ) + .with_workspace_id(&f.scope.workspace_id), + ) + .unwrap(); + // Any unexpected source file invalidates the empty proof, too. + std::fs::create_dir_all(f.root()).unwrap(); + std::fs::write(f.root().join("unattested-original"), "do not discard").unwrap(); + assert!(matches!( + f.runtime + .session_material_catalog_scoped(&f.scope, &f.request()), + Catalog::Unavailable { .. } + )); + std::fs::remove_file(f.root().join("unattested-original")).unwrap(); + f.runtime + .release_session_public_source_pin_scoped( + &f.scope, + &f.source(), + &f.session.to_string(), + pin, + ) + .unwrap(); + assert!(retention.pins(f.session).unwrap().is_empty()); + f.stop(); + let inventory = f + .runtime + .worker_retention_inventory(&f.scope.workspace_id, &f.worker) + .unwrap(); + assert_eq!(inventory.session_id, None); + assert_eq!(inventory.session_bytes, 0); + let removed = f + .runtime + .execute_worker_retention(&f.retention_request(SessionDisposition::Archive)) + .unwrap(); + let archive = removed.archive.unwrap(); + assert!(removed.source_removed); + assert_eq!(archive.source_session_id, f.session.to_string()); + assert_eq!(archive.content_file_count, 0); + assert!(archive.source_active_segment_id.is_none()); +} + #[test] fn canonical_pages_read_frozen_originals_after_tail_append_and_reject_forged_descriptors() { let f = Fixture::new(); @@ -524,14 +619,10 @@ fn explicit_erase_invalidates_pins_and_fences_old_readers_publishers_and_retry() SessionDeletionIntent::ErasePermanently, ) .unwrap(); - // Removing the remaining Worker registration/metadata still requires the - // caller's explicit erase intent; ordinary deletion may not infer it. + // Ordinary registration removal can resume a pre-existing permanent erase + // decision. It cannot establish one or recreate original Session material. f.runtime - .delete_worker_with_intent_scoped( - &f.scope, - &f.worker, - SessionDeletionIntent::ErasePermanently, - ) + .delete_worker_with_intent_scoped(&f.scope, &f.worker, SessionDeletionIntent::Ordinary) .unwrap(); assert!(!f.root().parent().unwrap().exists()); assert!(matches!( @@ -808,13 +899,47 @@ async fn signed_generated_client_reads_canonical_pages_and_pins_with_real_router .len(), 1 ); - client.session_material_release(pin).await.unwrap(); + client.session_material_release(pin.clone()).await.unwrap(); assert!( runtime_session_retention(f.temp.path()) .pins(f.session) .unwrap() .is_empty() ); + client.session_material_pin(pin).await.unwrap(); + f.stop(); + let erase = runtime_api::SessionMaterialEraseRequest { + workspace_id: f.scope.workspace_id.clone(), + source: f.source(), + expected_session_id: f.session.to_string(), + }; + for _ in 0..2 { + assert_eq!( + client + .session_material_erase(erase.clone()) + .await + .unwrap() + .session_id, + f.session.to_string() + ); + } + assert!(!f.root().exists()); + assert!(matches!( + client.session_material_catalog(f.request()).await.unwrap(), + Catalog::Unavailable { .. } + )); + assert!(matches!( + runtime_session_retention(f.temp.path()).ensure_live(f.session), + Err(session_store::StoreError::SessionErased(_)) + )); + // Register/lifecycle cleanup after erasure must not need a fresh archive or Job. + let result = f + .runtime + .execute_worker_retention(&f.retention_request(SessionDisposition::Purge)) + .unwrap(); + assert!(result.source_removed); + assert!(result.archive.is_none()); + assert!(!f.root().parent().unwrap().exists()); }) .await .expect("bounded signed client integration"); @@ -866,3 +991,39 @@ async fn canonical_http_rejects_replay_tampered_body_and_workspace_mismatch() { .unwrap(); assert_eq!(unsigned.status(), StatusCode::UNAUTHORIZED); } + +#[tokio::test] +async fn explicit_material_erasure_requires_delete_capability_and_matching_scope() { + let f = Fixture::new(); + f.stop(); + let identity = RuntimeIdentityMaterial::generate("erase-key").unwrap(); + let trust = workspace_trust(&identity); + let app = runtime_http_router_with_workspace_auth( + f.runtime.clone(), + None, + workspace_test_auth(vec![trust.clone()]), + ); + let path = "/v1/session-public/material/retention/erase"; + let mut request = runtime_api::SessionMaterialEraseRequest { + workspace_id: f.scope.workspace_id.clone(), + source: f.source(), + expected_session_id: f.session.to_string(), + }; + let body = serde_json::to_vec(&request).unwrap(); + let mut claims = signed_claims(&trust, "POST", path, &body); + claims.operation = "workers:read".into(); + let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert!(f.root().exists()); + request.workspace_id = "other-workspace".into(); + let body = serde_json::to_vec(&request).unwrap(); + let claims = signed_claims(&trust, "POST", path, &body); + let response = signed_request(app, &identity, &claims, body, "workspace-a").await; + assert_eq!(response.status(), StatusCode::FORBIDDEN); + assert!( + runtime_session_retention(f.temp.path()) + .ensure_live(f.session) + .is_ok() + ); + assert!(f.root().exists()); +} diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index 88e5e6fbc..a9f5c1cce 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -178,6 +178,40 @@ pub(crate) trait WorkerRetentionProvider: Send + Sync { pub use crate::session_retention::runtime_session_retention; +/// Prove the legitimate pre-publication state, rather than interpreting a missing +/// source as empty. A committed active pointer or any source file defeats this proof. +pub(crate) fn unmaterialized_session( + worker_dir: &Path, + workspace_id: &str, +) -> Result, RuntimeError> { + let metadata_path = worker_dir.join("metadata.json"); + let bytes = match fs::read(&metadata_path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(source) => { + return Err(RuntimeError::StoreIo { + operation: "inspect pending Session", + path: metadata_path, + source, + }); + } + }; + let metadata: session_store::WorkerMetadata = serde_json::from_slice(&bytes) + .map_err(|_| RuntimeError::InvalidRequest("Pending Session metadata is invalid".into()))?; + if metadata.workspace_id.as_deref() != Some(workspace_id) { + return Ok(None); + } + let Some(active) = metadata.active.filter(|active| active.segment_id.is_none()) else { + return Ok(None); + }; + let root = worker_dir.join("session"); + if collect_files(&root, "inspect pending Session")?.is_empty() { + Ok(Some(active.session_id)) + } else { + Ok(None) + } +} + pub(crate) fn aggregate_session_id( worker_dir: &Path, ) -> Result, RuntimeError> { @@ -303,6 +337,19 @@ pub(crate) fn delete_worker_aggregate( intent: session_store::retention::SessionDeletionIntent, ) -> Result<(), RuntimeError> { let session = aggregate_session_id(worker_dir)?; + let retention = runtime_session_retention(runtime_root); + // Ordinary policy never creates erase authority. A pre-existing permanent + // tombstone is an irrevocable decision whose physical cleanup may be resumed. + let intent = match session + .map(|session| retention.ensure_live(session)) + .transpose() + { + Err(session_store::StoreError::SessionErased(_)) => { + session_store::retention::SessionDeletionIntent::ErasePermanently + } + Err(error) => return Err(retention_error(error)), + Ok(_) => intent, + }; let remove = || -> Result<(), session_store::StoreError> { if intent == session_store::retention::SessionDeletionIntent::ErasePermanently { if let Some(session) = session { @@ -561,33 +608,36 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { return Err(RuntimeError::WorkerNotFound { worker_id }); } let session_dir = worker_dir.join("session"); - let (session_id, segment_ids, session_bytes) = if session_dir.is_dir() { - let manifest: CanonicalSessionManifest = read_json( - &session_dir.join("session.json"), - "inventory Worker retention", - )?; - let files = collect_files(&session_dir, "inventory Worker retention")?; - let mut segment_ids = BTreeSet::new(); - let mut bytes = 0_u64; - for (relative, path) in files { - bytes = bytes.saturating_add(file_len(&path, "inventory Worker retention")?); - if let Some(name) = relative.file_name().and_then(|value| value.to_str()) { - let segment = name - .strip_suffix(".trace.jsonl") - .or_else(|| name.strip_suffix(".jsonl")); - if let Some(segment) = segment { - segment_ids.insert(segment.to_string()); + let (session_id, segment_ids, session_bytes) = + if unmaterialized_session(&worker_dir, workspace_id)?.is_some() { + (None, Vec::new(), 0) + } else if session_dir.is_dir() { + let manifest: CanonicalSessionManifest = read_json( + &session_dir.join("session.json"), + "inventory Worker retention", + )?; + let files = collect_files(&session_dir, "inventory Worker retention")?; + let mut segment_ids = BTreeSet::new(); + let mut bytes = 0_u64; + for (relative, path) in files { + bytes = bytes.saturating_add(file_len(&path, "inventory Worker retention")?); + if let Some(name) = relative.file_name().and_then(|value| value.to_str()) { + let segment = name + .strip_suffix(".trace.jsonl") + .or_else(|| name.strip_suffix(".jsonl")); + if let Some(segment) = segment { + segment_ids.insert(segment.to_string()); + } } } - } - ( - Some(manifest.session_id), - segment_ids.into_iter().collect(), - bytes, - ) - } else { - (None, Vec::new(), 0) - }; + ( + Some(manifest.session_id), + segment_ids.into_iter().collect(), + bytes, + ) + } else { + (None, Vec::new(), 0) + }; let diagnostics_bytes = diagnostics_files(&worker_dir, "inventory Worker retention")? .into_iter() .try_fold(0_u64, |total, path| { @@ -909,14 +959,14 @@ fn commit_session_archive( RuntimeError::InvalidRequest("archive disposition requires archive_id".to_string()) })?; let session_dir = worker_dir.join("session"); - if !session_dir.is_dir() { - return Err(RuntimeError::StoreMissing { - operation: "archive Worker Session", - path: session_dir, - }); - } - let session: CanonicalSessionManifest = - read_json(&session_dir.join("session.json"), "archive Worker Session")?; + let pending = unmaterialized_session(worker_dir, &request.workspace_id)?; + let session: CanonicalSessionManifest = if let Some(session) = pending { + CanonicalSessionManifest { + session_id: session.to_string(), + } + } else { + read_json(&session_dir.join("session.json"), "archive Worker Session")? + }; let source_files = collect_files(&session_dir, "archive Worker Session")?; let (checksum, bytes, count) = checksum_files(&source_files, "archive Worker Session")?; let mut segment_ids = BTreeSet::new(); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 383cb0f70..25bd311b8 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -4842,15 +4842,20 @@ impl Runtime { expected_session_id: &str, pin: session_store::retention::SessionRetentionPin, ) -> Result<(), RuntimeError> { - let (root, _) = self.resolve_session_public_source(scope, source)?; + let (root, archive) = + self.resolve_session_public_source_with_validation(scope, source, false)?; let session = expected_session_id .parse::() .map_err(|_| RuntimeError::InvalidRequest("Invalid Session identity".into()))?; - if session_store::WorkerSessionStore::open_read_only(&root) - .and_then(|store| store.session_id()) - .map_err(crate::retention::retention_error)? - != Some(session) - { + let owned_session = match archive { + Some(manifest) => manifest.source_session_id.parse().ok(), + None => { + crate::retention::aggregate_session_id(root.parent().ok_or_else(|| { + RuntimeError::InvalidRequest("Session root unavailable".into()) + })?)? + } + }; + if owned_session != Some(session) { return Err(RuntimeError::InvalidRequest( "Session source identity mismatch".into(), )); @@ -4995,6 +5000,33 @@ impl Runtime { )) } + /// Prove an attributed but not yet materialized Session without treating a + /// missing committed log as empty. The source/Workspace binding is resolved first. + #[cfg(feature = "fs-store")] + pub(crate) fn session_not_materialized_scoped( + &self, + scope: &RuntimeWorkspaceScope, + source: &runtime_api::SessionPublicSource, + expected_session_id: &str, + ) -> Result { + let (root, archive) = self.resolve_session_public_source(scope, source)?; + if archive.is_some() { + return Ok(false); + } + let session = expected_session_id + .parse::() + .map_err(|_| RuntimeError::InvalidRequest("Invalid Session identity".into()))?; + let retention = self.session_retention_store()?; + let _guard = retention + .lock_live(session) + .map_err(crate::retention::retention_error)?; + Ok(crate::retention::unmaterialized_session( + root.parent() + .ok_or_else(|| RuntimeError::InvalidRequest("Session root unavailable".into()))?, + &scope.workspace_id, + )? == Some(session)) + } + /// Resolve the Backend-selected source without exposing local storage paths. /// The saved active pointer, never a caller-provided Segment, selects lineage. #[cfg(feature = "fs-store")] diff --git a/crates/worker-runtime/src/session_material.rs b/crates/worker-runtime/src/session_material.rs index 42b30462d..70b361a63 100644 --- a/crates/worker-runtime/src/session_material.rs +++ b/crates/worker-runtime/src/session_material.rs @@ -57,6 +57,18 @@ pub(crate) fn catalog( scope: &RuntimeWorkspaceScope, request: &SessionMaterialCatalogRequest, ) -> SessionMaterialCatalogAvailability { + if scope.workspace_id == request.workspace_id + && (1..=100).contains(&request.limit) + && request.offset == 0 + && request.expected_catalog_sha256.is_none() + && runtime + .session_not_materialized_scoped(scope, &request.source, &request.expected_session_id) + .is_ok_and(|empty| empty) + { + return SessionMaterialCatalogAvailability::NotMaterialized { + session_id: request.expected_session_id.clone(), + }; + } let result = (|| -> Result<_, SessionMaterialSourceError> { if scope.workspace_id != request.workspace_id { return Err("Session Workspace scope mismatch".into()); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 361106482..f276efad3 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -7812,7 +7812,10 @@ where }); } match state { - SubjektivSessionAttributionState::Confirmed { .. } => return Ok(()), + SubjektivSessionAttributionState::Confirmed { subject_id, .. } => { + return self + .retain_journal_source_before_publication(session_id, subject_id); + } SubjektivSessionAttributionState::OutcomeUnknown { .. } | SubjektivSessionAttributionState::Failed { .. } => true, SubjektivSessionAttributionState::LegacyUnknown { .. } => false, @@ -7967,8 +7970,44 @@ where } self.persist_subjektiv_session_attribution(SubjektivSessionAttributionState::Confirmed { session_id, - subject_id: output.subject_id, - }) + subject_id: output.subject_id.clone(), + })?; + self.retain_journal_source_before_publication(session_id, &output.subject_id) + } + + fn retain_journal_source_before_publication( + &self, + session_id: SessionId, + subject_id: &str, + ) -> Result<(), WorkerError> { + let Some(connection) = self.subjektiv_host_connection().map_err(|error| { + WorkerError::SubjektivSessionAttribution { + state: "retention_unconfirmed", + message: error.to_string(), + } + })? + else { + return Ok(()); + }; + if !connection.host.journal_recording_granted() { + return Ok(()); + } + let pin = session_store::retention::SessionRetentionPin( + match self.workspace_client().workspace_id() { + Some(workspace) => subjektiv::journal::source_retention_pin( + workspace, + subject_id, + &session_id.to_string(), + ), + None => session_id, + }, + ); + // Confirmation can be replayed after response loss. Persist this intent before + // exposing the Worker to input, not after the first RunCommitted notification. + // It does not claim that an original Session has materialized yet. + self.store + .retain_session_before_publication(session_id, pin)?; + Ok(()) } fn persist_subjektiv_session_attribution( @@ -11116,6 +11155,82 @@ mod build_summary_prompt_tests { assert!(worker.backend_job().is_none()); } + #[tokio::test] + async fn journal_attribution_protects_first_publication_before_any_admission_notification() { + for enabled in [false, true] { + let client = Arc::new(AttributionLifecycleClient::new([])); + let (mut worker, store, _temp) = attribution_lifecycle_worker(client.clone()).await; + if enabled { + worker.manifest.feature.subjektiv.profile.enabled = true; + worker + .manifest + .feature + .subjektiv + .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: "workspace-test".into(), + language: "English".into(), + }) + .unwrap(); + } + let session = worker.session_id(); + assert!(store.session_store.list_sessions().unwrap().is_empty()); + worker + .finalize_subjektiv_session_attribution( + true, + SubjektivSessionAttributionLifecycle::NewSession, + ) + .unwrap(); + let pin = session_store::retention::SessionRetentionPin( + subjektiv::journal::source_retention_pin( + "workspace-test", + "subject-test", + &session.to_string(), + ), + ); + let expected = if enabled { vec![pin] } else { vec![] }; + assert_eq!( + store.session_store.retention().pins(session).unwrap(), + expected + ); + // Reopen/replayed confirmation does not depend on a notification or add an owner. + worker + .finalize_subjektiv_session_attribution( + true, + SubjektivSessionAttributionLifecycle::RestoredSession, + ) + .unwrap(); + assert_eq!(client.requests.lock().unwrap().len(), 1); + assert_eq!( + store.session_store.retention().pins(session).unwrap(), + expected + ); + worker.materialize_durable_session_head().await.unwrap(); + let segment = worker.segment_id(); + store + .append( + session, + segment, + &LogEntry::TurnEnd { + ts: 1, + turn_count: 1, + }, + ) + .unwrap(); + // Simulate stop after the public commit but before RunCommitted/Host admission. + if enabled { + assert!( + matches!(store.session_store.delete_session(session), Err(StoreError::SessionPinned(id)) if id == session) + ); + assert!(store.session_store.exists(session, segment).unwrap()); + store + .session_store + .release_session_pin(session, pin) + .unwrap(); + } + assert!(store.session_store.delete_session(session).unwrap()); + } + } + #[tokio::test] async fn subject_maintenance_job_binding_is_immutable_and_has_no_subject_body_attribution() { for (subjektiv_consolidation, subjektiv_journal) in [(true, false), (false, true)] { diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 7ec01714f..4bae8aa3f 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1301,6 +1301,13 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { Err("Session source retention release is unavailable on this Runtime".into()) } + fn session_material_erase( + &self, + _request: runtime_api::SessionMaterialEraseRequest, + ) -> Result { + Err("Session source permanent erasure is unavailable on this Runtime".into()) + } + fn session_material_catalog( &self, _request: runtime_api::SessionMaterialCatalogRequest, @@ -2266,6 +2273,21 @@ impl RuntimeRegistry { }) } + pub fn session_material_erase( + &self, + runtime_id: &str, + request: runtime_api::SessionMaterialEraseRequest, + ) -> Result { + validate_backend_identifier("runtime_id", runtime_id)?; + self.runtime(runtime_id)? + .session_material_erase(request) + .map_err(|message| RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: runtime_id.into(), + code: "session_material_erase_failed".into(), + message, + }) + } + pub fn session_material_catalog( &self, runtime_id: &str, @@ -3324,6 +3346,27 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { }) } + fn session_material_erase( + &self, + request: runtime_api::SessionMaterialEraseRequest, + ) -> Result { + if request.workspace_id != self.workspace_id { + return Err("Session erase Workspace mismatch".into()); + } + let scope = RuntimeWorkspaceScope::new(&request.workspace_id, "embedded-backend"); + self.runtime + .delete_session_public_source_scoped( + &scope, + &request.source, + &request.expected_session_id, + session_store::retention::SessionDeletionIntent::ErasePermanently, + ) + .map_err(|e| e.to_string())?; + Ok(runtime_api::SessionMaterialEraseResponse { + session_id: request.expected_session_id, + }) + } + fn session_material_catalog( &self, request: runtime_api::SessionMaterialCatalogRequest, @@ -5503,6 +5546,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { .map_err(|diagnostic| diagnostic.message) } + fn session_material_erase( + &self, + request: runtime_api::SessionMaterialEraseRequest, + ) -> Result { + self.run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { client.session_material_erase(request).await }, + ) + .map_err(|diagnostic| diagnostic.message) + } + fn session_material_catalog( &self, request: runtime_api::SessionMaterialCatalogRequest, @@ -8262,12 +8317,58 @@ mod tests { ), "workers:read" ); + assert_eq!( + workspace_runtime_operation("POST", "/v1/session-public/material/retention/erase"), + "workers:delete" + ); assert_eq!( workspace_runtime_operation("POST", SSH_HOST_KEY_PROBE_PATH), SSH_HOST_KEY_PROBE_OPERATION ); } + #[test] + fn remote_session_erasure_uses_the_generated_management_transport() { + // The loopback fixture substitutes only the remote Runtime. This covers + // Registry/remote-adapter DTO and route wiring, not Runtime authorization or deletion. + let session_id = uuid::Uuid::new_v4().to_string(); + let (base_url, server) = serve_mock_http(vec![mock_response( + "POST", + "/v1/session-public/material/retention/erase", + true, + 200, + json!({"session_id": session_id}).to_string(), + )]); + let registry = RuntimeRegistry::new(Vec::new()); + registry.register( + RemoteWorkerRuntime::new( + RemoteRuntimeConfig::new( + "remote:erase", + "Erase transport fixture", + base_url, + Some("secret-token".into()), + ), + "workspace-test".into(), + "http://127.0.0.1:8787".into(), + ) + .unwrap(), + ); + let response = registry + .session_material_erase( + "remote:erase", + runtime_api::SessionMaterialEraseRequest { + workspace_id: "workspace-test".into(), + source: runtime_api::SessionPublicSource::Retained { + worker_id: EmbeddedWorkerId::from_legacy_u64(1), + }, + expected_session_id: session_id.clone(), + }, + ) + .unwrap(); + assert_eq!(response.session_id, session_id); + server.join().unwrap(); + } + #[test] fn remote_runtime_registry_routes_commands_without_browser_secret_leaks() { let worker_id = EmbeddedWorkerId::from_legacy_u64(1).to_string(); diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index ffb90b692..655b75ca0 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -63,6 +63,51 @@ pub struct WorkerRemovalPlanRequest { pub workspace_id: String, pub worker: RuntimeWorkerRef, pub reason: String, + /// Internal evidence cannot be supplied through serialized deletion requests. + #[serde(skip)] + pub(crate) journal_erasure: Option, +} + +const ERASED_JOURNAL_POLICY_ID: &str = "backend-journal-erased-session-v1"; + +/// Monotonic domain erasure proof, bound to the exact source Worker and Session. +/// The only constructor rereads the Journal tombstone; ordinary Purge grants no erase authority. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ErasedJournalSession { + workspace_id: String, + worker: RuntimeWorkerRef, + session_id: String, +} +impl ErasedJournalSession { + pub(crate) fn validated( + workspace_id: &str, + store: &crate::subjektiv::SubjektivStore, + attr: &crate::subjektiv::SubjectSessionAttribution, + ) -> crate::Result> { + let host = subjektiv::journal::JournalHostContext::validated_session(store, attr.clone()) + .map_err(|e| StoreError::Store(e.to_string()))?; + Ok(store + .journal_session_erased(&host) + .map_err(|e| StoreError::Store(e.to_string()))? + .then(|| Self { + workspace_id: workspace_id.into(), + worker: RuntimeWorkerRef::new(&attr.runtime_id, &attr.worker_id), + session_id: attr.session_id.clone(), + })) + } + + fn matches( + &self, + req: &WorkerRemovalPlanRequest, + inventory: &WorkerRetentionInventory, + ) -> bool { + self.workspace_id == req.workspace_id + && self.worker == req.worker + && inventory + .session_id + .as_ref() + .is_none_or(|id| *id == self.session_id) + } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -223,10 +268,20 @@ impl SqliteWorkspaceStore { inv: &WorkerRetentionInventory, ) -> Result { validate_plan(req, inv)?; + if req + .journal_erasure + .as_ref() + .is_some_and(|proof| !proof.matches(req, inv)) + { + return Err(WorkerRetentionError::Invalid( + "Journal erasure proof does not match removal source".into(), + )); + } + let erased_journal = req.journal_erasure.is_some(); let now = Utc::now().to_rfc3339(); self.with_conn_mut(|conn| { let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; - let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?; + let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker,erased_journal)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?; let worker=match load_worker(&tx,&req.workspace_id,&req.worker)? { Some(v)=>v, None=>{ @@ -272,7 +327,7 @@ impl SqliteWorkspaceStore { return Err(StoreError::InvalidInput(format!("stale:{plan_id}:plan state {} is not executable", state_s(plan.state)))); } if !plan.blockers.is_empty(){return Err(StoreError::InvalidInput(format!("blocked:{}",serde_json::to_string(&plan.blockers).unwrap())));} - let policy=load_removal_policy(&tx,workspace_id,&plan.worker)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?; + let policy=load_removal_policy(&tx,workspace_id,&plan.worker,plan.policy_id==ERASED_JOURNAL_POLICY_ID)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?; if policy.policy_id!=plan.policy_id||policy.policy_digest!=plan.policy_digest{ mark_stale(&tx,&plan,"policy content changed")?;tx.commit()?; return Err(stale_error(&plan,"policy content changed")); @@ -785,20 +840,28 @@ struct WorkerRow { fn load_worker(c: &Connection, w: &str, r: &RuntimeWorkerRef) -> crate::Result> { c.query_row("SELECT display_name,profile,retention_state,created_at,updated_at FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![w,r.runtime_id,r.worker_id],|x|Ok(WorkerRow{display_name:x.get(0)?,profile:x.get(1)?,retention_state:x.get(2)?,created_at:x.get(3)?,updated_at:x.get(4)?})).optional().map_err(StoreError::from) } -/// Recording Sessions may contain original text in tool results. Only a terminal, -/// Host-bound Journal attempt grants this exception; profile names grant nothing. +/// Recording Sessions may contain original text in tool results. A terminal, +/// Host-bound Journal attempt or a validated source-erasure proof grants this +/// exception; profile names and serialized deletion input grant nothing. +/// A saved erasure plan retains the monotonic decision across restart, but creates +/// no Runtime erase authority: the domain/physical tombstones must already exist. /// This plan-local policy never changes the Workspace conversation policy. fn load_removal_policy( c: &Connection, w: &str, worker: &RuntimeWorkerRef, + erased_journal: bool, ) -> crate::Result> { let Some(mut policy) = load_policy(c, w)? else { return Ok(None); }; - if crate::store::terminal_journal_job_worker(c, w, worker)? { + if erased_journal || crate::store::terminal_journal_job_worker(c, w, worker)? { let update = WorkerRetentionPolicyUpdate { - policy_id: "backend-journal-ephemeral-session-v1".into(), + policy_id: if erased_journal { + ERASED_JOURNAL_POLICY_ID.into() + } else { + "backend-journal-ephemeral-session-v1".into() + }, session_disposition: SessionDisposition::Purge, metadata_disposition: MetadataDisposition::Tombstone, archive_retention: ArchiveRetention::Forever, @@ -1194,6 +1257,7 @@ mod tests { worker_id: worker_id().to_string(), }, reason: "cleanup".into(), + journal_erasure: None, } } #[test] diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index e3d8b1a9f..10ad743c6 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2190,13 +2190,17 @@ impl WorkerRemovalService { )); } - if let Err(error) = subjektiv_journal::prepare_worker_removal(self, runtime, target) { - return Ok(worker_remove_error_response( - StatusCode::CONFLICT, - "journal_handoff_incomplete", - &error.to_string(), - )); - } + let journal_erasure = match subjektiv_journal::prepare_worker_removal(self, runtime, target) + { + Ok(proof) => proof, + Err(error) => { + return Ok(worker_remove_error_response( + StatusCode::CONFLICT, + "journal_handoff_incomplete", + &error.to_string(), + )); + } + }; let inventory = match runtime.worker_retention_inventory(&target) { Ok(inventory) => inventory, Err(_) => { @@ -2211,6 +2215,7 @@ impl WorkerRemovalService { workspace_id: self.workspace_id.clone(), worker: target.clone(), reason: reason.to_string(), + journal_erasure, }; let plan = match self.store.plan_worker_removal(&request, &inventory) { Ok(plan) => plan, diff --git a/crates/workspace-server/src/server/subjektiv_journal.rs b/crates/workspace-server/src/server/subjektiv_journal.rs index 96dd6c0d5..6de0b3e46 100644 --- a/crates/workspace-server/src/server/subjektiv_journal.rs +++ b/crates/workspace-server/src/server/subjektiv_journal.rs @@ -320,6 +320,22 @@ fn admit( )); } let mut id = store.journal_for_session(&host).map_err(domain)?; + let reader = material_reader(api, &attribution)?; + if reader.not_materialized { + if store.journal_retention_required(&host).map_err(domain)? { + return Err(Error::RepositoryConflict( + "Admitted Journal original Session is not materialized".into(), + )); + } + retain_source( + api.runtime.as_ref(), + api.workspace_id(), + &attribution, + material_source(api, &attribution)?, + true, + )?; + return Ok(JournalRecordingResponse::NoNewSources); + } retain_source( api.runtime.as_ref(), api.workspace_id(), @@ -327,7 +343,10 @@ fn admit( material_source(api, &attribution)?, false, )?; - let sources = canonical_catalog(api, &attribution)?; + let sources: Vec<_> = reader + .descriptors() + .map(|d| source_from_descriptor(&attribution.session_id, d)) + .collect(); for batch in sources.chunks(MAX_JOURNAL_RANGES) { id = Some( store @@ -374,7 +393,7 @@ pub(super) async fn recording_route( ) .into()); } - require_enabled(api.clone(), &worker)?; + require_enabled(&api, &worker)?; let store = store(&api)?; let attributions = store .subject_session_attributions(subject) @@ -424,7 +443,16 @@ pub(super) async fn job_route( .await } -fn require_enabled(api: WorkspaceApi, worker: &RuntimeWorkerRef) -> Result<()> { +fn require_enabled(api: &WorkspaceApi, worker: &RuntimeWorkerRef) -> Result<()> { + if !enabled(api, worker)? { + return Err(Error::WorkspacePermissionDenied( + "Journal Feature is disabled".into(), + )); + } + Ok(()) +} + +fn enabled(api: &WorkspaceApi, worker: &RuntimeWorkerRef) -> Result { let projection = api .store .worker_registry_projection(api.workspace_id(), worker)? @@ -459,12 +487,7 @@ fn require_enabled(api: WorkspaceApi, worker: &RuntimeWorkerRef) -> Result<()> { .ok_or_else(|| Error::Config("Journal Worker Profile bundle missing".into()))?; let manifest = crate::profile_settings::resolve_profile_manifest_from_config_bundle(&bundle, key)?; - if !manifest.feature.subjektiv.enabled() { - return Err(Error::WorkspacePermissionDenied( - "Journal Feature is disabled".into(), - )); - } - Ok(()) + Ok(manifest.feature.subjektiv.enabled()) } pub(super) fn drain_after_completion(api: &WorkspaceApi, request: &BackendJobRequest) { @@ -539,16 +562,6 @@ fn source_from_descriptor( }, } } -fn canonical_catalog( - api: &WorkspaceApi, - attribution: &crate::subjektiv::SubjectSessionAttribution, -) -> Result> { - let reader = material_reader(api, attribution)?; - Ok(reader - .descriptors() - .map(|d| source_from_descriptor(&attribution.session_id, d)) - .collect()) -} fn read_canonical( reader: &MaterialReader<'_>, source: &JournalSource, @@ -626,20 +639,14 @@ pub(super) fn recover(api: &WorkspaceApi) -> Result<()> { } if erased { if let Err(error) = material_source(api, attr).and_then(|source| { - retain_source( - api.runtime.as_ref(), - api.workspace_id(), - attr, - source, - true, - ) + erase_source(api.runtime.as_ref(), api.workspace_id(), attr, source) }) { - tracing::warn!(%error,session_id=%attr.session_id,"Erased Session source pin release remains recoverable"); + tracing::warn!(%error,session_id=%attr.session_id,"Erased Session physical cleanup remains recoverable"); } continue; } let source = RuntimeWorkerRef::new(&attr.runtime_id, &attr.worker_id); - if require_enabled(api.clone(), &source).is_err() { + if require_enabled(api, &source).is_err() { // Disablement or removal does not erase previously admitted work, but // also does not authorize collecting a new disabled Worker's history. let host = JournalHostContext::validated_session(&store, attr.clone()) @@ -710,16 +717,13 @@ fn removal_attribution( service: &WorkerRemovalService, target: &RuntimeWorkerRef, ) -> Result> { - let Some(p) = service + let Some(lease) = service .store - .worker_registry_projection(&service.workspace_id, target)? + .require_current_worker_singleton_owner(&service.workspace_id, target)? else { return Ok(None); }; - let Some(key) = p.resource_key.as_deref() else { - return Ok(None); - }; - let Some(subject) = key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) else { + let Some(subject) = lease.key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) else { return Ok(None); }; let store = crate::subjektiv::SubjektivStore::open( @@ -767,9 +771,9 @@ pub(super) fn prepare_worker_removal( service: &WorkerRemovalService, runtime: &RuntimeRegistry, target: &RuntimeWorkerRef, -) -> Result<()> { +) -> Result> { let Some(attr) = removal_attribution(service, target)? else { - return Ok(()); + return Ok(None); }; let store = crate::subjektiv::SubjektivStore::open( &service.journal_storage, @@ -781,31 +785,59 @@ pub(super) fn prepare_worker_removal( worker_id: worker_runtime::identity::WorkerId::parse(&target.worker_id) .ok_or_else(|| Error::Store("Worker identity invalid".into()))?, }; - retain_source(runtime, &service.workspace_id, &attr, source.clone(), false)?; + let api = service.journal_dispatch_api.upgrade().ok_or_else(|| { + Error::Store("Workspace API unavailable for Journal removal Feature gate".into()) + })?; + if let Some(proof) = + crate::retention::ErasedJournalSession::validated(&service.workspace_id, &store, &attr)? + { + // Erasure is monotonic. Retry the same physical intent, never admission or generation. + reconcile_session_jobs(&api, &attr, true)?; + erase_source(runtime, &service.workspace_id, &attr, source)?; + return Ok(Some(proof)); + } + let collect_new_sources = enabled(&api, target)?; + let mut id = store.journal_for_session(&host).map_err(domain)?; + // Disablement forbids new admission, not recovery of an already admitted prefix. + // With no admitted remainder there is no source reader or Job to create at removal. + if !collect_new_sources && !store.journal_retention_required(&host).map_err(domain)? { + return Ok(None); + } let reader = MaterialReader::open( runtime, &service.workspace_id, &target.runtime_id, - source, + source.clone(), &attr.session_id, )?; - let ranges: Vec<_> = reader - .descriptors() - .map(|d| source_from_descriptor(&attr.session_id, d)) - .collect(); - let mut id = store.journal_for_session(&host).map_err(domain)?; - for batch in ranges.chunks(MAX_JOURNAL_RANGES) { - id = Some( - store - .accept_journal_sources(&host, "Session activity", batch) - .map_err(domain)?, - ); + if reader.not_materialized { + if store.journal_retention_required(&host).map_err(domain)? { + return Err(Error::RepositoryConflict( + "Admitted Journal original Session is not materialized".into(), + )); + } + retain_source(runtime, &service.workspace_id, &attr, source, true)?; + return Ok(None); + } + retain_source(runtime, &service.workspace_id, &attr, source, false)?; + if collect_new_sources { + let ranges: Vec<_> = reader + .descriptors() + .map(|d| source_from_descriptor(&attr.session_id, d)) + .collect(); + for batch in ranges.chunks(MAX_JOURNAL_RANGES) { + id = Some( + store + .accept_journal_sources(&host, "Session activity", batch) + .map_err(domain)?, + ); + } } let Some(id) = id else { - return Ok(()); + return Ok(None); }; if !store.journal_retention_required(&host).map_err(domain)? { - return Ok(()); + return Ok(None); } if let Some(g) = store .active_journal_generation(&host, &id) @@ -817,7 +849,7 @@ pub(super) fn prepare_worker_removal( .get_backend_job(&service.workspace_id, &g.job.job_id)? .is_some() { - return Ok(()); + return Ok(None); } return Err(Error::Store( "Journal generation's existing Job intent is missing".into(), @@ -843,7 +875,7 @@ pub(super) fn prepare_worker_removal( // only starts the existing bounded Job; it never waits for Journal generation. api.drain_reserved_backend_jobs(); } - Ok(()) + Ok(None) } pub(super) fn validate_removal_retention( @@ -860,6 +892,22 @@ pub(super) fn validate_removal_retention( ) .map_err(|e| Error::Store(e.to_string()))?; let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?; + if store.journal_session_erased(&host).map_err(domain)? { + if prepared.runtime_request.session_disposition + != worker_runtime::retention::SessionDisposition::Purge + || prepared.runtime_request.diagnostics_disposition + != worker_runtime::retention::DiagnosticsDisposition::Purge + || prepared.runtime_request.archive_id.is_some() + { + return Err(Error::RepositoryConflict( + "Erased Journal Session requires source and diagnostics Purge, not an archive" + .into(), + )); + } + // Physical erasure precedes the plan. Runtime resumes aggregate cleanup with + // its retained permanent fence, including recovery after the aggregate is gone. + return Ok(()); + } let source = runtime_api::SessionPublicSource::Retained { worker_id: worker_runtime::identity::WorkerId::parse(&target.worker_id) .ok_or_else(|| Error::Store("Worker identity invalid".into()))?, @@ -926,6 +974,7 @@ struct MaterialReader<'a> { runtime_id: String, source: runtime_api::SessionPublicSource, session_id: String, + not_materialized: bool, descriptors: Vec, } fn material_descriptor( @@ -968,6 +1017,25 @@ impl<'a> MaterialReader<'a> { ) .map_err(|e| e.into_error())? { + runtime_api::SessionMaterialCatalogAvailability::NotMaterialized { + session_id: observed, + } => { + if observed != session_id || offset != 0 || condition.is_some() { + return Err(Error::RepositoryConflict( + "Journal Session materialization proof changed during catalog read" + .into(), + )); + } + return Ok(Self { + runtime, + workspace_id, + runtime_id: runtime_id.into(), + source, + session_id: session_id.into(), + not_materialized: true, + descriptors, + }); + } runtime_api::SessionMaterialCatalogAvailability::Page { page } => page, runtime_api::SessionMaterialCatalogAvailability::Unavailable { reason, @@ -1015,6 +1083,7 @@ impl<'a> MaterialReader<'a> { runtime_id: runtime_id.into(), source, session_id: session_id.into(), + not_materialized: false, descriptors, }) } @@ -1075,24 +1144,30 @@ impl<'a> MaterialReader<'a> { } } -pub(super) fn source_pin_id( +fn erase_source( + runtime: &RuntimeRegistry, workspace: &str, attr: &crate::subjektiv::SubjectSessionAttribution, -) -> String { - use sha2::{Digest, Sha256}; - let digest = Sha256::digest( - serde_json::to_vec(&( - "subjektiv-journal-retention-v1", - workspace, - &attr.subject_id, - &attr.session_id, - )) - .expect("string tuple serialization"), - ); - let mut bytes = [0; 16]; - bytes.copy_from_slice(&digest[..16]); - Uuid::from_bytes(bytes).to_string() + source: runtime_api::SessionPublicSource, +) -> Result<()> { + let response = runtime + .session_material_erase( + &attr.runtime_id, + runtime_api::SessionMaterialEraseRequest { + workspace_id: workspace.into(), + source, + expected_session_id: attr.session_id.clone(), + }, + ) + .map_err(|e| e.into_error())?; + if response.session_id != attr.session_id { + return Err(Error::RepositoryConflict( + "Session erase receipt differs from the requested identity".into(), + )); + } + Ok(()) } + fn retain_source( runtime: &RuntimeRegistry, workspace: &str, @@ -1100,7 +1175,9 @@ fn retain_source( source: runtime_api::SessionPublicSource, release: bool, ) -> Result<()> { - let pin_id = source_pin_id(workspace, attr); + let pin_id = + subjektiv::journal::source_retention_pin(workspace, &attr.subject_id, &attr.session_id) + .to_string(); let request = runtime_api::SessionMaterialPinRequest { workspace_id: workspace.into(), source, @@ -1340,8 +1417,10 @@ fn reconcile_session_jobs( } impl WorkspaceApi { - /// Host-only explicit erasure hook. The existing domain tombstone is the durable - /// cancellation intent; interrupted Worker Purge is retried by ordinary Job cleanup. + /// Host-only erasure saves the domain fence, cancels Jobs, then erases originals. + /// Runtime requires a stopped source Worker. Physical failure leaves the domain + /// tombstone and cancellations committed for exact retry or removal preflight; + /// terminal Job execution Sessions remain owned by ordinary Job cleanup. pub fn erase_subjektiv_journal_session( &self, session_id: &str, @@ -1359,12 +1438,11 @@ impl WorkspaceApi { .erase_journal_session(&host, request_id) .map_err(domain)?; reconcile_session_jobs(self, &attr, true)?; - retain_source( + erase_source( self.runtime.as_ref(), self.workspace_id(), &attr, material_source(self, &attr)?, - true, ) } } diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index 625acfb6c..061c29032 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs @@ -394,6 +394,24 @@ fn seed_session( session: session_store::SessionId, item: session_store::LoggedItem, origin: session_store::LoggedSessionHistoryOrigin, +) { + seed_session_history( + root, + session, + vec![session_store::LoggedHistoryEntry { + item, + metadata: session_store::LoggedSessionHistoryMetadata { + entry_id: session_store::LoggedSessionHistoryEntryId("read-source-original".into()), + origin, + derivation: None, + }, + }], + ); +} +fn seed_session_history( + root: &std::path::Path, + session: session_store::SessionId, + history: Vec, ) { use session_store::{LogEntry, Store as _, WorkerSessionStore}; let segment = Uuid::new_v4(); @@ -407,16 +425,7 @@ fn seed_session( session_id: session, system_prompt: None, config: agen::llm_client::RequestConfig::default(), - history: vec![session_store::LoggedHistoryEntry { - item, - metadata: session_store::LoggedSessionHistoryMetadata { - entry_id: session_store::LoggedSessionHistoryEntryId( - "read-source-original".into(), - ), - origin, - derivation: None, - }, - }], + history, forked_from: None, compacted_from: None, }], @@ -486,7 +495,12 @@ impl RuntimeFixture { workspace_id: TEST_WORKSPACE_ID.into(), source: src.clone(), expected_session_id: session.to_string(), - pin_id: subjektiv_journal::source_pin_id(TEST_WORKSPACE_ID, &attr), + pin_id: subjektiv::journal::source_retention_pin( + TEST_WORKSPACE_ID, + &attr.subject_id, + &attr.session_id, + ) + .to_string(), }, ) .unwrap(); @@ -559,6 +573,18 @@ impl RuntimeFixture { id, } } + fn stop_source(&self) { + self.api + .runtime + .stop_worker( + &self.source, + WorkerLifecycleRequest { + reason: Some("settle original source before permanent erase".into()), + ticket_assignment: None, + }, + ) + .unwrap(); + } fn dispatch_and_read(&self) -> BackendJobReservation { let dispatched = self .api @@ -664,9 +690,10 @@ impl RuntimeFixture { .unwrap() .is_none() ); - assert!( + assert_eq!( session_root(self.dir.path(), &self.source).exists(), - "the original conversation was not purged" + !self.store.journal_session_erased(&self.host).unwrap(), + "only explicit source erasure, never Job cleanup, removes the original conversation" ); } } @@ -713,6 +740,7 @@ async fn terminal_journal_attempts_purge_real_tool_result_sessions_without_chang .unwrap(); } "erased" => { + f.stop_source(); f.api .erase_subjektiv_journal_session( &f.session.to_string(), @@ -1126,6 +1154,7 @@ async fn journal_profile_name_without_host_job_binding_keeps_normal_archive_poli workspace_id: TEST_WORKSPACE_ID.into(), worker: f.source, reason: "profile is not purge authority".into(), + journal_erasure: None, }, &inventory, ) @@ -1364,6 +1393,7 @@ async fn failed_cancelled_and_erased_job_sessions_retry_purge_after_cleanup_fail ) .unwrap(); if outcome == "erased" { + f.stop_source(); f.api .erase_subjektiv_journal_session( &f.session.to_string(), @@ -1472,6 +1502,7 @@ async fn explicit_erasure_interrupts_completed_journal_cleanup_waiting_for_run_t session_root(f.dir.path(), worker).exists(), "normal cleanup still waits for tracked response teardown" ); + f.stop_source(); f.api .erase_subjektiv_journal_session(&f.session.to_string(), "erase-while-completed-run-busy") .unwrap(); @@ -1497,3 +1528,616 @@ async fn explicit_erasure_interrupts_completed_journal_cleanup_waiting_for_run_t assert!(f.store.journal_session_erased(&f.host).unwrap()); assert!(subjektiv_journal::execute(&f.api, worker, JournalJobRequest::ReadState).is_err()); } + +/// Removal uses the production embedded WorkerRemove forwarder, target-bound proof, +/// common reception, catalog/retention planner, Runtime and SQLite. Source creation +/// calls the create handler directly (no HTTP auth); canonical Session contents, +/// attribution and profile disablement are fixtures, not model execution or proofs +/// of the normal publication/attribution connection. +struct SubjectRemovalFixture { + dir: tempfile::TempDir, + api: WorkspaceApi, + dispatcher: Arc, + controller: RuntimeWorkerRef, + target: RuntimeWorkerRef, + store: crate::subjektiv::SubjektivStore, + attr: crate::subjektiv::SubjectSessionAttribution, + host: JournalHostContext, +} +impl SubjectRemovalFixture { + async fn new() -> Self { + Self::with_materialized_source(true).await + } + async fn with_materialized_source(materialized: bool) -> Self { + let dir = tempfile::tempdir().unwrap(); + let config = test_server_config(dir.path()); + let control = Arc::new(test_control_store(&config)); + seed_test_registered_workspace(control.as_ref(), &config) + .await + .unwrap(); + let dispatcher = Arc::new( + crate::worker_source::EmbeddedServerWorkerMutationDispatcher::new( + config.clone(), + control.clone(), + ), + ); + let api = WorkspaceApi::new_with_execution_backend_and_broker( + config, + control, + Arc::new(DeterministicExecutionBackend::default()), + BackendResourceBroker::default(), + HostExecutors::default(), + Some(dispatcher.clone()), + ) + .await + .unwrap(); + let Json(orchestrator) = scoped_start_workspace_orchestrator( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.into(), + }), + ) + .await + .unwrap(); + let controller = orchestrator.worker.unwrap(); + let controller = RuntimeWorkerRef::new(controller.runtime_id, controller.worker_id); + let store = open_subjektiv_store(&api).unwrap(); + let subject = store + .create_subject(crate::subjektiv::SubjectRole::new("removal-source").unwrap()) + .unwrap(); + let Json(created) = scoped_create_workspace_worker( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.into(), + }), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(), + display_name: "removal-source".into(), + singleton_key: None, + profile: Some("builtin:companion".into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest { + subjektiv: Some(server_api::SubjektivWorkerConnectionRequest { + subject_id: subject.id.clone(), + }), + }, + control_operation_id: None, + }), + ) + .await + .unwrap(); + let target = RuntimeWorkerRef::new(created.runtime_id, created.worker_id); + let session = store + .subject_session_attributions(&subject.id) + .unwrap() + .first() + .map(|attr| attr.session_id.parse().unwrap()) + .unwrap_or_else(session_store::new_session_id); + if materialized { + seed_session_history( + &session_root(dir.path(), &target), + session, + ["admitted-prefix", "disabled-new-history"] + .into_iter() + .map(|text| session_store::LoggedHistoryEntry { + item: message(text), + metadata: session_store::LoggedSessionHistoryMetadata { + entry_id: session_store::LoggedSessionHistoryEntryId(text.into()), + origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput, + derivation: None, + }, + }) + .collect(), + ); + } else { + use session_store::WorkerMetadataStore as _; + let root = session_root(dir.path(), &target); + let aggregate = root.parent().unwrap(); + let name = format!("worker-runtime-{}", target.worker_id); + session_store::WorkerAggregateStore::new(aggregate, &name) + .unwrap() + .write( + &session_store::WorkerMetadata::new( + &name, + Some(session_store::WorkerActiveSegmentRef::pending_segment( + session, + )), + ) + .with_workspace_id(TEST_WORKSPACE_ID), + ) + .unwrap(); + } + let attr = store + .record_session_attribution( + crate::subjektiv::SubjectSessionAttribution::new( + &subject.id, + &target.runtime_id, + &target.worker_id, + session.to_string(), + ) + .unwrap(), + ) + .unwrap(); + let host = JournalHostContext::validated_session(&store, attr.clone()).unwrap(); + seed_worker_control_grant(&api, &controller, &target, "subject-removal-fixture"); + api.runtime + .stop_worker( + &target, + WorkerLifecycleRequest { + reason: Some("test stopped Subject removal".into()), + ticket_assignment: None, + }, + ) + .unwrap(); + let summary = api.runtime.worker(&target).unwrap(); + assert_eq!(summary.state, "stopped"); + sync_worker_observation(&api, &summary).unwrap(); + Self { + dir, + api, + dispatcher, + controller, + target, + store, + attr, + host, + } + } + fn disable(&self) { + let mut registry = self + .api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &self.target) + .unwrap() + .unwrap(); + registry.profile = Some("builtin:default".into()); + self.api.store.upsert_worker_registry(®istry).unwrap(); + } + fn remove(&self) -> worker::WorkspaceResponse { + let key = self + .api + .store + .worker_resource_key(TEST_WORKSPACE_ID, &self.target) + .unwrap() + .unwrap(); + worker_runtime::worker_source::RuntimeWorkerMutationForwarder::embedded( + &self.controller.runtime_id, + worker_runtime::RuntimeWorkspaceScope::new(TEST_WORKSPACE_ID, "embedded-backend"), + &self.controller.worker_id, + self.dispatcher.clone(), + ) + .execute_worker_remove(&self.target.runtime_id, &key, "remove stopped Subject") + .unwrap() + } +} + +#[tokio::test] +async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a_job() { + let f = SubjectRemovalFixture::new().await; + f.disable(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .is_none() + ); + assert!(f.store.journal_for_session(&f.host).unwrap().is_none()); + assert!( + f.api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100) + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() { + let f = SubjectRemovalFixture::new().await; + let runtime_api::SessionMaterialCatalogAvailability::Page { page } = f + .api + .runtime + .session_material_catalog( + &f.target.runtime_id, + runtime_api::SessionMaterialCatalogRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + source: runtime_api::SessionPublicSource::Retained { + worker_id: worker_runtime::identity::WorkerId::parse(&f.target.worker_id) + .unwrap(), + }, + expected_session_id: f.attr.session_id.clone(), + expected_catalog_sha256: None, + offset: 0, + limit: 100, + }, + ) + .unwrap() + else { + panic!("canonical source catalog required") + }; + let ranges: Vec<_> = page + .descriptors + .into_iter() + .map(|d| JournalSourceRange { + source: JournalSource { + session_id: f.attr.session_id.clone(), + pointer: JournalSourcePointer { + segment_id: d.segment_id.parse().unwrap(), + entry_ref: d.entry_ref, + }, + order: d.canonical_order, + content_sha256: d.sha256, + total_bytes: d.total_bytes, + }, + bytes: JournalByteRange { + start: 0, + end: d.total_bytes, + }, + }) + .collect(); + assert_eq!(ranges.len(), 2); + let id = f + .store + .accept_journal_sources(&f.host, "accepted before disablement", &ranges[..1]) + .unwrap(); + f.disable(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .is_none() + ); + let jobs = f + .api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100) + .unwrap(); + assert_eq!(jobs.len(), 1); + assert_eq!( + jobs[0] + .request + .grants + .subjektiv_journal + .as_ref() + .unwrap() + .selected_ranges, + ranges[..1] + ); + assert_eq!( + f.store.journal_for_session(&f.host).unwrap().as_deref(), + Some(id.as_str()) + ); + assert!(f.store.journal_retention_required(&f.host).unwrap()); + let archive = f + .api + .config_store + .worker_session_archive(TEST_WORKSPACE_ID, &f.target, &f.attr.session_id) + .unwrap() + .unwrap(); + assert!(!session_root(f.dir.path(), &f.target).exists()); + let source = runtime_api::SessionPublicSource::Archived { + archive_id: archive.archive_id, + }; + let response = f + .api + .runtime + .session_material_catalog( + &f.target.runtime_id, + runtime_api::SessionMaterialCatalogRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + source, + expected_session_id: f.attr.session_id.clone(), + expected_catalog_sha256: None, + offset: 0, + limit: 100, + }, + ) + .unwrap(); + assert!(matches!( + response, + runtime_api::SessionMaterialCatalogAvailability::Page { .. } + )); +} + +#[tokio::test] +async fn attributed_unmaterialized_subject_removal_needs_no_journal_job() { + let f = SubjectRemovalFixture::with_materialized_source(false).await; + let retention = source_retention(f.dir.path()); + let session = f.attr.session_id.parse().unwrap(); + // Fixture the pre-publication pin without materializing the Session. + retention + .pin( + session, + session_store::retention::SessionRetentionPin( + subjektiv::journal::source_retention_pin( + TEST_WORKSPACE_ID, + &f.attr.subject_id, + &f.attr.session_id, + ) + .to_string() + .parse() + .unwrap(), + ), + || Ok(true), + ) + .unwrap(); + assert_eq!(retention.pins(session).unwrap().len(), 1); + let response = f.remove(); + assert!(retention.pins(session).unwrap().is_empty()); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .is_none() + ); + assert!(f.store.journal_for_session(&f.host).unwrap().is_none()); + assert!( + f.api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100) + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn missing_materialized_subject_source_blocks_removal_without_marking_it_empty() { + let f = SubjectRemovalFixture::new().await; + std::fs::remove_dir_all(session_root(f.dir.path(), &f.target)).unwrap(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::CONFLICT.as_u16(), + "{}", + response.body + ); + assert!(response.body.contains("journal_handoff_incomplete")); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .is_some() + ); + assert!(f.store.journal_for_session(&f.host).unwrap().is_none()); + assert!( + f.api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100) + .unwrap() + .is_empty() + ); +} + +fn source_retention(root: &std::path::Path) -> session_store::retention::SessionRetentionStore { + session_store::retention::SessionRetentionStore::new( + root.join(".test-embedded-runtime-store/retention/session-sources"), + ) +} + +#[tokio::test] +async fn explicitly_erased_subject_removal_purges_without_generating_or_archiving() { + let f = SubjectRemovalFixture::new().await; + let policy = f + .api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap(); + f.api + .erase_subjektiv_journal_session(&f.attr.session_id, "erase-before-normal-remove") + .unwrap(); + assert!(f.store.journal_session_erased(&f.host).unwrap()); + assert!(!session_root(f.dir.path(), &f.target).exists()); + // Exact erase replay must resume the same permanent intent even after source removal. + f.api + .erase_subjektiv_journal_session(&f.attr.session_id, "erase-before-normal-remove") + .unwrap(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .is_none() + ); + let saved = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!( + saved.plan.session_disposition, + worker_runtime::retention::SessionDisposition::Purge + ); + assert_eq!( + saved.plan.diagnostics_disposition, + worker_runtime::retention::DiagnosticsDisposition::Purge + ); + assert!(saved.plan.archive_id.is_none()); + assert!( + f.api + .config_store + .worker_session_archive(TEST_WORKSPACE_ID, &f.target, &f.attr.session_id) + .unwrap() + .is_none() + ); + assert!( + f.api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100) + .unwrap() + .is_empty() + ); + assert_eq!( + f.api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap(), + policy + ); + assert!(matches!( + source_retention(f.dir.path()).ensure_live(f.attr.session_id.parse().unwrap()), + Err(session_store::StoreError::SessionErased(_)) + )); +} + +#[tokio::test] +async fn erased_subject_removal_retries_interrupted_physical_erasure_under_the_permanent_fence() { + let f = SubjectRemovalFixture::new().await; + // Fixture boundary: interrupt the real retention deletion after its durable fence, + // before physical removal. WorkerRemove below uses the real Backend/Runtime erase path. + f.store + .erase_journal_session(&f.host, "erase-interrupted") + .unwrap(); + let retention = source_retention(f.dir.path()); + let session = f.attr.session_id.parse().unwrap(); + assert!( + retention + .delete( + session, + session_store::retention::SessionDeletionIntent::ErasePermanently, + || { + Err::<(), _>(session_store::StoreError::Io(std::io::Error::other( + "injected physical cleanup failure", + ))) + } + ) + .is_err() + ); + assert!(session_root(f.dir.path(), &f.target).exists()); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + assert!( + !session_root(f.dir.path(), &f.target) + .parent() + .unwrap() + .exists() + ); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .is_none() + ); + assert!( + f.api + .config_store + .worker_session_archive(TEST_WORKSPACE_ID, &f.target, &f.attr.session_id) + .unwrap() + .is_none() + ); + assert!( + f.api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100) + .unwrap() + .is_empty() + ); + assert!(matches!( + retention.ensure_live(session), + Err(session_store::StoreError::SessionErased(_)) + )); +} + +#[tokio::test] +async fn physical_erase_failure_keeps_domain_tombstone_and_job_cancellation_for_exact_retry() { + let f = RuntimeFixture::new().await; + let dispatched = f.dispatch_and_read(); + // Original source is not stopped: Runtime must reject physical erasure, after + // the Backend has durably fenced all Journal adoption and cancelled its Job. + assert!( + f.api + .erase_subjektiv_journal_session(&f.session.to_string(), "erase-unsettled-source") + .is_err() + ); + assert!(f.store.journal_session_erased(&f.host).unwrap()); + assert!(session_root(f.dir.path(), &f.source).exists()); + let job = f + .api + .store + .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id) + .unwrap() + .unwrap(); + assert_eq!(job.state, BackendJobState::Failed); + assert!( + subjektiv_journal::execute( + &f.api, + dispatched.attempt.worker.as_ref().unwrap(), + JournalJobRequest::ReadState + ) + .is_err() + ); + f.stop_source(); + f.api + .erase_subjektiv_journal_session(&f.session.to_string(), "erase-unsettled-source") + .unwrap(); + assert!(!session_root(f.dir.path(), &f.source).exists()); + assert_eq!( + f.api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.session.to_string(), None, 100) + .unwrap() + .len(), + 1 + ); +} + +#[tokio::test] +async fn serialized_removal_input_cannot_supply_journal_erasure_authority() { + let f = SubjectRemovalFixture::new().await; + let request: crate::retention::WorkerRemovalPlanRequest = serde_json::from_value(json!({ + "workspace_id": TEST_WORKSPACE_ID, "worker": f.target, "reason": "untrusted erase claim", + "journal_erasure": {"workspace_id": TEST_WORKSPACE_ID, "worker": f.target, "session_id": f.attr.session_id}, + })).unwrap(); + let inventory = f.api.runtime.worker_retention_inventory(&f.target).unwrap(); + let plan = f + .api + .store + .plan_worker_removal(&request, &inventory) + .unwrap(); + assert_eq!( + plan.session_disposition, + worker_runtime::retention::SessionDisposition::Archive + ); + assert!(!f.store.journal_session_erased(&f.host).unwrap()); + assert!(session_root(f.dir.path(), &f.target).exists()); +}