diff --git a/crates/session-store/src/worker_metadata.rs b/crates/session-store/src/worker_metadata.rs index e2a8ffe72..997419ba7 100644 --- a/crates/session-store/src/worker_metadata.rs +++ b/crates/session-store/src/worker_metadata.rs @@ -210,18 +210,6 @@ impl WorkerMetadata { } } -fn replace_active_session(metadata: &mut WorkerMetadata, active: Option) { - if let Some(session_id) = active.as_ref().map(|active| active.session_id) - && metadata - .subjektiv_session_attribution - .as_ref() - .is_some_and(|state| state.session_id() != session_id) - { - metadata.subjektiv_session_attribution = None; - } - metadata.active = active; -} - /// Sync persistence backend for Worker metadata. pub trait WorkerMetadataStore: Send + Sync { /// Create or replace metadata for its `worker_name` key. @@ -297,7 +285,7 @@ pub trait WorkerMetadataStore: Send + Sync { workspace_root: Option, ) -> Result { self.update_by_name(worker_name, |metadata| { - replace_active_session(metadata, active); + metadata.active = active; metadata.resolved_manifest_snapshot = resolved_manifest_snapshot; if let Some(workspace_id) = workspace_id { metadata.workspace_id = Some(workspace_id); @@ -317,7 +305,7 @@ pub trait WorkerMetadataStore: Send + Sync { workspace_id: Option, ) -> Result { self.update_by_name(worker_name, |metadata| { - replace_active_session(metadata, active); + metadata.active = active; metadata.resolved_manifest_snapshot = resolved_manifest_snapshot; if let Some(workspace_id) = workspace_id { metadata.workspace_id = Some(workspace_id); @@ -334,7 +322,7 @@ pub trait WorkerMetadataStore: Send + Sync { workspace_root: Option, ) -> Result { self.update_by_name(worker_name, |metadata| { - replace_active_session(metadata, active); + metadata.active = active; metadata.resolved_manifest_snapshot = resolved_manifest_snapshot; if let Some(workspace_root) = workspace_root { metadata.workspace_root = Some(workspace_root); @@ -342,18 +330,6 @@ pub trait WorkerMetadataStore: Send + Sync { }) } - /// Set the Host-owned Subjektiv Session attribution lifecycle state while - /// preserving the active Session pointer and every unrelated field. - fn set_subjektiv_session_attribution( - &self, - worker_name: &str, - state: SubjektivSessionAttributionState, - ) -> Result { - self.update_by_name(worker_name, |metadata| { - metadata.subjektiv_session_attribution = Some(state); - }) - } - /// Set spawned-child registry state while preserving active pointer and manifest snapshot. fn set_spawned_children( &self, @@ -1087,13 +1063,13 @@ mod tests { )) .unwrap(); store - .set_subjektiv_session_attribution( - "subject-worker", - SubjektivSessionAttributionState::OutcomeUnknown { - session_id, - diagnostic: "request interrupted".to_string(), - }, - ) + .update_by_name("subject-worker", |metadata| { + metadata.subjektiv_session_attribution = + Some(SubjektivSessionAttributionState::OutcomeUnknown { + session_id, + diagnostic: "request interrupted".to_string(), + }); + }) .unwrap(); store .set_active( @@ -1117,7 +1093,7 @@ mod tests { } #[test] - fn new_active_session_clears_prior_session_attribution_state() { + fn generic_active_update_does_not_interpret_host_attribution() { let tmp = tempfile::TempDir::new().unwrap(); let store = FsWorkerStore::new(tmp.path()).unwrap(); let old_session_id = crate::new_session_id(); @@ -1129,13 +1105,13 @@ mod tests { )) .unwrap(); store - .set_subjektiv_session_attribution( - "subject-worker", - SubjektivSessionAttributionState::Confirmed { - session_id: old_session_id, - subject_id: "subject-1".to_string(), - }, - ) + .update_by_name("subject-worker", |metadata| { + metadata.subjektiv_session_attribution = + Some(SubjektivSessionAttributionState::Confirmed { + session_id: old_session_id, + subject_id: "subject-1".to_string(), + }); + }) .unwrap(); store @@ -1151,7 +1127,9 @@ mod tests { restored.active, Some(WorkerActiveSegmentRef::pending_segment(new_session_id)) ); - assert!(restored.subjektiv_session_attribution.is_none()); + assert!( + matches!(restored.subjektiv_session_attribution, Some(SubjektivSessionAttributionState::Confirmed { session_id, .. }) if session_id == old_session_id) + ); } #[test] diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index a68b8f992..63e944e7b 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -5,6 +5,16 @@ //! an external plugin-loading surface. pub mod backend_job_result; + +/// Host metadata participants are independent of model-facing Tool enablement. +/// Disabling a Feature must not preserve stale Session-bound authority evidence. +pub(crate) fn session_metadata_lifecycles() +-> Vec> { + vec![std::sync::Arc::new( + crate::subjektiv_metadata::SessionAttributionMetadata, + )] +} + pub mod chat_invocation; pub mod drive; pub mod flow_transition; diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index f6f5869e6..ed90b8622 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -725,7 +725,7 @@ struct SubjectBehaviorRefreshState { struct WorkspaceSubjektivResidentSummarySource { host: SubjektivHostConnection, prompts: Arc>, - behavior_state: Mutex, + behavior_state: Arc>, } fn resident_summary_contribution( @@ -816,44 +816,49 @@ impl WorkspaceSubjektivResidentSummarySource { }, )) } - - fn remember_behavior(&self, output: &server_api::SubjektivResidentContextOutput) { - let mut state = self - .behavior_state - .lock() - .expect("subject behavior state poisoned"); - state.applied = Some(output.behavior_md.clone()); - } } /// Detect behavior changes against the state reflected in history, not the fetched state. /// -/// Initial load builds a prompt contribution; change refresh does not advance applied on fetch/render alone. -/// The interceptor confirms after durably appending a system item, allowing refetch if append fails. -/// Invalidate applied before rewrite so even identical bodies can be presented in the new history. +/// Fetch/render never advances applied state. Initial preparation captures a +/// source-local completion, invoked only after its prompt and active pointer commit. +/// Later refreshes are acknowledged by the interceptor after durable append. +/// Invalidate applied before rewrite so identical bodies can be represented again. #[async_trait] impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource { async fn load(&self) -> SystemPromptContribution { + self.load_initial().await.0 + } + + async fn load_initial( + &self, + ) -> ( + SystemPromptContribution, + Option, + ) { match self.fetch().await { Ok(output) => match self.render(&output) { Ok((rendered, _)) => { - self.remember_behavior(&output); - SystemPromptContribution::Ready(rendered) + let state = self.behavior_state.clone(); + let behavior = output.behavior_md; + ( + SystemPromptContribution::Ready(rendered), + Some(Box::new(move || { + state + .lock() + .expect("subject behavior state poisoned") + .applied = Some(behavior); + })), + ) } Err(error) => { tracing::error!(%error, "subject resident context render failed"); - SystemPromptContribution::Ready( - "The Host could not render the Subject resident context. Do not treat this as unset behavior or infer replacement behavior from Memory." - .to_string(), - ) + (SystemPromptContribution::Ready("The Host could not render the Subject resident context. Do not treat this as unset behavior or infer replacement behavior from Memory.".into()), None) } }, Err(error) => { tracing::error!(%error, "subject resident context fetch failed"); - SystemPromptContribution::Ready( - "The Host could not fetch the Subject resident context. Do not treat this as unset behavior or infer replacement behavior from Memory." - .to_string(), - ) + (SystemPromptContribution::Ready("The Host could not fetch the Subject resident context. Do not treat this as unset behavior or infer replacement behavior from Memory.".into()), None) } } } @@ -1218,7 +1223,7 @@ pub(crate) fn ordinary_subjektiv_resident_summary_source( Ok(Some(Arc::new(WorkspaceSubjektivResidentSummarySource { host: host.expect("checked explicit connection"), prompts, - behavior_state: Mutex::new(SubjectBehaviorRefreshState::default()), + behavior_state: Arc::new(Mutex::new(SubjectBehaviorRefreshState::default())), }))) } @@ -1835,8 +1840,16 @@ permission = "write" .unwrap() .unwrap(); - let initial = source.load().await; + let (initial, completion) = source.load_initial().await; assert!(matches!(initial, SystemPromptContribution::Ready(_))); + assert!( + source + .load_changed_resident_context() + .await + .unwrap() + .is_some() + ); + completion.expect("successful initial contribution must provide completion")(); assert!( source .load_changed_resident_context() @@ -1907,7 +1920,7 @@ permission = "write" // target-side bounded resident request contract. assert_eq!( client.timeouts.lock().unwrap().as_slice(), - &[RESIDENT_SUMMARY_REQUEST_TIMEOUT; 7] + &[RESIDENT_SUMMARY_REQUEST_TIMEOUT; 8] ); } diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 10cf5a6c5..13d61103b 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -15,6 +15,7 @@ pub mod hook; pub(crate) mod in_flight; pub mod ipc; pub mod job; +mod metadata_lifecycle; pub mod model_client; mod paste_artifact_tool; pub mod prompt; @@ -26,6 +27,7 @@ mod session_history; pub mod shared_state; mod shutdown_after_idle; mod subject_release; +mod subjektiv_metadata; pub use subject_release::SubjectReleaseReservation; pub mod skill; pub mod spawn; diff --git a/crates/worker/src/metadata_lifecycle.rs b/crates/worker/src/metadata_lifecycle.rs new file mode 100644 index 000000000..074e86d60 --- /dev/null +++ b/crates/worker/src/metadata_lifecycle.rs @@ -0,0 +1,69 @@ +//! Atomic metadata transitions with Host-owned Session lifecycle participants. +//! +//! Participants inspect domain evidence inside the metadata update, not after the +//! active pointer has been published. The generic writer never interprets it. + +use session_store::worker_metadata::{ + WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore, WorkerStoreError, +}; +use std::sync::Arc; + +pub(crate) trait SessionMetadataLifecycle: Send + Sync { + /// Runs under the metadata update lock. Only compare/transform the supplied + /// value: no I/O, reentrant store calls, lock waits, or Feature refresh. + fn active_changing(&self, metadata: &mut WorkerMetadata, next: Option<&WorkerActiveSegmentRef>); +} + +pub(crate) type WorkerMetadataWriter = + Arc Result<(), WorkerStoreError> + Send + Sync>; +pub(crate) type WorkerMetadataSegmentCas = Arc< + dyn Fn(&str, &WorkerActiveSegmentRef, WorkerActiveSegmentRef) -> Result + + Send + + Sync, +>; + +pub(crate) fn worker_metadata_writer_for_store(store: &St) -> WorkerMetadataWriter +where + St: WorkerMetadataStore + Clone + Send + Sync + 'static, +{ + let store = store.clone(); + let participants = crate::feature::builtin::session_metadata_lifecycles(); + Arc::new(move |next| { + store + .update_by_name(&next.worker_name, |metadata| { + for participant in &participants { + participant.active_changing(metadata, next.active.as_ref()); + } + metadata.active = next.active; + metadata.resolved_manifest_snapshot = next.resolved_manifest_snapshot; + if next.workspace_id.is_some() { + metadata.workspace_id = next.workspace_id; + } + if next.workspace_root.is_some() { + metadata.workspace_root = next.workspace_root; + } + }) + .map(|_| ()) + }) +} + +pub(crate) fn worker_metadata_segment_cas_for_store(store: &St) -> WorkerMetadataSegmentCas +where + St: WorkerMetadataStore + Clone + Send + Sync + 'static, +{ + let store = store.clone(); + let participants = crate::feature::builtin::session_metadata_lifecycles(); + Arc::new(move |worker_name, expected, replacement| { + let mut matched = false; + store.update_by_name(worker_name, |metadata| { + if metadata.active.as_ref() == Some(expected) { + for participant in &participants { + participant.active_changing(metadata, Some(&replacement)); + } + metadata.active = Some(replacement); + matched = true; + } + })?; + Ok(matched) + }) +} diff --git a/crates/worker/src/subjektiv_metadata.rs b/crates/worker/src/subjektiv_metadata.rs new file mode 100644 index 000000000..e24b4c926 --- /dev/null +++ b/crates/worker/src/subjektiv_metadata.rs @@ -0,0 +1,120 @@ +//! Host-owned attribution transitions. Metadata callbacks are synchronous, local +//! comparisons only: no network I/O, Feature lookup, or nested store access. + +use crate::metadata_lifecycle::SessionMetadataLifecycle; +use session_store::{ + SessionId, SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerMetadata, + WorkerMetadataStore, +}; + +pub(crate) struct SessionAttributionMetadata; + +impl SessionMetadataLifecycle for SessionAttributionMetadata { + fn active_changing( + &self, + metadata: &mut WorkerMetadata, + next: Option<&WorkerActiveSegmentRef>, + ) { + if metadata.active.as_ref().map(|active| active.session_id) + != next.map(|active| active.session_id) + { + metadata.subjektiv_session_attribution = None; + } + } +} + +pub(crate) fn compare_and_swap_attribution( + store: &St, + worker_name: &str, + session_id: SessionId, + expected: Option<&SubjektivSessionAttributionState>, + replacement: SubjektivSessionAttributionState, +) -> Result<(), crate::WorkerError> { + let mut matched = false; + store.update_by_name(worker_name, |metadata| { + if metadata.active.as_ref().map(|active| active.session_id) == Some(session_id) + && metadata.subjektiv_session_attribution.as_ref() == expected + && replacement.session_id() == session_id + { + metadata.subjektiv_session_attribution = Some(replacement); + matched = true; + } + })?; + if !matched { + return Err(crate::WorkerError::SubjektivSessionAttribution { + state: "attribution_conflict", + message: "active Session or observed attribution changed during Host confirmation" + .into(), + }); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + //! Connect the production metadata writer/CAS and Host lifecycle to a real + //! filesystem store; no metadata or persistence boundary is substituted. + use super::*; + use crate::metadata_lifecycle::{ + worker_metadata_segment_cas_for_store, worker_metadata_writer_for_store, + }; + + #[test] + fn host_transition_preserves_same_session_and_invalidates_new_session_atomically() { + let dir = tempfile::tempdir().unwrap(); + let store = session_store::FsWorkerStore::new(dir.path()).unwrap(); + let session = session_store::new_session_id(); + let mut metadata = WorkerMetadata::new( + "body", + Some(WorkerActiveSegmentRef::pending_segment(session)), + ); + metadata.workspace_id = Some("workspace".into()); + metadata.subjektiv_session_attribution = + Some(SubjektivSessionAttributionState::Confirmed { + session_id: session, + subject_id: "subject".into(), + }); + store.write(&metadata).unwrap(); + let writer = worker_metadata_writer_for_store(&store); + let cas = worker_metadata_segment_cas_for_store(&store); + let segment = + WorkerActiveSegmentRef::active_segment(session, session_store::new_segment_id()); + let mut next = WorkerMetadata::new("body", Some(segment.clone())); + writer(next.clone()).unwrap(); + let materialized = store.read_by_name("body").unwrap().unwrap(); + assert_eq!( + materialized.subjektiv_session_attribution, + metadata.subjektiv_session_attribution + ); + assert_eq!(materialized.workspace_id, metadata.workspace_id); + let compacted = + WorkerActiveSegmentRef::active_segment(session, session_store::new_segment_id()); + assert!(cas("body", &segment, compacted.clone()).unwrap()); + assert_eq!( + store + .read_by_name("body") + .unwrap() + .unwrap() + .subjektiv_session_attribution, + metadata.subjektiv_session_attribution + ); + let new_session = WorkerActiveSegmentRef::pending_segment(session_store::new_session_id()); + assert!(!cas("body", &segment, new_session.clone()).unwrap()); + assert!(cas("body", &compacted, new_session.clone()).unwrap()); + let switched = store.read_by_name("body").unwrap().unwrap(); + assert_eq!(switched.active, Some(new_session)); + assert!(switched.subjektiv_session_attribution.is_none()); + assert_eq!(switched.workspace_id, metadata.workspace_id); + // Returning to the old Session cannot recover the invalidated evidence. + next.active = metadata.active; + writer(next).unwrap(); + assert!( + store + .read_by_name("body") + .unwrap() + .unwrap() + .subjektiv_session_attribution + .is_none() + ); + } +} diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 805ffb044..aabb6314f 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -1676,41 +1676,10 @@ pub struct SegmentLocation { pub segment_id: SegmentId, } -type WorkerMetadataWriter = - Arc Result<(), WorkerStoreError> + Send + Sync>; -type WorkerMetadataSegmentCas = Arc< - dyn Fn(&str, &WorkerActiveSegmentRef, WorkerActiveSegmentRef) -> Result - + Send - + Sync, ->; - -fn worker_metadata_writer_for_store(store: &St) -> WorkerMetadataWriter -where - St: WorkerMetadataStore + Clone + Send + Sync + 'static, -{ - let store = store.clone(); - Arc::new(move |metadata| { - store - .set_active_with_workspace_context( - &metadata.worker_name, - metadata.active, - metadata.resolved_manifest_snapshot, - metadata.workspace_id, - metadata.workspace_root, - ) - .map(|_| ()) - }) -} - -fn worker_metadata_segment_cas_for_store(store: &St) -> WorkerMetadataSegmentCas -where - St: WorkerMetadataStore + Clone + Send + Sync + 'static, -{ - let store = store.clone(); - Arc::new(move |worker_name, expected, replacement| { - store.compare_and_swap_active(worker_name, expected, replacement) - }) -} +use crate::metadata_lifecycle::{ + WorkerMetadataSegmentCas, WorkerMetadataWriter, worker_metadata_segment_cas_for_store, + worker_metadata_writer_for_store, +}; /// Lock-free shared session/segment pointer. /// @@ -2938,6 +2907,8 @@ impl WorkerSession { } } +pub(crate) type PromptContributionCompletion = Box; + #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum SystemPromptContribution { Ready(String), @@ -2969,8 +2940,21 @@ pub(crate) trait SessionContextSource: Send + Sync { pub(crate) trait SystemPromptContributionSource: Send + Sync { async fn load(&self) -> SystemPromptContribution; - /// Returns a durable model-visible refresh only when user-managed resident - /// context changed since the last successful load. + /// Prepare an initial contribution with an optional source-bound completion. + /// Completion runs only after its exact prompt and active pointer commit, + /// under the publication/release fence. It must perform bounded local updates + /// only; no I/O, store reentry, or authority grants. Restore never replays it. + async fn load_initial( + &self, + ) -> ( + SystemPromptContribution, + Option, + ) { + (self.load().await, None) + } + + /// Returns a model-visible refresh only when user-managed resident context + /// changed since its last committed representation. async fn load_changed_resident_context( &self, ) -> Result, String> { @@ -3157,6 +3141,7 @@ pub struct Worker { /// Exact complete seed retained across ambiguous Store responses. The committed /// segment remains authoritative; metadata must be repaired before accepting input. prepared_initial_seed: Option>, + prepared_initial_prompt_completion: Option<(String, PromptContributionCompletion)>, subjektiv_host: Option>, /// One-shot generic Feature lifecycle notification required after restore. restore_feature_lifecycle_pending: bool, @@ -3427,6 +3412,7 @@ impl Worker { feature_resident_summary_source: None, session_context_source: None, prepared_initial_seed: None, + prepared_initial_prompt_completion: None, subjektiv_host: None, restore_feature_lifecycle_pending: false, feature_system_prompt_override: None, @@ -3482,6 +3468,7 @@ impl Worker { resident_summary_source: Option>, system_prompt_override: Option, ) { + self.prepared_initial_prompt_completion = None; self.feature_resident_summary_source = resident_summary_source; self.feature_system_prompt_override = system_prompt_override; } @@ -4671,16 +4658,19 @@ impl Worker { } } } + let mut completion = None; let resident_summary = if self.inject_resident_summary { match &self.feature_resident_summary_source { - Some(source) => match source.load().await { - SystemPromptContribution::Ready(summary) if !summary.is_empty() => { - Some(summary) + Some(source) => { + let (value, prepared) = source.load_initial().await; + match value { + SystemPromptContribution::Ready(summary) if !summary.is_empty() => { + completion = prepared; + Some(summary) + } + _ => None, } - SystemPromptContribution::Ready(_) | SystemPromptContribution::Unavailable => { - None - } - }, + } None => None, } } else { @@ -4707,6 +4697,12 @@ impl Worker { let rendered = template .render(&ctx) .map_err(|source| WorkerError::SystemPromptRender { source })?; + self.prepared_initial_prompt_completion = completion.and_then(|complete| { + resident_summary + .as_ref() + .filter(|body| rendered.contains(body.as_str())) + .map(|_| (rendered.clone(), complete)) + }); self.engine .as_mut() .expect("worker present") @@ -5959,6 +5955,11 @@ impl Worker { self.system_prompt_template = None; self.feature_system_prompt_override = None; self.prepared_initial_seed = None; + if let Some((prepared, complete)) = self.prepared_initial_prompt_completion.take() { + if system_prompt.as_ref() == Some(&prepared) { + complete(); + } + } Ok(()) } @@ -7999,6 +8000,7 @@ where feature_resident_summary_source: None, session_context_source: None, prepared_initial_seed: None, + prepared_initial_prompt_completion: None, subjektiv_host: None, restore_feature_lifecycle_pending: false, feature_system_prompt_override: None, @@ -8029,6 +8031,7 @@ where subjektiv_attached: bool, lifecycle: SubjektivSessionAttributionLifecycle, ) -> Result<(), WorkerError> { + let _admission = self.segment_state.mutation_authority.enter()?; let worker_name = self.manifest.worker.name.as_str(); let metadata = self.store.read_by_name(worker_name)?.ok_or_else(|| { WorkerError::SubjektivSessionAttribution { @@ -8107,11 +8110,15 @@ where } else { "legacy Session has no local attribution evidence; immutable Backend verification started" }; - self.persist_subjektiv_session_attribution(incomplete_subjektiv_attribution_state( + let pending_attribution = incomplete_subjektiv_attribution_state( session_id, create_if_missing, initial_diagnostic.to_string(), - ))?; + ); + self.persist_subjektiv_session_attribution( + metadata.subjektiv_session_attribution.as_ref(), + pending_attribution.clone(), + )?; let output = if self.subjektiv_host.is_some() { let connection = self .subjektiv_host_connection() @@ -8144,7 +8151,7 @@ where message.clone(), ) }; - self.persist_subjektiv_session_attribution(state)?; + self.persist_subjektiv_session_attribution(Some(&pending_attribution), state)?; return Err(WorkerError::SubjektivSessionAttribution { state: if create_if_missing && rejected { "failed" @@ -8170,6 +8177,7 @@ where "Session attribution transport outcome is unknown: {error}" )); self.persist_subjektiv_session_attribution( + Some(&pending_attribution), incomplete_subjektiv_attribution_state( session_id, create_if_missing, @@ -8198,7 +8206,7 @@ where diagnostic: message.clone(), } }; - self.persist_subjektiv_session_attribution(state)?; + self.persist_subjektiv_session_attribution(Some(&pending_attribution), state)?; return Err(WorkerError::SubjektivSessionAttribution { state: if create_if_missing { "failed" @@ -8216,6 +8224,7 @@ where "Session attribution response outcome is unknown: {error}" )); self.persist_subjektiv_session_attribution( + Some(&pending_attribution), incomplete_subjektiv_attribution_state( session_id, create_if_missing, @@ -8237,20 +8246,26 @@ where let message = "Session attribution response did not confirm the requested immutable identity" .to_string(); - self.persist_subjektiv_session_attribution(incomplete_subjektiv_attribution_state( - session_id, - create_if_missing, - message.clone(), - ))?; + self.persist_subjektiv_session_attribution( + Some(&pending_attribution), + incomplete_subjektiv_attribution_state( + session_id, + create_if_missing, + message.clone(), + ), + )?; return Err(WorkerError::SubjektivSessionAttribution { state: incomplete_subjektiv_attribution_label(create_if_missing), message, }); } - self.persist_subjektiv_session_attribution(SubjektivSessionAttributionState::Confirmed { - session_id, - subject_id: output.subject_id.clone(), - })?; + self.persist_subjektiv_session_attribution( + Some(&pending_attribution), + SubjektivSessionAttributionState::Confirmed { + session_id, + subject_id: output.subject_id.clone(), + }, + )?; self.retain_journal_source_before_publication(session_id, &output.subject_id) } @@ -8291,11 +8306,16 @@ where fn persist_subjektiv_session_attribution( &self, + expected: Option<&SubjektivSessionAttributionState>, state: SubjektivSessionAttributionState, ) -> Result<(), WorkerError> { - self.store - .set_subjektiv_session_attribution(&self.manifest.worker.name, state)?; - Ok(()) + crate::subjektiv_metadata::compare_and_swap_attribution( + &self.store, + &self.manifest.worker.name, + self.session_id(), + expected, + state, + ) } /// Build an in-process Internal Worker without machine-wide allocation or durable Worker metadata. @@ -8380,6 +8400,7 @@ where feature_resident_summary_source: None, session_context_source: None, prepared_initial_seed: None, + prepared_initial_prompt_completion: None, subjektiv_host: None, restore_feature_lifecycle_pending: false, feature_system_prompt_override: None, @@ -8512,6 +8533,7 @@ where feature_resident_summary_source: None, session_context_source: None, prepared_initial_seed: None, + prepared_initial_prompt_completion: None, subjektiv_host: None, restore_feature_lifecycle_pending: false, feature_system_prompt_override: None, @@ -8923,6 +8945,7 @@ where feature_resident_summary_source: None, session_context_source: None, prepared_initial_seed: None, + prepared_initial_prompt_completion: None, subjektiv_host: None, restore_feature_lifecycle_pending: true, feature_system_prompt_override: None, @@ -11358,9 +11381,9 @@ mod build_summary_prompt_tests { TransportUnknown, } - #[derive(Debug)] struct AttributionLifecycleClient { outcomes: Mutex>, + before_response: Mutex>>, requests: Mutex>, } @@ -11368,11 +11391,18 @@ mod build_summary_prompt_tests { fn new(outcomes: impl IntoIterator) -> Self { Self { outcomes: Mutex::new(outcomes.into_iter().collect()), + before_response: Mutex::new(None), requests: Mutex::new(Vec::new()), } } } + impl std::fmt::Debug for AttributionLifecycleClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("AttributionLifecycleClient") + } + } + impl WorkspaceClient for AttributionLifecycleClient { fn workspace_id(&self) -> Option<&str> { Some("workspace-test") @@ -11391,6 +11421,9 @@ mod build_summary_prompt_tests { request: WorkspaceRequest, ) -> Result { self.requests.lock().unwrap().push(request.clone()); + if let Some(change) = self.before_response.lock().unwrap().take() { + change(); + } match self .outcomes .lock() @@ -11638,6 +11671,110 @@ mod build_summary_prompt_tests { )); } + // Production lifecycle, atomic Host CAS, and filesystem metadata store. Only + // the external Workspace transport is substituted to interleave a newer owner. + #[tokio::test] + async fn delayed_attribution_response_cannot_overwrite_new_session_or_subject_evidence() { + for switch_session in [true, false] { + let client = Arc::new(AttributionLifecycleClient::new([ + AttributionOutcome::Success, + ])); + let (worker, store, _temp) = attribution_lifecycle_worker(client.clone()).await; + let worker_name = worker.manifest().worker.name.clone(); + let replacement_session = if switch_session { + session_store::new_session_id() + } else { + worker.session_id() + }; + let changed = store.clone(); + let name = worker_name.clone(); + *client.before_response.lock().unwrap() = Some(Box::new(move || { + if switch_session { + worker_metadata_writer_for_store(&changed)(WorkerMetadata::new( + name, + Some(WorkerActiveSegmentRef::pending_segment(replacement_session)), + )) + .unwrap(); + } else { + changed + .update_by_name(&name, |metadata| { + metadata.subjektiv_session_attribution = + Some(SubjektivSessionAttributionState::Confirmed { + session_id: replacement_session, + subject_id: "new-subject".into(), + }); + }) + .unwrap(); + } + })); + let error = worker + .finalize_subjektiv_session_attribution( + true, + SubjektivSessionAttributionLifecycle::NewSession, + ) + .unwrap_err(); + assert!( + error.to_string().contains("attribution_conflict"), + "{error}" + ); + let saved = store.read_by_name(&worker_name).unwrap().unwrap(); + assert_eq!( + saved.active.as_ref().unwrap().session_id, + replacement_session + ); + assert_eq!( + saved.subjektiv_session_attribution, + if switch_session { + None + } else { + Some(SubjektivSessionAttributionState::Confirmed { + session_id: replacement_session, + subject_id: "new-subject".into(), + }) + } + ); + assert_eq!(client.requests.lock().unwrap().len(), 1); + } + } + + #[tokio::test] + async fn released_body_cannot_reconfirm_attribution_or_contact_host() { + let client = Arc::new(AttributionLifecycleClient::new([ + AttributionOutcome::Success, + ])); + let (worker, store, _temp) = attribution_lifecycle_worker(client.clone()).await; + let before = store + .read_by_name(&worker.manifest().worker.name) + .unwrap() + .unwrap(); + let control = crate::subject_release::ReleaseControl::new(worker.mutation_authority()); + let request = tokio::spawn({ + let control = control.clone(); + async move { control.request().await } + }); + control.notified().await; + control.respond(|| Ok(())); + let mut reservation = request.await.unwrap().unwrap(); + reservation.commit_disconnect().unwrap(); + drop(reservation); + assert!( + worker + .finalize_subjektiv_session_attribution( + true, + SubjektivSessionAttributionLifecycle::NewSession + ) + .is_err() + ); + assert!(client.requests.lock().unwrap().is_empty()); + assert_eq!( + store + .read_by_name(&worker.manifest().worker.name) + .unwrap() + .unwrap(), + before + ); + } + #[tokio::test] async fn unknown_and_failed_attribution_remain_incomplete_until_lifecycle_retry() { for (first, expected_state) in [ @@ -11750,13 +11887,13 @@ mod build_summary_prompt_tests { ) .unwrap(); store - .set_subjektiv_session_attribution( - &manifest.worker.name, - SubjektivSessionAttributionState::OutcomeUnknown { - session_id, - diagnostic: "response lost before restart".to_string(), - }, - ) + .update_by_name(&manifest.worker.name, |metadata| { + metadata.subjektiv_session_attribution = + Some(SubjektivSessionAttributionState::OutcomeUnknown { + session_id, + diagnostic: "response lost before restart".to_string(), + }); + }) .unwrap(); let client = Arc::new(AttributionLifecycleClient::new([ AttributionOutcome::Success, @@ -17024,6 +17161,110 @@ permission = "write" assert_eq!(worker.history().len(), 1); } + struct PreparedPromptSource { + loads: Arc, + completions: Arc, + } + #[async_trait::async_trait] + impl SystemPromptContributionSource for PreparedPromptSource { + async fn load(&self) -> SystemPromptContribution { + panic!("initial load must use its prepared completion") + } + async fn load_initial( + &self, + ) -> ( + SystemPromptContribution, + Option, + ) { + self.loads.fetch_add(1, Ordering::SeqCst); + let completed = self.completions.clone(); + ( + SystemPromptContribution::Ready("PREPARED_RESIDENT_BODY".into()), + Some(Box::new(move || { + completed.fetch_add(1, Ordering::SeqCst); + })), + ) + } + } + + // Production initial-head preparation/create/reconciliation and real FsStore. + // Only returned write failures and the Feature completion counter are injected. + #[tokio::test] + async fn initial_prompt_completion_waits_for_commit_and_reconciliation_and_never_replays_to_replacement() + { + for failure in [3, 4, 0] { + for replace_source in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let store = FailRunTerminalStore { + inner: session_store::FsStore::new(temp.path().join("sessions")).unwrap(), + fail_run_terminal: Arc::new(AtomicBool::new(false)), + invocation_failure: Arc::new(AtomicUsize::new(failure)), + }; + let mut worker = Worker::new( + minimal_manifest(), + Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient), + store.clone(), + WorkerWorkspaceContext::no_workspace(), + WorkerFilesystemAuthority::None, + Scope::empty(), + ) + .await + .unwrap(); + worker.set_system_prompt_template( + SystemPromptTemplate::parse("default", PromptCatalogSource::builtins_only()) + .unwrap(), + ); + let loads = Arc::new(AtomicUsize::new(0)); + let completed = Arc::new(AtomicUsize::new(0)); + worker.install_system_prompt_contribution( + Some(Arc::new(PreparedPromptSource { + loads: loads.clone(), + completions: completed.clone(), + })), + None, + ); + if failure == 0 { + let fail = Arc::new(AtomicBool::new(true)); + worker.worker_metadata_writer = Some(Arc::new(move |_| { + if fail.swap(false, Ordering::SeqCst) { + Err(std::io::Error::other("initial active pointer write failed").into()) + } else { + Ok(()) + } + })); + } + assert!(worker.materialize_durable_session_head().await.is_err()); + assert_eq!(completed.load(Ordering::SeqCst), 0); + let new_completed = Arc::new(AtomicUsize::new(0)); + let new_loads = Arc::new(AtomicUsize::new(0)); + if replace_source { + worker.install_system_prompt_contribution( + Some(Arc::new(PreparedPromptSource { + loads: new_loads.clone(), + completions: new_completed.clone(), + })), + None, + ); + } + worker.materialize_durable_session_head().await.unwrap(); + worker.materialize_durable_session_head().await.unwrap(); + assert_eq!(loads.load(Ordering::SeqCst), 1); + assert_eq!( + completed.load(Ordering::SeqCst), + usize::from(!replace_source) + ); + assert_eq!(new_completed.load(Ordering::SeqCst), 0); + assert_eq!(new_loads.load(Ordering::SeqCst), 0); + let saved = store + .read_all(worker.session_id(), worker.segment_id()) + .unwrap(); + assert!( + matches!(&saved[0], LogEntry::AnnotatedSegmentStart { system_prompt: Some(prompt), .. } if prompt.contains("PREPARED_RESIDENT_BODY")) + ); + } + } + } + #[tokio::test] async fn initial_metadata_failure_repairs_pointer_without_refetch_or_user_append() { let temp = tempfile::tempdir().unwrap();