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

fix: harden subject memory delivery

Browse Source
This commit is contained in:
Hare
2026-10-02 09:05:11 +09:00
parent 07f2b0e8a3
commit f50a0caea9
6 changed files with 1170 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/agen/src/tool.rs
+3
View File
@@ -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),
crates/server-api/src/lib.rs
+26 -1
View File
@@ -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<memory::extract::StagingEvidence>,
#[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
pub evidence_total: usize,
pub evidence_truncated: bool,
pub source_refs: Vec<memory::schema::SourceEvidenceRef>,
#[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<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
@@ -4781,6 +4805,7 @@ pub enum SubjektivMemoryBackendResponse {
Read(SubjektivMemoryReadResponse),
ListRevisions(SubjektivMemoryListRevisionsResponse),
ProposalValidated(SubjektivMemoryProposalValidationResponse),
ReceiptStatus(SubjektivMemoryReceiptStatusResponse),
Staged(SubjektivMemoryStageExplicitResponse),
}
crates/worker/src/feature/builtin/subjektiv_memory.rs
+591 -135
View File
@@ -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<dyn WorkspaceClient>,
capture: CommittedSessionCaptureHandle,
pending: Arc<Mutex<HashMap<String, PendingExplicit>>>,
}
#[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::<server_api::RepositoryApiError>(&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<server_api::SubjektivMemoryReceiptStatusResponse, ToolError> {
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<server_api::SubjektivMemoryReceiptStatusResponse, ToolError> {
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,49 +607,81 @@ struct PendingCommitHook {
#[async_trait]
impl Hook<RunCommitted> 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::<Vec<_>>();
let mut completed = Vec::new();
replay_pending_committed_memory(&self.state)
}
}
struct PendingRetryHook {
state: SubjektivMemoryState,
}
#[async_trait]
impl Hook<PreLlmRequest> for PendingRetryHook {
async fn call(&self, _input: &PreRequestContext) -> Result<HookPreRequestAction, HookError> {
replay_pending_committed_memory(&self.state)?;
Ok(HookPreRequestAction::Continue)
}
}
struct PendingRewriteHook {
state: SubjektivMemoryState,
}
#[async_trait]
impl Hook<BeforeSessionRewrite> for PendingRewriteHook {
async fn call(
&self,
_input: &BeforeSessionRewriteContext,
) -> Result<BeforeSessionRewriteAction, HookError> {
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 item in pending {
let Some(evidence) = view.evidence_for_tool_call(&item.call_id) else {
// The operation belongs to an uncommitted or different Session.
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 self.state.stage(
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;
}
};
if let Err(error) = state.stage(
&item.receipt_id,
&capture.session_id,
item.kind.clone(),
item.kind,
item.claim,
item.why_useful,
item.staleness,
vec![evidence],
vec![call.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);
failures.push(format!("{receipt_id}: {error}"));
}
}
if failures.is_empty() {
@@ -581,7 +695,76 @@ impl Hook<RunCommitted> for PendingCommitHook {
),
))
}
}
fn pending_from_committed_call(
state: &SubjektivMemoryState,
receipt_id: &str,
tool_name: &str,
arguments: &str,
) -> Result<Option<PendingExplicit>, 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<String, ToolError> {
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<T: JsonSchema>() -> serde_json::Value {
serde_json::to_value(schemars::schema_for!(T)).unwrap_or_else(|_| serde_json::json!({}))
}
fn structured_conflict_output(error: ToolError) -> Result<ToolOutput, ToolError> {
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<ToolOutput, ToolError> {
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<WorkspaceResponse, WorkspaceClientError> {
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<WorkspaceResponse, WorkspaceClientError> {
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<Vec<WorkspaceRequest>>,
staged: Mutex<HashSet<String>>,
fail_stages: AtomicUsize,
}
impl WorkspaceClient for RecordingWorkspaceClient {
@@ -754,19 +1000,52 @@ mod tests {
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
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,16 +1193,153 @@ 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(
history: vec![
crate::session_history::history_entry(
agen::Item::tool_call(
"call-commit",
REMEMBER_TOOL,
@@ -908,7 +1348,18 @@ mod tests {
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::<SubjektivMemoryBackendRequest>(
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())
);
}
}
crates/worker/src/session_capture.rs
+57
View File
@@ -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<CommittedToolCall> {
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::<serde_json::Value>(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::<HashSet<_>>();
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<SessionEntryEvidence> {
let entry = self.index.iter().find(|entry| entry.id.as_str() == id)?;
let excerpt = self
crates/workspace-server/src/server.rs
+442 -9
View File
@@ -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<server_api::SubjektivMemoryStageExplicitResponse> {
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::<Vec<_>>().join(" ");
let mut chars = normalized.chars();
@@ -28772,12 +28932,27 @@ impl From<Error> 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<RuntimeDiagnostic>) -> 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::<Vec<_>>();
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::<Vec<_>>();
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,
docs/design/subjektiv-store.md
+23 -10
View File
@@ -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