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

refactor: bind Subject metadata and prompt confirmation to Host lifecycle

Browse Source
This commit is contained in:
Hare
2026-10-20 03:52:55 +09:00
parent 6e729ed06c
commit 86602240fa
7 changed files with 569 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/session-store/src/worker_metadata.rs
+21 -43
View File
@@ -210,18 +210,6 @@ impl WorkerMetadata {
}
}
fn replace_active_session(metadata: &mut WorkerMetadata, active: Option<WorkerActiveSegmentRef>) {
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<PathBuf>,
) -> Result<WorkerMetadata, WorkerStoreError> {
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<String>,
) -> Result<WorkerMetadata, WorkerStoreError> {
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<PathBuf>,
) -> Result<WorkerMetadata, WorkerStoreError> {
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<WorkerMetadata, WorkerStoreError> {
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]
crates/worker/src/feature/builtin.rs
+10
View File
@@ -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<std::sync::Arc<dyn crate::metadata_lifecycle::SessionMetadataLifecycle>> {
vec![std::sync::Arc::new(
crate::subjektiv_metadata::SessionAttributionMetadata,
)]
}
pub mod chat_invocation;
pub mod drive;
pub mod flow_transition;
crates/worker/src/feature/builtin/memory.rs
+38 -25
View File
@@ -725,7 +725,7 @@ struct SubjectBehaviorRefreshState {
struct WorkspaceSubjektivResidentSummarySource {
host: SubjektivHostConnection,
prompts: Arc<ArcSwap<PromptCatalog>>,
behavior_state: Mutex<SubjectBehaviorRefreshState>,
behavior_state: Arc<Mutex<SubjectBehaviorRefreshState>>,
}
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<crate::worker::PromptContributionCompletion>,
) {
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]
);
}
crates/worker/src/lib.rs
+2
View File
@@ -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;
crates/worker/src/metadata_lifecycle.rs
+69
View File
@@ -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<dyn Fn(WorkerMetadata) -> Result<(), WorkerStoreError> + Send + Sync>;
pub(crate) type WorkerMetadataSegmentCas = Arc<
dyn Fn(&str, &WorkerActiveSegmentRef, WorkerActiveSegmentRef) -> Result<bool, WorkerStoreError>
+ Send
+ Sync,
>;
pub(crate) fn worker_metadata_writer_for_store<St>(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<St>(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)
})
}
crates/worker/src/subjektiv_metadata.rs
+120
View File
@@ -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<St: WorkerMetadataStore>(
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()
);
}
}
crates/worker/src/worker.rs
+309 -68
View File
@@ -1676,41 +1676,10 @@ pub struct SegmentLocation {
pub segment_id: SegmentId,
}
type WorkerMetadataWriter =
Arc<dyn Fn(WorkerMetadata) -> Result<(), WorkerStoreError> + Send + Sync>;
type WorkerMetadataSegmentCas = Arc<
dyn Fn(&str, &WorkerActiveSegmentRef, WorkerActiveSegmentRef) -> Result<bool, WorkerStoreError>
+ Send
+ Sync,
>;
fn worker_metadata_writer_for_store<St>(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<St>(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<dyn FnOnce() + Send + Sync>;
#[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<PromptContributionCompletion>,
) {
(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<Option<ResidentContextRefresh>, String> {
@@ -3157,6 +3141,7 @@ pub struct Worker<C: LlmClient, St: Store> {
/// Exact complete seed retained across ambiguous Store responses. The committed
/// segment remains authoritative; metadata must be repaired before accepting input.
prepared_initial_seed: Option<Vec<LogEntry>>,
prepared_initial_prompt_completion: Option<(String, PromptContributionCompletion)>,
subjektiv_host: Option<Arc<dyn crate::subjektiv::SubjektivHost>>,
/// One-shot generic Feature lifecycle notification required after restore.
restore_feature_lifecycle_pending: bool,
@@ -3427,6 +3412,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
resident_summary_source: Option<Arc<dyn SystemPromptContributionSource>>,
system_prompt_override: Option<String>,
) {
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
}
}
}
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
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<VecDeque<AttributionOutcome>>,
before_response: Mutex<Option<Box<dyn FnOnce() + Send>>>,
requests: Mutex<Vec<WorkspaceRequest>>,
}
@@ -11368,11 +11391,18 @@ mod build_summary_prompt_tests {
fn new(outcomes: impl IntoIterator<Item = AttributionOutcome>) -> 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<WorkspaceResponse, WorkspaceClientError> {
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<AtomicUsize>,
completions: Arc<AtomicUsize>,
}
#[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<PromptContributionCompletion>,
) {
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();