1566 lines
55 KiB
Rust
1566 lines
55 KiB
Rust
//! Integration of real standalone Subject lifecycle, dialog, Journal and Memory services.
|
|
//! Only model streams are scripted; temporary storage and test-owned subprocesses isolate
|
|
//! persistence and kernel leases. Run with `cargo test -p standalone --test subjektiv`.
|
|
//! No Workspace or live model provider is used; this is not a continuity meaning evaluation.
|
|
|
|
use std::collections::{BTreeMap, VecDeque};
|
|
use std::io::{Read, Write};
|
|
use std::path::{Path, PathBuf};
|
|
use std::pin::Pin;
|
|
use std::process::{Child, Command, Stdio};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use agen::llm_client::{
|
|
LlmClient,
|
|
error::ClientError,
|
|
event::{Event as LlmEvent, StopReason},
|
|
types::{Item, Request},
|
|
};
|
|
use async_trait::async_trait;
|
|
use feature_storage::FeatureStorage;
|
|
use futures::{Stream, stream};
|
|
use manifest::{ProfileRegistrySource, ProfileSelector};
|
|
use protocol::{Event, Method};
|
|
use serde_json::{Value, json};
|
|
use standalone::subjektiv::StandaloneSubjects;
|
|
use standalone::{
|
|
ResolvedStandaloneLaunch, StandaloneHost, StandaloneLaunchConfig, StandaloneLaunchError,
|
|
StandaloneStartupError, StandaloneWorkerStore,
|
|
};
|
|
use subjektiv::{SubjectStagingRecord, SubjektivStore};
|
|
|
|
#[derive(Clone)]
|
|
struct ScriptedClient {
|
|
responses: Arc<Mutex<VecDeque<Vec<LlmEvent>>>>,
|
|
requests: Arc<Mutex<Vec<Request>>>,
|
|
consolidate: bool,
|
|
extract: bool,
|
|
commit_guard: Option<(PathBuf, String)>,
|
|
fail_after_apply: bool,
|
|
job_ids: Arc<Mutex<Vec<String>>>,
|
|
job_started: Arc<tokio::sync::Notify>,
|
|
}
|
|
|
|
impl ScriptedClient {
|
|
fn new(responses: Vec<Vec<LlmEvent>>) -> Self {
|
|
Self {
|
|
responses: Arc::new(Mutex::new(responses.into())),
|
|
requests: Arc::new(Mutex::new(Vec::new())),
|
|
consolidate: false,
|
|
extract: false,
|
|
commit_guard: None,
|
|
fail_after_apply: false,
|
|
job_ids: Arc::new(Mutex::new(Vec::new())),
|
|
job_started: Arc::new(tokio::sync::Notify::new()),
|
|
}
|
|
}
|
|
|
|
fn consolidating(responses: Vec<Vec<LlmEvent>>, fail_after_apply: bool) -> Self {
|
|
Self {
|
|
consolidate: true,
|
|
fail_after_apply,
|
|
..Self::new(responses)
|
|
}
|
|
}
|
|
|
|
async fn first_job(&self) -> String {
|
|
tokio::time::timeout(Duration::from_secs(30), self.job_started.notified())
|
|
.await
|
|
.expect("lifecycle must auto-start a generic Job");
|
|
self.job_ids.lock().unwrap()[0].clone()
|
|
}
|
|
|
|
fn job_response(&self, request: &Request) -> Result<Vec<LlmEvent>, ClientError> {
|
|
assert!(
|
|
request
|
|
.system_prompt
|
|
.as_deref()
|
|
.unwrap()
|
|
.contains("# Memory staging consolidater"),
|
|
"generic Job must render the instruction of its selected consolidation Profile"
|
|
);
|
|
for forbidden in [
|
|
"Read",
|
|
"Bash",
|
|
"SubWorkerSpawn",
|
|
"SubjektivMemoryRemember",
|
|
"SubjektivSessionRead",
|
|
] {
|
|
assert!(
|
|
!request.tools.iter().any(|tool| tool.name == forbidden),
|
|
"Job received {forbidden}"
|
|
);
|
|
}
|
|
let envelope = request
|
|
.items
|
|
.iter()
|
|
.filter_map(Item::as_text)
|
|
.find(|text| text.contains("input_json: "))
|
|
.expect("immutable Job envelope");
|
|
let input: Value = serde_json::from_str(
|
|
envelope
|
|
.lines()
|
|
.find_map(|line| line.strip_prefix("input_json: "))
|
|
.unwrap(),
|
|
)
|
|
.unwrap();
|
|
let id = envelope
|
|
.lines()
|
|
.find_map(|line| line.strip_prefix("job_id: "))
|
|
.unwrap()
|
|
.to_owned();
|
|
let mut ids = self.job_ids.lock().unwrap();
|
|
if !ids.contains(&id) {
|
|
ids.push(id);
|
|
self.job_started.notify_one();
|
|
}
|
|
drop(ids);
|
|
let result = |call: &str| {
|
|
request.items.iter().find_map(|item| match item {
|
|
Item::ToolResult {
|
|
call_id,
|
|
content,
|
|
is_error,
|
|
summary,
|
|
..
|
|
} if call_id == call => {
|
|
assert!(!is_error, "{call}: {summary}; {content:?}");
|
|
Some(content.as_deref().expect("structured domain result"))
|
|
}
|
|
_ => None,
|
|
})
|
|
};
|
|
if let Some(Item::ToolResult {
|
|
is_error,
|
|
summary,
|
|
content,
|
|
..
|
|
}) = request.items.iter().find(
|
|
|item| matches!(item, Item::ToolResult { call_id, .. } if call_id == "job-result"),
|
|
) {
|
|
assert!(
|
|
!is_error,
|
|
"structured result rejected: {summary}; {content:?}"
|
|
);
|
|
assert_eq!(summary, "Host Job result accepted");
|
|
// SubmitJobResult intentionally returns summary-only. Durable
|
|
// structured acceptance is asserted against host.jobs().wait().
|
|
return Ok(prose());
|
|
}
|
|
if let Some(applied) = result("apply") {
|
|
let response: server_api::SubjektivMemoryBackendResponse =
|
|
serde_json::from_str(applied).unwrap();
|
|
let server_api::SubjektivMemoryBackendResponse::CandidateDecided(decision) = response
|
|
else {
|
|
panic!("candidate apply rejected: {applied}");
|
|
};
|
|
assert_eq!(
|
|
decision.action,
|
|
server_api::SubjektivMemoryCandidateResolutionAction::Applied
|
|
);
|
|
assert_eq!(decision.affected_memory.len(), 1);
|
|
if self.fail_after_apply {
|
|
return Err(ClientError::Config(
|
|
"simulated model transport loss after durable candidate apply".into(),
|
|
));
|
|
}
|
|
return Ok(tool(
|
|
"job-result",
|
|
"SubmitJobResult",
|
|
json!({"result":input}),
|
|
));
|
|
}
|
|
if let Some(candidate) = result("candidate") {
|
|
let response: server_api::SubjektivMemoryBackendResponse =
|
|
serde_json::from_str(candidate).unwrap();
|
|
let server_api::SubjektivMemoryBackendResponse::Candidate(candidate) = response else {
|
|
panic!("candidate read rejected: {candidate}");
|
|
};
|
|
assert_eq!(input["candidate_ids"], json!([candidate.candidate_id]));
|
|
assert!(
|
|
!candidate.evidence.is_empty(),
|
|
"Job must receive committed provenance"
|
|
);
|
|
return Ok(tool(
|
|
"apply",
|
|
"MemoryApplyCandidate",
|
|
json!({
|
|
"request_id":format!("apply:{}", candidate.candidate_id),
|
|
"candidate_id":candidate.candidate_id, "reason":"Adopt the committed lesson",
|
|
"decision":{"kind":"apply", "target":{"kind":"create"}, "memory":{
|
|
"kind":candidate.kind, "state":"active", "claim":candidate.claim,
|
|
"body_md":candidate.claim, "why_useful":candidate.why_useful,
|
|
"change_reason":"Adopt committed standalone evidence"
|
|
}}
|
|
}),
|
|
));
|
|
}
|
|
if let Some(list) = result("candidates") {
|
|
let response: server_api::SubjektivMemoryBackendResponse =
|
|
serde_json::from_str(list).unwrap();
|
|
let server_api::SubjektivMemoryBackendResponse::Candidates(list) = response else {
|
|
panic!("candidate list rejected: {list}");
|
|
};
|
|
assert_eq!(
|
|
input["candidate_ids"],
|
|
json!(
|
|
list.items
|
|
.iter()
|
|
.map(|c| &c.candidate_id)
|
|
.collect::<Vec<_>>()
|
|
)
|
|
);
|
|
if let Some(candidate) = list.items.first() {
|
|
return Ok(tool(
|
|
"candidate",
|
|
"MemoryStagingRead",
|
|
json!({"candidate_id":candidate.candidate_id}),
|
|
));
|
|
}
|
|
return Ok(tool(
|
|
"job-result",
|
|
"SubmitJobResult",
|
|
json!({"result":input}),
|
|
));
|
|
}
|
|
Ok(tool("candidates", "MemoryStagingList", json!({})))
|
|
}
|
|
|
|
fn surface_response(&self, request: &Request) -> Vec<LlmEvent> {
|
|
assert_eq!(
|
|
request
|
|
.tools
|
|
.iter()
|
|
.map(|t| t.name.as_str())
|
|
.collect::<Vec<_>>(),
|
|
["SubmitMemorySurface"]
|
|
);
|
|
assert!(
|
|
!request.items.iter().any(
|
|
|item| matches!(item, Item::ToolCall { name, .. } if name == "MemoryApplyCandidate")
|
|
),
|
|
"surface editing must use a clean context, not consolidation history"
|
|
);
|
|
if let Some(Item::ToolResult {
|
|
is_error,
|
|
summary,
|
|
content,
|
|
..
|
|
}) = request
|
|
.items
|
|
.iter()
|
|
.find(|item| matches!(item, Item::ToolResult { call_id, .. } if call_id == "surface"))
|
|
{
|
|
assert!(!is_error, "surface rejected: {summary}; {content:?}");
|
|
assert_eq!(summary, "Accepted 1 grounded Memory surface point(s).");
|
|
return prose();
|
|
}
|
|
let input: Value = request
|
|
.items
|
|
.iter()
|
|
.filter_map(Item::as_text)
|
|
.filter_map(|text| serde_json::from_str::<Value>(text).ok())
|
|
.find(|value| value.get("materials").is_some())
|
|
.expect("bounded confirmed surface materials");
|
|
let materials = input["materials"].as_array().unwrap();
|
|
assert_eq!(
|
|
materials.len(),
|
|
1,
|
|
"only the confirmed lesson belongs in this surface"
|
|
);
|
|
let material = &materials[0];
|
|
tool(
|
|
"surface",
|
|
"SubmitMemorySurface",
|
|
json!({"points":[{
|
|
"body_md":format!("- {}", material["body_md"].as_str().unwrap()),
|
|
"memory_refs":[{"memory_id":material["memory_id"], "change_id":material["change_id"]}]
|
|
}]}),
|
|
)
|
|
}
|
|
|
|
fn extract_response(&self, request: &Request) -> Vec<LlmEvent> {
|
|
assert!(self.extract, "unexpected extract request");
|
|
for forbidden in [
|
|
"Bash",
|
|
"SubWorkerSpawn",
|
|
"MemoryApplyCandidate",
|
|
"SubmitJobResult",
|
|
"SubjektivMemoryRemember",
|
|
] {
|
|
assert!(
|
|
!request.tools.iter().any(|tool| tool.name == forbidden),
|
|
"extract received {forbidden}"
|
|
);
|
|
}
|
|
let result = |id: &str| {
|
|
request.items.iter().find_map(|item| match item {
|
|
Item::ToolResult {
|
|
call_id,
|
|
is_error,
|
|
summary,
|
|
..
|
|
} if call_id == id => {
|
|
assert!(!is_error, "extract {id}: {summary}");
|
|
Some(summary)
|
|
}
|
|
_ => None,
|
|
})
|
|
};
|
|
if result("extract-finish").is_some() {
|
|
return prose();
|
|
}
|
|
if result("extract-stage").is_some() {
|
|
return tool(
|
|
"extract-finish",
|
|
"FinishMemoryExtraction",
|
|
json!({"staged_count":1}),
|
|
);
|
|
}
|
|
let input = request
|
|
.items
|
|
.iter()
|
|
.filter_map(Item::as_text)
|
|
.find(|text| text.contains("# Initial session entry index"))
|
|
.expect("committed extract view");
|
|
let entry = input
|
|
.lines()
|
|
.filter(|line| line.contains("A useful lesson is worth retaining across Workers."))
|
|
.find_map(|line| {
|
|
line.strip_prefix("- [").and_then(|line| {
|
|
let (entry, rest) = line.split_once(' ')?;
|
|
(rest.starts_with("user ") || rest.starts_with("user]")).then_some(entry)
|
|
})
|
|
})
|
|
.expect("user input reference");
|
|
tool(
|
|
"extract-stage",
|
|
"StageMemoryCandidate",
|
|
json!({
|
|
"kind":"lesson", "claim":"Extract a committed lesson through the normal lifecycle",
|
|
"why_useful":"Normal extraction must retain Host-validated provenance", "entry_refs":[entry]
|
|
}),
|
|
)
|
|
}
|
|
|
|
fn requests(&self) -> Vec<Request> {
|
|
self.requests.lock().unwrap().clone()
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl LlmClient for ScriptedClient {
|
|
async fn stream(
|
|
&self,
|
|
request: Request,
|
|
) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
|
|
{
|
|
// Journal is a separate scripted model boundary. Successful production recording
|
|
// is required before these Memory scenarios may hand the Subject to a new body.
|
|
if request
|
|
.tools
|
|
.iter()
|
|
.any(|tool| tool.name == "ReadJournalState")
|
|
{
|
|
return Ok(Box::pin(stream::iter(
|
|
journal_response(&request).into_iter().map(Ok),
|
|
)));
|
|
}
|
|
// Lifecycle-triggered consolidation uses a separate Internal Worker. It
|
|
// must not consume the dialog script or silently become a live request.
|
|
let job = request
|
|
.tools
|
|
.iter()
|
|
.any(|tool| tool.name == "SubmitJobResult");
|
|
self.requests.lock().unwrap().push(request.clone());
|
|
if let Some((state, subject)) = &self.commit_guard {
|
|
if request
|
|
.items
|
|
.iter()
|
|
.any(|item| matches!(item,Item::ToolResult {call_id,..} if call_id=="remember"))
|
|
{
|
|
inspect_store(state, |store| {
|
|
assert!(
|
|
store
|
|
.pending_staging_candidates(subject, 100)
|
|
.unwrap()
|
|
.is_empty(),
|
|
"the open Run must not yet publish an explicit candidate"
|
|
)
|
|
});
|
|
}
|
|
}
|
|
let response = if job {
|
|
if !self.consolidate {
|
|
return Err(ClientError::Config("unscripted background Job".into()));
|
|
}
|
|
self.job_response(&request)?
|
|
} else if request
|
|
.tools
|
|
.iter()
|
|
.any(|tool| tool.name == "StageMemoryCandidate")
|
|
{
|
|
self.extract_response(&request)
|
|
} else if request
|
|
.tools
|
|
.iter()
|
|
.any(|tool| tool.name == "SubmitMemorySurface")
|
|
{
|
|
assert!(self.consolidate, "unexpected surface model invocation");
|
|
self.surface_response(&request)
|
|
} else {
|
|
self.responses
|
|
.lock()
|
|
.unwrap()
|
|
.pop_front()
|
|
.expect("unexpected model request: dialog script exhausted")
|
|
};
|
|
Ok(Box::pin(stream::iter(response.into_iter().map(Ok))))
|
|
}
|
|
|
|
fn clone_boxed(&self) -> Box<dyn LlmClient> {
|
|
Box::new(self.clone())
|
|
}
|
|
}
|
|
|
|
// The outer model script is not a continuity meaning evaluation. Persistence, complete
|
|
// source delivery and the handoff gate remain the production implementations.
|
|
fn journal_response(request: &Request) -> Vec<LlmEvent> {
|
|
let result = |id: &str| -> Option<Value> {
|
|
request.items.iter().find_map(|item| match item {
|
|
Item::ToolResult {
|
|
call_id,
|
|
content,
|
|
is_error,
|
|
summary,
|
|
..
|
|
} if call_id == id => {
|
|
assert!(!is_error, "{id}: {summary}; {content:?}");
|
|
Some(
|
|
content
|
|
.as_deref()
|
|
.map(|v| serde_json::from_str(v).unwrap())
|
|
.unwrap_or(Value::Null),
|
|
)
|
|
}
|
|
_ => None,
|
|
})
|
|
};
|
|
if result("journal-submit").is_some() {
|
|
return prose();
|
|
}
|
|
if result("journal-checkpoint").is_some() {
|
|
return tool("journal-submit", "SubmitJobResult", json!({"result":{}}));
|
|
}
|
|
let Some(state) = result("journal-state") else {
|
|
return tool("journal-state", "ReadJournalState", json!({}));
|
|
};
|
|
let state = &state["value"];
|
|
let generation = state["generation_id"].as_str().unwrap();
|
|
if let Some(position) = result("journal-position") {
|
|
return tool(
|
|
"journal-checkpoint",
|
|
"CheckpointJournal",
|
|
json!({
|
|
"request_id":format!("checkpoint:{generation}"), "expected_state":position["value"]["state_token"],
|
|
"action":{"action":"checkpoint","record_operation_ids":[format!("record:{generation}")]},
|
|
"covered_ranges":state["selected_ranges"]
|
|
}),
|
|
);
|
|
}
|
|
if let Some(record) = result("journal-record") {
|
|
return tool(
|
|
"journal-position",
|
|
"UpdateJournalPosition",
|
|
json!({
|
|
"request_id":format!("position:{generation}"), "expected_state":record["value"]["state_token"],
|
|
"action":{"action":"update_position","current_position":"Memory scenario recorded; represented work may remain unfinished"}, "covered_ranges":[]
|
|
}),
|
|
);
|
|
}
|
|
let input: Value = request
|
|
.items
|
|
.iter()
|
|
.filter_map(Item::as_text)
|
|
.find_map(|text| {
|
|
text.lines()
|
|
.find_map(|line| line.strip_prefix("input_json: "))
|
|
})
|
|
.map(|json| serde_json::from_str(json).unwrap())
|
|
.expect("saved direct Journal input");
|
|
let input = subjektiv::journal::recording::JournalInput::decode(&input).unwrap();
|
|
assert_eq!(
|
|
serde_json::to_value(&input.selected_ranges).unwrap(),
|
|
state["selected_ranges"]
|
|
);
|
|
assert!(
|
|
input
|
|
.capture
|
|
.entries
|
|
.iter()
|
|
.all(|entry| input.capture.recording_text(entry.order).is_ok())
|
|
);
|
|
let (name, action) = if state["current_open"].is_null() {
|
|
(
|
|
"CreateJournalEntry",
|
|
json!({"action":"create","title":"Memory conversation","body":"Conversation recorded by the scripted model fixture","ongoing":"Unfinished work is not declared complete","state":"open","source_end":null,"correction_refs":[]}),
|
|
)
|
|
} else {
|
|
(
|
|
"RewriteJournalEntry",
|
|
json!({"action":"rewrite","entry_id":state["current_open"]["id"],"title":"Memory conversation","body":"Further conversation recorded by the scripted model fixture","ongoing":"Unfinished work is not declared complete"}),
|
|
)
|
|
};
|
|
tool(
|
|
"journal-record",
|
|
name,
|
|
json!({
|
|
"request_id":format!("record:{generation}"), "expected_state":state["current_journal"]["state_token"],
|
|
"action":action,"covered_ranges":state["selected_ranges"]
|
|
}),
|
|
)
|
|
}
|
|
|
|
async fn disconnect_recorded(host: &mut StandaloneHost) {
|
|
let mut last_error = None;
|
|
let completed = tokio::time::timeout(Duration::from_secs(30), async {
|
|
loop {
|
|
match host.disconnect_subject().await {
|
|
Ok(()) => return,
|
|
Err(error) => {
|
|
last_error = Some(error.to_string());
|
|
tokio::task::yield_now().await;
|
|
}
|
|
}
|
|
}
|
|
})
|
|
.await;
|
|
assert!(
|
|
completed.is_ok(),
|
|
"actual Idle/Journal completion is required before Memory scenario handoff: {last_error:?}"
|
|
);
|
|
}
|
|
|
|
fn prose() -> Vec<LlmEvent> {
|
|
vec![
|
|
LlmEvent::text_block_start(0),
|
|
LlmEvent::text_delta(0, "scripted answer"),
|
|
LlmEvent::text_block_stop(0, Some(StopReason::EndTurn)),
|
|
]
|
|
}
|
|
|
|
fn tool(id: &str, name: &str, input: Value) -> Vec<LlmEvent> {
|
|
vec![
|
|
LlmEvent::tool_use_start(0, id, name),
|
|
LlmEvent::tool_input_delta(0, input.to_string()),
|
|
LlmEvent::tool_use_stop(0),
|
|
]
|
|
}
|
|
|
|
fn launch(cwd: &Path, state: &Path, subject: &str, name: &str) -> ResolvedStandaloneLaunch {
|
|
StandaloneLaunchConfig::new(
|
|
cwd,
|
|
state,
|
|
ProfileSelector::source_named(ProfileRegistrySource::Builtin, "standalone-subjektiv"),
|
|
name,
|
|
)
|
|
.with_subject(subject.to_owned())
|
|
.resolve()
|
|
.expect("resolve selected Subject launch")
|
|
}
|
|
|
|
async fn run(host: &StandaloneHost, text: &str) -> BTreeMap<String, Value> {
|
|
let mut client = host.connect();
|
|
client
|
|
.send(&Method::submit_text(
|
|
protocol::new_submission_request_id(),
|
|
text,
|
|
))
|
|
.await
|
|
.expect("submit dialog input");
|
|
let mut event_trace = Vec::new();
|
|
let outcome = tokio::time::timeout(Duration::from_secs(30), async {
|
|
let mut results = BTreeMap::new();
|
|
let mut answered = false;
|
|
loop {
|
|
let event = client
|
|
.next_event()
|
|
.await
|
|
.expect("protocol stream")
|
|
.expect("event");
|
|
if event_trace.len() < 100 {
|
|
event_trace.push(format!("{event:?}"));
|
|
}
|
|
match event {
|
|
Event::ToolResult {
|
|
id,
|
|
output,
|
|
summary,
|
|
is_error,
|
|
..
|
|
} => {
|
|
assert!(!is_error, "{id}: {summary}; {output:?}");
|
|
let value = serde_json::from_str(&output.expect("structured tool output"))
|
|
.expect("tool output JSON");
|
|
results.insert(id, value);
|
|
}
|
|
Event::Error { code, message } => {
|
|
panic!("dialog {text:?} failed: {code:?}: {message}")
|
|
}
|
|
Event::TextDelta { text } if text.contains("scripted answer") => answered = true,
|
|
Event::RunEnd { .. } => {
|
|
assert!(
|
|
answered,
|
|
"a failed/interrupted Run is not successful execution"
|
|
);
|
|
return results;
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
})
|
|
.await;
|
|
outcome.unwrap_or_else(|error| {
|
|
panic!("dialog {text:?} did not complete: {error}; events: {event_trace:#?}")
|
|
})
|
|
}
|
|
|
|
// The common typed store is only an assertion oracle here. Mutations under
|
|
// test always go through operator admin or the Host/model capability.
|
|
fn inspect_store<T>(state: &Path, inspect: impl FnOnce(&SubjektivStore) -> T) -> T {
|
|
let root = state.join("subjektiv");
|
|
let scope_id: String =
|
|
serde_json::from_slice(&std::fs::read(root.join("scope.json")).unwrap()).unwrap();
|
|
let manager = FeatureStorage::new(root.join("features"));
|
|
let scope = manager.scope(&scope_id).unwrap();
|
|
let registration = SubjektivStore::register(&scope).unwrap();
|
|
let store = SubjektivStore::open(&scope, ®istration).unwrap();
|
|
let result = inspect(&store);
|
|
drop(store);
|
|
manager.shutdown().unwrap();
|
|
result
|
|
}
|
|
|
|
#[test]
|
|
fn operator_subject_catalog_survives_close_and_reopen() {
|
|
let temp = tempfile::tempdir().unwrap();
|
|
let state = temp.path().join("state");
|
|
let catalog = StandaloneSubjects::open(&state).unwrap();
|
|
assert!(catalog.list().unwrap().is_empty());
|
|
let first = catalog
|
|
.create("coding", Some("Keep precise evidence."))
|
|
.unwrap();
|
|
let second = catalog.create("coding", None).unwrap();
|
|
assert_ne!(
|
|
first.id, second.id,
|
|
"role/Profile/cwd must not derive Subject identity"
|
|
);
|
|
uuid::Uuid::parse_str(
|
|
first
|
|
.id
|
|
.strip_prefix("subject-")
|
|
.expect("common Subject issuer"),
|
|
)
|
|
.unwrap();
|
|
uuid::Uuid::parse_str(
|
|
second
|
|
.id
|
|
.strip_prefix("subject-")
|
|
.expect("common Subject issuer"),
|
|
)
|
|
.unwrap();
|
|
drop(catalog);
|
|
let reopened = StandaloneSubjects::open(&state).unwrap();
|
|
let items = reopened.list().unwrap();
|
|
assert_eq!(items.len(), 2);
|
|
assert_eq!(items.iter().find(|s| s.id == first.id).unwrap(), &first);
|
|
assert_eq!(items.iter().find(|s| s.id == second.id).unwrap(), &second);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn enabled_profile_without_subject_is_inert_and_disabled_selection_is_rejected() {
|
|
let temp = tempfile::tempdir().unwrap();
|
|
for profile in ["standalone-subjektiv", "standalone"] {
|
|
let state = temp.path().join(profile);
|
|
let launch = StandaloneLaunchConfig::new(
|
|
temp.path(),
|
|
&state,
|
|
ProfileSelector::source_named(ProfileRegistrySource::Builtin, profile),
|
|
"unattached",
|
|
)
|
|
.resolve()
|
|
.unwrap();
|
|
let model = ScriptedClient::new(vec![prose()]);
|
|
let host = StandaloneHost::start_with_model_client(launch, model.clone())
|
|
.await
|
|
.unwrap();
|
|
assert!(host.record().subject.is_none());
|
|
run(
|
|
&host,
|
|
"An unattached policy must not manufacture a Subject.",
|
|
)
|
|
.await;
|
|
let request = &model.requests()[0];
|
|
assert!(request.tools.iter().any(|t| t.name == "Read"));
|
|
assert!(
|
|
!request
|
|
.tools
|
|
.iter()
|
|
.any(|t| t.name.starts_with("Subjektiv"))
|
|
);
|
|
assert!(host.jobs().pending().unwrap().is_empty());
|
|
host.shutdown().await.unwrap();
|
|
assert!(
|
|
!state.join("subjektiv").exists(),
|
|
"policy alone must have no Subject storage side effect"
|
|
);
|
|
}
|
|
let state = temp.path().join("rejected");
|
|
let selected = uuid::Uuid::now_v7().to_string();
|
|
let error = StandaloneLaunchConfig::new(
|
|
temp.path(),
|
|
&state,
|
|
ProfileSelector::source_named(ProfileRegistrySource::Builtin, "standalone"),
|
|
"disabled",
|
|
)
|
|
.with_subject(selected)
|
|
.resolve()
|
|
.err()
|
|
.expect("disabled selection rejected");
|
|
assert_eq!(error, StandaloneLaunchError::InvalidSubjectPolicy);
|
|
assert!(!state.exists(), "admission rejection precedes persistence");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn explicit_remember_is_committed_staging_and_sessions_follow_subject_across_workers() {
|
|
let temp = tempfile::tempdir().unwrap();
|
|
let state = temp.path().join("state");
|
|
let catalog = StandaloneSubjects::open(&state).unwrap();
|
|
let subject = catalog
|
|
.create("coding", Some("Keep committed references."))
|
|
.unwrap();
|
|
let foreign = catalog.create("other", None).unwrap();
|
|
drop(catalog);
|
|
let claim = "Use immutable committed references for test evidence";
|
|
let mut model = ScriptedClient::new(vec![
|
|
tool(
|
|
"remember",
|
|
"SubjektivMemoryRemember",
|
|
json!({
|
|
"kind":"lesson", "claim":claim, "why_useful":"Avoid uncommitted evidence"
|
|
}),
|
|
),
|
|
tool("query", "SubjektivMemoryQuery", json!({"query":claim})),
|
|
prose(),
|
|
]);
|
|
model.commit_guard = Some((state.clone(), subject.id.clone()));
|
|
let host = StandaloneHost::start_with_model_client(
|
|
launch(temp.path(), &state, &subject.id, "first-body"),
|
|
model.clone(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let binding = host.record().subject.clone().unwrap();
|
|
let worker_id = host.worker_id();
|
|
let session_id = host.record().active_session_id;
|
|
assert_eq!(binding.subject_id, subject.id);
|
|
assert!(
|
|
host.record()
|
|
.manifest
|
|
.feature
|
|
.subjektiv
|
|
.workspace_settings
|
|
.is_none()
|
|
);
|
|
let results = run(
|
|
&host,
|
|
"Remember this lesson using committed model-origin evidence.",
|
|
)
|
|
.await;
|
|
assert_eq!(results["remember"]["status"], "pending_commit");
|
|
assert_eq!(
|
|
results["query"]["items"],
|
|
json!([]),
|
|
"remember never writes confirmed Memory"
|
|
);
|
|
let request = &model.requests()[0];
|
|
for name in [
|
|
"SubjektivMemoryRemember",
|
|
"SubjektivMemoryQuery",
|
|
"SubjektivSessionSearch",
|
|
"SubjektivSessionRead",
|
|
] {
|
|
assert!(
|
|
request.tools.iter().any(|t| t.name == name),
|
|
"missing {name}"
|
|
);
|
|
}
|
|
assert!(
|
|
!request
|
|
.tools
|
|
.iter()
|
|
.any(|t| t.name == "MemoryApplyCandidate"),
|
|
"body must not receive consolidation authority"
|
|
);
|
|
host.shutdown().await.unwrap();
|
|
|
|
let staged: SubjectStagingRecord = inspect_store(&state, |store| {
|
|
assert!(store.list_memories(&subject.id).unwrap().is_empty());
|
|
let candidates = store.pending_staging_candidates(&subject.id, 100).unwrap();
|
|
assert_eq!(
|
|
candidates.len(),
|
|
1,
|
|
"committed remember must become staging, not disappear"
|
|
);
|
|
let attribution = store
|
|
.session_attribution(&session_id.to_string())
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(attribution.subject_id, subject.id);
|
|
assert_eq!(attribution.runtime_id, binding.scope_id);
|
|
assert_eq!(attribution.worker_id, worker_id.to_string());
|
|
candidates.into_iter().next().unwrap()
|
|
});
|
|
assert_eq!(staged.claim, claim);
|
|
assert_eq!(staged.subject_id, subject.id);
|
|
assert!(!staged.evidence.is_empty());
|
|
assert_eq!(staged.evidence.len(), staged.source_refs.len());
|
|
let workers = StandaloneWorkerStore::open(&state).unwrap();
|
|
let index = session_store::public_index::read_fs_session_public_index(
|
|
&workers.sessions_dir(worker_id),
|
|
session_id,
|
|
Default::default(),
|
|
)
|
|
.unwrap();
|
|
for evidence in &staged.evidence {
|
|
let source = staged
|
|
.source_refs
|
|
.iter()
|
|
.find(|s| s.evidence_id.as_deref() == Some(&evidence.id))
|
|
.unwrap();
|
|
assert_eq!(
|
|
source.session_id.as_deref(),
|
|
Some(session_id.to_string().as_str())
|
|
);
|
|
let entry = index
|
|
.segments
|
|
.iter()
|
|
.filter(|segment| source.segment_id.as_deref() == Some(segment.segment_id.as_str()))
|
|
.flat_map(|segment| &segment.entries)
|
|
.find(|entry| entry.entry_ref == evidence.id)
|
|
.expect("exact committed public reference");
|
|
assert!(entry.compact_text.contains(claim));
|
|
let [start, end] = evidence.entry_range.expect("committed range");
|
|
assert!(
|
|
start <= end,
|
|
"committed source ranges use inclusive endpoints"
|
|
);
|
|
assert_eq!(source.entry_range, evidence.entry_range);
|
|
let origin = evidence.origin.as_ref().unwrap();
|
|
assert_eq!(origin.kind, memory::schema::EvidenceOriginKind::ModelOutput);
|
|
assert!(
|
|
origin.runtime_id.is_none(),
|
|
"local scope is not a Runtime identity"
|
|
);
|
|
assert_eq!(
|
|
origin.worker_id.as_deref(),
|
|
Some(worker_id.to_string().as_str())
|
|
);
|
|
assert!(origin.workspace_id.is_none());
|
|
assert!(origin.account_id.is_none());
|
|
assert_eq!(source.origin.as_ref(), Some(origin));
|
|
}
|
|
let source = &staged.source_refs[0];
|
|
let read_input = json!({
|
|
"session_id":session_id.to_string(), "segment_id":source.segment_id,
|
|
"entry_ref":source.evidence_id,
|
|
});
|
|
let restored_model = ScriptedClient::new(vec![
|
|
tool("recall", "SubjektivSessionRead", read_input.clone()),
|
|
tool("query-restored", "SubjektivMemoryQuery", json!({})),
|
|
prose(),
|
|
]);
|
|
let mut restored =
|
|
StandaloneHost::restore_with_model_client(state.clone(), worker_id, restored_model.clone())
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(restored.record().subject.as_ref(), Some(&binding));
|
|
assert_eq!(restored.record().active_session_id, session_id);
|
|
let results = run(&restored, "Resume and recall the committed evidence.").await;
|
|
assert_eq!(results["recall"]["status"], "ok");
|
|
assert!(results["recall"].to_string().contains(claim));
|
|
assert_eq!(results["query-restored"]["items"], json!([]));
|
|
assert!(format!("{:?}", restored_model.requests()[0].items).contains("Remember this lesson"));
|
|
disconnect_recorded(&mut restored).await;
|
|
restored.shutdown().await.unwrap();
|
|
|
|
let second_model = ScriptedClient::new(vec![
|
|
tool("prior-session", "SubjektivSessionRead", read_input.clone()),
|
|
tool("sessions", "SubjektivSessionList", json!({})),
|
|
prose(),
|
|
]);
|
|
let second = StandaloneHost::start_with_model_client(
|
|
launch(temp.path(), &state, &subject.id, "second-body"),
|
|
second_model,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_ne!(second.worker_id(), worker_id);
|
|
assert_eq!(second.record().subject.as_ref(), Some(&binding));
|
|
let second_session = second.record().active_session_id;
|
|
let results = run(
|
|
&second,
|
|
"Recall from another Worker, without a new Memory candidate.",
|
|
)
|
|
.await;
|
|
assert_eq!(results["prior-session"]["status"], "ok");
|
|
assert!(results["prior-session"].to_string().contains(claim));
|
|
let sessions = results["sessions"].to_string();
|
|
assert!(sessions.contains(&session_id.to_string()));
|
|
assert!(
|
|
sessions.contains(&second_session.to_string()),
|
|
"Sessions with no candidates must be attributed too"
|
|
);
|
|
second.shutdown().await.unwrap();
|
|
inspect_store(&state, |store| {
|
|
assert_eq!(
|
|
serde_json::to_value(store.pending_staging_candidates(&subject.id, 100).unwrap())
|
|
.unwrap(),
|
|
json!([staged])
|
|
);
|
|
assert_eq!(
|
|
store
|
|
.session_attribution(&second_session.to_string())
|
|
.unwrap()
|
|
.unwrap()
|
|
.subject_id,
|
|
subject.id
|
|
);
|
|
});
|
|
|
|
let other = StandaloneHost::start_with_model_client(
|
|
launch(temp.path(), &state, &foreign.id, "foreign-body"),
|
|
ScriptedClient::new(vec![
|
|
tool("foreign", "SubjektivSessionRead", read_input),
|
|
prose(),
|
|
]),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let results = run(
|
|
&other,
|
|
"A known foreign Session reference is not authority.",
|
|
)
|
|
.await;
|
|
assert_eq!(results["foreign"]["status"], "error");
|
|
assert!(!results["foreign"].to_string().contains(claim));
|
|
other.shutdown().await.unwrap();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn restore_keeps_saved_subject_and_rejects_replaced_storage_scope() {
|
|
let temp = tempfile::tempdir().unwrap();
|
|
let state = temp.path().join("state");
|
|
let catalog = StandaloneSubjects::open(&state).unwrap();
|
|
let subject = catalog.create("original", None).unwrap();
|
|
let other = catalog.create("not-a-resume-selection", None).unwrap();
|
|
drop(catalog);
|
|
let host = StandaloneHost::start_with_model_client(
|
|
launch(temp.path(), &state, &subject.id, "saved-binding"),
|
|
ScriptedClient::new(Vec::new()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let id = host.worker_id();
|
|
let binding = host.record().subject.clone().unwrap();
|
|
host.shutdown().await.unwrap();
|
|
|
|
// Replace only the operator-managed storage root with a genuine, separately
|
|
// issued scope. The saved Worker/cwd/session remains otherwise untouched.
|
|
let foreign_state = temp.path().join("foreign-state");
|
|
let catalog = StandaloneSubjects::open(&foreign_state).unwrap();
|
|
catalog.create("foreign", None).unwrap();
|
|
drop(catalog);
|
|
let original_root = temp.path().join("saved-subjektiv");
|
|
std::fs::rename(state.join("subjektiv"), &original_root).unwrap();
|
|
std::fs::rename(foreign_state.join("subjektiv"), state.join("subjektiv")).unwrap();
|
|
let error = StandaloneHost::restore_with_model_client(
|
|
state.clone(),
|
|
id,
|
|
ScriptedClient::new(Vec::new()),
|
|
)
|
|
.await
|
|
.err()
|
|
.expect("foreign scope rejected");
|
|
assert!(
|
|
matches!(error, StandaloneStartupError::Subject(ref message) if message.contains("storage scope changed")),
|
|
"{error:?}"
|
|
);
|
|
assert_eq!(
|
|
StandaloneWorkerStore::open(&state)
|
|
.unwrap()
|
|
.load(id)
|
|
.unwrap()
|
|
.subject
|
|
.as_ref(),
|
|
Some(&binding)
|
|
);
|
|
std::fs::rename(state.join("subjektiv"), foreign_state.join("subjektiv")).unwrap();
|
|
std::fs::rename(&original_root, state.join("subjektiv")).unwrap();
|
|
let restored =
|
|
StandaloneHost::restore_with_model_client(state, id, ScriptedClient::new(Vec::new()))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(restored.record().subject.as_ref(), Some(&binding));
|
|
assert_ne!(
|
|
restored.record().subject.as_ref().unwrap().subject_id,
|
|
other.id
|
|
);
|
|
restored.shutdown().await.unwrap();
|
|
}
|
|
|
|
// A failing test never leaves its test-owned process running. This guard can
|
|
// only address the Child returned by spawn, never a parent or dogfood Worker.
|
|
struct OwnedChild(Child);
|
|
impl Drop for OwnedChild {
|
|
fn drop(&mut self) {
|
|
if self.0.try_wait().ok().flatten().is_none() {
|
|
let _ = self.0.kill();
|
|
}
|
|
let _ = self.0.wait();
|
|
}
|
|
}
|
|
|
|
fn child_command(mode: &str, cwd: &Path, state: &Path, subject: &str, ready: &Path) -> Command {
|
|
let mut command = Command::new(std::env::current_exe().unwrap());
|
|
command
|
|
.args(["--exact", "subject_process_child", "--nocapture"])
|
|
.env("YOI_SUBJEKTIV_TEST_CHILD_MODE", mode)
|
|
.env("YOI_SUBJEKTIV_TEST_CWD", cwd)
|
|
.env("YOI_SUBJEKTIV_TEST_STATE", state)
|
|
.env("YOI_SUBJEKTIV_TEST_SUBJECT", subject)
|
|
.env("YOI_SUBJEKTIV_TEST_READY", ready)
|
|
.stdin(Stdio::piped());
|
|
command
|
|
}
|
|
|
|
fn wait_child(child: &mut OwnedChild) -> std::process::ExitStatus {
|
|
let deadline = Instant::now() + Duration::from_secs(30);
|
|
loop {
|
|
if let Some(status) = child.0.try_wait().unwrap() {
|
|
return status;
|
|
}
|
|
assert!(Instant::now() < deadline, "test-owned child did not exit");
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn subject_process_child() {
|
|
// The ordinary harness also discovers this subprocess entry point. Only a
|
|
// parent semantic test supplies the private fixture paths and mode.
|
|
let Ok(mode) = std::env::var("YOI_SUBJEKTIV_TEST_CHILD_MODE") else {
|
|
return;
|
|
};
|
|
let cwd = PathBuf::from(std::env::var_os("YOI_SUBJEKTIV_TEST_CWD").unwrap());
|
|
let state = PathBuf::from(std::env::var_os("YOI_SUBJEKTIV_TEST_STATE").unwrap());
|
|
let subject = std::env::var("YOI_SUBJEKTIV_TEST_SUBJECT").unwrap();
|
|
let result = StandaloneHost::start_with_model_client(
|
|
launch(&cwd, &state, &subject, "test-process"),
|
|
ScriptedClient::new(Vec::new()),
|
|
)
|
|
.await;
|
|
match mode.as_str() {
|
|
"contender" => {
|
|
let error = result
|
|
.err()
|
|
.expect("another process owns the Subject kernel lease");
|
|
assert!(
|
|
matches!(error, StandaloneStartupError::Subject(ref message) if message.contains("already active")),
|
|
"{error:?}"
|
|
);
|
|
}
|
|
"abnormal-owner" => {
|
|
let host = result.expect("child owns Subject");
|
|
let ready = PathBuf::from(std::env::var_os("YOI_SUBJEKTIV_TEST_READY").unwrap());
|
|
let mut file = std::fs::File::create(&ready).unwrap();
|
|
file.write_all(host.worker_id().to_string().as_bytes())
|
|
.unwrap();
|
|
file.sync_all().unwrap();
|
|
let mut byte = [0];
|
|
std::io::stdin().read_exact(&mut byte).unwrap();
|
|
assert_eq!(byte, [b'x']);
|
|
// Deliberately omit Host shutdown and all Rust destructors. The
|
|
// parent requests this exit only after proving cross-process lock.
|
|
std::process::exit(86);
|
|
}
|
|
other => panic!("unknown child mode {other}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn kernel_subject_lease_excludes_other_processes_and_recovers_after_abnormal_exit() {
|
|
let temp = tempfile::tempdir().unwrap();
|
|
let state = temp.path().join("state");
|
|
let parent_cwd = temp.path().join("parent-project");
|
|
let child_cwd = temp.path().join("child-project");
|
|
std::fs::create_dir(&parent_cwd).unwrap();
|
|
std::fs::create_dir(&child_cwd).unwrap();
|
|
let catalog = StandaloneSubjects::open(&state).unwrap();
|
|
let subject = catalog.create("singleton", None).unwrap();
|
|
let independent = catalog.create("independent", None).unwrap();
|
|
drop(catalog);
|
|
let ready = temp.path().join("ready");
|
|
let mut parent = StandaloneHost::start_with_model_client(
|
|
launch(&parent_cwd, &state, &subject.id, "parent-owner"),
|
|
ScriptedClient::new(Vec::new()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let mut contender = OwnedChild(
|
|
child_command("contender", &child_cwd, &state, &subject.id, &ready)
|
|
.spawn()
|
|
.unwrap(),
|
|
);
|
|
assert!(
|
|
wait_child(&mut contender).success(),
|
|
"a different cwd must not evade the Subject lock"
|
|
);
|
|
let independent_host = StandaloneHost::start_with_model_client(
|
|
launch(&child_cwd, &state, &independent.id, "independent-owner"),
|
|
ScriptedClient::new(Vec::new()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
independent_host.shutdown().await.unwrap();
|
|
disconnect_recorded(&mut parent).await;
|
|
parent.shutdown().await.unwrap();
|
|
|
|
let mut owner = OwnedChild(
|
|
child_command("abnormal-owner", &child_cwd, &state, &subject.id, &ready)
|
|
.spawn()
|
|
.unwrap(),
|
|
);
|
|
let deadline = Instant::now() + Duration::from_secs(30);
|
|
let id = loop {
|
|
if ready.exists() {
|
|
let text = std::fs::read_to_string(&ready).unwrap();
|
|
if let Ok(id) = text.parse::<standalone::WorkerId>() {
|
|
break id;
|
|
}
|
|
}
|
|
assert!(
|
|
owner.0.try_wait().unwrap().is_none(),
|
|
"owner exited before ready"
|
|
);
|
|
assert!(Instant::now() < deadline, "owner did not become ready");
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
};
|
|
let mut contender = OwnedChild(
|
|
child_command("contender", &parent_cwd, &state, &subject.id, &ready)
|
|
.spawn()
|
|
.unwrap(),
|
|
);
|
|
assert!(wait_child(&mut contender).success());
|
|
owner.0.stdin.as_mut().unwrap().write_all(b"x").unwrap();
|
|
assert_eq!(
|
|
wait_child(&mut owner).code(),
|
|
Some(86),
|
|
"exercise process death, not graceful shutdown"
|
|
);
|
|
let recovered =
|
|
StandaloneHost::restore_with_model_client(state, id, ScriptedClient::new(Vec::new()))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(recovered.worker_id(), id);
|
|
assert_eq!(
|
|
recovered.record().subject.as_ref().unwrap().subject_id,
|
|
subject.id
|
|
);
|
|
recovered.shutdown().await.unwrap();
|
|
}
|
|
|
|
fn remember_script(claim: &str) -> Vec<Vec<LlmEvent>> {
|
|
vec![
|
|
tool(
|
|
"remember",
|
|
"SubjektivMemoryRemember",
|
|
json!({
|
|
"kind":"lesson", "claim":claim, "why_useful":"Retain standalone evidence"
|
|
}),
|
|
),
|
|
prose(),
|
|
]
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn automatic_generic_consolidation_applies_candidate_publishes_surface_and_follows_subject() {
|
|
let temp = tempfile::tempdir().unwrap();
|
|
let state = temp.path().join("state");
|
|
let catalog = StandaloneSubjects::open(&state).unwrap();
|
|
let subject = catalog.create("consolidation", None).unwrap();
|
|
drop(catalog);
|
|
let claim = "Validate the immutable result before reporting success";
|
|
let model = ScriptedClient::consolidating(remember_script(claim), false);
|
|
let mut host = StandaloneHost::start_with_model_client(
|
|
launch(temp.path(), &state, &subject.id, "consolidation-body"),
|
|
model.clone(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let worker_id = host.worker_id();
|
|
let results = run(
|
|
&host,
|
|
"Remember this lesson and let the Host consolidate its committed candidate.",
|
|
)
|
|
.await;
|
|
assert_eq!(results["remember"]["status"], "pending_commit");
|
|
let job_id = model.first_job().await;
|
|
let completed = tokio::time::timeout(Duration::from_secs(30), host.jobs().wait(&job_id))
|
|
.await
|
|
.expect("Job cleanup completes")
|
|
.unwrap();
|
|
assert_eq!(
|
|
completed.state,
|
|
standalone::jobs::JobState::Completed,
|
|
"{completed:?}"
|
|
);
|
|
assert_eq!(
|
|
completed.request.profile,
|
|
"builtin:standalone-subjektiv-consolidation"
|
|
);
|
|
assert_eq!(completed.request.purpose, "subjektiv_consolidation");
|
|
assert!(
|
|
completed.domain_grant.is_some(),
|
|
"policy is not a domain capability"
|
|
);
|
|
let ids = completed.request.input["candidate_ids"].as_array().unwrap();
|
|
assert_eq!(ids.len(), 1, "Job must snapshot the committed candidate");
|
|
let accepted = completed
|
|
.result
|
|
.clone()
|
|
.expect("structured result is durable success authority");
|
|
assert_eq!(accepted["subject_id"], subject.id);
|
|
assert_eq!(accepted["candidate_ids"], json!(ids));
|
|
assert_eq!(
|
|
accepted["candidate_dispositions"],
|
|
json!([{"candidate_id":ids[0], "action":"applied"}])
|
|
);
|
|
assert_eq!(accepted["surface"]["availability"], "ready");
|
|
assert!(accepted["surface"]["generation_id"].is_string());
|
|
assert!(accepted["surface"]["snapshot_id"].is_string());
|
|
// The Subject consumer may already have acknowledged successful cleanup.
|
|
// Acknowledgement timing is not success authority; the immutable accepted
|
|
// result above remains durable regardless of that scheduling race.
|
|
let requests = model.requests();
|
|
assert!(requests.iter().any(|r| {
|
|
r.tools
|
|
.iter()
|
|
.any(|tool| tool.name == "MemoryApplyCandidate")
|
|
}));
|
|
assert!(requests.iter().any(|r| {
|
|
r.tools
|
|
.iter()
|
|
.any(|tool| tool.name == "SubmitMemorySurface")
|
|
}));
|
|
assert!(
|
|
state
|
|
.join("subjektiv/jobs")
|
|
.join(&subject.id)
|
|
.join("jobs.sqlite3")
|
|
.exists()
|
|
);
|
|
assert!(
|
|
!state
|
|
.join(worker_id.to_string())
|
|
.join("jobs.sqlite3")
|
|
.exists(),
|
|
"selected Subject owns the shared generic Job ledger"
|
|
);
|
|
disconnect_recorded(&mut host).await;
|
|
host.shutdown().await.unwrap();
|
|
let memory = inspect_store(&state, |store| {
|
|
assert!(
|
|
store
|
|
.pending_staging_candidates(&subject.id, 100)
|
|
.unwrap()
|
|
.is_empty()
|
|
);
|
|
let memory = store.list_memories(&subject.id).unwrap().pop().unwrap();
|
|
assert_eq!(memory.claim, claim);
|
|
assert!(!memory.change_id.is_empty());
|
|
assert_eq!(memory.source_candidate_ids, vec![ids[0].as_str().unwrap()]);
|
|
let surface = store.resident_surface(&subject.id).unwrap();
|
|
assert_eq!(surface.availability, subjektiv::SurfaceAvailability::Ready);
|
|
let snapshot = surface.snapshot.unwrap();
|
|
assert_eq!(
|
|
snapshot.id,
|
|
accepted["surface"]["snapshot_id"].as_str().unwrap()
|
|
);
|
|
assert!(snapshot.body_md.contains(claim));
|
|
assert_eq!(
|
|
snapshot.memory_refs,
|
|
vec![subjektiv::MemoryChangeRef {
|
|
memory_id: memory.id.clone(),
|
|
change_id: memory.change_id.clone()
|
|
}]
|
|
);
|
|
assert_eq!(
|
|
snapshot.built_from_memory_fingerprint,
|
|
accepted["surface"]["memory_fingerprint"].as_str().unwrap()
|
|
);
|
|
memory
|
|
});
|
|
let next_model = ScriptedClient::new(vec![
|
|
tool("confirmed", "SubjektivMemoryQuery", json!({"query":claim})),
|
|
prose(),
|
|
]);
|
|
let next = StandaloneHost::start_with_model_client(
|
|
launch(temp.path(), &state, &subject.id, "next-body"),
|
|
next_model.clone(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_ne!(next.worker_id(), worker_id);
|
|
let retained = next.jobs().get(&job_id).unwrap();
|
|
assert_eq!(retained.request, completed.request);
|
|
assert_eq!(retained.state, completed.state);
|
|
assert_eq!(
|
|
retained.result, completed.result,
|
|
"result follows the Subject, not the old Worker"
|
|
);
|
|
let results = run(&next, "Recall confirmed Memory in the next Worker.").await;
|
|
assert_eq!(results["confirmed"]["items"].as_array().unwrap().len(), 1);
|
|
assert!(results["confirmed"].to_string().contains(&memory.id));
|
|
assert!(
|
|
next_model.requests()[0]
|
|
.system_prompt
|
|
.as_deref()
|
|
.unwrap()
|
|
.contains(claim),
|
|
"fresh resident surface must be injected at new Worker creation"
|
|
);
|
|
assert_eq!(
|
|
next.jobs().acknowledge(&job_id).unwrap().result,
|
|
Some(accepted)
|
|
);
|
|
next.shutdown().await.unwrap();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn transport_loss_after_candidate_apply_preserves_memory_and_later_job_only_rebuilds_surface()
|
|
{
|
|
let temp = tempfile::tempdir().unwrap();
|
|
let state = temp.path().join("state");
|
|
let catalog = StandaloneSubjects::open(&state).unwrap();
|
|
let subject = catalog.create("partial-apply", None).unwrap();
|
|
drop(catalog);
|
|
let claim = "Candidate receipts survive transport failure";
|
|
let losing = ScriptedClient::consolidating(remember_script(claim), true);
|
|
let mut host = StandaloneHost::start_with_model_client(
|
|
launch(temp.path(), &state, &subject.id, "losing-body"),
|
|
losing.clone(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
run(
|
|
&host,
|
|
"Stage a lesson whose Job loses its transport after applying it.",
|
|
)
|
|
.await;
|
|
let first_id = losing.first_job().await;
|
|
let failed = tokio::time::timeout(Duration::from_secs(30), host.jobs().wait(&first_id))
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(
|
|
failed.state,
|
|
standalone::jobs::JobState::Failed,
|
|
"{failed:?}"
|
|
);
|
|
assert!(
|
|
failed.result.is_none(),
|
|
"partial apply is not an accepted Job result"
|
|
);
|
|
disconnect_recorded(&mut host).await;
|
|
host.shutdown().await.unwrap();
|
|
let original = inspect_store(&state, |store| {
|
|
assert!(
|
|
store
|
|
.pending_staging_candidates(&subject.id, 100)
|
|
.unwrap()
|
|
.is_empty()
|
|
);
|
|
let memory = store.list_memories(&subject.id).unwrap();
|
|
assert_eq!(memory.len(), 1);
|
|
assert_eq!(memory[0].claim, claim);
|
|
assert!(!memory[0].change_id.is_empty());
|
|
assert_ne!(
|
|
store.resident_surface(&subject.id).unwrap().availability,
|
|
subjektiv::SurfaceAvailability::Ready
|
|
);
|
|
serde_json::to_value(&memory[0]).unwrap()
|
|
});
|
|
let rebuilding = ScriptedClient::consolidating(vec![prose()], false);
|
|
let next = StandaloneHost::start_with_model_client(
|
|
launch(temp.path(), &state, &subject.id, "rebuild-body"),
|
|
rebuilding.clone(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(next.jobs().get(&first_id).unwrap(), failed);
|
|
run(
|
|
&next,
|
|
"A later explicit Host interaction rebuilds the missing surface.",
|
|
)
|
|
.await;
|
|
let second_id = rebuilding.first_job().await;
|
|
assert_ne!(
|
|
first_id, second_id,
|
|
"do not silently replay the failed immutable Job"
|
|
);
|
|
let completed = tokio::time::timeout(Duration::from_secs(30), next.jobs().wait(&second_id))
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(
|
|
completed.state,
|
|
standalone::jobs::JobState::Completed,
|
|
"{completed:?}"
|
|
);
|
|
assert_eq!(
|
|
completed.request.input["candidate_ids"],
|
|
json!([]),
|
|
"resolved candidates cannot be applied again"
|
|
);
|
|
assert_eq!(
|
|
completed.result.as_ref().unwrap()["surface"]["availability"],
|
|
"ready"
|
|
);
|
|
assert!(
|
|
!rebuilding
|
|
.requests()
|
|
.iter()
|
|
.any(|request| request.items.iter().any(
|
|
|item| matches!(item, Item::ToolCall { name, .. } if name == "MemoryApplyCandidate")
|
|
))
|
|
);
|
|
assert_eq!(
|
|
next.jobs().get(&first_id).unwrap(),
|
|
failed,
|
|
"later success does not rewrite the original failed outcome"
|
|
);
|
|
next.shutdown().await.unwrap();
|
|
inspect_store(&state, |store| {
|
|
let memories = store.list_memories(&subject.id).unwrap();
|
|
assert_eq!(memories.len(), 1);
|
|
assert_eq!(
|
|
serde_json::to_value(&memories[0]).unwrap(),
|
|
original,
|
|
"confirmed Memory is immutable across surface repair"
|
|
);
|
|
assert_eq!(
|
|
store.resident_surface(&subject.id).unwrap().availability,
|
|
subjektiv::SurfaceAvailability::Ready
|
|
);
|
|
});
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn normal_extract_stages_committed_sources_and_consolidates_through_generic_job() {
|
|
let temp = tempfile::tempdir().unwrap();
|
|
let state = temp.path().join("state");
|
|
let catalog = StandaloneSubjects::open(&state).unwrap();
|
|
let subject = catalog.create("extract", None).unwrap();
|
|
drop(catalog);
|
|
let mut model = ScriptedClient::consolidating(vec![prose()], false);
|
|
model.extract = true;
|
|
let mut selected = launch(temp.path(), &state, &subject.id, "extract-body");
|
|
selected
|
|
.profile
|
|
.manifest
|
|
.feature
|
|
.subjektiv
|
|
.profile
|
|
.extraction
|
|
.threshold = Some(1);
|
|
// An explicit model override selects a different provider by contract.
|
|
// Keep this mechanism test entirely on the injected scripted transport.
|
|
selected
|
|
.profile
|
|
.manifest
|
|
.feature
|
|
.subjektiv
|
|
.profile
|
|
.extraction
|
|
.model = None;
|
|
let host = StandaloneHost::start_with_model_client(selected, model.clone())
|
|
.await
|
|
.unwrap();
|
|
let session = host.record().active_session_id;
|
|
run(&host, "A useful lesson is worth retaining across Workers.").await;
|
|
let job_id = model.first_job().await;
|
|
let finished = tokio::time::timeout(Duration::from_secs(30), host.jobs().wait(&job_id))
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(
|
|
finished.state,
|
|
standalone::jobs::JobState::Completed,
|
|
"{finished:?}"
|
|
);
|
|
assert_eq!(
|
|
finished.result.as_ref().unwrap()["candidate_dispositions"][0]["action"],
|
|
"applied"
|
|
);
|
|
let worker = host.worker_id();
|
|
host.shutdown().await.unwrap();
|
|
inspect_store(&state, |store| {
|
|
let memory = store.list_memories(&subject.id).unwrap().pop().unwrap();
|
|
assert_eq!(
|
|
memory.claim,
|
|
"Extract a committed lesson through the normal lifecycle"
|
|
);
|
|
let candidate = store
|
|
.staging_candidate(&subject.id, &memory.source_candidate_ids[0])
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(
|
|
candidate.source_refs[0].session_id,
|
|
Some(session.to_string())
|
|
);
|
|
assert!(
|
|
candidate
|
|
.evidence
|
|
.iter()
|
|
.all(|e| e.origin.as_ref().unwrap().kind
|
|
== memory::schema::EvidenceOriginKind::HumanInput),
|
|
"{candidate:#?}"
|
|
);
|
|
assert_eq!(
|
|
store.resident_surface(&subject.id).unwrap().availability,
|
|
subjektiv::SurfaceAvailability::Ready
|
|
);
|
|
});
|
|
assert!(
|
|
model
|
|
.requests()
|
|
.iter()
|
|
.any(|r| r.tools.iter().any(|t| t.name == "StageMemoryCandidate")),
|
|
"extract must use the injected scripted transport"
|
|
);
|
|
let log_root = state
|
|
.join(worker.to_string())
|
|
.join("sessions")
|
|
.join(session.to_string());
|
|
let text = std::fs::read_dir(log_root)
|
|
.unwrap()
|
|
.map(|entry| std::fs::read_to_string(entry.unwrap().path()).unwrap())
|
|
.collect::<String>();
|
|
assert!(
|
|
text.contains("subjektiv.extract.v1"),
|
|
"normal extraction must durably advance the existing append-only pointer"
|
|
);
|
|
}
|