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

Add persistent Subject behavior context

Browse Source
This commit is contained in:
Hare
2026-10-05 16:27:23 +09:00
parent cc97987482
commit 8891236db3
28 changed files with 1999 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+1
View File
@@ -1375,6 +1375,7 @@ where
crate::feature::builtin::memory::ordinary_subjektiv_resident_summary_source(
worker.manifest(),
worker.workspace_client_handle(),
worker.prompts(),
)?
{
let workspace_id = worker
crates/worker/src/feature/builtin/memory.rs
+352 -63
View File
@@ -4,7 +4,8 @@
//! authority. In that case model-visible Memory tools must go through the
//! workspace backend instead of resolving `.yoi/memory` from a Worker workdir.
use std::sync::Arc;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
use arc_swap::ArcSwap;
@@ -23,11 +24,15 @@ use crate::feature::{
FeatureDescriptor, FeatureHookPoint, FeatureInstallContext, FeatureInstallError, FeatureModule,
HookDeclaration, ToolContribution, ToolDeclaration,
};
use crate::hook::{Hook, HookError, HookErrorCategory, HookExecutionPolicy, WorkerRestored};
use crate::hook::{
BeforeSessionRewrite, BeforeSessionRewriteAction, BeforeSessionRewriteContext, Hook, HookError,
HookErrorCategory, HookExecutionPolicy, HookPreRequestAction, PreLlmRequest, PreRequestContext,
WorkerRestored,
};
use crate::prompt::catalog::{PromptCatalog, WorkerPrompt};
use crate::worker::{
SystemPromptContribution, SystemPromptContributionSource, WorkspaceClient,
WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
ResidentContextRefresh, SystemPromptContribution, SystemPromptContributionSource,
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
};
#[derive(Clone, Debug)]
@@ -73,6 +78,8 @@ pub enum WorkspaceMemoryBackendError {
Decode(#[from] serde_json::Error),
#[error("workspace memory backend rejected operation: {0}")]
Backend(String),
#[error("workspace memory backend returned an invalid response: {0}")]
InvalidResponse(String),
}
impl dyn WorkspaceClient + '_ {
@@ -653,12 +660,17 @@ struct WorkspaceResidentSummarySource {
client: Arc<dyn WorkspaceClient>,
}
struct WorkspaceSubjektivResidentSummarySource {
client: Arc<dyn WorkspaceClient>,
#[derive(Default)]
struct SubjectBehaviorRefreshState {
applied: Option<(u64, String)>,
pending: HashMap<u64, String>,
}
const READY_EMPTY_RESIDENT_SURFACE: &str =
"The current ready resident memory surface is intentionally empty.";
struct WorkspaceSubjektivResidentSummarySource {
client: Arc<dyn WorkspaceClient>,
prompts: Arc<ArcSwap<PromptCatalog>>,
behavior_state: Mutex<SubjectBehaviorRefreshState>,
}
fn resident_summary_contribution(
output: memory::backend::MemoryResidentSummaryOutput,
@@ -696,36 +708,132 @@ impl SystemPromptContributionSource for WorkspaceResidentSummarySource {
}
}
impl WorkspaceSubjektivResidentSummarySource {
async fn fetch(
&self,
) -> Result<server_api::SubjektivResidentContextOutput, WorkspaceMemoryBackendError> {
match execute_subjektiv_memory_operation(
self.client.as_ref(),
server_api::SubjektivMemoryBackendOperation::ResidentContext(Default::default()),
)
.await?
{
server_api::SubjektivMemoryBackendResponse::ResidentContext(output) => Ok(output),
other => Err(WorkspaceMemoryBackendError::InvalidResponse(format!(
"unexpected subject resident context response: {other:?}"
))),
}
}
fn render(
&self,
output: &server_api::SubjektivResidentContextOutput,
) -> Result<String, String> {
use memory::backend::MemoryResidentSummaryAvailability as Availability;
let availability = match output.memory_surface.availability {
Availability::Ready => "ready",
Availability::Ungenerated => "ungenerated",
Availability::Stale => "stale",
Availability::Failed => "failed",
Availability::Unavailable => "unavailable",
};
self.prompts
.load()
.subjektiv_resident_context(
&output.behavior_md,
availability,
output.memory_surface.content.as_deref(),
)
.map_err(|error| error.to_string())
}
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_revision, output.behavior_md.clone()));
state.pending.remove(&output.behavior_revision);
}
}
#[async_trait]
impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource {
async fn load(&self) -> SystemPromptContribution {
match execute_subjektiv_memory_operation(
self.client.as_ref(),
server_api::SubjektivMemoryBackendOperation::ResidentSummary(Default::default()),
)
.await
{
Ok(server_api::SubjektivMemoryBackendResponse::ResidentSummary(output)) => {
match resident_summary_contribution(output) {
SystemPromptContribution::Ready(content) if content.trim().is_empty() => {
SystemPromptContribution::Ready(READY_EMPTY_RESIDENT_SURFACE.to_string())
}
contribution => contribution,
match self.fetch().await {
Ok(output) => match self.render(&output) {
Ok(rendered) => {
self.remember_behavior(&output);
SystemPromptContribution::Ready(rendered)
}
}
Ok(other) => {
tracing::debug!(?other, "unexpected subject resident Memory response");
SystemPromptContribution::Unavailable
}
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(),
)
}
},
Err(error) => {
tracing::debug!(%error, "subject resident Memory unavailable");
SystemPromptContribution::Unavailable
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(),
)
}
}
}
async fn load_changed_resident_context(
&self,
) -> Result<Option<ResidentContextRefresh>, String> {
let output = self.fetch().await.map_err(|error| error.to_string())?;
let changed = self
.behavior_state
.lock()
.expect("subject behavior state poisoned")
.applied
.as_ref()
.is_none_or(|(revision, body)| {
*revision != output.behavior_revision || body != &output.behavior_md
});
if !changed {
return Ok(None);
}
let rendered = self.render(&output)?;
let revision = output.behavior_revision;
self.behavior_state
.lock()
.expect("subject behavior state poisoned")
.pending
.insert(revision, output.behavior_md);
Ok(Some(ResidentContextRefresh {
body: rendered,
revision,
}))
}
fn confirm_resident_context_revision(&self, revision: u64) {
let mut state = self
.behavior_state
.lock()
.expect("subject behavior state poisoned");
if let Some(body) = state.pending.remove(&revision) {
state.applied = Some((revision, body));
}
}
fn invalidate_resident_context_representation(&self) {
self.behavior_state
.lock()
.expect("subject behavior state poisoned")
.applied = None;
}
}
const SUBJEKTIV_RESTORE_REFRESH_HOOK: &str = "refresh-resident-surface-after-restore";
const SUBJEKTIV_BEHAVIOR_REFRESH_HOOK: &str = "refresh-subject-behavior-before-request";
const SUBJEKTIV_BEHAVIOR_REWRITE_HOOK: &str = "invalidate-subject-behavior-before-rewrite";
/// subjektiv-owned restore behavior. Initial prompt contribution remains a
/// shared host facility, while loading and rendering a post-restore refresh is
@@ -773,6 +881,14 @@ impl FeatureModule for SubjektivResidentRestoreRefreshFeature {
SUBJEKTIV_RESTORE_REFRESH_HOOK,
FeatureHookPoint::WorkerRestored,
))
.with_hook(HookDeclaration::new(
SUBJEKTIV_BEHAVIOR_REFRESH_HOOK,
FeatureHookPoint::PreLlmRequest,
))
.with_hook(HookDeclaration::new(
SUBJEKTIV_BEHAVIOR_REWRITE_HOOK,
FeatureHookPoint::BeforeSessionRewrite,
))
}
fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
@@ -784,10 +900,64 @@ impl FeatureModule for SubjektivResidentRestoreRefreshFeature {
prompts: Arc::clone(&self.prompts),
workspace_id: self.workspace_id.clone(),
},
)?;
context.hooks().add_pre_llm_request(
SUBJEKTIV_BEHAVIOR_REFRESH_HOOK,
HookExecutionPolicy::fail_closed(),
SubjektivBehaviorRefreshHook {
source: Arc::clone(&self.source),
},
)?;
context.hooks().add_before_session_rewrite(
SUBJEKTIV_BEHAVIOR_REWRITE_HOOK,
HookExecutionPolicy::fail_closed(),
SubjektivBehaviorRewriteHook {
source: Arc::clone(&self.source),
},
)
}
}
struct SubjektivBehaviorRewriteHook {
source: Arc<dyn SystemPromptContributionSource>,
}
#[async_trait]
impl Hook<BeforeSessionRewrite> for SubjektivBehaviorRewriteHook {
async fn call(
&self,
_context: &BeforeSessionRewriteContext,
) -> Result<BeforeSessionRewriteAction, HookError> {
self.source.invalidate_resident_context_representation();
Ok(BeforeSessionRewriteAction::Continue)
}
}
struct SubjektivBehaviorRefreshHook {
source: Arc<dyn SystemPromptContributionSource>,
}
#[async_trait]
impl Hook<PreLlmRequest> for SubjektivBehaviorRefreshHook {
async fn call(&self, context: &PreRequestContext) -> Result<HookPreRequestAction, HookError> {
let refresh = self
.source
.load_changed_resident_context()
.await
.map_err(|error| HookError::new(HookErrorCategory::Dependency, error))?;
if let Some(refresh) = refresh {
let system_items = context.system_items().ok_or_else(|| {
HookError::new(
HookErrorCategory::Dependency,
"Subject behavior refresh requires durable session append authority",
)
})?;
system_items.append_subject_behavior_refresh(refresh.body, refresh.revision);
}
Ok(HookPreRequestAction::Continue)
}
}
struct SubjektivResidentRestoreRefreshHook {
source: Arc<dyn SystemPromptContributionSource>,
prompts: Arc<ArcSwap<PromptCatalog>>,
@@ -931,6 +1101,7 @@ pub(crate) fn ordinary_subjektiv_features_enabled(manifest: &manifest::WorkerMan
pub(crate) fn ordinary_subjektiv_resident_summary_source(
manifest: &manifest::WorkerManifest,
client: Arc<dyn WorkspaceClient>,
prompts: Arc<ArcSwap<PromptCatalog>>,
) -> std::io::Result<Option<Arc<dyn SystemPromptContributionSource>>> {
if !ordinary_subjektiv_features_enabled(manifest) {
return Ok(None);
@@ -968,6 +1139,8 @@ pub(crate) fn ordinary_subjektiv_resident_summary_source(
}
Ok(Some(Arc::new(WorkspaceSubjektivResidentSummarySource {
client,
prompts,
behavior_state: Mutex::new(SubjectBehaviorRefreshState::default()),
})))
}
@@ -1100,6 +1273,10 @@ mod tests {
use agen::tool::ToolDefinition;
use std::sync::Mutex;
fn test_prompts() -> Arc<ArcSwap<PromptCatalog>> {
Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap()))
}
fn test_client() -> Arc<dyn WorkspaceClient> {
Arc::new(crate::worker::TestWorkspaceHttpClient::new(
"workspace",
@@ -1112,6 +1289,7 @@ mod tests {
availability: memory::backend::MemoryResidentSummaryAvailability,
content: Option<String>,
scope_allowed: bool,
behavior: Mutex<(u64, String)>,
paths: Mutex<Vec<String>>,
}
@@ -1124,6 +1302,7 @@ mod tests {
availability,
content: content.map(str::to_string),
scope_allowed: true,
behavior: Mutex::new((3, "Be deliberate.".to_string())),
paths: Mutex::new(Vec::new()),
}
}
@@ -1133,6 +1312,7 @@ mod tests {
availability: memory::backend::MemoryResidentSummaryAvailability::Ready,
content: Some("must not be injected".into()),
scope_allowed: false,
behavior: Mutex::new((3, "must not be injected".to_string())),
paths: Mutex::new(Vec::new()),
}
}
@@ -1161,7 +1341,7 @@ mod tests {
serde_json::from_str(request.body.as_deref().unwrap_or_default()).unwrap();
assert!(matches!(
request.operation,
server_api::SubjektivMemoryBackendOperation::ResidentSummary(_)
server_api::SubjektivMemoryBackendOperation::ResidentContext(_)
));
if !self.scope_allowed {
return Ok(WorkspaceResponse {
@@ -1169,13 +1349,18 @@ mod tests {
body: "subject scope unavailable".into(),
});
}
let (behavior_revision, behavior_md) = self.behavior.lock().unwrap().clone();
Ok(WorkspaceResponse {
status: 200,
body: serde_json::to_string(
&server_api::SubjektivMemoryBackendResponse::ResidentSummary(
memory::backend::MemoryResidentSummaryOutput {
availability: self.availability,
content: self.content.clone(),
&server_api::SubjektivMemoryBackendResponse::ResidentContext(
server_api::SubjektivResidentContextOutput {
behavior_md,
behavior_revision,
memory_surface: memory::backend::MemoryResidentSummaryOutput {
availability: self.availability,
content: self.content.clone(),
},
},
),
)
@@ -1479,13 +1664,19 @@ permission = "write"
memory::backend::MemoryResidentSummaryAvailability::Ready,
Some("current subject surface"),
));
let ready = ordinary_subjektiv_resident_summary_source(&manifest, ready_client.clone())
.unwrap()
.unwrap();
assert_eq!(
ready.load().await,
SystemPromptContribution::Ready("current subject surface".into())
);
let ready = ordinary_subjektiv_resident_summary_source(
&manifest,
ready_client.clone(),
test_prompts(),
)
.unwrap()
.unwrap();
let loaded = ready.load().await;
let SystemPromptContribution::Ready(loaded) = loaded else {
panic!("subject resident context must be present");
};
assert!(loaded.contains("Be deliberate."));
assert!(loaded.contains("current subject surface"));
assert_eq!(
ready_client.paths.lock().unwrap().clone(),
vec!["/api/w/workspace/subjektiv/memory".to_string()]
@@ -1495,13 +1686,18 @@ permission = "write"
memory::backend::MemoryResidentSummaryAvailability::Ready,
None,
));
let ready_empty = ordinary_subjektiv_resident_summary_source(&manifest, ready_empty_client)
.unwrap()
.unwrap();
assert_eq!(
ready_empty.load().await,
SystemPromptContribution::Ready(READY_EMPTY_RESIDENT_SURFACE.into())
);
let ready_empty = ordinary_subjektiv_resident_summary_source(
&manifest,
ready_empty_client,
test_prompts(),
)
.unwrap()
.unwrap();
let SystemPromptContribution::Ready(ready_empty) = ready_empty.load().await else {
panic!("empty surface must retain Subject behavior");
};
assert!(ready_empty.contains("Be deliberate."));
assert!(ready_empty.contains("intentionally empty"));
for availability in [
memory::backend::MemoryResidentSummaryAvailability::Ungenerated,
@@ -1514,22 +1710,101 @@ permission = "write"
availability,
Some("must not be injected"),
)),
test_prompts(),
)
.unwrap()
.unwrap();
assert_eq!(source.load().await, SystemPromptContribution::Unavailable);
let SystemPromptContribution::Ready(loaded) = source.load().await else {
panic!("Subject behavior must survive unavailable surfaces");
};
assert!(loaded.contains("Be deliberate."));
assert!(!loaded.contains("must not be injected"));
}
}
#[tokio::test]
async fn subject_resident_source_emits_only_changed_behavior_revisions() {
let manifest = subject_manifest("workspace");
let client = Arc::new(SubjectResidentClient::new(
memory::backend::MemoryResidentSummaryAvailability::Ungenerated,
None,
));
let source =
ordinary_subjektiv_resident_summary_source(&manifest, client.clone(), test_prompts())
.unwrap()
.unwrap();
let initial = source.load().await;
assert!(matches!(initial, SystemPromptContribution::Ready(_)));
assert!(
source
.load_changed_resident_context()
.await
.unwrap()
.is_none()
);
*client.behavior.lock().unwrap() = (4, "Ask when uncertain.".to_string());
let refresh = source
.load_changed_resident_context()
.await
.unwrap()
.expect("changed behavior must produce one refresh");
assert_eq!(refresh.revision, 4);
assert!(refresh.body.contains("Ask when uncertain."));
source.confirm_resident_context_revision(refresh.revision);
assert!(
source
.load_changed_resident_context()
.await
.unwrap()
.is_none()
);
*client.behavior.lock().unwrap() = (5, String::new());
let cleared = source
.load_changed_resident_context()
.await
.unwrap()
.expect("cleared behavior must produce one refresh");
assert_eq!(cleared.revision, 5);
assert!(cleared.body.contains("No user-managed behavior is set"));
source.confirm_resident_context_revision(cleared.revision);
assert!(
source
.load_changed_resident_context()
.await
.unwrap()
.is_none()
);
source.invalidate_resident_context_representation();
let replay_after_rewrite = source
.load_changed_resident_context()
.await
.unwrap()
.expect("history rewrite must require a durable current representation");
assert_eq!(replay_after_rewrite.revision, 5);
assert!(
replay_after_rewrite
.body
.contains("No user-managed behavior is set")
);
}
#[tokio::test]
async fn subject_resident_source_fails_closed_without_trusted_subject_scope() {
let manifest = subject_manifest("workspace");
let client = Arc::new(SubjectResidentClient::denied());
let source = ordinary_subjektiv_resident_summary_source(&manifest, client.clone())
.unwrap()
.unwrap();
let source =
ordinary_subjektiv_resident_summary_source(&manifest, client.clone(), test_prompts())
.unwrap()
.unwrap();
assert_eq!(source.load().await, SystemPromptContribution::Unavailable);
let SystemPromptContribution::Ready(failure) = source.load().await else {
panic!("fetch failure must be explicit in resident context");
};
assert!(failure.contains("could not fetch"));
assert_eq!(
client.paths.lock().unwrap().clone(),
vec!["/api/w/workspace/subjektiv/memory".to_string()]
@@ -1557,7 +1832,7 @@ permission = "write"
)
.unwrap();
assert!(
ordinary_subjektiv_resident_summary_source(&disabled, test_client())
ordinary_subjektiv_resident_summary_source(&disabled, test_client(), test_prompts())
.unwrap()
.is_none()
);
@@ -1569,22 +1844,30 @@ permission = "write"
Some("policy-only surface must not load"),
));
assert!(
ordinary_subjektiv_resident_summary_source(&policy_only, unattached_client.clone())
.unwrap()
.is_none()
ordinary_subjektiv_resident_summary_source(
&policy_only,
unattached_client.clone(),
test_prompts()
)
.unwrap()
.is_none()
);
assert!(unattached_client.paths.lock().unwrap().is_empty());
let foreign = subject_manifest("other-workspace");
assert!(ordinary_subjektiv_resident_summary_source(&foreign, test_client()).is_err());
assert!(
ordinary_subjektiv_resident_summary_source(&foreign, test_client(), test_prompts())
.is_err()
);
}
#[test]
fn subject_restore_refresh_is_suppressed_for_internal_hosts() {
let manifest = subject_manifest("workspace");
let source = ordinary_subjektiv_resident_summary_source(&manifest, test_client())
.unwrap()
.unwrap();
let source =
ordinary_subjektiv_resident_summary_source(&manifest, test_client(), test_prompts())
.unwrap()
.unwrap();
let prompts = Arc::new(ArcSwap::from(
crate::prompt::catalog::PromptCatalog::builtins_only().unwrap(),
));
@@ -1598,9 +1881,10 @@ permission = "write"
#[test]
fn subject_restore_refresh_feature_declares_only_restore_lifecycle() {
let manifest = subject_manifest("workspace");
let source = ordinary_subjektiv_resident_summary_source(&manifest, test_client())
.unwrap()
.unwrap();
let source =
ordinary_subjektiv_resident_summary_source(&manifest, test_client(), test_prompts())
.unwrap()
.unwrap();
let feature = SubjektivResidentRestoreRefreshFeature::new(
source,
Arc::new(ArcSwap::from(
@@ -1611,8 +1895,13 @@ permission = "write"
let descriptor = feature.descriptor();
assert!(descriptor.tools.is_empty());
assert_eq!(descriptor.hooks.len(), 1);
assert_eq!(descriptor.hooks.len(), 3);
assert_eq!(descriptor.hooks[0].point, FeatureHookPoint::WorkerRestored);
assert_eq!(descriptor.hooks[1].point, FeatureHookPoint::PreLlmRequest);
assert_eq!(
descriptor.hooks[2].point,
FeatureHookPoint::BeforeSessionRewrite
);
}
#[test]
crates/worker/src/hook.rs
+14
View File
@@ -281,6 +281,20 @@ impl SystemItemAppendHandle {
.expect("system-item append queue poisoned")
.push(item);
}
/// Queue a revisioned Subject behavior refresh. 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_revision: u64) {
self.pending
.lock()
.expect("system-item append queue poisoned")
.push(SystemItem::SubjectBehaviorRefresh {
body: body.into(),
behavior_revision,
prompt_provenance: None,
});
}
}
/// Host-created narrow append authority exposed only at a restore lifecycle
crates/worker/src/ipc/interceptor.rs
+159 -1
View File
@@ -38,7 +38,7 @@ use crate::hook::{
use crate::ipc::notify_buffer::{NotifyBuffer, build_system_item_with_provenance};
use crate::prompt::catalog::PromptCatalog;
use crate::session_history::SessionHistoryMetadata;
use crate::worker::SystemItemCommitter;
use crate::worker::{SystemItemCommitter, SystemPromptContributionSource};
use agen::HistoryEntry;
use agen::token_counter::total_tokens;
@@ -109,6 +109,7 @@ pub(crate) struct WorkerInterceptor {
/// worker. `None` in tests / `Worker::new` paths where no writer is
/// attached.
log_writer: Option<Arc<dyn SystemItemCommitter>>,
resident_context_source: Option<Arc<dyn SystemPromptContributionSource>>,
pending_committed_history: Arc<Mutex<VecDeque<HistoryEntry<SessionHistoryMetadata>>>>,
/// Next turn index assigned by `on_prompt_submit`.
next_turn_index: AtomicUsize,
@@ -163,6 +164,7 @@ impl WorkerInterceptor {
prompts,
prompt_workspace_id: None,
log_writer,
resident_context_source: None,
pending_committed_history,
next_turn_index: AtomicUsize::new(0),
tool_calls_this_turn: AtomicUsize::new(0),
@@ -179,11 +181,31 @@ impl WorkerInterceptor {
self
}
pub(crate) fn with_resident_context_source(
mut self,
source: Option<Arc<dyn SystemPromptContributionSource>>,
) -> Self {
self.resident_context_source = source;
self
}
pub(crate) fn with_prompt_workspace_id(mut self, workspace_id: Option<String>) -> Self {
self.prompt_workspace_id = workspace_id;
self
}
fn confirm_resident_context_item(&self, item: &SystemItem) {
let Some(source) = self.resident_context_source.as_ref() else {
return;
};
if let SystemItem::SubjectBehaviorRefresh {
behavior_revision, ..
} = item
{
source.confirm_resident_context_revision(*behavior_revision);
}
}
/// Commit each `SystemItem` as its own `LogEntry::AnnotatedSystemItem`
/// entry through the attached writer (no-op when no writer is
/// wired). Sync — writes complete before the matching
@@ -211,6 +233,7 @@ impl WorkerInterceptor {
.lock()
.expect("pending committed history poisoned")
.push_back(entry);
self.confirm_resident_context_item(item);
}
Ok(())
}
@@ -350,6 +373,11 @@ 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() => {
@@ -763,6 +791,7 @@ mod tests {
HookTurnEndAction, OnTurnEnd, PostToolCall, PreLlmRequest, PreToolCall,
};
use crate::session_history::{WorkerHistoryProvenance, history_entry};
use crate::worker::SystemPromptContribution;
fn test_prompts() -> Arc<ArcSwap<PromptCatalog>> {
Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap()))
@@ -859,6 +888,53 @@ mod tests {
}
}
struct AppendingSubjectBehaviorHook;
#[async_trait]
impl Hook<PreLlmRequest> for AppendingSubjectBehaviorHook {
async fn call(
&self,
input: &PreRequestContext,
) -> Result<HookPreRequestAction, crate::hook::HookError> {
input
.system_items()
.expect("test interceptor has durable append authority")
.append_subject_behavior_refresh("current Subject context", 7);
Ok(HookPreRequestAction::Continue)
}
}
struct ConfirmRecordingResidentSource {
confirmed: Arc<Mutex<Vec<u64>>>,
}
#[async_trait]
impl SystemPromptContributionSource for ConfirmRecordingResidentSource {
async fn load(&self) -> SystemPromptContribution {
SystemPromptContribution::Unavailable
}
fn confirm_resident_context_revision(&self, revision: u64) {
self.confirmed.lock().unwrap().push(revision);
}
}
struct FailingSystemItemCommitter;
impl SystemItemCommitter for FailingSystemItemCommitter {
fn commit_log_entry(
&self,
entry: session_store::LogEntry,
) -> Result<(), session_store::StoreError> {
if matches!(entry, session_store::LogEntry::AnnotatedSystemItem { .. }) {
return Err(session_store::StoreError::Io(std::io::Error::other(
"synthetic system-item failure",
)));
}
Ok(())
}
}
fn task_tool_call_info(
name: &str,
input: serde_json::Value,
@@ -1346,6 +1422,88 @@ mod tests {
}
}
#[tokio::test]
async fn subject_behavior_revision_is_confirmed_only_after_durable_commit() {
let mut builder = HookRegistryBuilder::new();
builder.add_pre_llm_request(AppendingSubjectBehaviorHook);
let committed = Arc::new(Mutex::new(Vec::new()));
let confirmed = Arc::new(Mutex::new(Vec::new()));
let source: Arc<dyn SystemPromptContributionSource> =
Arc::new(ConfirmRecordingResidentSource {
confirmed: Arc::clone(&confirmed),
});
let interceptor = WorkerInterceptor::new(
Arc::new(builder.build()),
None,
None,
NotifyBuffer::new(),
Arc::new(Mutex::new(Vec::new())),
test_prompts(),
Some(Arc::new(RecordingSystemItemCommitter {
committed: Arc::clone(&committed),
})),
)
.with_resident_context_source(Some(source));
let mut ctx = Vec::new();
let action = interceptor
.pre_llm_request(PreLlmRequestContext {
invocation: Default::default(),
items: &mut ctx,
history: &[],
})
.await
.unwrap();
assert!(matches!(action, PreRequestAction::ContinueWith(_)));
assert_eq!(*confirmed.lock().unwrap(), [7]);
assert!(matches!(
committed.lock().unwrap().as_slice(),
[SystemItem::SubjectBehaviorRefresh {
behavior_revision: 7,
prompt_provenance: Some(_),
..
}]
));
}
#[tokio::test]
async fn failed_subject_behavior_persistence_does_not_confirm_revision() {
let mut builder = HookRegistryBuilder::new();
builder.add_pre_llm_request(AppendingSubjectBehaviorHook);
let confirmed = Arc::new(Mutex::new(Vec::new()));
let source: Arc<dyn SystemPromptContributionSource> =
Arc::new(ConfirmRecordingResidentSource {
confirmed: Arc::clone(&confirmed),
});
let interceptor = WorkerInterceptor::new(
Arc::new(builder.build()),
None,
None,
NotifyBuffer::new(),
Arc::new(Mutex::new(Vec::new())),
test_prompts(),
Some(Arc::new(FailingSystemItemCommitter)),
)
.with_resident_context_source(Some(source));
let mut ctx = Vec::new();
let action = interceptor
.pre_llm_request(PreLlmRequestContext {
invocation: Default::default(),
items: &mut ctx,
history: &[],
})
.await
.unwrap();
assert!(matches!(
action,
PreRequestAction::Cancel(reason) if reason.contains("session persistence failed")
));
assert!(confirmed.lock().unwrap().is_empty());
}
#[tokio::test]
async fn pre_llm_request_without_log_writer_does_not_expose_system_item_handle() {
let saw_handle = Arc::new(AtomicBool::new(false));
crates/worker/src/prompt/catalog.rs
+18
View File
@@ -105,6 +105,7 @@ pub enum WorkerPrompt {
AgentsMdSection,
ResidentMemorySummarySection,
ResidentMemoryRestoreSection,
SubjektivResidentContext,
WorkerOrchestrationGuidanceSection,
SubWorkerSpawnToolDescription,
}
@@ -128,6 +129,7 @@ impl WorkerPrompt {
Self::AgentsMdSection => "internal.agents_md_section",
Self::ResidentMemorySummarySection => "internal.resident_memory_summary_section",
Self::ResidentMemoryRestoreSection => "internal.resident_memory_restore_section",
Self::SubjektivResidentContext => "internal.subjektiv_resident_context",
Self::WorkerOrchestrationGuidanceSection => {
"internal.worker_orchestration_guidance_section"
}
@@ -150,6 +152,7 @@ impl WorkerPrompt {
WorkerPrompt::AgentsMdSection,
WorkerPrompt::ResidentMemorySummarySection,
WorkerPrompt::ResidentMemoryRestoreSection,
WorkerPrompt::SubjektivResidentContext,
WorkerPrompt::WorkerOrchestrationGuidanceSection,
WorkerPrompt::SubWorkerSpawnToolDescription,
];
@@ -518,6 +521,21 @@ impl PromptCatalog {
},
)
}
pub fn subjektiv_resident_context(
&self,
behavior_md: &str,
surface_availability: &str,
surface_body: Option<&str>,
) -> Result<String, CatalogError> {
self.render(
WorkerPrompt::SubjektivResidentContext,
minijinja::context! {
behavior_md => behavior_md,
surface_availability => surface_availability,
surface_body => surface_body.unwrap_or_default(),
},
)
}
pub fn worker_orchestration_guidance_section(&self) -> Result<String, CatalogError> {
self.render(
WorkerPrompt::WorkerOrchestrationGuidanceSection,
crates/worker/src/worker.rs
+53 -9
View File
@@ -2346,9 +2346,30 @@ pub(crate) enum SystemPromptContribution {
Unavailable,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ResidentContextRefresh {
pub body: String,
pub revision: u64,
}
#[async_trait::async_trait]
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.
async fn load_changed_resident_context(
&self,
) -> Result<Option<ResidentContextRefresh>, String> {
Ok(None)
}
/// Acknowledges that the revision's typed refresh was durably committed.
fn confirm_resident_context_revision(&self, _revision: u64) {}
/// Invalidates the in-memory representation fence before history may be
/// compacted or rewound.
fn invalidate_resident_context_representation(&self) {}
}
#[derive(Debug, Clone)]
@@ -3825,6 +3846,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
)
.with_usage_tracker(self.usage_tracker.clone())
.with_metrics_tracker(self.metrics_tracker.clone())
.with_resident_context_source(self.feature_resident_summary_source.clone())
.with_prompt_workspace_id(
self.workspace_context
.workspace_id()
@@ -9191,16 +9213,20 @@ mod build_summary_prompt_tests {
serde_json::from_str(request.body.as_deref().unwrap_or_default()).unwrap();
assert!(matches!(
request.operation,
server_api::SubjektivMemoryBackendOperation::ResidentSummary(_)
server_api::SubjektivMemoryBackendOperation::ResidentContext(_)
));
self.load_count.fetch_add(1, Ordering::SeqCst);
Ok(WorkspaceResponse {
status: 200,
body: serde_json::to_string(
&server_api::SubjektivMemoryBackendResponse::ResidentSummary(
memory::backend::MemoryResidentSummaryOutput {
availability: self.availability,
content: self.content.clone(),
&server_api::SubjektivMemoryBackendResponse::ResidentContext(
server_api::SubjektivResidentContextOutput {
behavior_md: "restore behavior".to_string(),
behavior_revision: 1,
memory_surface: memory::backend::MemoryResidentSummaryOutput {
availability: self.availability,
content: self.content.clone(),
},
},
),
)
@@ -9221,7 +9247,9 @@ mod build_summary_prompt_tests {
load_count: Arc::clone(&load_count),
});
let source = crate::feature::builtin::memory::ordinary_subjektiv_resident_summary_source(
manifest, client,
manifest,
client,
Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())),
)
.unwrap()
.unwrap();
@@ -12186,9 +12214,10 @@ permission = "write"
.unwrap()
.to_string();
assert!(
empty_boundary.contains("current ready resident memory surface is intentionally empty")
empty_boundary
.contains("current generated Memory surface is ready and intentionally empty")
);
assert!(!empty_boundary.contains("No current ready resident memory surface"));
assert!(!empty_boundary.contains("No generated Memory surface exists yet"));
drop(empty_restore);
for availability in [
@@ -12240,7 +12269,22 @@ permission = "write"
.as_text()
.unwrap()
.to_string();
assert!(tombstone.contains("No current ready resident memory surface"));
let unavailable_message = match availability {
memory::backend::MemoryResidentSummaryAvailability::Ungenerated => {
"No generated Memory surface exists yet"
}
memory::backend::MemoryResidentSummaryAvailability::Stale => {
"generated Memory surface is stale"
}
memory::backend::MemoryResidentSummaryAvailability::Failed => {
"Generation of the Memory surface failed"
}
memory::backend::MemoryResidentSummaryAvailability::Unavailable => {
"Host could not fetch the Subject resident context"
}
memory::backend::MemoryResidentSummaryAvailability::Ready => unreachable!(),
};
assert!(tombstone.contains(unavailable_message));
assert!(!tombstone.contains("latest subject surface"));
assert!(!tombstone.contains("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH"));
let entries = store.read_all(session_id, segment_id).unwrap();