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

refactor: isolate Feature System input migration and compaction policy

Browse Source
This commit is contained in:
Hare
2026-10-20 03:21:22 +09:00
parent bb5c50f9ab
commit 6e729ed06c
27 changed files with 2226 additions and 646 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+3 -3
View File
@@ -1656,10 +1656,10 @@ where
if ordinary_subjektiv_features_enabled
&& lifecycle_enabled
&& let Some(host) = subjektiv_host.as_ref()
&& host.host.journal_startup_granted()
&& (host.host.journal_startup_granted() || host.host.journal_compaction_granted())
{
worker.install_initial_context_source(Arc::new(
crate::feature::builtin::journal::JournalStartupContextSource::new(
worker.install_session_context_source(Arc::new(
crate::feature::builtin::journal::JournalContextSource::new(
host.clone(),
worker.prompts(),
),
crates/worker/src/feature/builtin/journal.rs
+143 -26
View File
@@ -23,7 +23,7 @@ use crate::hook::{
use crate::subjektiv::{SubjektivHostConnection, SubjektivHostError};
use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
use async_trait::async_trait;
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::sync::Arc;
use subjektiv::journal::transport::*;
@@ -32,12 +32,73 @@ use subjektiv::journal::{
MAX_JOURNAL_RESPONSE_BYTES,
};
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct JournalStartupInput {
pub snapshot: subjektiv::journal::startup::JournalStartupSnapshot,
pub snapshot_sha256: String,
pub prompt_provenance: session_store::PromptRenderProvenance,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum JournalReadStatus {
Available,
Absent,
Disabled,
Unavailable,
}
impl JournalReadStatus {
fn as_str(self) -> &'static str {
match self {
Self::Available => "available",
Self::Absent => "absent",
Self::Disabled => "disabled",
Self::Unavailable => "unavailable",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct JournalCompactionInput {
status: JournalReadStatus,
snapshot: Option<subjektiv::journal::startup::JournalStartupSnapshot>,
snapshot_sha256: Option<String>,
prompt_provenance: session_store::PromptRenderProvenance,
}
fn journal_input(
body: String,
payload: session_store::FeatureInputPayload,
kind: &str,
) -> Result<session_store::SystemInputEnvelope, String> {
use session_store::*;
let public = PublicSystemItem {
kind: kind.into(),
body: body.clone(),
presentation: PublicSystemPresentation::Context,
};
let metadata = SystemInputMetadata::new(
SystemInputUsePolicy {
recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden,
compaction: InputCompactionUse::ReferenceInput,
},
Some(payload),
Some(public),
)
.map_err(|e| e.to_string())?;
SystemInputEnvelope::new(body, metadata).map_err(|e| e.to_string())
}
/// Subjektiv owns startup projection; Core only commits its typed initial items.
pub(crate) struct JournalStartupContextSource {
pub(crate) struct JournalContextSource {
host: SubjektivHostConnection,
prompts: Arc<arc_swap::ArcSwap<crate::prompt::catalog::PromptCatalog>>,
}
impl JournalStartupContextSource {
impl JournalContextSource {
pub(crate) fn new(
host: SubjektivHostConnection,
prompts: Arc<arc_swap::ArcSwap<crate::prompt::catalog::PromptCatalog>>,
@@ -46,9 +107,12 @@ impl JournalStartupContextSource {
}
}
#[async_trait]
impl crate::worker::InitialContextSource for JournalStartupContextSource {
impl crate::worker::SessionContextSource for JournalContextSource {
async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
use sha2::{Digest, Sha256};
if !self.host.host.journal_startup_granted() {
return Ok(Vec::new());
}
let snapshot = self
.host
.journal_startup_context()
@@ -58,8 +122,8 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
let empty = snapshot.journals.is_empty() && !snapshot.omitted;
// Hash the same canonical JSON value retained by SystemItem, so replay
// can verify provenance without depending on domain struct field order.
let snapshot = serde_json::to_value(snapshot).map_err(|e| e.to_string())?;
let snapshot_json = serde_json::to_string(&snapshot).map_err(|e| e.to_string())?;
let snapshot_value = serde_json::to_value(&snapshot).map_err(|e| e.to_string())?;
let snapshot_json = serde_json::to_string(&snapshot_value).map_err(|e| e.to_string())?;
if snapshot_json.len() > subjektiv::journal::startup::STARTUP_MAX_BYTES {
return Err("Host Journal startup snapshot exceeds independent byte budget".into());
}
@@ -74,9 +138,10 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
return Err("Journal startup prompt exceeds rendered byte budget".into());
}
let projection = prompts.projection();
Ok(vec![session_store::LoggedSystemHistoryEntry {
item: session_store::SystemItem::JournalStartup {
body,
let payload = session_store::FeatureInputPayload::encode(
"journal:startup",
1,
&JournalStartupInput {
snapshot,
snapshot_sha256: Sha256::digest(snapshot_json.as_bytes())
.iter()
@@ -89,11 +154,38 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
logical_name: "internal.subjektiv_journal_startup".into(),
},
},
metadata: crate::session_history::metadata(
session_store::LoggedSessionHistoryOrigin::JournalContext,
None,
),
}])
)
.map_err(|e| e.to_string())?;
Ok(vec![session_store::LoggedSystemHistoryEntry::from_input(
session_store::LoggedSessionHistoryOrigin::Provided {
producer: "subjektiv.journal".into(),
},
None,
journal_input(body, payload, "journal_startup")?,
)])
}
async fn load_compaction_context(
&self,
session: session_store::SessionId,
) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
// A changed Session must never reuse the old source's Host ownership.
let connection = if self.host.context.session_id == session.to_string() {
Ok(Some(self.host.clone()))
} else {
Err(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"context source belongs to a different Session",
))
};
let input = freeze_compaction_journal(connection, &self.prompts.load_full())?;
Ok(vec![session_store::LoggedSystemHistoryEntry::from_input(
session_store::LoggedSessionHistoryOrigin::Provided {
producer: "subjektiv.journal".into(),
},
None,
input,
)])
}
}
@@ -103,7 +195,8 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
pub(crate) fn freeze_compaction_journal(
connection: std::io::Result<Option<SubjektivHostConnection>>,
prompts: &crate::prompt::catalog::PromptCatalog,
) -> Result<String, crate::prompt::catalog::CatalogError> {
) -> Result<session_store::SystemInputEnvelope, String> {
use JournalReadStatus::*;
let workspace_id = connection.as_ref().ok().and_then(|connection| {
connection
.as_ref()
@@ -116,24 +209,25 @@ pub(crate) fn freeze_compaction_journal(
Ok(_) => None,
Err(_) => Some(Err(())),
};
let snapshot = read.as_ref().and_then(|read| read.as_ref().ok()).cloned();
let (status, json) = match read {
None => ("disabled", String::new()),
None => (Disabled, String::new()),
Some(Ok(snapshot)) => {
let status = if snapshot.journals.is_empty() {
"absent"
Absent
} else {
"available"
Available
};
match serde_json::to_string(&snapshot) {
Ok(json) if json.len() <= subjektiv::journal::startup::STARTUP_MAX_BYTES => {
(status, json)
}
_ => ("unavailable", String::new()),
_ => (Unavailable, String::new()),
}
}
Some(Err(())) => ("unavailable", String::new()),
Some(Err(())) => (Unavailable, String::new()),
};
if status == "unavailable" {
if status == Unavailable {
tracing::warn!("Journal compaction snapshot unavailable; continuing without Journal");
}
use sha2::{Digest, Sha256};
@@ -143,12 +237,35 @@ pub(crate) fn freeze_compaction_journal(
.map(|byte| format!("{byte:02x}"))
.collect()
});
prompts.subjektiv_journal_compaction(
status,
&json,
snapshot_sha256.as_deref(),
workspace_id.as_deref(),
let body = prompts
.subjektiv_journal_compaction(
status.as_str(),
&json,
snapshot_sha256.as_deref(),
workspace_id.as_deref(),
)
.map_err(|e| e.to_string())?;
let payload = session_store::FeatureInputPayload::encode(
"journal:compaction",
1,
&JournalCompactionInput {
status,
prompt_provenance: session_store::PromptRenderProvenance {
workspace_id,
source_digest: prompts.projection().source_digest.clone(),
projection_digest: prompts.projection().catalog_digest.clone(),
logical_name: "internal.subjektiv_journal_compaction".into(),
},
snapshot: if status == Unavailable {
None
} else {
snapshot
},
snapshot_sha256,
},
)
.map_err(|e| e.to_string())?;
journal_input(body, payload, "journal_compaction")
}
#[derive(Clone)]
crates/worker/src/feature/builtin/memory.rs
+145 -47
View File
@@ -43,6 +43,80 @@ use crate::worker::{
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
};
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct SubjectBehaviorInput {
pub behavior_md: String,
pub prompt_provenance: Option<session_store::PromptRenderProvenance>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct ResidentSummaryInput {
prompt_provenance: Option<session_store::PromptRenderProvenance>,
}
fn resident_input(
body: String,
schema: &str,
payload: &impl serde::Serialize,
kind: &str,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> {
use session_store::*;
let input = SystemInputEnvelope::new(
body.clone(),
SystemInputMetadata::new(
SystemInputUsePolicy {
recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden,
compaction: InputCompactionUse::Retain,
},
Some(FeatureInputPayload::encode(schema, 1, payload)?),
Some(PublicSystemItem {
kind: kind.into(),
body,
presentation: PublicSystemPresentation::Context,
}),
)?,
)?;
Ok(LoggedSystemHistoryEntry::from_input(
LoggedSessionHistoryOrigin::Provided {
producer: "subjektiv.resident-context".into(),
},
None,
input,
))
}
pub(crate) fn subject_behavior_input(
body: String,
behavior_md: String,
prompt_provenance: Option<session_store::PromptRenderProvenance>,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> {
resident_input(
body,
"subject:behavior",
&SubjectBehaviorInput {
behavior_md,
prompt_provenance,
},
"subject_behavior_refresh",
)
}
pub(crate) fn resident_summary_input(
body: String,
prompt_provenance: Option<session_store::PromptRenderProvenance>,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> {
resident_input(
body,
"memory:resident-summary",
&ResidentSummaryInput { prompt_provenance },
"resident_summary_refresh",
)
}
#[derive(Clone, Debug)]
pub struct WorkspaceHttpMemoryBackend {
client: Arc<dyn WorkspaceClient>,
@@ -714,7 +788,7 @@ impl WorkspaceSubjektivResidentSummarySource {
fn render(
&self,
output: &server_api::SubjektivResidentContextOutput,
) -> Result<String, String> {
) -> Result<(String, session_store::PromptRenderProvenance), String> {
use memory::backend::MemoryResidentSummaryAvailability as Availability;
let availability = match output.memory_surface.availability {
Availability::Ready => "ready",
@@ -723,14 +797,24 @@ impl WorkspaceSubjektivResidentSummarySource {
Availability::Failed => "failed",
Availability::Unavailable => "unavailable",
};
self.prompts
.load()
let prompts = self.prompts.load_full();
let body = prompts
.subjektiv_resident_context(
&output.behavior_md,
availability,
output.memory_surface.content.as_deref(),
)
.map_err(|error| error.to_string())
.map_err(|error| error.to_string())?;
let projection = prompts.projection();
Ok((
body,
session_store::PromptRenderProvenance {
workspace_id: self.host.host.workspace_id(),
source_digest: projection.source_digest.clone(),
projection_digest: projection.catalog_digest.clone(),
logical_name: "internal.subjektiv_resident_context".into(),
},
))
}
fn remember_behavior(&self, output: &server_api::SubjektivResidentContextOutput) {
@@ -752,7 +836,7 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource
async fn load(&self) -> SystemPromptContribution {
match self.fetch().await {
Ok(output) => match self.render(&output) {
Ok(rendered) => {
Ok((rendered, _)) => {
self.remember_behavior(&output);
SystemPromptContribution::Ready(rendered)
}
@@ -788,20 +872,30 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource
if !changed {
return Ok(None);
}
let rendered = self.render(&output)?;
let (rendered, prompt_provenance) = self.render(&output)?;
Ok(Some(ResidentContextRefresh {
body: rendered,
behavior_md: output.behavior_md,
prompt_provenance: Some(prompt_provenance),
}))
}
fn system_item_committed(
&self,
item: &session_store::SystemItem,
_metadata: &crate::session_history::SessionHistoryMetadata,
_item: &session_store::SystemItem,
metadata: &crate::session_history::SessionHistoryMetadata,
) {
if let session_store::SystemItem::SubjectBehaviorRefresh { behavior_md, .. } = item {
self.confirm_resident_context_behavior(behavior_md);
if let Some(input) = metadata
.system_input
.as_ref()
.and_then(|input| input.feature_payload())
.and_then(|payload| {
payload
.decode::<SubjectBehaviorInput>("subject:behavior", 1)
.ok()
})
{
self.confirm_resident_context_behavior(&input.behavior_md);
}
}
@@ -954,7 +1048,13 @@ impl Hook<PreLlmRequest> for SubjektivBehaviorRefreshHook {
"Subject behavior refresh requires durable session append authority",
)
})?;
system_items.append_subject_behavior_refresh(refresh.body, refresh.behavior_md);
let input = subject_behavior_input(
refresh.body,
refresh.behavior_md,
refresh.prompt_provenance,
)
.map_err(|error| HookError::new(HookErrorCategory::Dependency, error.to_string()))?;
system_items.append_input(input);
}
Ok(HookPreRequestAction::Continue)
}
@@ -979,16 +1079,18 @@ impl Hook<WorkerRestored> for SubjektivResidentRestoreRefreshHook {
.resident_memory_restore_section(surface_ready, summary)
.map_err(|error| HookError::new(HookErrorCategory::Internal, error.to_string()))?;
let projection = prompts.projection();
context.system_items().append_resident_summary_refresh(
let input = resident_summary_input(
body,
session_store::PromptRenderProvenance {
Some(session_store::PromptRenderProvenance {
workspace_id: self.workspace_id.clone(),
source_digest: projection.source_digest.clone(),
projection_digest: projection.catalog_digest.clone(),
logical_name: WorkerPrompt::ResidentMemoryRestoreSection.key().to_string(),
},
);
}),
)
.map_err(|error| HookError::new(HookErrorCategory::Dependency, error.to_string()))?;
context.system_items().append_input(input);
Ok(())
}
}
@@ -1751,14 +1853,13 @@ permission = "write"
.expect("changed behavior must produce one refresh");
assert_eq!(refresh.behavior_md, "Ask when uncertain.");
assert!(refresh.body.contains("Ask when uncertain."));
source.system_item_committed(
&session_store::SystemItem::SubjectBehaviorRefresh {
body: refresh.body.clone(),
behavior_md: refresh.behavior_md.clone(),
prompt_provenance: None,
},
&session_store::LoggedSessionHistoryMetadata::legacy_unknown(),
);
let input = subject_behavior_input(
refresh.body.clone(),
refresh.behavior_md.clone(),
refresh.prompt_provenance.clone(),
)
.unwrap();
source.system_item_committed(&input.item, &input.metadata);
assert!(
source
.load_changed_resident_context()
@@ -1775,14 +1876,13 @@ permission = "write"
.expect("cleared behavior must produce one refresh");
assert_eq!(cleared.behavior_md, "");
assert!(cleared.body.contains("No user-managed behavior is set"));
source.system_item_committed(
&session_store::SystemItem::SubjectBehaviorRefresh {
body: cleared.body.clone(),
behavior_md: cleared.behavior_md.clone(),
prompt_provenance: None,
},
&session_store::LoggedSessionHistoryMetadata::legacy_unknown(),
);
let input = subject_behavior_input(
cleared.body.clone(),
cleared.behavior_md.clone(),
cleared.prompt_provenance.clone(),
)
.unwrap();
source.system_item_committed(&input.item, &input.metadata);
assert!(
source
.load_changed_resident_context()
@@ -1840,14 +1940,13 @@ permission = "write"
.await
.unwrap()
.unwrap();
source.system_item_committed(
&session_store::SystemItem::SubjectBehaviorRefresh {
body: first.body.clone(),
behavior_md: first.behavior_md.clone(),
prompt_provenance: None,
},
&session_store::LoggedSessionHistoryMetadata::legacy_unknown(),
);
let input = subject_behavior_input(
first.body.clone(),
first.behavior_md.clone(),
first.prompt_provenance.clone(),
)
.unwrap();
source.system_item_committed(&input.item, &input.metadata);
assert_eq!(
source
.load_changed_resident_context()
@@ -1856,14 +1955,13 @@ permission = "write"
.unwrap(),
second
);
source.system_item_committed(
&session_store::SystemItem::SubjectBehaviorRefresh {
body: second.body.clone(),
behavior_md: second.behavior_md.clone(),
prompt_provenance: None,
},
&session_store::LoggedSessionHistoryMetadata::legacy_unknown(),
);
let input = subject_behavior_input(
second.body.clone(),
second.behavior_md.clone(),
second.prompt_provenance.clone(),
)
.unwrap();
source.system_item_committed(&input.item, &input.metadata);
assert!(
source
.load_changed_resident_context()
crates/worker/src/feature/builtin/task/mod.rs
+6 -4
View File
@@ -276,7 +276,9 @@ mod tests {
use super::*;
use crate::hook::{PreRequestInfo, SystemItemAppendHandle};
fn pre_request_context(pending: Arc<Mutex<Vec<SystemItem>>>) -> PreRequestContext {
fn pre_request_context(
pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>,
) -> PreRequestContext {
PreRequestContext::new(
PreRequestInfo {
item_count: 1,
@@ -315,7 +317,7 @@ mod tests {
let queued = pending.lock().expect("pending queue poisoned");
assert_eq!(queued.len(), 1);
let SystemItem::TaskReminder { body, .. } = &queued[0] else {
let SystemItem::TaskReminder { body, .. } = &queued[0].item else {
panic!("unexpected system item: {:?}", queued[0]);
};
assert!(body.starts_with("Current session steps are listed below."));
@@ -339,7 +341,7 @@ mod tests {
}
let queued = pending.lock().expect("pending queue poisoned");
let SystemItem::TaskReminder { source, body, .. } = &queued[0] else {
let SystemItem::TaskReminder { source, body, .. } = &queued[0].item else {
panic!("unexpected system item: {:?}", queued[0]);
};
assert_eq!(*source, SystemReminderSource::TaskInactivity);
@@ -452,7 +454,7 @@ mod tests {
}
let queued = pending.lock().expect("pending queue poisoned");
assert_eq!(queued.len(), 1);
let SystemItem::TaskReminder { body, .. } = &queued[0] else {
let SystemItem::TaskReminder { body, .. } = &queued[0].item else {
panic!("unexpected system item: {:?}", queued[0]);
};
assert!(body.contains("taskid 1 (completed) done"));
crates/worker/src/feature/invocation.rs
+39
View File
@@ -39,6 +39,45 @@ const MAX_INVOCATION_DESCRIPTOR_BYTES: usize =
protocol::invocation::MAX_FEATURE_INVOCATION_DESCRIPTOR_BYTES;
const MAX_COMPLETION_RESPONSE_BYTES: usize = 64 * 1024;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct InvocationResultInput {
pub result: FeatureInvocationResult,
}
pub(crate) fn result_input(
result: FeatureInvocationResult,
body: String,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> {
use session_store::*;
let payload = FeatureInputPayload::encode(
"feature:invocation-result",
1,
&InvocationResultInput { result },
)?;
let metadata = SystemInputMetadata::new(
SystemInputUsePolicy {
recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden,
compaction: InputCompactionUse::Retain,
},
Some(payload),
Some(PublicSystemItem {
kind: "feature_invocation_result".into(),
body: body.clone(),
presentation: PublicSystemPresentation::Context,
}),
)?;
Ok(LoggedSystemHistoryEntry::from_input(
LoggedSessionHistoryOrigin::Provided {
producer: "feature.invocation".into(),
},
None,
SystemInputEnvelope::new(body, metadata)?,
))
}
pub(crate) fn invocation_result_is_bounded(result: &FeatureInvocationResult) -> bool {
result.message.len() <= MAX_INVOCATION_MESSAGE_BYTES
&& result
crates/worker/src/hook.rs
+37 -43
View File
@@ -25,7 +25,9 @@ use agen::tool::{ToolOutput, ToolResult};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use session_store::{SystemItem, SystemReminder};
#[cfg(test)]
use session_store::SystemItem;
use session_store::SystemReminder;
use thiserror::Error;
use crate::SessionEntryRef;
@@ -254,7 +256,7 @@ impl From<HookTurnEndAction> for TurnEndAction {
// Hook context handles
// =============================================================================
/// Host-created handle for appending approved durable [`SystemItem`] requests.
/// Host-created handle for appending approved durable [`session_store::SystemItem`] requests.
///
/// Hook code can use this handle only when the Worker host includes it in an
/// event-specific context. The handle queues typed requests; the host drains the
@@ -263,11 +265,11 @@ impl From<HookTurnEndAction> for TurnEndAction {
/// raw `agen::Item`, history writer, event sender, `Worker`, `Engine`, or
/// notification buffer.
pub struct SystemItemAppendHandle {
pending: Arc<Mutex<Vec<SystemItem>>>,
pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>,
}
impl SystemItemAppendHandle {
pub(crate) fn new(pending: Arc<Mutex<Vec<SystemItem>>>) -> Self {
pub(crate) fn new(pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>) -> Self {
Self { pending }
}
@@ -279,53 +281,44 @@ impl SystemItemAppendHandle {
self.pending
.lock()
.expect("system-item append queue poisoned")
.push(item);
.push(session_store::LoggedSystemHistoryEntry {
item,
metadata: crate::session_history::metadata(
session_store::LoggedSessionHistoryOrigin::BackendInstruction {
operation_id: None,
},
None,
),
});
}
/// Queue a Subject behavior refresh with its exact represented behavior. The caller receives this
/// narrow operation only at a pre-request boundary, and the host commits it
/// before exposing it to the model.
pub fn append_subject_behavior_refresh(
&self,
body: impl Into<String>,
behavior_md: impl Into<String>,
) {
/// Queue producer-validated System context. Metadata and body are committed together.
/// The Host exposes this capability only at a pre-request boundary.
pub fn append_input(&self, input: session_store::LoggedSystemHistoryEntry) {
self.pending
.lock()
.expect("system-item append queue poisoned")
.push(SystemItem::SubjectBehaviorRefresh {
body: body.into(),
behavior_md: behavior_md.into(),
prompt_provenance: None,
});
.push(input);
}
}
/// Host-created narrow append authority exposed only at a restore lifecycle
/// boundary. It cannot emit arbitrary system items.
/// boundary. No model-facing tool receives this capability.
pub struct RestoreSystemItemAppendHandle {
pending: Arc<Mutex<Vec<SystemItem>>>,
pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>,
}
impl RestoreSystemItemAppendHandle {
pub(crate) fn new(pending: Arc<Mutex<Vec<SystemItem>>>) -> Self {
pub(crate) fn new(pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>) -> Self {
Self { pending }
}
/// Queue a Feature-rendered resident-summary refresh for durable append at
/// the restore lifecycle boundary.
pub fn append_resident_summary_refresh(
&self,
body: impl Into<String>,
prompt_provenance: session_store::PromptRenderProvenance,
) {
/// Queue producer-validated System context at the Host restore boundary.
pub fn append_input(&self, input: session_store::LoggedSystemHistoryEntry) {
self.pending
.lock()
.expect("restore system-item append queue poisoned")
.push(SystemItem::ResidentSummaryRefresh {
body: body.into(),
prompt_provenance: Some(prompt_provenance),
});
.push(input);
}
}
@@ -983,7 +976,7 @@ mod tests {
let queued = pending.lock().expect("pending queue poisoned");
assert_eq!(queued.len(), 1);
match &queued[0] {
match &queued[0].item {
SystemItem::TaskReminder { body, .. } => {
assert_eq!(body, "remember tasks");
}
@@ -992,7 +985,7 @@ mod tests {
}
#[test]
fn restore_append_handle_queues_only_resident_refresh_items() {
fn restore_append_handle_preserves_producer_metadata() {
let pending = Arc::new(Mutex::new(Vec::new()));
let handle = RestoreSystemItemAppendHandle::new(Arc::clone(&pending));
let provenance = session_store::PromptRenderProvenance {
@@ -1003,15 +996,16 @@ mod tests {
logical_name: "internal.resident_memory_restore_section".into(),
};
handle.append_resident_summary_refresh("current surface", provenance.clone());
assert!(matches!(
pending.lock().unwrap().as_slice(),
[SystemItem::ResidentSummaryRefresh {
body,
prompt_provenance: Some(actual),
}] if body == "current surface" && actual == &provenance
));
let input = crate::feature::builtin::memory::resident_summary_input(
"current surface".into(),
Some(provenance),
)
.unwrap();
let expected = serde_json::to_value(&input).unwrap();
handle.append_input(input);
let queued = pending.lock().unwrap();
assert_eq!(queued.len(), 1);
assert_eq!(serde_json::to_value(&queued[0]).unwrap(), expected);
}
#[test]
crates/worker/src/ipc/interceptor.rs
+65 -25
View File
@@ -262,6 +262,31 @@ impl WorkerInterceptor {
)
}
fn commit_system_entries(
&self,
entries: &[session_store::LoggedSystemHistoryEntry],
) -> Result<(), session_store::StoreError> {
let Some(writer) = &self.log_writer else {
return Ok(());
};
for entry in entries {
writer.commit_annotated_system_item_then(
entry.clone(),
Vec::new(),
&mut |history| {
self.pending_committed_history
.lock()
.expect("pending committed history poisoned")
.push_back(history.clone());
if let Some(source) = &self.resident_context_source {
source.system_item_committed(&entry.item, &history.annotation);
}
},
)?;
}
Ok(())
}
fn current_turn_index(&self) -> usize {
self.next_turn_index
.load(Ordering::Relaxed)
@@ -387,11 +412,6 @@ impl WorkerInterceptor {
} if prompt_provenance.is_none() => {
*prompt_provenance = Some(provenance("internal.task_reminder"));
}
SystemItem::SubjectBehaviorRefresh {
prompt_provenance, ..
} if prompt_provenance.is_none() => {
*prompt_provenance = Some(provenance("internal.subjektiv_resident_context"));
}
SystemItem::Interrupt {
prompt_provenance, ..
} if prompt_provenance.is_none() => {
@@ -617,15 +637,17 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
return Ok(PreRequestAction::Yield);
}
let mut system_items: Vec<SystemItem> = std::mem::take(
let mut system_items: Vec<session_store::LoggedSystemHistoryEntry> = std::mem::take(
&mut *pending_hook_system_items
.lock()
.expect("pending hook system-item queue poisoned"),
);
self.attach_prompt_provenance(&mut system_items);
for entry in &mut system_items {
self.attach_prompt_provenance(std::slice::from_mut(&mut entry.item));
}
let mut appended_items: Vec<Item> = system_items
.iter()
.map(SystemItem::to_history_item)
.map(|entry| entry.item.to_history_item())
.collect();
let effective_context = if appended_items.is_empty() {
Cow::Borrowed(context.as_slice())
@@ -639,7 +661,7 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
let compact_decision =
self.request_compact_decision(current_tokens, effective_context.as_ref());
if !matches!(compact_decision, AutomaticCompactDecision::Continue) {
if let Err(error) = self.commit_system_items(&system_items) {
if let Err(error) = self.commit_system_entries(&system_items) {
return Ok(PreRequestAction::Cancel(format!(
"session persistence failed: {error}"
)));
@@ -683,7 +705,7 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
if appended_items.is_empty() {
return Ok(PreRequestAction::Continue);
}
Ok(match self.commit_system_items(&system_items) {
Ok(match self.commit_system_entries(&system_items) {
Ok(()) => PreRequestAction::ContinueWith(appended_items),
Err(error) => PreRequestAction::Cancel(format!("session persistence failed: {error}")),
})
@@ -983,7 +1005,14 @@ mod tests {
input
.system_items()
.expect("test interceptor has durable append authority")
.append_subject_behavior_refresh("current Subject context", "Ask when uncertain.");
.append_input(
crate::feature::builtin::memory::subject_behavior_input(
"current Subject context".into(),
"Ask when uncertain.".into(),
None,
)
.unwrap(),
);
Ok(HookPreRequestAction::Continue)
}
}
@@ -1000,11 +1029,23 @@ mod tests {
fn system_item_committed(
&self,
item: &SystemItem,
_metadata: &crate::session_history::SessionHistoryMetadata,
_item: &SystemItem,
metadata: &crate::session_history::SessionHistoryMetadata,
) {
if let SystemItem::SubjectBehaviorRefresh { behavior_md, .. } = item {
self.confirmed.lock().unwrap().push(behavior_md.to_owned());
if let Some(input) = metadata
.system_input
.as_ref()
.and_then(|input| input.feature_payload())
.and_then(|payload| {
payload
.decode::<crate::feature::builtin::memory::SubjectBehaviorInput>(
"subject:behavior",
1,
)
.ok()
})
{
self.confirmed.lock().unwrap().push(input.behavior_md);
}
}
}
@@ -1549,11 +1590,7 @@ mod tests {
assert_eq!(*confirmed.lock().unwrap(), ["Ask when uncertain."]);
assert!(matches!(
committed.lock().unwrap().as_slice(),
[SystemItem::SubjectBehaviorRefresh {
behavior_md,
prompt_provenance: Some(_),
..
}] if behavior_md == "Ask when uncertain."
[SystemItem::Input { body }] if body == "current Subject context"
));
}
@@ -1654,12 +1691,15 @@ mod tests {
})),
)
.with_resident_context_source(Some(source));
let items = ["raw behavior", ""].map(|behavior_md| SystemItem::SubjectBehaviorRefresh {
body: "same rendered context".into(),
behavior_md: behavior_md.into(),
prompt_provenance: None,
let items = ["raw behavior", ""].map(|behavior_md| {
crate::feature::builtin::memory::subject_behavior_input(
"same rendered context".into(),
behavior_md.into(),
None,
)
.unwrap()
});
interceptor.commit_system_items(&items).unwrap();
interceptor.commit_system_entries(&items).unwrap();
assert_eq!(*confirmed.lock().unwrap(), ["raw behavior", ""]);
assert_eq!(committed.lock().unwrap().len(), 2);
}
crates/worker/src/ipc/protocol_session.rs
+18 -30
View File
@@ -340,22 +340,13 @@ mod tests {
#[test]
fn system_item_log_entry_keeps_identity_in_live_event() {
let entry_id = session_store::LoggedSessionHistoryEntryId::new();
let mut input =
crate::feature::builtin::memory::resident_summary_input("refresh".into(), None)
.unwrap();
input.metadata.entry_id = entry_id.clone();
let event = live_log_entry_event(LogEntry::AnnotatedSystemItem {
ts: session_store::segment_log::now_millis(),
entry: session_store::LoggedSystemHistoryEntry {
item: session_store::SystemItem::ResidentSummaryRefresh {
body: "refresh".to_string(),
prompt_provenance: None,
},
metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None,
entry_id: entry_id.clone(),
origin: session_store::LoggedSessionHistoryOrigin::BackendInstruction {
operation_id: None,
},
derivation: None,
},
},
entry: input,
extensions: Vec::new(),
})
.expect("SystemItem must be live-relevant");
@@ -381,25 +372,22 @@ mod tests {
logical_name: "private-template".into(),
};
let items = [
session_store::SystemItem::SubjectBehaviorRefresh {
body: "displayed context".into(),
behavior_md: "private-raw-behavior".into(),
prompt_provenance: Some(provenance.clone()),
},
session_store::SystemItem::JournalStartup {
body: "displayed reference".into(),
snapshot: serde_json::json!({ "internal": "private-snapshot" }),
snapshot_sha256: "private-snapshot-digest".into(),
prompt_provenance: provenance,
},
crate::feature::builtin::memory::subject_behavior_input(
"displayed context".into(),
"private-raw-behavior".into(),
Some(provenance.clone()),
)
.unwrap(),
crate::feature::builtin::memory::resident_summary_input(
"displayed reference".into(),
Some(provenance),
)
.unwrap(),
];
for item in items {
for input in items {
let entry = LogEntry::AnnotatedSystemItem {
ts: 7,
entry: session_store::LoggedSystemHistoryEntry {
item,
metadata: session_store::LoggedSessionHistoryMetadata::legacy_unknown(),
},
entry: input,
extensions: vec![],
};
let persisted: LogEntry =
crates/worker/src/session_history.rs
+14 -16
View File
@@ -67,6 +67,16 @@ pub(crate) fn metadata(
}
}
pub(crate) fn compaction_use(
metadata: &SessionHistoryMetadata,
) -> session_store::InputCompactionUse {
metadata
.system_input
.as_ref()
.map(|input| input.usage().compaction)
.unwrap_or(session_store::InputCompactionUse::Retain)
}
#[cfg(test)]
pub(crate) fn history_entry(
item: Item,
@@ -134,22 +144,10 @@ pub(crate) fn restore_history_entries(
}
*target = from_logged(entry);
}
LogEntry::AnnotatedSystemItem { entry, .. } => {
history.push(HistoryEntry::new(entry.item.to_history_item(), {
let mut metadata = entry.metadata.clone();
// Old typed startup records used generic placeholder origins.
// Preserve explicitly supplied metadata on current contributions.
if matches!(
metadata.origin,
WorkerHistoryProvenance::DerivedSummary
| WorkerHistoryProvenance::LegacyUnknown
) && matches!(entry.item, session_store::SystemItem::JournalStartup { .. })
{
metadata.origin = WorkerHistoryProvenance::JournalContext;
}
metadata
}))
}
LogEntry::AnnotatedSystemItem { entry, .. } => history.push(HistoryEntry::new(
entry.item.to_history_item(),
entry.metadata.clone(),
)),
_ => {}
}
}
crates/worker/src/skill.rs
+44
View File
@@ -8,6 +8,50 @@ pub use server_api::{
SkillProjectionStatus, SkillProvenance, SkillResourceRef, SkillSourceKind,
};
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct SkillActivationInput {
pub name: String,
}
pub(crate) fn activation_input(
activation: &SkillActivationResponse,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> {
use session_store::*;
let body = format!(
"Agent Skill `{}` activated from {}.\n\n{}",
activation.name, activation.provenance.id, activation.body
);
let payload = FeatureInputPayload::encode(
"skill:activation",
1,
&SkillActivationInput {
name: activation.name.clone(),
},
)?;
let metadata = SystemInputMetadata::new(
SystemInputUsePolicy {
recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden,
compaction: InputCompactionUse::Retain,
},
Some(payload),
Some(PublicSystemItem {
kind: "skill_activation".into(),
body: body.clone(),
presentation: PublicSystemPresentation::Context,
}),
)?;
Ok(LoggedSystemHistoryEntry::from_input(
LoggedSessionHistoryOrigin::Provided {
producer: "skill.activation".into(),
},
None,
SystemInputEnvelope::new(body, metadata)?,
))
}
#[derive(Debug, thiserror::Error)]
pub enum SkillClientError {
#[error("workspace client is unavailable: {0}")]
crates/worker/src/subjektiv.rs
+53 -13
View File
@@ -26,6 +26,30 @@ use server_api::*;
use crate::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceServerOperation};
impl<C: agen::llm_client::client::LlmClient + 'static, St: session_store::Store>
crate::Worker<C, St>
{
/// Bind live Host capability and its context provider, not authority inherited
/// from a previous binding. Ordinary Feature/lifecycle gates also apply to embedders.
pub fn bind_subjektiv_host(&mut self, host: Option<Arc<dyn SubjektivHost>>) {
self.set_subjektiv_host(host);
if self.manifest_lifecycle_features_enabled()
&& self.backend_job().is_none()
&& crate::feature::builtin::memory::ordinary_subjektiv_features_enabled(self.manifest())
&& let Ok(Some(connection)) = self.subjektiv_host_connection()
&& (connection.host.journal_startup_granted()
|| connection.host.journal_compaction_granted())
{
self.install_session_context_source(Arc::new(
crate::feature::builtin::journal::JournalContextSource::new(
connection,
self.prompts(),
),
));
}
}
}
/// Host-issued execution context, never deserialized from model arguments.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SubjektivHostContext {
@@ -1130,7 +1154,12 @@ enabled = true
let initial = store.read_all(session, segment).unwrap();
let journal = initial.iter().find_map(|entry| match entry {
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::JournalStartup { .. }) =>
if entry
.metadata
.system_input
.as_ref()
.and_then(|input| input.feature_payload())
.is_some_and(|payload| payload.schema() == "journal:startup") =>
{
Some(entry)
}
@@ -1140,22 +1169,33 @@ enabled = true
let body = journal.map(|journal| {
assert_eq!(
journal.metadata.origin,
LoggedSessionHistoryOrigin::JournalContext
LoggedSessionHistoryOrigin::Provided {
producer: "subjektiv.journal".into()
}
);
let SystemItem::JournalStartup {
body,
snapshot: saved,
snapshot_sha256,
prompt_provenance,
} = &journal.item
else {
let SystemItem::Input { body } = &journal.item else {
unreachable!()
};
assert_eq!(*saved, serde_json::to_value(&snapshot).unwrap());
let payload = journal
.metadata
.system_input
.as_ref()
.unwrap()
.feature_payload()
.unwrap()
.decode::<crate::feature::builtin::journal::JournalStartupInput>(
"journal:startup",
1,
)
.unwrap();
let saved = serde_json::to_value(&payload.snapshot).unwrap();
let snapshot_sha256 = &payload.snapshot_sha256;
let prompt_provenance = &payload.prompt_provenance;
assert_eq!(saved, serde_json::to_value(&snapshot).unwrap());
use sha2::{Digest, Sha256};
assert_eq!(
*snapshot_sha256,
Sha256::digest(serde_json::to_vec(saved).unwrap())
Sha256::digest(serde_json::to_vec(&saved).unwrap())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>()
@@ -1168,7 +1208,7 @@ enabled = true
body.trim_end(),
format!(
"## Prior Journal context\n\n```json\n{}\n```",
serde_json::to_string(saved).unwrap()
serde_json::to_string(&saved).unwrap()
)
);
body.as_str()
@@ -1237,7 +1277,7 @@ enabled = true
}
assert_eq!(host.startup_calls.load(Ordering::SeqCst), 1);
let logs = store.read_all(session, segment).unwrap();
assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if matches!(entry.item, SystemItem::JournalStartup { .. }))).count(), usize::from(contributes_context));
assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if entry.metadata.system_input.as_ref().and_then(|input| input.feature_payload()).is_some_and(|payload| payload.schema() == "journal:startup"))).count(), usize::from(contributes_context));
controller
.handle
.send(Method::Shutdown {
crates/worker/src/worker.rs
+229 -165
View File
@@ -2757,11 +2757,25 @@ pub trait SystemItemCommitter: Send + Sync {
.unwrap_or(WorkerHistoryProvenance::BackendInstruction { operation_id: None }),
None,
);
let history = HistoryEntry::new(item.to_history_item(), metadata.clone());
self.commit_annotated_system_item_then(
session_store::LoggedSystemHistoryEntry { item, metadata },
extensions,
after_commit,
)
}
fn commit_annotated_system_item_then(
&self,
entry: session_store::LoggedSystemHistoryEntry,
extensions: Vec<SessionExtension>,
after_commit: &mut dyn FnMut(&HistoryEntry<SessionHistoryMetadata>),
) -> Result<HistoryEntry<SessionHistoryMetadata>, StoreError> {
entry.validate_system_input()?;
let history = HistoryEntry::new(entry.item.to_history_item(), entry.metadata.clone());
self.commit_usage_boundary(
LogEntry::AnnotatedSystemItem {
ts: segment_log::now_millis(),
entry: session_store::LoggedSystemHistoryEntry { item, metadata },
entry,
extensions,
},
&mut || after_commit(&history),
@@ -2934,14 +2948,21 @@ pub(crate) enum SystemPromptContribution {
pub(crate) struct ResidentContextRefresh {
pub body: String,
pub behavior_md: String,
pub prompt_provenance: Option<session_store::PromptRenderProvenance>,
}
/// Initial-only System messages and Feature-owned metadata. Core persists and
/// replays each pair unchanged; it does not infer provenance from the payload.
/// Prepared with the head, never at restore/request hooks.
/// Feature-owned System context at explicit creation and compaction boundaries.
/// Core persists and replays each pair unchanged; normal restore never reloads it.
#[async_trait::async_trait]
pub(crate) trait InitialContextSource: Send + Sync {
pub(crate) trait SessionContextSource: Send + Sync {
async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String>;
async fn load_compaction_context(
&self,
_session: SessionId,
) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
Ok(Vec::new())
}
}
#[async_trait::async_trait]
@@ -3132,7 +3153,7 @@ pub struct Worker<C: LlmClient, St: Store> {
inject_resident_summary: bool,
/// Deferred resident prompt source installed by an enabled Feature.
feature_resident_summary_source: Option<Arc<dyn SystemPromptContributionSource>>,
initial_context_source: Option<Arc<dyn InitialContextSource>>,
session_context_source: Option<Arc<dyn SessionContextSource>>,
/// 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>>,
@@ -3404,7 +3425,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
prompts,
inject_resident_summary: true,
feature_resident_summary_source: None,
initial_context_source: None,
session_context_source: None,
prepared_initial_seed: None,
subjektiv_host: None,
restore_feature_lifecycle_pending: false,
@@ -3452,8 +3473,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
self.inject_resident_summary = enabled;
}
pub(crate) fn install_initial_context_source(&mut self, source: Arc<dyn InitialContextSource>) {
self.initial_context_source = Some(source);
pub(crate) fn install_session_context_source(&mut self, source: Arc<dyn SessionContextSource>) {
self.session_context_source = Some(source);
}
pub(crate) fn install_system_prompt_contribution(
@@ -3629,10 +3650,12 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
self.workspace_context.client_handle()
}
/// Bind a live capability before Feature installation. This is deliberately
/// separate from Workspace authority and is not inherited by SubWorkers.
pub fn bind_subjektiv_host(&mut self, host: Option<Arc<dyn crate::subjektiv::SubjektivHost>>) {
pub(crate) fn set_subjektiv_host(
&mut self,
host: Option<Arc<dyn crate::subjektiv::SubjektivHost>>,
) {
self.subjektiv_host = host;
self.session_context_source = None;
}
pub fn subjektiv_host_connection(
@@ -3700,26 +3723,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
{
let activation = self.workspace_client().activate_skill(name)?;
self.ensure_segment_head().await?;
let body = format!(
"Agent Skill `{}` activated from {}.\n\n{}",
activation.name, activation.provenance.id, activation.body
);
let skill_metadata = new_history_metadata(
WorkerHistoryProvenance::BackendInstruction { operation_id: None },
None,
);
let input = crate::skill::activation_input(&activation).map_err(StoreError::from)?;
let history_entry = HistoryEntry::new(input.item.to_history_item(), input.metadata.clone());
self.commit_entry(LogEntry::AnnotatedSystemItem {
ts: segment_log::now_millis(),
entry: session_store::LoggedSystemHistoryEntry {
item: SystemItem::SkillActivation {
name: activation.name.clone(),
body: body.clone(),
},
metadata: skill_metadata.clone(),
},
entry: input,
extensions: Vec::new(),
})?;
let history_entry = HistoryEntry::new(agen::Item::system_message(body), skill_metadata);
let mut annotate = history_annotator(
self.log_writer_handle(),
vec![history_entry.clone()],
@@ -4594,18 +4604,12 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.lock()
.expect("restore system-item queue poisoned"),
);
for item in system_items {
let history_item = item.to_history_item();
let metadata = new_history_metadata(
WorkerHistoryProvenance::BackendInstruction { operation_id: None },
None,
);
for entry in system_items {
let history_item = entry.item.to_history_item();
let metadata = entry.metadata.clone();
self.commit_entry(LogEntry::AnnotatedSystemItem {
ts: segment_log::now_millis(),
entry: session_store::LoggedSystemHistoryEntry {
item,
metadata: metadata.clone(),
},
entry,
extensions: Vec::new(),
})?;
let history_entry = HistoryEntry::new(history_item, metadata);
@@ -5368,17 +5372,14 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.invocation_receipts
.push(InvocationReceipt::new(invocation, Some(result.clone())));
}
let item = SystemItem::FeatureInvocationResult { result, body };
let metadata = new_history_metadata(
WorkerHistoryProvenance::BackendInstruction { operation_id: None },
None,
);
let history = HistoryEntry::new(item.to_history_item(), metadata.clone());
let input =
crate::feature::invocation::result_input(result, body).map_err(StoreError::from)?;
let history = HistoryEntry::new(input.item.to_history_item(), input.metadata.clone());
handle
.writer
.append_entry_locked(LogEntry::AnnotatedSystemItem {
ts: segment_log::now_millis(),
entry: session_store::LoggedSystemHistoryEntry { item, metadata },
entry: input,
extensions: vec![pending_activation_extension(&completed)],
})?;
*state = completed;
@@ -5882,7 +5883,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
} else {
if self.prepared_initial_seed.is_none() {
self.ensure_system_prompt_materialized().await?;
let items = match &self.initial_context_source {
let items = match &self.session_context_source {
Some(source) => source.load().await.map_err(WorkerError::InvalidState)?,
None => Vec::new(),
};
@@ -6939,11 +6940,18 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// Freeze one Host-owned read before starting the compactor. Background recording
// remains independent; never re-read when building the replacement segment.
let journal_context = crate::feature::builtin::journal::freeze_compaction_journal(
self.subjektiv_host_connection(),
&self.prompts.load_full(),
)
.map_err(WorkerError::PromptCatalog)?;
let frozen_context = match &self.session_context_source {
Some(source) => source
.load_compaction_context(self.session_id())
.await
.map_err(WorkerError::InvalidState)?,
None => Vec::new(),
};
for entry in &frozen_context {
entry
.validate_system_input()
.map_err(|e| WorkerError::InvalidState(e.to_string()))?;
}
let history_entries = self.session.history().entries();
let retain_from = cut.index.min(history_entries.len());
let initial_summary = initial_compaction_summary(history_entries)?;
@@ -6953,11 +6961,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
let retained_history_entries = history_entries[retain_from..]
.iter()
.filter(|entry| {
!matches!(
entry.annotation.origin,
WorkerHistoryProvenance::CompactionSummary
| WorkerHistoryProvenance::JournalContext
)
crate::session_history::compaction_use(&entry.annotation)
== session_store::InputCompactionUse::Retain
})
.cloned()
.collect::<Vec<_>>();
@@ -6976,11 +6981,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
let mut entries_to_summarise = history_entries[..retain_from]
.iter()
.filter(|entry| {
!matches!(
entry.annotation.origin,
WorkerHistoryProvenance::CompactionSummary
| WorkerHistoryProvenance::JournalContext
)
crate::session_history::compaction_use(&entry.annotation)
== session_store::InputCompactionUse::Retain
})
.cloned()
.collect::<Vec<_>>();
@@ -7067,24 +7069,29 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// Journals. In particular, a failed/empty read must not silently lose context
// that the previous summary intentionally delegated to its injected Journal.
// Carry them only into the compactor, never alongside the new reference seed.
for entry in history_entries
.iter()
.filter(|entry| entry.annotation.origin == WorkerHistoryProvenance::JournalContext)
{
for entry in history_entries.iter().filter(|entry| {
crate::session_history::compaction_use(&entry.annotation)
== session_store::InputCompactionUse::ReferenceInput
}) {
if let Some(body) = entry.item.as_text() {
summary_input.text.push_str(
"\n\n## Previous Journal reference (historical, not automatically retained)\n",
"\n\n## Previous reference input (historical, not automatically retained)\n",
);
summary_input.text.push_str(body);
summary_input
.text
.push_str("\n## End previous Journal reference\n");
.push_str("\n## End previous reference input\n");
}
}
// Mandatory context, outside overview truncation. The internal Worker's normal
// request budget checks include this exact body and the full previous summary.
summary_input.text.push_str("\n\n");
summary_input.text.push_str(&journal_context);
for (index, entry) in frozen_context.iter().enumerate() {
if index != 0 {
summary_input.text.push_str("\n\n");
}
summary_input.text.push_str(&entry.item.history_text());
}
if summary_input.warning_exceeded {
self.alert(
AlertLevel::Warn,
@@ -7400,8 +7407,11 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
crate::session_history::COMPACTION_SUMMARY_HEADER
));
compact_introduced_system_messages.push(summary_message.clone());
let journal_message = Item::system_message(journal_context);
compact_introduced_system_messages.push(journal_message.clone());
let context_messages = frozen_context
.iter()
.map(|entry| entry.item.to_history_item())
.collect::<Vec<_>>();
compact_introduced_system_messages.extend(context_messages.iter().cloned());
compact_introduced_system_messages.extend(auto_read_messages.iter().cloned());
if let Some(msg) = reference_message.as_ref() {
compact_introduced_system_messages.push(msg.clone());
@@ -7414,7 +7424,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
compact_introduced_system_messages.push(task_snapshot_message.clone());
new_history.push(summary_message);
new_history.push(journal_message);
new_history.extend(context_messages);
new_history.extend(auto_read_messages);
if let Some(msg) = reference_message {
new_history.push(msg);
@@ -7454,6 +7464,32 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.iter()
.map(|entry| entry.annotation.entry_id.clone())
.collect::<Vec<_>>();
let generic_summary_input = session_store::SystemInputEnvelope::new(
new_history[0]
.as_text()
.expect("compaction summary is text"),
session_store::SystemInputMetadata::new(
session_store::SystemInputUsePolicy {
recording_material: session_store::InputEvidenceUse::Exclude,
memory_evidence: session_store::InputEvidenceUse::Exclude,
observation: session_store::InputObservationUse::Hidden,
compaction: session_store::InputCompactionUse::OpeningSummary,
},
None,
None,
)
.map_err(|e| WorkerError::InvalidState(e.to_string()))?,
)
.map_err(|e| WorkerError::InvalidState(e.to_string()))?;
let summary_entry = session_store::LoggedSystemHistoryEntry::from_input(
WorkerHistoryProvenance::Provided {
producer: "worker.compaction".into(),
},
Some(SessionHistoryDerivation {
sources: derived_sources.clone(),
}),
generic_summary_input,
);
let mut original_cursor = 0usize;
let compacted_history_entries = new_history
.iter()
@@ -7461,25 +7497,9 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.enumerate()
.map(|(index, item)| {
if index == 0 {
HistoryEntry::new(
item,
new_history_metadata(
WorkerHistoryProvenance::CompactionSummary,
Some(SessionHistoryDerivation {
sources: derived_sources.clone(),
}),
),
)
} else if index == 1 {
HistoryEntry::new(
item,
new_history_metadata(
WorkerHistoryProvenance::JournalContext,
Some(SessionHistoryDerivation {
sources: derived_sources.clone(),
}),
),
)
HistoryEntry::new(item, summary_entry.metadata.clone())
} else if let Some(context) = frozen_context.get(index - 1) {
HistoryEntry::new(item, context.metadata.clone())
} else if let Some((offset, original)) = original_entries[original_cursor..]
.iter()
.enumerate()
@@ -7977,7 +7997,7 @@ where
prompts: common.prompts,
inject_resident_summary: true,
feature_resident_summary_source: None,
initial_context_source: None,
session_context_source: None,
prepared_initial_seed: None,
subjektiv_host: None,
restore_feature_lifecycle_pending: false,
@@ -8358,7 +8378,7 @@ where
prompts: common.prompts,
inject_resident_summary: true,
feature_resident_summary_source: None,
initial_context_source: None,
session_context_source: None,
prepared_initial_seed: None,
subjektiv_host: None,
restore_feature_lifecycle_pending: false,
@@ -8490,7 +8510,7 @@ where
prompts: common.prompts,
inject_resident_summary: true,
feature_resident_summary_source: None,
initial_context_source: None,
session_context_source: None,
prepared_initial_seed: None,
subjektiv_host: None,
restore_feature_lifecycle_pending: false,
@@ -8901,7 +8921,7 @@ where
prompts: common.prompts,
inject_resident_summary: true,
feature_resident_summary_source: None,
initial_context_source: None,
session_context_source: None,
prepared_initial_seed: None,
subjektiv_host: None,
restore_feature_lifecycle_pending: true,
@@ -9296,9 +9316,10 @@ fn summary_shrink_prompt(
fn initial_compaction_summary(
entries: &[HistoryEntry<SessionHistoryMetadata>],
) -> Result<Option<String>, WorkerError> {
let mut summaries = entries
.iter()
.filter(|entry| entry.annotation.origin == WorkerHistoryProvenance::CompactionSummary);
let mut summaries = entries.iter().filter(|entry| {
crate::session_history::compaction_use(&entry.annotation)
== session_store::InputCompactionUse::OpeningSummary
});
let Some(entry) = summaries.next() else {
return Ok(None);
};
@@ -10739,13 +10760,32 @@ mod build_summary_prompt_tests {
#[test]
fn malformed_or_duplicate_host_summaries_fail_instead_of_omitting_context() {
use crate::session_history::history_entry;
let valid = history_entry(
Item::system_message(format!(
"{}body",
crate::session_history::COMPACTION_SUMMARY_HEADER
)),
WorkerHistoryProvenance::CompactionSummary,
);
let opening = |item| {
let mut entry = history_entry(
item,
WorkerHistoryProvenance::Provided {
producer: "arbitrary-summary-source".into(),
},
);
entry.annotation.system_input = Some(
session_store::SystemInputMetadata::new(
session_store::SystemInputUsePolicy {
recording_material: session_store::InputEvidenceUse::Exclude,
memory_evidence: session_store::InputEvidenceUse::Exclude,
observation: session_store::InputObservationUse::Hidden,
compaction: session_store::InputCompactionUse::OpeningSummary,
},
None,
None,
)
.unwrap(),
);
entry
};
let valid = opening(Item::system_message(format!(
"{}body",
crate::session_history::COMPACTION_SUMMARY_HEADER
)));
assert_eq!(
initial_compaction_summary(std::slice::from_ref(&valid))
.unwrap()
@@ -10757,13 +10797,7 @@ mod build_summary_prompt_tests {
Item::user_message("pasted"),
Item::system_message("missing marker"),
] {
assert!(
initial_compaction_summary(&[history_entry(
item,
WorkerHistoryProvenance::CompactionSummary
)])
.is_err()
);
assert!(initial_compaction_summary(&[opening(item)]).is_err());
}
}
@@ -11190,7 +11224,7 @@ mod build_summary_prompt_tests {
1
}
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::FeatureInvocationResult { .. }) =>
if input_schema(entry) == Some("feature:invocation-result") =>
{
2
}
@@ -14162,13 +14196,7 @@ permission = "write"
let entries = store.read_all(session_id, segment_id).unwrap();
assert!(matches!(
entries.last(),
Some(LogEntry::AnnotatedSystemItem {
entry: session_store::LoggedSystemHistoryEntry {
item: SystemItem::ResidentSummaryRefresh { .. },
..
},
..
})
Some(LogEntry::AnnotatedSystemItem { entry, .. }) if input_schema(entry) == Some("memory:resident-summary")
));
}
}
@@ -14465,15 +14493,10 @@ permission = "write"
assert!(entries.iter().any(|entry| {
matches!(
entry,
LogEntry::AnnotatedSystemItem {
entry: session_store::LoggedSystemHistoryEntry {
item: SystemItem::SkillActivation { name, body },
..
},
..
} if name == "triage-errors"
&& body.contains("# Triage Errors")
&& body == history_text
LogEntry::AnnotatedSystemItem { entry, .. }
if entry.metadata.system_input.as_ref().and_then(|input| input.feature_payload()).and_then(|payload| payload.decode::<crate::skill::SkillActivationInput>("skill:activation", 1).ok()).is_some_and(|payload| payload.name == "triage-errors")
&& entry.item.history_text().contains("# Triage Errors")
&& entry.item.history_text() == history_text
)
}));
}
@@ -15867,7 +15890,7 @@ permission = "write"
.position(|entry| {
matches!(entry,
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::FeatureInvocationResult { .. })
if input_schema(entry) == Some("feature:invocation-result")
)
})
.expect("invocation result must be durable before the model request");
@@ -15876,7 +15899,7 @@ permission = "write"
.position(|entry| {
matches!(entry,
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::SubjectBehaviorRefresh { .. })
if input_schema(entry) == Some("subject:behavior")
)
})
.expect("Subject behavior refresh must be durable before the model request");
@@ -16115,14 +16138,14 @@ permission = "write"
.iter()
.filter(|entry| matches!(entry,
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::FeatureInvocationResult { .. })
if input_schema(entry) == Some("feature:invocation-result")
))
.count(),
2 // Explicit replay displays the cached result; it does not repeat the effect.
);
assert!(entries.iter().any(|entry| matches!(entry,
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::ResidentSummaryRefresh { .. })
if input_schema(entry) == Some("memory:resident-summary")
)));
}
@@ -16165,7 +16188,7 @@ permission = "write"
LogEntry::AnnotatedSystemItem {
entry, extensions, ..
} => {
matches!(entry.item, SystemItem::FeatureInvocationResult { .. })
input_schema(entry) == Some("feature:invocation-result")
&& extensions.iter().any(|extension| {
extension.domain == SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN
&& serde_json::from_value::<PendingActivationState>(
@@ -16756,34 +16779,70 @@ permission = "write"
}));
}
fn startup_test_item() -> SystemItem {
SystemItem::JournalStartup {
body: "PRIOR_JOURNAL_REFERENCE".into(),
snapshot: serde_json::json!({"subject_id":"subject", "journals":[], "omitted":false}),
snapshot_sha256: "snapshot-digest".into(),
prompt_provenance: PromptRenderProvenance {
workspace_id: None,
source_digest: "source".into(),
projection_digest: "projection".into(),
logical_name: "internal.subjektiv_journal_startup".into(),
fn input_schema(entry: &session_store::LoggedSystemHistoryEntry) -> Option<&str> {
entry
.metadata
.system_input
.as_ref()?
.feature_payload()
.map(|payload| payload.schema())
}
fn startup_test_input() -> session_store::LoggedSystemHistoryEntry {
use session_store::*;
let payload = FeatureInputPayload::encode(
"journal:startup",
1,
&crate::feature::builtin::journal::JournalStartupInput {
snapshot: subjektiv::journal::startup::JournalStartupSnapshot {
subject_id: "subject".into(),
journals: vec![],
omitted: false,
},
snapshot_sha256: "snapshot-digest".into(),
prompt_provenance: PromptRenderProvenance {
workspace_id: None,
source_digest: "source".into(),
projection_digest: "projection".into(),
logical_name: "internal.subjektiv_journal_startup".into(),
},
},
}
)
.unwrap();
LoggedSystemHistoryEntry::from_input(
LoggedSessionHistoryOrigin::Provided {
producer: "arbitrary-context-provider".into(),
},
None,
SystemInputEnvelope::new(
"PRIOR_JOURNAL_REFERENCE",
SystemInputMetadata::new(
SystemInputUsePolicy {
recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden,
compaction: InputCompactionUse::ReferenceInput,
},
Some(payload),
None,
)
.unwrap(),
)
.unwrap(),
)
}
struct StartupTestSource {
calls: Arc<AtomicUsize>,
fail: Arc<AtomicBool>,
}
#[async_trait]
impl InitialContextSource for StartupTestSource {
impl SessionContextSource for StartupTestSource {
async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
self.calls.fetch_add(1, Ordering::SeqCst);
if self.fail.load(Ordering::SeqCst) {
return Err("startup read failed".into());
}
Ok(vec![session_store::LoggedSystemHistoryEntry {
item: startup_test_item(),
metadata: new_history_metadata(WorkerHistoryProvenance::JournalContext, None),
}])
Ok(vec![startup_test_input()])
}
}
@@ -16791,7 +16850,7 @@ permission = "write"
async fn initial_system_messages_preserve_feature_metadata_without_payload_classification() {
struct Source(Vec<session_store::LoggedSystemHistoryEntry>);
#[async_trait]
impl InitialContextSource for Source {
impl SessionContextSource for Source {
async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
Ok(self.0.clone())
}
@@ -16807,16 +16866,19 @@ permission = "write"
},
metadata: new_history_metadata(WorkerHistoryProvenance::JournalContext, None),
},
session_store::LoggedSystemHistoryEntry {
item: startup_test_item(),
metadata: new_history_metadata(
{
let mut input = startup_test_input();
let custom = new_history_metadata(
WorkerHistoryProvenance::BackendInstruction {
operation_id: Some("feature-selected-origin".into()),
},
Some(SessionHistoryDerivation {
sources: vec![SessionHistoryEntryId::new()],
}),
),
);
input.metadata.origin = custom.origin;
input.metadata.derivation = custom.derivation;
input
},
];
let mut worker = Worker::new(
@@ -16829,7 +16891,7 @@ permission = "write"
)
.await
.unwrap();
worker.install_initial_context_source(Arc::new(Source(expected.clone())));
worker.install_session_context_source(Arc::new(Source(expected.clone())));
worker.materialize_durable_session_head().await.unwrap();
let log = store
.read_all(worker.session_id(), worker.segment_id())
@@ -16886,7 +16948,7 @@ permission = "write"
.await
.unwrap();
let calls = Arc::new(AtomicUsize::new(0));
worker.install_initial_context_source(Arc::new(StartupTestSource {
worker.install_session_context_source(Arc::new(StartupTestSource {
calls: calls.clone(),
fail: Arc::new(AtomicBool::new(false)),
}));
@@ -16907,7 +16969,9 @@ permission = "write"
assert_eq!(replay[0].item.as_text(), Some("PRIOR_JOURNAL_REFERENCE"));
assert_eq!(
replay[0].annotation.origin,
WorkerHistoryProvenance::JournalContext
WorkerHistoryProvenance::Provided {
producer: "arbitrary-context-provider".into()
}
);
let view =
crate::session_capture::SessionCapture::from_history_entries("startup", replay);
@@ -16918,7 +16982,7 @@ permission = "write"
let entries = store
.read_all(worker.session_id(), worker.segment_id())
.unwrap();
let journal = entries.iter().position(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if matches!(entry.item, SystemItem::JournalStartup { .. }))).unwrap();
let journal = entries.iter().position(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if input_schema(entry) == Some("journal:startup"))).unwrap();
let user = entries
.iter()
.position(|entry| matches!(entry, LogEntry::AnnotatedUserInput { .. }))
@@ -16944,7 +17008,7 @@ permission = "write"
.unwrap();
let calls = Arc::new(AtomicUsize::new(0));
let fail = Arc::new(AtomicBool::new(true));
worker.install_initial_context_source(Arc::new(StartupTestSource {
worker.install_session_context_source(Arc::new(StartupTestSource {
calls: calls.clone(),
fail: fail.clone(),
}));
@@ -16975,7 +17039,7 @@ permission = "write"
.await
.unwrap();
let calls = Arc::new(AtomicUsize::new(0));
worker.install_initial_context_source(Arc::new(StartupTestSource {
worker.install_session_context_source(Arc::new(StartupTestSource {
calls: calls.clone(),
fail: Arc::new(AtomicBool::new(false)),
}));
@@ -17030,7 +17094,7 @@ permission = "write"
.unwrap();
worker.enable_worker_metadata_write_through().unwrap();
let calls = Arc::new(AtomicUsize::new(0));
worker.install_initial_context_source(Arc::new(StartupTestSource {
worker.install_session_context_source(Arc::new(StartupTestSource {
calls: calls.clone(),
fail: Arc::new(AtomicBool::new(false)),
}));
@@ -17052,7 +17116,7 @@ permission = "write"
&manifest.worker.name, manifest.clone(), store.clone(),
PromptCatalogSource::builtins_only(), WorkerWorkspaceContext::no_workspace(),
WorkerFilesystemAuthority::None, Some(Box::new(NoopClient))).await.unwrap();
restored.install_initial_context_source(Arc::new(StartupTestSource {
restored.install_session_context_source(Arc::new(StartupTestSource {
calls: calls.clone(),
fail: Arc::new(AtomicBool::new(true)),
}));
@@ -17377,7 +17441,7 @@ permission = "write"
.iter()
.filter_map(|entry| match entry {
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::JournalStartup { .. }) =>
if input_schema(entry) == Some("journal:startup") =>
{
Some(entry.item.clone())
}
@@ -17428,7 +17492,7 @@ permission = "write"
.iter()
.filter_map(|entry| match entry {
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::JournalStartup { .. }) =>
if input_schema(entry) == Some("journal:startup") =>
{
Some(&entry.item)
}
crates/worker/tests/compact_events_test.rs
+137 -43
View File
@@ -2220,10 +2220,48 @@ fn summary_entry(
origin: session_store::LoggedSessionHistoryOrigin,
) -> session_store::LoggedHistoryEntry {
let mut entry = annotated(item);
entry.metadata.system_input = matches!(
&entry.item,
session_store::LoggedItem::Message {
role: session_store::LoggedRole::System,
..
}
)
.then(|| {
session_store::SystemInputMetadata::new(
session_store::SystemInputUsePolicy {
recording_material: session_store::InputEvidenceUse::Exclude,
memory_evidence: session_store::InputEvidenceUse::Exclude,
observation: session_store::InputObservationUse::Hidden,
compaction: match origin {
session_store::LoggedSessionHistoryOrigin::CompactionSummary => {
session_store::InputCompactionUse::OpeningSummary
}
session_store::LoggedSessionHistoryOrigin::JournalContext => {
session_store::InputCompactionUse::ReferenceInput
}
_ => session_store::InputCompactionUse::Retain,
},
},
None,
None,
)
.unwrap()
});
entry.metadata.origin = origin;
entry
}
fn compaction_use(
metadata: &session_store::LoggedSessionHistoryMetadata,
) -> session_store::InputCompactionUse {
metadata
.system_input
.as_ref()
.map(|input| input.usage().compaction)
.unwrap_or(session_store::InputCompactionUse::Retain)
}
fn long_summary(label: &str) -> String {
format!(
"## Completed Tasks\n{label}: investigated the transport boundary.\n\n\
@@ -2359,7 +2397,8 @@ async fn full_initial_summary_survives_small_overview_budgets_and_both_retention
assert_eq!(
replacement
.iter()
.filter(|entry| entry.metadata.origin == CompactionSummary)
.filter(|entry| compaction_use(&entry.metadata)
== session_store::InputCompactionUse::OpeningSummary)
.count(),
1
);
@@ -2604,7 +2643,9 @@ async fn three_chained_compactions_transport_each_complete_predecessor_without_r
let seed = replacement_seed(&fixture.worker);
let inherited = seed
.iter()
.filter(|entry| entry.metadata.origin == CompactionSummary)
.filter(|entry| {
compaction_use(&entry.metadata) == session_store::InputCompactionUse::OpeningSummary
})
.collect::<Vec<_>>();
assert_eq!(inherited.len(), 1);
assert_eq!(
@@ -2892,7 +2933,7 @@ impl worker::subjektiv::SubjektivHost for CompactionJournalHost {
}
fn journal_startup_granted(&self) -> bool {
true
false
}
fn journal_startup_context(
@@ -2996,8 +3037,8 @@ fn compaction_snapshot(generation: usize) -> subjektiv::journal::startup::Journa
fn journal_body(worker: &Worker<MockClient, TestStore>) -> String {
let seed = replacement_seed(worker);
assert_eq!(
seed[1].metadata.origin,
session_store::LoggedSessionHistoryOrigin::JournalContext
compaction_use(&seed[1].metadata),
session_store::InputCompactionUse::ReferenceInput
);
let item = Item::from(seed[1].item.clone());
assert!(matches!(
@@ -3009,8 +3050,8 @@ fn journal_body(worker: &Worker<MockClient, TestStore>) -> String {
));
assert_eq!(
seed.iter()
.filter(|entry| entry.metadata.origin
== session_store::LoggedSessionHistoryOrigin::JournalContext)
.filter(|entry| compaction_use(&entry.metadata)
== session_store::InputCompactionUse::ReferenceInput)
.count(),
1
);
@@ -3119,20 +3160,34 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
projection.source_digest = "a".repeat(64);
let initial_catalog = Arc::new(worker::PromptCatalog::from_projection(projection).unwrap());
fixture.worker.prompts().store(initial_catalog.clone());
let legacy_startup = session_store::LoggedSystemHistoryEntry {
item: session_store::SystemItem::JournalStartup {
body: "OLD_TYPED_STARTUP_JOURNAL".into(),
snapshot: serde_json::json!({"journals": []}),
snapshot_sha256: "legacy-startup-digest".into(),
prompt_provenance: session_store::PromptRenderProvenance {
workspace_id: None,
source_digest: "legacy-source".into(),
projection_digest: "legacy-projection".into(),
logical_name: "internal.subjektiv_journal_startup".into(),
},
},
metadata: summary_entry(Item::system_message(""), DerivedSummary).metadata,
};
// Already migrated generic input with unrelated provenance/payload; Core
// must select it by explicit policy, not identify a Journal Feature.
let legacy_startup = session_store::LoggedSystemHistoryEntry::from_input(
DerivedSummary,
None,
session_store::SystemInputEnvelope::new(
"OLD_TYPED_STARTUP_JOURNAL",
session_store::SystemInputMetadata::new(
session_store::SystemInputUsePolicy {
recording_material: session_store::InputEvidenceUse::Exclude,
memory_evidence: session_store::InputEvidenceUse::Exclude,
observation: session_store::InputObservationUse::Hidden,
compaction: session_store::InputCompactionUse::ReferenceInput,
},
Some(
session_store::FeatureInputPayload::encode(
"renamed-owner:reference",
8,
&serde_json::json!({"private": "retained opaque context"}),
)
.unwrap(),
),
None,
)
.unwrap(),
)
.unwrap(),
);
let legacy_id = legacy_startup.metadata.entry_id.clone();
fixture
.restore_seed_with_system(
@@ -3166,7 +3221,7 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
.unwrap()
.annotation
.origin,
JournalContext
DerivedSummary
);
let session = fixture.worker.session_id();
let original_segment = fixture.worker.segment_id();
@@ -3191,8 +3246,8 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
.unwrap()
.annotation
.origin,
JournalContext,
"legacy startup classification survives a normal turn"
DerivedSummary,
"generic source provenance survives a normal turn unchanged"
);
}
assert_eq!(
@@ -3243,7 +3298,7 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
assert_eq!(host.reads.lock().unwrap().len(), index + 1);
let history = fixture.worker.history();
let overview = input
.split("## Previous Journal reference (historical, not automatically retained)")
.split("## Previous reference input (historical, not automatically retained)")
.next()
.unwrap();
for old in &old_bodies {
@@ -3336,6 +3391,17 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
.run_text("normal post-compaction turn")
.await
.unwrap();
let frozen_seed = replacement_seed(&fixture.worker)
.into_iter()
.filter(|entry| {
matches!(
compaction_use(&entry.metadata),
session_store::InputCompactionUse::ReferenceInput
| session_store::InputCompactionUse::OpeningSummary
)
})
.collect::<Vec<_>>();
assert_eq!(frozen_seed.len(), 2);
let records = fixture
.worker
.store()
@@ -3375,6 +3441,20 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
.run_text("normal turn after typed history restore")
.await
.unwrap();
for frozen in frozen_seed {
let replayed = fixture
.worker
.session_history()
.entries()
.iter()
.find(|entry| entry.annotation.entry_id == frozen.metadata.entry_id)
.expect("frozen entry identity survives re-restore");
assert_eq!(replayed.item, Item::from(frozen.item));
assert_eq!(
replayed.annotation, frozen.metadata,
"origin, derivation, public allowlist, policy and private payload survive atomic seed/restore"
);
}
assert_eq!(
host.reads.lock().unwrap().len(),
3,
@@ -3445,27 +3525,41 @@ async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversat
)
.await;
fixture.worker.compact(0).await.unwrap();
let body = journal_body(&fixture.worker);
let expected = match case {
"no-host" | "disabled" => "disabled",
"absent" => "absent",
_ => "unavailable",
};
assert!(
body.contains(&format!("Journal read status: {expected}.")),
"{case}"
);
let input = first_compactor_input(&observed.requests()[1]);
assert_frozen_journal_transport(&input, &body);
assert_journal_provenance(
&body,
expected_json.as_deref(),
&fixture.worker.prompts().load_full(),
(case != "no-host").then_some(COMPACTION_HOST_WORKSPACE_ID),
);
let body = if matches!(case, "disabled" | "no-host") {
assert_eq!(
replacement_seed(&fixture.worker)
.iter()
.filter(|entry| compaction_use(&entry.metadata)
== session_store::InputCompactionUse::ReferenceInput)
.count(),
0
);
assert!(!input.contains("Journal reference frozen at this compaction boundary"));
String::new()
} else {
let body = journal_body(&fixture.worker);
let expected = if case == "absent" {
"absent"
} else {
"unavailable"
};
assert!(
body.contains(&format!("Journal read status: {expected}.")),
"{case}"
);
assert_frozen_journal_transport(&input, &body);
assert_journal_provenance(
&body,
expected_json.as_deref(),
&fixture.worker.prompts().load_full(),
Some(COMPACTION_HOST_WORKSPACE_ID),
);
body
};
assert_full_initial_summary(&input, &inherited);
assert!(input.contains("UNRECORDED_FALLBACK_GOAL"), "{case}");
assert!(input.contains("## Previous Journal reference (historical, not automatically retained)\nPREVIOUS_JOURNAL_DELEGATED_CONTEXT"));
assert!(input.contains("## Previous reference input (historical, not automatically retained)\nPREVIOUS_JOURNAL_DELEGATED_CONTEXT"));
assert!(
fixture.worker.history().iter().all(|item| !item
.as_text()
crates/worker/tests/feature_invocation_test.rs
+15 -5
View File
@@ -14,7 +14,7 @@ use protocol::{
InvocationArgumentDescriptor, InvocationArgumentType, InvocationArgumentValue,
InvocationCompletion, InvocationValue, Segment,
};
use session_store::{CombinedStore, FsStore, FsWorkerStore, LogEntry, SystemItem};
use session_store::{CombinedStore, FsStore, FsWorkerStore, LogEntry};
use worker::feature::{
FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureInvocationContext,
FeatureInvocationHandler, FeatureInvocationHandlerError, FeatureModule, FeatureRegistryBuilder,
@@ -224,10 +224,20 @@ fn results(worker: &Worker<Trace, TestStore>) -> Vec<FeatureInvocationResult> {
.0
.into_iter()
.filter_map(|entry| match entry {
LogEntry::AnnotatedSystemItem { entry, .. } => match entry.item {
SystemItem::FeatureInvocationResult { result, .. } => Some(result),
_ => None,
},
LogEntry::AnnotatedSystemItem { entry, .. } => {
#[derive(serde::Deserialize)]
struct SavedResult {
result: FeatureInvocationResult,
}
entry
.metadata
.system_input
.as_ref()?
.feature_payload()?
.decode::<SavedResult>("feature:invocation-result", 1)
.ok()
.map(|input| input.result)
}
_ => None,
})
.collect()