diff --git a/crates/agen/src/tool.rs b/crates/agen/src/tool.rs index 7557d9787..29fbb026a 100644 --- a/crates/agen/src/tool.rs +++ b/crates/agen/src/tool.rs @@ -30,6 +30,9 @@ pub enum ToolError { /// Internal error #[error("Internal error: {0}")] Internal(String), + /// Machine-readable domain conflict that a tool adapter may project as structured output. + #[error("{code}: {message}")] + StructuredConflict { code: String, message: String }, /// Cooperative cancellation completed with bounded terminal output. #[error("Tool execution cancelled")] Cancelled(ToolOutput), diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 2dadb6ef2..bdeb31135 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -4619,6 +4619,12 @@ pub struct SubjektivMemoryValidateProposalRequest { pub intent: SubjektivMemoryRevisionIntent, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryReceiptStatusRequest { + pub receipt_id: String, +} + /// Host-resolved staging input. Subject scope and evidence origin are never /// accepted from model-visible tool arguments. #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] @@ -4646,6 +4652,7 @@ pub enum SubjektivMemoryBackendOperation { Read(SubjektivMemoryReadRequest), ListRevisions(SubjektivMemoryListRevisionsRequest), ValidateProposal(SubjektivMemoryValidateProposalRequest), + ReceiptStatus(SubjektivMemoryReceiptStatusRequest), StageExplicit(SubjektivMemoryStageExplicitRequest), } @@ -4682,7 +4689,13 @@ pub struct SubjektivMemoryQueryResponse { pub struct SubjektivMemoryEvidenceCandidate { pub candidate_id: String, pub evidence: Vec, + #[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))] + pub evidence_total: usize, + pub evidence_truncated: bool, pub source_refs: Vec, + #[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))] + pub source_refs_total: usize, + pub source_refs_truncated: bool, } #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] @@ -4756,10 +4769,21 @@ pub struct SubjektivMemoryProposalValidationResponse { pub intent: SubjektivMemoryRevisionIntent, } -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub enum SubjektivMemoryReceiptStatus { + PendingCommit, Staged, + Missing, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryReceiptStatusResponse { + pub receipt_id: String, + pub status: SubjektivMemoryReceiptStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub candidate_id: Option, } #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] @@ -4781,6 +4805,7 @@ pub enum SubjektivMemoryBackendResponse { Read(SubjektivMemoryReadResponse), ListRevisions(SubjektivMemoryListRevisionsResponse), ProposalValidated(SubjektivMemoryProposalValidationResponse), + ReceiptStatus(SubjektivMemoryReceiptStatusResponse), Staged(SubjektivMemoryStageExplicitResponse), } diff --git a/crates/worker/src/feature/builtin/subjektiv_memory.rs b/crates/worker/src/feature/builtin/subjektiv_memory.rs index 20c469752..05b9e0b66 100644 --- a/crates/worker/src/feature/builtin/subjektiv_memory.rs +++ b/crates/worker/src/feature/builtin/subjektiv_memory.rs @@ -4,8 +4,8 @@ //! become immutable staging candidates; T-670 remains the only authority that may //! adopt a candidate as a confirmed revision. -use std::collections::{HashMap, HashSet}; -use std::sync::{Arc, Mutex}; +use std::collections::HashSet; +use std::sync::Arc; use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput}; use async_trait::async_trait; @@ -15,7 +15,8 @@ use serde::Deserialize; use server_api::{ SubjektivMemoryBackendOperation, SubjektivMemoryBackendRequest, SubjektivMemoryBackendResponse, SubjektivMemoryListRevisionsRequest, SubjektivMemoryQueryRequest, SubjektivMemoryReadRequest, - SubjektivMemoryReceiptStatus, SubjektivMemoryRevisionIntent, SubjektivMemoryRevisionProposal, + SubjektivMemoryReceiptStatus, SubjektivMemoryReceiptStatusRequest, + SubjektivMemoryRevisionIntent, SubjektivMemoryRevisionProposal, SubjektivMemoryStageExplicitRequest, SubjektivMemoryStageExplicitResponse, SubjektivMemoryValidateProposalRequest, }; @@ -27,8 +28,9 @@ use crate::feature::{ HookDeclaration, ToolContribution, ToolDeclaration, }; use crate::hook::{ - Hook, HookError, HookErrorCategory, HookExecutionPolicy, HookFailurePolicy, RunCommitted, - RunCommittedContext, + BeforeSessionRewrite, BeforeSessionRewriteAction, BeforeSessionRewriteContext, Hook, HookError, + HookErrorCategory, HookExecutionPolicy, HookFailurePolicy, HookPreRequestAction, PreLlmRequest, + PreRequestContext, RunCommitted, RunCommittedContext, }; use crate::session_capture::{SessionCapture, SessionEntryEvidence}; use crate::worker::{WorkspaceClient, WorkspaceServerOperation}; @@ -39,6 +41,9 @@ const LIST_REVISIONS_TOOL: &str = "SubjektivMemoryListRevisions"; const REMEMBER_TOOL: &str = "SubjektivMemoryRemember"; const PROPOSE_REVISION_TOOL: &str = "SubjektivMemoryProposeRevision"; const COMMIT_HOOK: &str = "stage-explicit-memory-after-commit"; +const RETRY_HOOK: &str = "retry-committed-explicit-memory"; +const REWRITE_HOOK: &str = "flush-explicit-memory-before-rewrite"; +const MAX_EVIDENCE_REFS: usize = 20; const QUERY_DESCRIPTION: &str = "Search the connected subject's current Memory revisions. Omitted states means active only; results are bounded and cursors are bound to the subject, filters, order, and store snapshot."; const READ_DESCRIPTION: &str = "Read one current or immutable historical Memory revision for the connected subject, with line-bounded Markdown and paged provenance anchors. Continue partial reads using the returned exact revision."; @@ -55,13 +60,11 @@ pub(crate) struct SubjektivMemoryFeature { struct SubjektivMemoryState { client: Arc, capture: CommittedSessionCaptureHandle, - pending: Arc>>, } #[derive(Clone)] struct PendingExplicit { receipt_id: String, - call_id: String, kind: CandidateKind, claim: String, why_useful: String, @@ -88,11 +91,7 @@ impl SubjektivMemoryFeature { )); } Ok(Some(Self { - state: SubjektivMemoryState { - client, - capture, - pending: Arc::new(Mutex::new(HashMap::new())), - }, + state: SubjektivMemoryState { client, capture }, })) } } @@ -116,6 +115,14 @@ impl FeatureModule for SubjektivMemoryFeature { COMMIT_HOOK, FeatureHookPoint::RunCommitted, )) + .with_hook(HookDeclaration::new( + RETRY_HOOK, + FeatureHookPoint::PreLlmRequest, + )) + .with_hook(HookDeclaration::new( + REWRITE_HOOK, + FeatureHookPoint::BeforeSessionRewrite, + )) } fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { @@ -157,6 +164,19 @@ impl FeatureModule for SubjektivMemoryFeature { PendingCommitHook { state: self.state.clone(), }, + )?; + context.hooks().add_pre_request( + RETRY_HOOK, + PendingRetryHook { + state: self.state.clone(), + }, + )?; + context.hooks().add_before_session_rewrite( + REWRITE_HOOK, + HookExecutionPolicy::new(HookFailurePolicy::FailClosed, 30_000), + PendingRewriteHook { + state: self.state.clone(), + }, ) } } @@ -217,7 +237,10 @@ impl Tool for SubjektivReadTool { LIST_REVISIONS_TOOL, )?), }; - let response = self.state.execute(operation)?; + let response = match self.state.execute(operation) { + Ok(response) => response, + Err(error) => return structured_conflict_output(error), + }; let (summary, value) = match response { SubjektivMemoryBackendResponse::Query(value) => ( format!("Found {} Memory item(s).", value.items.len()), @@ -346,14 +369,18 @@ impl Tool for SubjektivExplicitTool { )); } let validation = - self.state + match self + .state .execute(SubjektivMemoryBackendOperation::ValidateProposal( SubjektivMemoryValidateProposalRequest { memory_id: params.memory_id.clone(), expected_revision: params.expected_revision, intent: params.intent, }, - ))?; + )) { + Ok(response) => response, + Err(error) => return structured_conflict_output(error), + }; let validated = match validation { SubjektivMemoryBackendResponse::ProposalValidated(value) => value, other => { @@ -379,6 +406,11 @@ impl Tool for SubjektivExplicitTool { } }; validate_candidate_text(&claim, &why_useful)?; + if entry_refs.len() > MAX_EVIDENCE_REFS { + return Err(ToolError::InvalidArgument(format!( + "entry_refs is limited to {MAX_EVIDENCE_REFS} committed references" + ))); + } if proposal .as_ref() .is_some_and(|proposal| proposal.change_reason.trim().is_empty()) @@ -395,28 +427,11 @@ impl Tool for SubjektivExplicitTool { .into(), )); } - let pending = PendingExplicit { - receipt_id: receipt_id.clone(), - call_id, - kind, - claim, - why_useful, - staleness, - proposal: proposal.clone(), - }; - let mut queue = self.state.pending.lock().map_err(|_| { - ToolError::ExecutionFailed("subjektiv pending queue poisoned".into()) - })?; - match queue.get(&receipt_id) { - Some(existing) if !pending_equivalent(existing, &pending) => { - return Err(ToolError::InvalidArgument( - "this tool operation was retried with different content".into(), - )); - } - Some(_) => {} - None => { - queue.insert(receipt_id.clone(), pending); - } + let status = self.state.receipt_status(&receipt_id)?; + if status.status == SubjektivMemoryReceiptStatus::Staged { + let value = serde_json::to_value(status) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + return json_output("Explicit Memory candidate is already staged.", &value); } let value = serde_json::json!({ "receipt_id": receipt_id, @@ -436,7 +451,7 @@ impl Tool for SubjektivExplicitTool { let entries = resolve_entry_refs(&capture, &entry_refs)?; ensure_preference_evidence(&kind, &entries)?; - let response = self.state.stage( + let response = match self.state.stage( &receipt_id, &capture.session_id, kind, @@ -445,7 +460,10 @@ impl Tool for SubjektivExplicitTool { staleness, entries, proposal, - )?; + ) { + Ok(response) => response, + Err(error) => return structured_conflict_output(error), + }; staged_output(response) } } @@ -462,9 +480,31 @@ impl SubjektivMemoryState { )) .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; if !response.is_success() { + let parsed = + serde_json::from_str::(&response.body).ok(); + if let Some((code, message)) = parsed.as_ref().and_then(|error| { + error + .diagnostics + .iter() + .find(|diagnostic| { + matches!( + diagnostic.code.as_str(), + "revision_conflict" | "stale_cursor" + ) + }) + .map(|diagnostic| (diagnostic.code.as_str(), error.message.as_str())) + }) { + return Err(ToolError::StructuredConflict { + code: code.to_string(), + message: message.to_string(), + }); + } + let detail = parsed + .map(|error| error.message) + .unwrap_or_else(|| response.body.clone()); let message = format!( - "subjektiv Memory backend returned HTTP {}: {}", - response.status, response.body + "subjektiv Memory backend returned HTTP {}: {detail}", + response.status ); return if matches!(response.status, 400 | 404 | 409 | 422) { Err(ToolError::InvalidArgument(message)) @@ -477,6 +517,48 @@ impl SubjektivMemoryState { }) } + fn backend_receipt_status( + &self, + expected_receipt_id: &str, + ) -> Result { + match self.execute(SubjektivMemoryBackendOperation::ReceiptStatus( + SubjektivMemoryReceiptStatusRequest { + receipt_id: expected_receipt_id.to_string(), + }, + ))? { + SubjektivMemoryBackendResponse::ReceiptStatus(value) => Ok(value), + other => Err(ToolError::ExecutionFailed(format!( + "unexpected receipt status response: {other:?}" + ))), + } + } + + fn receipt_status( + &self, + expected_receipt_id: &str, + ) -> Result { + let mut status = self.backend_receipt_status(expected_receipt_id)?; + if status.status == SubjektivMemoryReceiptStatus::Missing { + let capture = self + .capture + .capture() + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + let session_id = capture.session_id.clone(); + let view = SessionCapture::from_history_entries(capture.segment_id, capture.history); + let committed = view + .committed_pending_tool_calls(&[REMEMBER_TOOL, PROPOSE_REVISION_TOOL]) + .into_iter() + .any(|call| { + receipt_id(&session_id, &call.call_id) + .is_ok_and(|committed_receipt| committed_receipt == expected_receipt_id) + }); + if committed { + status.status = SubjektivMemoryReceiptStatus::PendingCommit; + } + } + Ok(status) + } + #[allow(clippy::too_many_arguments)] fn stage( &self, @@ -525,63 +607,164 @@ struct PendingCommitHook { #[async_trait] impl Hook for PendingCommitHook { async fn call(&self, _input: &RunCommittedContext) -> Result<(), HookError> { - let capture = - self.state.capture.capture().map_err(|error| { - HookError::new(HookErrorCategory::Dependency, error.to_string()) - })?; - let view = SessionCapture::from_history_entries( - capture.segment_id.clone(), - capture.history.clone(), - ); - let pending = self - .state - .pending - .lock() - .map_err(|_| HookError::new(HookErrorCategory::Internal, "pending queue poisoned"))? - .values() - .cloned() - .collect::>(); - let mut completed = Vec::new(); - let mut failures = Vec::new(); - for item in pending { - let Some(evidence) = view.evidence_for_tool_call(&item.call_id) else { - // The operation belongs to an uncommitted or different Session. + replay_pending_committed_memory(&self.state) + } +} + +struct PendingRetryHook { + state: SubjektivMemoryState, +} + +#[async_trait] +impl Hook for PendingRetryHook { + async fn call(&self, _input: &PreRequestContext) -> Result { + replay_pending_committed_memory(&self.state)?; + Ok(HookPreRequestAction::Continue) + } +} + +struct PendingRewriteHook { + state: SubjektivMemoryState, +} + +#[async_trait] +impl Hook for PendingRewriteHook { + async fn call( + &self, + _input: &BeforeSessionRewriteContext, + ) -> Result { + replay_pending_committed_memory(&self.state)?; + Ok(BeforeSessionRewriteAction::Continue) + } +} + +fn replay_pending_committed_memory(state: &SubjektivMemoryState) -> Result<(), HookError> { + let capture = state + .capture + .capture() + .map_err(|error| HookError::new(HookErrorCategory::Dependency, error.to_string()))?; + let view = + SessionCapture::from_history_entries(capture.segment_id.clone(), capture.history.clone()); + let mut failures = Vec::new(); + for call in view.committed_pending_tool_calls(&[REMEMBER_TOOL, PROPOSE_REVISION_TOOL]) { + let receipt_id = match receipt_id(&capture.session_id, &call.call_id) { + Ok(value) => value, + Err(error) => { + failures.push(error.to_string()); continue; + } + }; + match state.backend_receipt_status(&receipt_id) { + Ok(status) if status.status == SubjektivMemoryReceiptStatus::Staged => continue, + Ok(_) => {} + Err(error) => { + failures.push(format!("{receipt_id}: {error}")); + continue; + } + } + let item = + match pending_from_committed_call(state, &receipt_id, &call.name, &call.arguments) { + Ok(Some(value)) => value, + Ok(None) => continue, + Err(error) => { + failures.push(format!("{receipt_id}: {error}")); + continue; + } }; - match self.state.stage( - &item.receipt_id, - &capture.session_id, - item.kind.clone(), - item.claim, - item.why_useful, - item.staleness, - vec![evidence], - item.proposal, - ) { - Ok(_) => completed.push(item.receipt_id), - Err(error) => failures.push(error.to_string()), - } - } - if !completed.is_empty() { - let mut queue = self.state.pending.lock().map_err(|_| { - HookError::new(HookErrorCategory::Internal, "pending queue poisoned") - })?; - for receipt_id in completed { - queue.remove(&receipt_id); - } - } - if failures.is_empty() { - Ok(()) - } else { - Err(HookError::new( - HookErrorCategory::Dependency, - format!( - "explicit Memory staging after commit failed: {}", - failures.join("; ") - ), - )) + if let Err(error) = state.stage( + &item.receipt_id, + &capture.session_id, + item.kind, + item.claim, + item.why_useful, + item.staleness, + vec![call.evidence], + item.proposal, + ) { + failures.push(format!("{receipt_id}: {error}")); } } + if failures.is_empty() { + Ok(()) + } else { + Err(HookError::new( + HookErrorCategory::Dependency, + format!( + "explicit Memory staging after commit failed: {}", + failures.join("; ") + ), + )) + } +} + +fn pending_from_committed_call( + state: &SubjektivMemoryState, + receipt_id: &str, + tool_name: &str, + arguments: &str, +) -> Result, ToolError> { + let (kind, claim, why_useful, staleness, entry_refs, proposal) = match tool_name { + REMEMBER_TOOL => { + let params: RememberParams = parse(arguments, REMEMBER_TOOL)?; + ( + params.kind, + params.claim, + params.why_useful, + params.staleness, + params.entry_refs, + None, + ) + } + PROPOSE_REVISION_TOOL => { + let params: ProposeRevisionParams = parse(arguments, PROPOSE_REVISION_TOOL)?; + let response = state.execute(SubjektivMemoryBackendOperation::ValidateProposal( + SubjektivMemoryValidateProposalRequest { + memory_id: params.memory_id.clone(), + expected_revision: params.expected_revision, + intent: params.intent, + }, + ))?; + let validated = match response { + SubjektivMemoryBackendResponse::ProposalValidated(value) => value, + other => { + return Err(ToolError::ExecutionFailed(format!( + "unexpected proposal validation response: {other:?}" + ))); + } + }; + ( + validated.kind, + params.claim, + params.why_useful, + params.staleness, + params.entry_refs, + Some(SubjektivMemoryRevisionProposal { + memory_id: params.memory_id, + expected_revision: params.expected_revision, + intent: params.intent, + change_reason: params.change_reason, + }), + ) + } + _ => return Ok(None), + }; + if !entry_refs.is_empty() { + return Ok(None); + } + validate_candidate_text(&claim, &why_useful)?; + if kind == CandidateKind::Preference { + return Err(ToolError::InvalidArgument( + "preference candidates cannot use model-origin implicit evidence".into(), + )); + } + Ok(Some(PendingExplicit { + receipt_id: receipt_id.to_string(), + kind, + claim, + why_useful, + staleness, + proposal, + })) } fn resolve_entry_refs( @@ -636,16 +819,6 @@ fn receipt_id(session_id: &str, call_id: &str) -> Result { Ok(value) } -fn pending_equivalent(left: &PendingExplicit, right: &PendingExplicit) -> bool { - left.receipt_id == right.receipt_id - && left.call_id == right.call_id - && left.kind == right.kind - && left.claim == right.claim - && left.why_useful == right.why_useful - && left.staleness == right.staleness - && left.proposal == right.proposal -} - fn validate_candidate_text(claim: &str, why_useful: &str) -> Result<(), ToolError> { required(claim, "claim")?; required(why_useful, "why_useful")?; @@ -669,6 +842,22 @@ fn schema_for() -> serde_json::Value { serde_json::to_value(schemars::schema_for!(T)).unwrap_or_else(|_| serde_json::json!({})) } +fn structured_conflict_output(error: ToolError) -> Result { + match error { + ToolError::StructuredConflict { code, message } => json_output( + format!("Subjektiv Memory operation rejected with {code}."), + &serde_json::json!({ + "status": "error", + "error": { + "code": code, + "message": message, + }, + }), + ), + other => Err(other), + } +} + fn staged_output(response: SubjektivMemoryStageExplicitResponse) -> Result { if !matches!(response.status, SubjektivMemoryReceiptStatus::Staged) { return Err(ToolError::ExecutionFailed( @@ -705,6 +894,9 @@ mod tests { use super::*; use crate::feature::session::{CommittedRunExit, CommittedSessionCapture}; use crate::worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse}; + use std::collections::HashSet; + use std::sync::Mutex; + use std::sync::atomic::{AtomicUsize, Ordering}; #[derive(Debug)] struct TestWorkspaceClient; @@ -722,19 +914,73 @@ mod tests { true } + fn execute( + &self, + request: WorkspaceRequest, + ) -> Result { + let request: SubjektivMemoryBackendRequest = + serde_json::from_str(request.body.as_deref().unwrap()).unwrap(); + let SubjektivMemoryBackendOperation::ReceiptStatus(input) = request.operation else { + return Err(WorkspaceClientError::Request( + "test client only supports receipt lookup".into(), + )); + }; + Ok(WorkspaceResponse { + status: 200, + body: serde_json::to_string(&SubjektivMemoryBackendResponse::ReceiptStatus( + server_api::SubjektivMemoryReceiptStatusResponse { + receipt_id: input.receipt_id, + status: SubjektivMemoryReceiptStatus::Missing, + candidate_id: None, + }, + )) + .unwrap(), + }) + } + } + + #[derive(Debug)] + struct ConflictWorkspaceClient(&'static str); + + impl WorkspaceClient for ConflictWorkspaceClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-test") + } + + fn kind(&self) -> &str { + "conflict-test" + } + + fn is_available(&self) -> bool { + true + } + fn execute( &self, _request: WorkspaceRequest, ) -> Result { - Err(WorkspaceClientError::Request( - "test client has no backend".into(), - )) + Ok(WorkspaceResponse { + status: 409, + body: serde_json::to_string(&server_api::RepositoryApiError::new( + 409, + "Conflict", + format!("{}: retry with a fresh snapshot", self.0), + vec![server_api::Diagnostic { + code: self.0.to_string(), + severity: server_api::DiagnosticSeverity::Error, + message: "typed conflict".into(), + }], + )) + .unwrap(), + }) } } #[derive(Debug, Default)] struct RecordingWorkspaceClient { requests: Mutex>, + staged: Mutex>, + fail_stages: AtomicUsize, } impl WorkspaceClient for RecordingWorkspaceClient { @@ -754,19 +1000,52 @@ mod tests { &self, request: WorkspaceRequest, ) -> Result { - self.requests.lock().unwrap().push(request); - Ok(WorkspaceResponse { - status: 200, - body: serde_json::to_string(&SubjektivMemoryBackendResponse::Staged( - SubjektivMemoryStageExplicitResponse { - candidate_id: "explicit:session-1:call-commit".into(), - receipt_id: "explicit:session-1:call-commit".into(), + self.requests.lock().unwrap().push(request.clone()); + let request: SubjektivMemoryBackendRequest = + serde_json::from_str(request.body.as_deref().unwrap()).unwrap(); + let response = match request.operation { + SubjektivMemoryBackendOperation::ReceiptStatus(input) => { + let staged = self.staged.lock().unwrap().contains(&input.receipt_id); + SubjektivMemoryBackendResponse::ReceiptStatus( + server_api::SubjektivMemoryReceiptStatusResponse { + receipt_id: input.receipt_id.clone(), + status: if staged { + SubjektivMemoryReceiptStatus::Staged + } else { + SubjektivMemoryReceiptStatus::Missing + }, + candidate_id: staged.then_some(input.receipt_id), + }, + ) + } + SubjektivMemoryBackendOperation::StageExplicit(input) => { + if self + .fail_stages + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok() + { + return Ok(WorkspaceResponse { + status: 503, + body: r#"{"error":"Unavailable","message":"retry","diagnostics":[]}"# + .into(), + }); + } + self.staged.lock().unwrap().insert(input.receipt_id.clone()); + SubjektivMemoryBackendResponse::Staged(SubjektivMemoryStageExplicitResponse { + candidate_id: input.receipt_id.clone(), + receipt_id: input.receipt_id, status: SubjektivMemoryReceiptStatus::Staged, target: None, intent: None, - }, - )) - .unwrap(), + }) + } + other => panic!("unexpected test operation: {other:?}"), + }; + Ok(WorkspaceResponse { + status: 200, + body: serde_json::to_string(&response).unwrap(), }) } } @@ -788,7 +1067,6 @@ mod tests { state: SubjektivMemoryState { client: Arc::new(TestWorkspaceClient), capture, - pending: Arc::new(Mutex::new(HashMap::new())), }, } } @@ -811,9 +1089,16 @@ mod tests { PROPOSE_REVISION_TOOL, ] ); - assert_eq!(descriptor.hooks.len(), 1); + assert_eq!(descriptor.hooks.len(), 3); assert_eq!(descriptor.hooks[0].name, COMMIT_HOOK); assert_eq!(descriptor.hooks[0].point, FeatureHookPoint::RunCommitted); + assert_eq!(descriptor.hooks[1].name, RETRY_HOOK); + assert_eq!(descriptor.hooks[1].point, FeatureHookPoint::PreLlmRequest); + assert_eq!(descriptor.hooks[2].name, REWRITE_HOOK); + assert_eq!( + descriptor.hooks[2].point, + FeatureHookPoint::BeforeSessionRewrite + ); } #[test] @@ -845,6 +1130,27 @@ mod tests { assert!(parsed.entry_refs.is_empty()); } + #[tokio::test] + async fn backend_conflict_discriminants_survive_as_structured_tool_output() { + for code in ["revision_conflict", "stale_cursor"] { + let tool = SubjektivReadTool { + state: SubjektivMemoryState { + client: Arc::new(ConflictWorkspaceClient(code)), + capture: feature().state.capture, + }, + operation: ReadOperation::Query, + }; + let output = tool + .execute("{}", ToolExecutionContext::new("call", "batch", 0)) + .await + .unwrap(); + let value: serde_json::Value = + serde_json::from_str(output.content.as_deref().unwrap()).unwrap(); + assert_eq!(value["status"], "error"); + assert_eq!(value["error"]["code"], code); + } + } + #[test] fn receipt_identity_is_stable_and_rejects_invalid_values() { assert_eq!( @@ -872,12 +1178,9 @@ mod tests { serde_json::from_str(output.content.as_deref().unwrap()).unwrap(); assert_eq!(value["status"], "pending_commit"); assert_eq!(value["receipt_id"], "explicit:session-1:call-1"); - assert_eq!(feature.state.pending.lock().unwrap().len(), 1); - tool.execute(input, ToolExecutionContext::new("call-1", "batch-1", 0)) .await .unwrap(); - assert_eq!(feature.state.pending.lock().unwrap().len(), 1); let preference = tool .execute( @@ -890,25 +1193,173 @@ mod tests { } #[tokio::test] - async fn committed_tool_call_stages_pending_candidate_with_host_evidence() { + async fn cancelled_or_uncommitted_calls_are_not_replayed() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let capture = CommittedSessionCaptureHandle::new(|| { + Ok(CommittedSessionCapture { + session_id: "session-1".into(), + segment_id: "segment-1".into(), + session_revision: 1, + entry_count: 3, + run_exit: CommittedRunExit::Interrupted, + history: vec![ + crate::session_history::history_entry( + agen::Item::tool_call( + "call-uncommitted", + REMEMBER_TOOL, + r#"{"kind":"lesson","claim":"No result","why_useful":"Never stage"}"#, + ), + crate::session_history::WorkerHistoryProvenance::ModelOutput { + worker: crate::session_history::worker_subject(Default::default()), + }, + ), + crate::session_history::history_entry( + agen::Item::tool_call( + "call-cancelled", + REMEMBER_TOOL, + r#"{"kind":"lesson","claim":"Cancelled","why_useful":"Never stage"}"#, + ), + crate::session_history::WorkerHistoryProvenance::ModelOutput { + worker: crate::session_history::worker_subject(Default::default()), + }, + ), + crate::session_history::history_entry( + agen::Item::tool_result_item("call-cancelled", "cancelled", None, true), + crate::session_history::WorkerHistoryProvenance::ToolOutput { + worker: crate::session_history::worker_subject(Default::default()), + }, + ), + ], + usage_history: Vec::new(), + extensions: Vec::new(), + }) + }); + PendingCommitHook { + state: SubjektivMemoryState { + client: client.clone(), + capture, + }, + } + .call(&RunCommittedContext { + invocation: crate::hook::HookInvocationContext::default(), + exit: crate::hook::RunCommittedExit::Interrupted, + committed_history: crate::hook::HookHistoryRange::default(), + }) + .await + .unwrap(); + assert!(client.requests.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn failed_commit_hook_retries_from_durable_history() { + let client = Arc::new(RecordingWorkspaceClient::default()); + client.fail_stages.store(1, Ordering::SeqCst); + let capture = CommittedSessionCaptureHandle::new(|| { + Ok(CommittedSessionCapture { + session_id: "session-1".into(), + segment_id: "segment-1".into(), + session_revision: 2, + entry_count: 2, + run_exit: CommittedRunExit::Finished, + history: vec![ + crate::session_history::history_entry( + agen::Item::tool_call( + "call-retry", + REMEMBER_TOOL, + r#"{"kind":"lesson","claim":"Retry","why_useful":"Recover after failure"}"#, + ), + crate::session_history::WorkerHistoryProvenance::ModelOutput { + worker: crate::session_history::worker_subject(Default::default()), + }, + ), + crate::session_history::history_entry( + agen::Item::tool_result_with_content( + "call-retry", + "pending commit", + r#"{"status":"pending_commit"}"#, + ), + crate::session_history::WorkerHistoryProvenance::ToolOutput { + worker: crate::session_history::worker_subject(Default::default()), + }, + ), + ], + usage_history: Vec::new(), + extensions: Vec::new(), + }) + }); + let hook = PendingCommitHook { + state: SubjektivMemoryState { + client: client.clone(), + capture, + }, + }; + let context = RunCommittedContext { + invocation: crate::hook::HookInvocationContext::default(), + exit: crate::hook::RunCommittedExit::Finished, + committed_history: crate::hook::HookHistoryRange::default(), + }; + assert!(hook.call(&context).await.is_err()); + hook.call(&context).await.unwrap(); + hook.call(&context).await.unwrap(); + assert!( + client + .staged + .lock() + .unwrap() + .contains("explicit:session-1:call-retry") + ); + let stage_attempts = client + .requests + .lock() + .unwrap() + .iter() + .filter(|request| { + let request: SubjektivMemoryBackendRequest = + serde_json::from_str(request.body.as_deref().unwrap()).unwrap(); + matches!( + request.operation, + SubjektivMemoryBackendOperation::StageExplicit(_) + ) + }) + .count(); + assert_eq!( + stage_attempts, 2, + "a staged receipt must not be replayed again" + ); + } + + #[tokio::test] + async fn committed_tool_call_is_replayed_after_restart_with_host_evidence() { let client = Arc::new(RecordingWorkspaceClient::default()); let capture = CommittedSessionCaptureHandle::new(|| { Ok(CommittedSessionCapture { session_id: "session-1".into(), segment_id: "segment-1".into(), session_revision: 2, - entry_count: 1, + entry_count: 2, run_exit: CommittedRunExit::Finished, - history: vec![crate::session_history::history_entry( - agen::Item::tool_call( - "call-commit", - REMEMBER_TOOL, - r#"{"kind":"lesson","claim":"Committed","why_useful":"Evidence"}"#, + history: vec![ + crate::session_history::history_entry( + agen::Item::tool_call( + "call-commit", + REMEMBER_TOOL, + r#"{"kind":"lesson","claim":"Committed","why_useful":"Evidence"}"#, + ), + crate::session_history::WorkerHistoryProvenance::ModelOutput { + worker: crate::session_history::worker_subject(Default::default()), + }, ), - crate::session_history::WorkerHistoryProvenance::ModelOutput { - worker: crate::session_history::worker_subject(Default::default()), - }, - )], + crate::session_history::history_entry( + agen::Item::tool_result_with_content( + "call-commit", + "pending commit", + r#"{"status":"pending_commit"}"#, + ), + crate::session_history::WorkerHistoryProvenance::ToolOutput { + worker: crate::session_history::worker_subject(Default::default()), + }, + ), + ], usage_history: Vec::new(), extensions: Vec::new(), }) @@ -916,47 +1367,52 @@ mod tests { let state = SubjektivMemoryState { client: client.clone(), capture, - pending: Arc::new(Mutex::new(HashMap::new())), }; - let tool = SubjektivExplicitTool { - state: state.clone(), - operation: ExplicitOperation::Remember, - }; - tool.execute( - r#"{"kind":"lesson","claim":"Committed","why_useful":"Evidence"}"#, - ToolExecutionContext::new("call-commit", "batch-1", 0), - ) - .await - .unwrap(); - assert_eq!(state.pending.lock().unwrap().len(), 1); + assert_eq!( + state + .receipt_status("explicit:session-1:call-commit") + .unwrap() + .status, + SubjektivMemoryReceiptStatus::PendingCommit + ); - PendingCommitHook { + PendingRetryHook { state: state.clone(), } - .call(&RunCommittedContext { - invocation: crate::hook::HookInvocationContext::default(), - exit: crate::hook::RunCommittedExit::Finished, - committed_history: crate::hook::HookHistoryRange::default(), - }) + .call(&crate::hook::PreRequestContext::new( + crate::hook::PreRequestInfo { + item_count: 2, + estimated_tokens: None, + turn_index: 0, + tool_calls_this_turn: 0, + }, + None, + )) .await .unwrap(); - assert!(state.pending.lock().unwrap().is_empty()); let requests = client.requests.lock().unwrap(); - assert_eq!(requests.len(), 1); - let body = requests[0].body.as_deref().unwrap(); - let request: SubjektivMemoryBackendRequest = serde_json::from_str(body).unwrap(); - let SubjektivMemoryBackendOperation::StageExplicit(staged) = request.operation else { - panic!("expected explicit stage request"); - }; - assert_eq!(staged.session_id, "session-1"); - assert_eq!(staged.evidence.len(), 1); + let request = requests + .iter() + .filter_map(|request| { + serde_json::from_str::( + request.body.as_deref().unwrap(), + ) + .ok() + }) + .find_map(|request| match request.operation { + SubjektivMemoryBackendOperation::StageExplicit(staged) => Some(staged), + _ => None, + }) + .expect("restart replay must stage the committed receipt"); + assert_eq!(request.session_id, "session-1"); + assert_eq!(request.evidence.len(), 1); assert_eq!( - staged.evidence[0].origin.as_ref().unwrap().kind, + request.evidence[0].origin.as_ref().unwrap().kind, memory::schema::EvidenceOriginKind::ModelOutput ); assert_eq!( - staged.source_refs[0].evidence_id.as_deref(), - Some(staged.evidence[0].id.as_str()) + request.source_refs[0].evidence_id.as_deref(), + Some(request.evidence[0].id.as_str()) ); } } diff --git a/crates/worker/src/session_capture.rs b/crates/worker/src/session_capture.rs index 86c9fb911..ad7a25adf 100644 --- a/crates/worker/src/session_capture.rs +++ b/crates/worker/src/session_capture.rs @@ -4,6 +4,7 @@ //! assigns append-stable `SessionEntryRef` values, and provides sparse overview, bounded //! range/search, read, and generic evidence projections without granting mutation authority. +use std::collections::HashSet; use std::sync::Arc; use crate::session_history::{SessionHistoryMetadata, WorkerHistoryProvenance}; @@ -207,6 +208,14 @@ pub(crate) struct ReadResult { pub truncated: bool, } +#[derive(Debug, Clone)] +pub(crate) struct CommittedToolCall { + pub call_id: String, + pub name: String, + pub arguments: String, + pub evidence: SessionEntryEvidence, +} + #[derive(Debug, Clone)] pub(crate) struct SessionEntryEvidence { pub segment_id: String, @@ -622,6 +631,54 @@ impl SessionCapture { self.evidence_for(reference.id.as_str()) } + pub(crate) fn committed_pending_tool_calls(&self, names: &[&str]) -> Vec { + let pending_call_ids = self + .entries + .iter() + .filter_map(|entry| { + let Item::ToolResult { + call_id, + content: Some(content), + is_error: false, + .. + } = &entry.item + else { + return None; + }; + (serde_json::from_str::(content) + .ok() + .and_then(|value| value.get("status").cloned()) + .as_ref() + .and_then(serde_json::Value::as_str) + == Some("pending_commit")) + .then_some(call_id.as_str()) + }) + .collect::>(); + self.entries + .iter() + .filter_map(|entry| { + let Item::ToolCall { + call_id, + name, + arguments, + .. + } = &entry.item + else { + return None; + }; + if !names.contains(&name.as_str()) || !pending_call_ids.contains(call_id.as_str()) { + return None; + } + Some(CommittedToolCall { + call_id: call_id.clone(), + name: name.clone(), + arguments: arguments.clone(), + evidence: self.evidence_for_tool_call(call_id)?, + }) + }) + .collect() + } + pub(crate) fn evidence_for(&self, id: &str) -> Option { let entry = self.index.iter().find(|entry| entry.id.as_str() == id)?; let excerpt = self diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index ec8d813ab..fdb939ee3 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -16776,6 +16776,9 @@ const SUBJEKTIV_QUERY_MAX_LIMIT: usize = 100; const SUBJEKTIV_BODY_DEFAULT_LINES: usize = 200; const SUBJEKTIV_BODY_MAX_LINES: usize = 1_000; const SUBJEKTIV_EVIDENCE_PAGE_SIZE: usize = 20; +const SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT: usize = 20; +const SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES: usize = 64; +const SUBJEKTIV_NESTED_EVIDENCE_MAX_BYTES: usize = 32 * 1024; #[derive(Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -16840,6 +16843,23 @@ async fn scoped_subjektiv_memory_backend( subjektiv_memory_validate_proposal(&store, &subject_id, input)?, ) } + server_api::SubjektivMemoryBackendOperation::ReceiptStatus(input) => { + validate_subjektiv_receipt_id(&input.receipt_id)?; + let candidate = store + .staging_candidate(&subject_id, &input.receipt_id) + .map_err(subjektiv_store_error)?; + server_api::SubjektivMemoryBackendResponse::ReceiptStatus( + server_api::SubjektivMemoryReceiptStatusResponse { + receipt_id: input.receipt_id, + status: if candidate.is_some() { + server_api::SubjektivMemoryReceiptStatus::Staged + } else { + server_api::SubjektivMemoryReceiptStatus::Missing + }, + candidate_id: candidate.map(|record| record.id), + }, + ) + } server_api::SubjektivMemoryBackendOperation::StageExplicit(input) => { // Re-derive the same subject and immutable Session attribution for the // write. The request never carries subject, Worker, Runtime, origin, @@ -16891,7 +16911,7 @@ fn subjektiv_memory_query( } if cursor.store_revision != subject.store_revision { return Err(Error::RepositoryConflict( - "subjektiv query cursor is stale because Memory changed; restart the query".into(), + "stale_cursor: subjektiv query cursor is stale because Memory changed; restart the query".into(), ) .into()); } @@ -16983,6 +17003,12 @@ fn subjektiv_memory_read( .into()); } let body_offset = input.offset.unwrap_or(0); + if body_offset > 0 && input.revision.is_none() { + return Err(Error::InvalidInput( + "body continuation requires the exact revision returned by the first read".into(), + ) + .into()); + } let lines = memory_body_lines(&record.body_md); if body_offset > lines.len() { return Err(Error::InvalidInput("body line offset exceeds document length".into()).into()); @@ -17021,6 +17047,8 @@ fn subjektiv_memory_read( let mut source_candidate_ids = Vec::new(); let mut source_candidates = Vec::new(); let mut derived_from = Vec::new(); + let mut evidence_budget = SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT; + let mut source_ref_budget = SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT; for index in evidence_offset..evidence_end { if let Some(candidate_id) = record.source_candidate_ids.get(index) { let candidate = store @@ -17032,10 +17060,30 @@ fn subjektiv_memory_read( )) })?; source_candidate_ids.push(candidate_id.clone()); + let evidence_total = candidate.evidence.len(); + let source_refs_total = candidate.source_refs.len(); + let evidence_take = evidence_total.min(evidence_budget); + let source_refs_take = source_refs_total.min(source_ref_budget); + evidence_budget -= evidence_take; + source_ref_budget -= source_refs_take; source_candidates.push(server_api::SubjektivMemoryEvidenceCandidate { candidate_id: candidate.id, - evidence: candidate.evidence, - source_refs: candidate.source_refs, + evidence: candidate + .evidence + .into_iter() + .take(evidence_take) + .map(bounded_subjektiv_staging_evidence) + .collect(), + evidence_total, + evidence_truncated: evidence_total > evidence_take, + source_refs: candidate + .source_refs + .into_iter() + .take(source_refs_take) + .map(bounded_subjektiv_source_ref) + .collect(), + source_refs_total, + source_refs_truncated: source_refs_total > source_refs_take, }); } else { let derived_index = index - record.source_candidate_ids.len(); @@ -17046,6 +17094,7 @@ fn subjektiv_memory_read( }); } } + bound_subjektiv_nested_evidence(&mut source_candidates)?; let evidence_has_more = evidence_end < total_evidence; let evidence_next_cursor = evidence_has_more .then(|| { @@ -17180,21 +17229,35 @@ fn subjektiv_memory_validate_proposal( }) } +fn validate_subjektiv_receipt_id(receipt_id: &str) -> ApiResult<()> { + let receipt_id = receipt_id.trim(); + if receipt_id.is_empty() || receipt_id.len() > 512 || receipt_id.chars().any(char::is_control) { + return Err(Error::InvalidInput("explicit Memory receipt_id is invalid".into()).into()); + } + Ok(()) +} + fn subjektiv_memory_stage_explicit( store: &crate::subjektiv::SubjektivStore, mut input: server_api::SubjektivMemoryStageExplicitRequest, attribution: crate::subjektiv::SubjectSessionAttribution, ) -> ApiResult { + validate_subjektiv_receipt_id(&input.receipt_id)?; let receipt_id = input.receipt_id.trim(); - if receipt_id.is_empty() || receipt_id.len() > 512 || receipt_id.chars().any(char::is_control) { - return Err(Error::InvalidInput("explicit Memory receipt_id is invalid".into()).into()); - } if input.evidence.is_empty() || input.source_refs.is_empty() { return Err(Error::InvalidInput( "explicit Memory staging requires host-resolved committed evidence".into(), ) .into()); } + if input.evidence.len() > SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT + || input.source_refs.len() > SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT + { + return Err(Error::InvalidInput(format!( + "explicit Memory evidence/source_refs are limited to {SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT} items each" + )) + .into()); + } if matches!(input.kind, memory::extract::CandidateKind::Preference) && input.evidence.iter().any(|evidence| { !matches!( @@ -17454,6 +17517,103 @@ fn memory_body_lines(body: &str) -> Vec<&str> { } } +fn bound_subjektiv_nested_evidence( + candidates: &mut [server_api::SubjektivMemoryEvidenceCandidate], +) -> ApiResult<()> { + loop { + let serialized = serde_json::to_vec(candidates) + .map_err(|error| Error::Store(format!("encode subjektiv evidence page: {error}")))?; + if serialized.len() <= SUBJEKTIV_NESTED_EVIDENCE_MAX_BYTES { + return Ok(()); + } + let mut removed = false; + for candidate in candidates.iter_mut().rev() { + if candidate.source_refs.pop().is_some() { + candidate.source_refs_truncated = true; + removed = true; + break; + } + if candidate.evidence.pop().is_some() { + candidate.evidence_truncated = true; + removed = true; + break; + } + } + if !removed { + return Err(Error::Store( + "subjektiv evidence metadata exceeds its response byte budget".into(), + ) + .into()); + } + } +} + +fn bounded_subjektiv_staging_evidence( + mut evidence: memory::extract::StagingEvidence, +) -> memory::extract::StagingEvidence { + evidence.kind = memory::schema::EvidenceKind::new(bounded_utf8_bytes( + evidence.kind.as_str().to_string(), + SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES, + )); + evidence.origin = evidence.origin.map(bounded_subjektiv_evidence_origin); + evidence.excerpt = evidence + .excerpt + .map(|value| bounded_utf8_bytes(value, SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES)); + evidence.summary = evidence + .summary + .map(|value| bounded_utf8_bytes(value, SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES)); + evidence +} + +fn bounded_subjektiv_source_ref( + mut source: memory::schema::SourceEvidenceRef, +) -> memory::schema::SourceEvidenceRef { + source.origin = source.origin.map(bounded_subjektiv_evidence_origin); + source.evidence_kind = source.evidence_kind.map(|kind| { + memory::schema::EvidenceKind::new(bounded_utf8_bytes( + kind.as_str().to_string(), + SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES, + )) + }); + source.label = source + .label + .map(|value| bounded_utf8_bytes(value, SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES)); + source.summary = source + .summary + .map(|value| bounded_utf8_bytes(value, SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES)); + source +} + +fn bounded_subjektiv_evidence_origin( + mut origin: memory::schema::EvidenceOrigin, +) -> memory::schema::EvidenceOrigin { + for field in [ + &mut origin.account_id, + &mut origin.workspace_id, + &mut origin.runtime_id, + &mut origin.worker_id, + &mut origin.flow_selector, + &mut origin.flow_definition_id, + ] { + if let Some(value) = field.take() { + *field = Some(bounded_utf8_bytes(value, SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES)); + } + } + origin +} + +fn bounded_utf8_bytes(value: String, max_bytes: usize) -> String { + if value.len() <= max_bytes { + return value; + } + let marker = "…"; + let mut end = max_bytes.saturating_sub(marker.len()); + while end > 0 && !value.is_char_boundary(end) { + end -= 1; + } + format!("{}{marker}", &value[..end]) +} + fn bounded_memory_excerpt(body: &str, max_chars: usize) -> String { let normalized = body.split_whitespace().collect::>().join(" "); let mut chars = normalized.chars(); @@ -28772,12 +28932,27 @@ impl From for ApiError { severity: HostDiagnosticSeverity::Error, message: sanitize_backend_error(&ticket_error.to_string()), }], + Error::RepositoryConflict(message) => subjektiv_conflict_code(message) + .map(|code| { + vec![RuntimeDiagnostic { + code: code.to_string(), + severity: HostDiagnosticSeverity::Error, + message: sanitize_backend_error(message), + }] + }) + .unwrap_or_default(), _ => Vec::new(), }; Self { error, diagnostics } } } +fn subjektiv_conflict_code(message: &str) -> Option<&'static str> { + ["revision_conflict", "stale_cursor"] + .into_iter() + .find(|code| message.starts_with(code)) +} + impl ApiError { fn with_diagnostics(error: Error, diagnostics: Vec) -> Self { Self { error, diagnostics } @@ -34968,12 +35143,46 @@ mod tests { excerpt: Some("Correct the first rule".to_string()), summary: Some("Human correction".to_string()), }; + let oversized = scoped_subjektiv_memory_backend( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + context.clone(), + Json(server_api::SubjektivMemoryBackendRequest { + operation: server_api::SubjektivMemoryBackendOperation::StageExplicit( + server_api::SubjektivMemoryStageExplicitRequest { + receipt_id: "explicit:session-1:oversized".to_string(), + session_id: "session-1".to_string(), + kind: memory::extract::CandidateKind::Decision, + claim: "Too many anchors".to_string(), + why_useful: "Prove response bounds".to_string(), + staleness: None, + evidence: vec![ + explicit_evidence.clone(); + SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT + 1 + ], + source_refs: vec![memory::schema::SourceEvidenceRef { + segment_id: Some("segment-1".to_string()), + entry_range: Some([0, 0]), + evidence_id: Some("E00000000".to_string()), + ..Default::default() + }], + proposal: None, + }, + ), + }), + ) + .await + .unwrap_err(); + assert!(matches!(oversized.error, Error::InvalidInput(_))); + let Json(explicit) = scoped_subjektiv_memory_backend( State(api.clone()), AxumPath(ScopedWorkspacePath { workspace_id: TEST_WORKSPACE_ID.to_string(), }), - context, + context.clone(), Json(server_api::SubjektivMemoryBackendRequest { operation: server_api::SubjektivMemoryBackendOperation::StageExplicit( server_api::SubjektivMemoryStageExplicitRequest { @@ -35007,6 +35216,34 @@ mod tests { other => panic!("unexpected explicit response: {other:?}"), }; assert_eq!(staged.candidate_id, "explicit:session-1:call-1"); + let Json(status) = scoped_subjektiv_memory_backend( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + context, + Json(server_api::SubjektivMemoryBackendRequest { + operation: server_api::SubjektivMemoryBackendOperation::ReceiptStatus( + server_api::SubjektivMemoryReceiptStatusRequest { + receipt_id: staged.receipt_id.clone(), + }, + ), + }), + ) + .await + .unwrap(); + let status = match status { + server_api::SubjektivMemoryBackendResponse::ReceiptStatus(status) => status, + other => panic!("unexpected receipt response: {other:?}"), + }; + assert_eq!( + status.status, + server_api::SubjektivMemoryReceiptStatus::Staged + ); + assert_eq!( + status.candidate_id.as_deref(), + Some(staged.candidate_id.as_str()) + ); let candidate = store .staging_candidate(&subject.id, &staged.candidate_id) .unwrap() @@ -35067,6 +35304,18 @@ mod tests { store .revise_memory(&subject.id, &second.id, 1, resolved) .unwrap(); + let revision_conflict = subjektiv_memory_validate_proposal( + &store, + &subject.id, + server_api::SubjektivMemoryValidateProposalRequest { + memory_id: second.id.clone(), + expected_revision: 1, + intent: server_api::SubjektivMemoryRevisionIntent::Revise, + }, + ) + .unwrap_err(); + let revision_api = ApiError::from(revision_conflict.error).into_repository_api_error(); + assert_eq!(revision_api.diagnostics[0].code, "revision_conflict"); let default_page = subjektiv_memory_query( &store, @@ -35142,7 +35391,51 @@ mod tests { }, ) .unwrap_err(); - assert!(matches!(stale.error, Error::RepositoryConflict(_))); + assert!(matches!(&stale.error, Error::RepositoryConflict(_))); + let stale_api = ApiError::from(stale.error).into_repository_api_error(); + assert_eq!(stale_api.diagnostics[0].code, "stale_cursor"); + + let initial_read = subjektiv_memory_read( + &store, + &subject.id, + server_api::SubjektivMemoryReadRequest { + memory_id: first.id.clone(), + revision: None, + offset: None, + limit: Some(1), + evidence_cursor: None, + }, + ) + .unwrap(); + assert_eq!(initial_read.revision, 1); + assert_eq!(initial_read.body_next_offset, Some(1)); + store + .revise_memory( + &subject.id, + &first.id, + 1, + crate::subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Decision, + "Choose stable cursors", + "replacement one\nreplacement two\n", + "Prevents mixed snapshots", + "replace current body", + ), + ) + .unwrap(); + let unbound_continuation = subjektiv_memory_read( + &store, + &subject.id, + server_api::SubjektivMemoryReadRequest { + memory_id: first.id.clone(), + revision: None, + offset: initial_read.body_next_offset, + limit: Some(1), + evidence_cursor: None, + }, + ) + .unwrap_err(); + assert!(matches!(unbound_continuation.error, Error::InvalidInput(_))); let read = subjektiv_memory_read( &store, @@ -35160,7 +35453,147 @@ mod tests { assert_eq!(read.body_next_offset, Some(2)); assert!(read.body_truncated); assert_eq!(read.revision, 1); - assert_eq!(read.current_revision, 1); + assert_eq!(read.current_revision, 2); + + let nested_count = SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT + 5; + let nested_evidence = (0..nested_count) + .map(|index| memory::extract::StagingEvidence { + id: format!("nested-{index}-{}", "i".repeat(220)), + kind: memory::schema::EvidenceKind::new(format!( + "custom-{index}-{}", + "k".repeat(4_000) + )), + entry_range: Some([index as u64, index as u64]), + origin: Some(memory::schema::EvidenceOrigin { + kind: memory::schema::EvidenceOriginKind::ModelOutput, + account_id: Some(format!("account-{index}-{}", "a".repeat(4_000))), + workspace_id: None, + runtime_id: Some(format!("runtime-{index}-{}", "r".repeat(4_000))), + worker_id: Some(format!("worker-{index}-{}", "w".repeat(4_000))), + flow_selector: Some(format!("flow-{index}-{}", "f".repeat(4_000))), + flow_definition_id: Some(format!("definition-{index}-{}", "d".repeat(4_000))), + flow_definition_revision: None, + }), + excerpt: Some(format!("evidence-{index}-{}", "x".repeat(4_000))), + summary: Some(format!("evidence-summary-{index}-{}", "q".repeat(4_000))), + }) + .collect::>(); + let nested_refs = (0..nested_count) + .map(|index| memory::schema::SourceEvidenceRef { + session_id: Some(format!("session-{index}-{}", "u".repeat(220))), + segment_id: Some(format!("segment-{index}-{}", "g".repeat(220))), + entry_range: Some([index as u64, index as u64]), + evidence_id: Some(format!("nested-{index}-{}", "i".repeat(220))), + evidence_kind: Some(memory::schema::EvidenceKind::new(format!( + "source-custom-{index}-{}", + "s".repeat(4_000) + ))), + label: Some(format!("source-{index}-{}", "y".repeat(4_000))), + summary: Some(format!("summary-{index}-{}", "z".repeat(4_000))), + origin: Some(memory::schema::EvidenceOrigin { + kind: memory::schema::EvidenceOriginKind::ModelOutput, + account_id: Some(format!("source-account-{index}-{}", "a".repeat(4_000))), + workspace_id: None, + runtime_id: Some(format!("source-runtime-{index}-{}", "r".repeat(4_000))), + worker_id: Some(format!("source-worker-{index}-{}", "w".repeat(4_000))), + flow_selector: Some(format!("source-flow-{index}-{}", "f".repeat(4_000))), + flow_definition_id: Some(format!( + "source-definition-{index}-{}", + "d".repeat(4_000) + )), + flow_definition_revision: Some(index as u64), + }), + ..Default::default() + }) + .collect::>(); + let nested_candidate = memory::extract::StagingRecord::from_candidate( + "candidate-nested", + "nested-dedup", + memory::schema::SourceRef { + segment_id: "segment-nested".to_string(), + range: [0, nested_count as u64 - 1], + }, + memory::extract::ExtractedCandidate { + kind: memory::extract::CandidateKind::Lesson, + claim: "Keep nested provenance bounded".to_string(), + why_useful: "Prevents unbounded read responses".to_string(), + staleness: None, + evidence_ids: nested_evidence.iter().map(|item| item.id.clone()).collect(), + }, + nested_evidence, + nested_refs, + ); + let nested_candidate = store + .stage_candidate(crate::subjektiv::SubjectStagingRecord::attach( + &subject.id, + nested_candidate, + )) + .unwrap(); + let (nested_memory, _) = store + .apply_candidate( + &subject.id, + &nested_candidate.id, + crate::subjektiv::MemoryRevisionTarget::Create, + crate::subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Lesson, + "Keep nested provenance bounded", + "bounded body", + "Prevents unbounded read responses", + "adopt oversized historical candidate", + ), + "adopt for response-bound regression", + ) + .unwrap(); + let nested_read = subjektiv_memory_read( + &store, + &subject.id, + server_api::SubjektivMemoryReadRequest { + memory_id: nested_memory.id, + revision: Some(1), + offset: None, + limit: None, + evidence_cursor: None, + }, + ) + .unwrap(); + assert_eq!(nested_read.source_candidates.len(), 1); + let nested = &nested_read.source_candidates[0]; + assert_eq!(nested.evidence.len(), SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT); + assert_eq!(nested.evidence_total, nested_count); + assert!(nested.evidence_truncated); + assert!(!nested.source_refs.is_empty()); + assert!(nested.source_refs.len() <= SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT); + assert_eq!(nested.source_refs_total, nested_count); + assert!(nested.source_refs_truncated); + assert!(nested.evidence.iter().all(|item| { + item.excerpt + .as_ref() + .is_none_or(|value| value.len() <= SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES) + })); + assert!(nested.evidence.iter().all(|item| { + item.kind.as_str().len() <= SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES + && item + .origin + .as_ref() + .and_then(|origin| origin.worker_id.as_ref()) + .is_none_or(|value| value.len() <= SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES) + })); + assert!(nested.source_refs.iter().all(|item| { + item.label + .as_ref() + .is_none_or(|value| value.len() <= SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES) + && item + .evidence_kind + .as_ref() + .is_none_or(|kind| kind.as_str().len() <= SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES) + })); + assert!( + serde_json::to_vec(&nested_read.source_candidates) + .unwrap() + .len() + <= SUBJEKTIV_NESTED_EVIDENCE_MAX_BYTES + ); + assert!(serde_json::to_vec(&nested_read).unwrap().len() < 64 * 1024); let revisions = subjektiv_memory_list_revisions( &store, diff --git a/docs/design/subjektiv-store.md b/docs/design/subjektiv-store.md index 156713fff..04726b1bc 100644 --- a/docs/design/subjektiv-store.md +++ b/docs/design/subjektiv-store.md @@ -175,28 +175,41 @@ empty `states` or `kinds` is invalid. Output defaults to 20 and is capped at 100 ordered by `updated_at DESC, memory_id ASC`. Its opaque cursor binds the subject, canonical filters, offset, and subject `store_revision`; any confirmed-Memory change makes it a typed stale-cursor conflict rather than silently mixing -snapshots. Revision history is ordered by revision descending. Its first page +snapshots. Revision conflicts and stale cursors retain their diagnostic code as a +structured `{status: "error", error: {code, message}}` tool result instead of +requiring models to parse prose. Revision history is ordered by revision +descending. Its first page fixes the maximum revision, so later revisions neither duplicate nor displace old page members. Read accepts a positive exact revision or resolves the current revision once. A missing historical revision never falls back to current. Markdown pagination is line-based (default 200, maximum 1000) and reports `body_truncated` plus -`body_next_offset`. Provenance uses a separate immutable-revision-bound cursor and -returns at most 20 candidate/derivation references per page. Candidate evidence is -the bounded host-resolved anchor saved in staging; raw Session bodies are not -copied into Memory responses. Resolved, retracted, and historical revisions remain +`body_next_offset`; every nonzero continuation offset must also supply the exact +revision returned by the first page, so a current-revision change cannot mix body +versions. Provenance uses a separate immutable-revision-bound cursor and returns at +most 20 candidate/derivation references per page. Expanded candidates share one +page-wide budget of 20 evidence records and 20 source references, plus a hard 32 KiB +serialized budget for the expanded candidate array; displayed anchor text is capped +at 64 UTF-8 bytes and total/truncation metadata reports omissions. +Explicit staging rejects larger anchor sets. Candidate evidence is the bounded host-resolved anchor saved in staging; raw Session bodies are not copied +into Memory responses. Resolved, retracted, and historical revisions remain addressable by ID. Remember and ProposeRevision only stage candidates. Entry references are resolved against the host's committed Session capture, and preference candidates continue to require exclusively `HumanInput` evidence. When no entry is supplied, a non-preference request returns `pending_commit`; after the run commits, the host -uses the committed tool-call entry itself as model-origin evidence. An operation -that never commits is not staged. Receipt identity is derived from Session and -tool-call identity, and exact backend retries return the first candidate rather -than creating another one. Neither path changes `memory_records`, -`memory_revisions`, `store_revision`, or a surface snapshot. +reconstructs pending receipts from the durable tool-call/tool-result history and +uses the committed tool-call entry itself as model-origin evidence. Receipt lookup +reports `pending_commit`, `staged`, or `missing`; the post-commit hook retries +immediately, the pre-request hook replays durable pending receipts after Worker +restore, and the fail-closed pre-rewrite hook prevents compaction from dropping an +unstaged receipt. Hook failure and process restart therefore retain an idempotent +retry path, while an operation that never commits is not staged. Receipt identity is derived from Session and tool-call identity, +and exact backend retries return the first candidate rather than creating another +one. Neither path changes `memory_records`, `memory_revisions`, `store_revision`, +or a surface snapshot. Revision proposals add optional `revision_proposal` metadata to the existing v2 `SubjectStagingRecord` envelope; automatic extraction remains proposal-free and