diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index cf2fbdb9c..2dadb6ef2 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -1176,6 +1176,20 @@ pub trait ServerApi { #[path] workspace_id: String, #[body] request: SubjektivRecordSessionRequest, ) -> Result; + #[post( + "/api/w/{workspace_id}/subjektiv/memory", + operation_id = "subjektiv_memory_backend", + status = 200, + error_status = 400, + additional_error_statuses = [401, 403, 404, 409, 500], + openapi = false, + )] + async fn subjektiv_memory_backend( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[body] request: SubjektivMemoryBackendRequest, + ) -> Result; #[get( "/api/w/{workspace_id}/skills", @@ -4506,6 +4520,270 @@ pub struct SubjektivRecordSessionResponse { pub session_id: String, } +/// Current lifecycle state of one confirmed subject Memory revision. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum SubjektivMemoryState { + Active, + Resolved, + Retracted, +} + +/// Requested state/change semantics for a staging-only revision proposal. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum SubjektivMemoryRevisionIntent { + Revise, + Resolve, + Retract, + Reopen, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryRevisionRef { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub revision: u64, +} + +/// Typed metadata retained with a revision proposal candidate for T-670. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryRevisionProposal { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub expected_revision: u64, + pub intent: SubjektivMemoryRevisionIntent, + pub change_reason: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryProposalTarget { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub expected_revision: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryQueryRequest { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub query: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub kinds: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub states: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schemars(range(min = 1, max = 100))] + pub limit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cursor: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryReadRequest { + pub memory_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub revision: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))] + pub offset: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schemars(range(min = 1, max = 1000))] + pub limit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub evidence_cursor: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryListRevisionsRequest { + pub memory_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schemars(range(min = 1, max = 100))] + pub limit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cursor: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryValidateProposalRequest { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub expected_revision: u64, + pub intent: SubjektivMemoryRevisionIntent, +} + +/// Host-resolved staging input. Subject scope and evidence origin are never +/// accepted from model-visible tool arguments. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryStageExplicitRequest { + pub receipt_id: String, + pub session_id: String, + pub kind: memory::extract::CandidateKind, + pub claim: String, + pub why_useful: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub staleness: Option, + #[serde(default)] + pub evidence: Vec, + #[serde(default)] + pub source_refs: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub proposal: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(tag = "operation", content = "input", rename_all = "snake_case")] +pub enum SubjektivMemoryBackendOperation { + Query(SubjektivMemoryQueryRequest), + Read(SubjektivMemoryReadRequest), + ListRevisions(SubjektivMemoryListRevisionsRequest), + ValidateProposal(SubjektivMemoryValidateProposalRequest), + StageExplicit(SubjektivMemoryStageExplicitRequest), +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryBackendRequest { + pub operation: SubjektivMemoryBackendOperation, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryQueryItem { + pub id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub revision: u64, + pub kind: memory::extract::CandidateKind, + pub state: SubjektivMemoryState, + pub claim: String, + pub excerpt: String, + pub updated_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryQueryResponse { + pub items: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_cursor: Option, + pub has_more: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryEvidenceCandidate { + pub candidate_id: String, + pub evidence: Vec, + pub source_refs: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryReadResponse { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub revision: u64, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub current_revision: u64, + pub kind: memory::extract::CandidateKind, + pub state: SubjektivMemoryState, + pub claim: String, + /// A line-bounded slice, never an implicit full-document claim. + pub body_md: String, + pub why_useful: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub staleness: Option, + pub change_reason: String, + pub created_at: String, + pub updated_at: String, + #[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))] + pub body_offset: usize, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))] + pub body_next_offset: Option, + pub body_truncated: bool, + /// Candidate ids on this evidence page. The corresponding bounded anchors + /// are returned in `source_candidates` with the same order. + pub source_candidate_ids: Vec, + pub source_candidates: Vec, + /// Derivation references on this evidence page. + pub derived_from: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub evidence_next_cursor: Option, + pub evidence_has_more: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryRevisionItem { + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub revision: u64, + pub kind: memory::extract::CandidateKind, + pub state: SubjektivMemoryState, + pub claim: String, + pub change_reason: String, + pub updated_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryListRevisionsResponse { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub current_revision: u64, + pub items: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_cursor: Option, + pub has_more: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryProposalValidationResponse { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub current_revision: u64, + pub kind: memory::extract::CandidateKind, + pub state: SubjektivMemoryState, + pub intent: SubjektivMemoryRevisionIntent, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum SubjektivMemoryReceiptStatus { + Staged, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryStageExplicitResponse { + pub candidate_id: String, + pub receipt_id: String, + pub status: SubjektivMemoryReceiptStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub target: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub intent: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(tag = "result", content = "data", rename_all = "snake_case")] +pub enum SubjektivMemoryBackendResponse { + Query(SubjektivMemoryQueryResponse), + Read(SubjektivMemoryReadResponse), + ListRevisions(SubjektivMemoryListRevisionsResponse), + ProposalValidated(SubjektivMemoryProposalValidationResponse), + Staged(SubjektivMemoryStageExplicitResponse), +} + pub const WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES: usize = 128; pub const WORKSPACE_DELETION_MAX_REVISION_BYTES: usize = 128; pub const WORKSPACE_DELETION_MAX_CONFIRMATION_BYTES: usize = 256; @@ -11673,6 +11951,7 @@ mod openapi_artifact_tests { "worker_observation_sessions", "subjektiv_stage_candidate", "subjektiv_record_session", + "subjektiv_memory_backend", "workspace_worker_discovery", "workspace_worker_remove", ]; diff --git a/crates/worker-runtime/src/worker_source.rs b/crates/worker-runtime/src/worker_source.rs index 3c6dbfccf..db1ae1c11 100644 --- a/crates/worker-runtime/src/worker_source.rs +++ b/crates/worker-runtime/src/worker_source.rs @@ -576,6 +576,11 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { client.subjektiv_record_session(workspace_id, request).await, ) } + WorkspaceServerOperation::SubjektivMemory(request) => { + generated_workspace_response( + client.subjektiv_memory_backend(workspace_id, request).await, + ) + } } }) }) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 0da10dcf0..0fee205bb 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1287,6 +1287,15 @@ where if let Some(plan) = memory_install_plan { feature_registry.add_module(plan.module); } + if let Some(subjektiv_memory) = + crate::feature::builtin::subjektiv_memory::SubjektivMemoryFeature::from_resolved_config( + &worker.manifest().feature.subjektiv, + worker.committed_session_capture_handle(), + worker.workspace_client_handle(), + )? + { + feature_registry.add_module(subjektiv_memory); + } if let Some(memory_config) = memory_lifecycle_config && let Some(memory_lifecycle) = crate::feature::builtin::memory_lifecycle::MemoryLifecycleFeature::from_resolved_config( diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index 417c9ca80..2469afd82 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -16,6 +16,7 @@ pub mod objective; pub mod orchestration; mod resource_projection; pub mod session_explore; +pub(crate) mod subjektiv_memory; pub mod task; pub mod ticket; pub mod worker_observation; diff --git a/crates/worker/src/feature/builtin/memory_staging_output.rs b/crates/worker/src/feature/builtin/memory_staging_output.rs index 2b9b7a09b..e51939a3a 100644 --- a/crates/worker/src/feature/builtin/memory_staging_output.rs +++ b/crates/worker/src/feature/builtin/memory_staging_output.rs @@ -410,7 +410,7 @@ fn evidence_origin(origin: &protocol::SessionEntryProvenance) -> EvidenceOrigin } } -fn staging_evidence(entry: &SessionEntryEvidence) -> StagingEvidence { +pub(crate) fn staging_evidence(entry: &SessionEntryEvidence) -> StagingEvidence { StagingEvidence { id: entry.entry_ref.to_string(), kind: evidence_kind(entry), @@ -421,7 +421,7 @@ fn staging_evidence(entry: &SessionEntryEvidence) -> StagingEvidence { } } -fn source_evidence_ref(entry: &SessionEntryEvidence) -> SourceEvidenceRef { +pub(crate) fn source_evidence_ref(entry: &SessionEntryEvidence) -> SourceEvidenceRef { SourceEvidenceRef { segment_id: Some(entry.segment_id.clone()), entry_range: Some(entry.entry_range), diff --git a/crates/worker/src/feature/builtin/subjektiv_memory.rs b/crates/worker/src/feature/builtin/subjektiv_memory.rs new file mode 100644 index 000000000..20c469752 --- /dev/null +++ b/crates/worker/src/feature/builtin/subjektiv_memory.rs @@ -0,0 +1,962 @@ +//! Host-scoped model tools for revisioned subjektiv Memory. +//! +//! Confirmed Memory is read-only here. Explicit remembering and revision requests +//! 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 agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput}; +use async_trait::async_trait; +use memory::extract::CandidateKind; +use schemars::JsonSchema; +use serde::Deserialize; +use server_api::{ + SubjektivMemoryBackendOperation, SubjektivMemoryBackendRequest, SubjektivMemoryBackendResponse, + SubjektivMemoryListRevisionsRequest, SubjektivMemoryQueryRequest, SubjektivMemoryReadRequest, + SubjektivMemoryReceiptStatus, SubjektivMemoryRevisionIntent, SubjektivMemoryRevisionProposal, + SubjektivMemoryStageExplicitRequest, SubjektivMemoryStageExplicitResponse, + SubjektivMemoryValidateProposalRequest, +}; + +use crate::feature::builtin::memory_staging_output::{source_evidence_ref, staging_evidence}; +use crate::feature::session::CommittedSessionCaptureHandle; +use crate::feature::{ + FeatureDescriptor, FeatureHookPoint, FeatureInstallContext, FeatureInstallError, FeatureModule, + HookDeclaration, ToolContribution, ToolDeclaration, +}; +use crate::hook::{ + Hook, HookError, HookErrorCategory, HookExecutionPolicy, HookFailurePolicy, RunCommitted, + RunCommittedContext, +}; +use crate::session_capture::{SessionCapture, SessionEntryEvidence}; +use crate::worker::{WorkspaceClient, WorkspaceServerOperation}; + +const QUERY_TOOL: &str = "SubjektivMemoryQuery"; +const READ_TOOL: &str = "SubjektivMemoryRead"; +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 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."; +const LIST_REVISIONS_DESCRIPTION: &str = "List immutable revisions of one connected-subject Memory in descending revision order using snapshot-bound pagination."; +const REMEMBER_DESCRIPTION: &str = "Explicitly stage a new Memory candidate for the connected subject. This never writes confirmed Memory. With no entry_refs, the committed tool call itself becomes model-origin evidence after the run commits and status is pending_commit."; +const PROPOSE_DESCRIPTION: &str = "Stage a typed correction, resolution, retraction, or reopen proposal for an exact current Memory revision. This validates optimistic concurrency now but never creates a confirmed revision."; + +#[derive(Clone)] +pub(crate) struct SubjektivMemoryFeature { + state: SubjektivMemoryState, +} + +#[derive(Clone)] +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, + staleness: Option, + proposal: Option, +} + +impl SubjektivMemoryFeature { + pub(crate) fn from_resolved_config( + config: &manifest::ResolvedSubjektivFeatureConfig, + capture: CommittedSessionCaptureHandle, + client: Arc, + ) -> std::io::Result> { + if !config.profile.enabled { + return Ok(None); + } + config + .validate_execution() + .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?; + if !client.is_available() || client.workspace_id().is_none() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "subjektiv Memory tools require Backend Workspace API authority", + )); + } + Ok(Some(Self { + state: SubjektivMemoryState { + client, + capture, + pending: Arc::new(Mutex::new(HashMap::new())), + }, + })) + } +} + +impl FeatureModule for SubjektivMemoryFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("subjektiv-memory-tools", "subjektiv Memory Tools") + .with_description("Subject-scoped recall and staging-only explicit Memory tools.") + .with_tool(ToolDeclaration::new(QUERY_TOOL, QUERY_DESCRIPTION)) + .with_tool(ToolDeclaration::new(READ_TOOL, READ_DESCRIPTION)) + .with_tool(ToolDeclaration::new( + LIST_REVISIONS_TOOL, + LIST_REVISIONS_DESCRIPTION, + )) + .with_tool(ToolDeclaration::new(REMEMBER_TOOL, REMEMBER_DESCRIPTION)) + .with_tool(ToolDeclaration::new( + PROPOSE_REVISION_TOOL, + PROPOSE_DESCRIPTION, + )) + .with_hook(HookDeclaration::new( + COMMIT_HOOK, + FeatureHookPoint::RunCommitted, + )) + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + for (name, description, operation) in [ + (QUERY_TOOL, QUERY_DESCRIPTION, ReadOperation::Query), + (READ_TOOL, READ_DESCRIPTION, ReadOperation::Read), + ( + LIST_REVISIONS_TOOL, + LIST_REVISIONS_DESCRIPTION, + ReadOperation::ListRevisions, + ), + ] { + context.tools().register(ToolContribution::new( + name, + read_tool_definition(name, description, self.state.clone(), operation), + ))?; + } + context.tools().register(ToolContribution::new( + REMEMBER_TOOL, + explicit_tool_definition( + REMEMBER_TOOL, + REMEMBER_DESCRIPTION, + self.state.clone(), + ExplicitOperation::Remember, + ), + ))?; + context.tools().register(ToolContribution::new( + PROPOSE_REVISION_TOOL, + explicit_tool_definition( + PROPOSE_REVISION_TOOL, + PROPOSE_DESCRIPTION, + self.state.clone(), + ExplicitOperation::Propose, + ), + ))?; + context.hooks().add_run_committed( + COMMIT_HOOK, + HookExecutionPolicy::new(HookFailurePolicy::AttentionRequired, 30_000), + PendingCommitHook { + state: self.state.clone(), + }, + ) + } +} + +#[derive(Clone, Copy)] +enum ReadOperation { + Query, + Read, + ListRevisions, +} + +fn read_tool_definition( + name: &'static str, + description: &'static str, + state: SubjektivMemoryState, + operation: ReadOperation, +) -> ToolDefinition { + Arc::new(move || { + let schema = match operation { + ReadOperation::Query => schema_for::(), + ReadOperation::Read => schema_for::(), + ReadOperation::ListRevisions => schema_for::(), + }; + let tool: Arc = Arc::new(SubjektivReadTool { + state: state.clone(), + operation, + }); + ( + ToolMeta::new(name) + .description(description) + .input_schema(schema), + tool, + ) + }) +} + +struct SubjektivReadTool { + state: SubjektivMemoryState, + operation: ReadOperation, +} + +#[async_trait] +impl Tool for SubjektivReadTool { + async fn execute( + &self, + input_json: &str, + _context: ToolExecutionContext, + ) -> Result { + let operation = match self.operation { + ReadOperation::Query => { + SubjektivMemoryBackendOperation::Query(parse(input_json, QUERY_TOOL)?) + } + ReadOperation::Read => { + SubjektivMemoryBackendOperation::Read(parse(input_json, READ_TOOL)?) + } + ReadOperation::ListRevisions => SubjektivMemoryBackendOperation::ListRevisions(parse( + input_json, + LIST_REVISIONS_TOOL, + )?), + }; + let response = self.state.execute(operation)?; + let (summary, value) = match response { + SubjektivMemoryBackendResponse::Query(value) => ( + format!("Found {} Memory item(s).", value.items.len()), + serde_json::to_value(value), + ), + SubjektivMemoryBackendResponse::Read(value) => ( + format!( + "Read Memory {} revision {}.", + value.memory_id, value.revision + ), + serde_json::to_value(value), + ), + SubjektivMemoryBackendResponse::ListRevisions(value) => ( + format!( + "Listed {} revision(s) for Memory {}.", + value.items.len(), + value.memory_id + ), + serde_json::to_value(value), + ), + other => { + return Err(ToolError::ExecutionFailed(format!( + "unexpected subjektiv backend response: {other:?}" + ))); + } + }; + let value = value.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + Ok(json_output(summary, &value)?) + } +} + +#[derive(Clone, Copy)] +enum ExplicitOperation { + Remember, + Propose, +} + +fn explicit_tool_definition( + name: &'static str, + description: &'static str, + state: SubjektivMemoryState, + operation: ExplicitOperation, +) -> ToolDefinition { + Arc::new(move || { + let schema = match operation { + ExplicitOperation::Remember => schema_for::(), + ExplicitOperation::Propose => schema_for::(), + }; + let tool: Arc = Arc::new(SubjektivExplicitTool { + state: state.clone(), + operation, + }); + ( + ToolMeta::new(name) + .description(description) + .input_schema(schema), + tool, + ) + }) +} + +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +struct RememberParams { + kind: CandidateKind, + claim: String, + why_useful: String, + #[serde(default)] + staleness: Option, + #[serde(default)] + entry_refs: Vec, +} + +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +struct ProposeRevisionParams { + memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + expected_revision: u64, + intent: SubjektivMemoryRevisionIntent, + claim: String, + why_useful: String, + change_reason: String, + #[serde(default)] + staleness: Option, + #[serde(default)] + entry_refs: Vec, +} + +struct SubjektivExplicitTool { + state: SubjektivMemoryState, + operation: ExplicitOperation, +} + +#[async_trait] +impl Tool for SubjektivExplicitTool { + async fn execute( + &self, + input_json: &str, + context: ToolExecutionContext, + ) -> Result { + let call_id = required(&context.call_id, "tool call_id")?.to_string(); + let capture = self + .state + .capture + .capture() + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + let receipt_id = receipt_id(&capture.session_id, &call_id)?; + let (kind, claim, why_useful, staleness, entry_refs, proposal) = match self.operation { + ExplicitOperation::Remember => { + let params: RememberParams = parse(input_json, REMEMBER_TOOL)?; + ( + params.kind, + params.claim, + params.why_useful, + params.staleness, + params.entry_refs, + None, + ) + } + ExplicitOperation::Propose => { + let params: ProposeRevisionParams = parse(input_json, PROPOSE_REVISION_TOOL)?; + if params.expected_revision == 0 { + return Err(ToolError::InvalidArgument( + "expected_revision must be a positive integer".into(), + )); + } + let validation = + self.state + .execute(SubjektivMemoryBackendOperation::ValidateProposal( + SubjektivMemoryValidateProposalRequest { + memory_id: params.memory_id.clone(), + expected_revision: params.expected_revision, + intent: params.intent, + }, + ))?; + let validated = match validation { + SubjektivMemoryBackendResponse::ProposalValidated(value) => value, + other => { + return Err(ToolError::ExecutionFailed(format!( + "unexpected proposal validation response: {other:?}" + ))); + } + }; + let proposal = SubjektivMemoryRevisionProposal { + memory_id: params.memory_id, + expected_revision: params.expected_revision, + intent: params.intent, + change_reason: params.change_reason, + }; + ( + validated.kind, + params.claim, + params.why_useful, + params.staleness, + params.entry_refs, + Some(proposal), + ) + } + }; + validate_candidate_text(&claim, &why_useful)?; + if proposal + .as_ref() + .is_some_and(|proposal| proposal.change_reason.trim().is_empty()) + { + return Err(ToolError::InvalidArgument( + "change_reason must not be empty".into(), + )); + } + + if entry_refs.is_empty() { + if kind == CandidateKind::Preference { + return Err(ToolError::InvalidArgument( + "preference candidates require committed HumanInput entry_refs; the model-authored tool call is not preference authority" + .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 value = serde_json::json!({ + "receipt_id": receipt_id, + "status": "pending_commit", + "target": proposal.as_ref().map(|value| serde_json::json!({ + "memory_id": value.memory_id, + "expected_revision": value.expected_revision, + })), + "intent": proposal.as_ref().map(|value| value.intent), + "message": "Candidate is not confirmed Memory. It will be staged only after this tool call is durably committed; an uncommitted/cancelled operation is discarded.", + }); + return json_output( + "Accepted explicit Memory candidate pending Session commit.", + &value, + ); + } + + let entries = resolve_entry_refs(&capture, &entry_refs)?; + ensure_preference_evidence(&kind, &entries)?; + let response = self.state.stage( + &receipt_id, + &capture.session_id, + kind, + claim, + why_useful, + staleness, + entries, + proposal, + )?; + staged_output(response) + } +} + +impl SubjektivMemoryState { + fn execute( + &self, + operation: SubjektivMemoryBackendOperation, + ) -> Result { + let response = self + .client + .execute_server_operation(WorkspaceServerOperation::SubjektivMemory( + SubjektivMemoryBackendRequest { operation }, + )) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + if !response.is_success() { + let message = format!( + "subjektiv Memory backend returned HTTP {}: {}", + response.status, response.body + ); + return if matches!(response.status, 400 | 404 | 409 | 422) { + Err(ToolError::InvalidArgument(message)) + } else { + Err(ToolError::ExecutionFailed(message)) + }; + } + serde_json::from_str(&response.body).map_err(|error| { + ToolError::ExecutionFailed(format!("decode subjektiv response: {error}")) + }) + } + + #[allow(clippy::too_many_arguments)] + fn stage( + &self, + receipt_id: &str, + session_id: &str, + kind: CandidateKind, + claim: String, + why_useful: String, + staleness: Option, + entries: Vec, + proposal: Option, + ) -> Result { + if entries.is_empty() { + return Err(ToolError::ExecutionFailed( + "explicit staging requires committed evidence".into(), + )); + } + let evidence = entries.iter().map(staging_evidence).collect(); + let source_refs = entries.iter().map(source_evidence_ref).collect(); + let response = self.execute(SubjektivMemoryBackendOperation::StageExplicit( + SubjektivMemoryStageExplicitRequest { + receipt_id: receipt_id.to_string(), + session_id: session_id.to_string(), + kind, + claim, + why_useful, + staleness, + evidence, + source_refs, + proposal, + }, + ))?; + match response { + SubjektivMemoryBackendResponse::Staged(value) => Ok(value), + other => Err(ToolError::ExecutionFailed(format!( + "unexpected explicit staging response: {other:?}" + ))), + } + } +} + +struct PendingCommitHook { + state: SubjektivMemoryState, +} + +#[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. + 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("; ") + ), + )) + } + } +} + +fn resolve_entry_refs( + capture: &crate::feature::session::CommittedSessionCapture, + entry_refs: &[String], +) -> Result, ToolError> { + let view = + SessionCapture::from_history_entries(capture.segment_id.clone(), capture.history.clone()); + let mut seen = HashSet::new(); + entry_refs + .iter() + .map(|entry_ref| { + if !seen.insert(entry_ref) { + return Err(ToolError::InvalidArgument(format!( + "duplicate SessionEntryRef {entry_ref:?}" + ))); + } + view.evidence_for(entry_ref).ok_or_else(|| { + ToolError::InvalidArgument(format!( + "unknown SessionEntryRef {entry_ref:?} for the committed Session" + )) + }) + }) + .collect() +} + +fn ensure_preference_evidence( + kind: &CandidateKind, + entries: &[SessionEntryEvidence], +) -> Result<(), ToolError> { + if kind == &CandidateKind::Preference + && entries + .iter() + .any(|entry| !matches!(entry.origin, protocol::SessionEntryProvenance::HumanInput)) + { + return Err(ToolError::InvalidArgument( + "preference candidates require exclusively HumanInput evidence".into(), + )); + } + Ok(()) +} + +fn receipt_id(session_id: &str, call_id: &str) -> Result { + required(session_id, "Session id")?; + required(call_id, "tool call_id")?; + let value = format!("explicit:{session_id}:{call_id}"); + if value.len() > 512 || value.chars().any(char::is_control) { + return Err(ToolError::InvalidArgument( + "explicit Memory receipt identity is invalid".into(), + )); + } + 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")?; + Ok(()) +} + +fn required<'a>(value: &'a str, field: &str) -> Result<&'a str, ToolError> { + if value.trim().is_empty() || value.chars().any(char::is_control) { + Err(ToolError::InvalidArgument(format!("{field} is invalid"))) + } else { + Ok(value) + } +} + +fn parse(input: &str, tool: &str) -> Result { + serde_json::from_str(input) + .map_err(|error| ToolError::InvalidArgument(format!("invalid {tool} input: {error}"))) +} + +fn schema_for() -> serde_json::Value { + serde_json::to_value(schemars::schema_for!(T)).unwrap_or_else(|_| serde_json::json!({})) +} + +fn staged_output(response: SubjektivMemoryStageExplicitResponse) -> Result { + if !matches!(response.status, SubjektivMemoryReceiptStatus::Staged) { + return Err(ToolError::ExecutionFailed( + "subjektiv backend returned a non-staged receipt".into(), + )); + } + let value = serde_json::to_value(&response) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + json_output( + format!( + "Staged explicit Memory candidate {}.", + response.candidate_id + ), + &value, + ) +} + +fn json_output( + summary: impl Into, + value: &serde_json::Value, +) -> Result { + Ok(ToolOutput { + summary: summary.into(), + content: Some( + serde_json::to_string_pretty(value) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?, + ), + attachments: Vec::new(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::feature::session::{CommittedRunExit, CommittedSessionCapture}; + use crate::worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse}; + + #[derive(Debug)] + struct TestWorkspaceClient; + + impl WorkspaceClient for TestWorkspaceClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-test") + } + + fn kind(&self) -> &str { + "test" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &self, + _request: WorkspaceRequest, + ) -> Result { + Err(WorkspaceClientError::Request( + "test client has no backend".into(), + )) + } + } + + #[derive(Debug, Default)] + struct RecordingWorkspaceClient { + requests: Mutex>, + } + + impl WorkspaceClient for RecordingWorkspaceClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-test") + } + + fn kind(&self) -> &str { + "recording-test" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &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(), + status: SubjektivMemoryReceiptStatus::Staged, + target: None, + intent: None, + }, + )) + .unwrap(), + }) + } + } + + fn feature() -> SubjektivMemoryFeature { + let capture = CommittedSessionCaptureHandle::new(|| { + Ok(CommittedSessionCapture { + session_id: "session-1".into(), + segment_id: "segment-1".into(), + session_revision: 1, + entry_count: 0, + run_exit: CommittedRunExit::Finished, + history: Vec::new(), + usage_history: Vec::new(), + extensions: Vec::new(), + }) + }); + SubjektivMemoryFeature { + state: SubjektivMemoryState { + client: Arc::new(TestWorkspaceClient), + capture, + pending: Arc::new(Mutex::new(HashMap::new())), + }, + } + } + + #[test] + fn descriptor_registers_exact_five_tools_and_commit_hook() { + let descriptor = feature().descriptor(); + let names = descriptor + .tools + .iter() + .map(|tool| tool.name.as_str()) + .collect::>(); + assert_eq!( + names, + vec![ + QUERY_TOOL, + READ_TOOL, + LIST_REVISIONS_TOOL, + REMEMBER_TOOL, + PROPOSE_REVISION_TOOL, + ] + ); + assert_eq!(descriptor.hooks.len(), 1); + assert_eq!(descriptor.hooks[0].name, COMMIT_HOOK); + assert_eq!(descriptor.hooks[0].point, FeatureHookPoint::RunCommitted); + } + + #[test] + fn model_schemas_exclude_host_scope_and_allow_optional_evidence_refs() { + let remember = schema_for::(); + let remember_text = serde_json::to_string(&remember).unwrap(); + assert!(!remember_text.contains("subject_id")); + assert!(!remember_text.contains("session_id")); + assert!(!remember_text.contains("origin")); + assert!( + !remember["required"] + .as_array() + .unwrap() + .iter() + .any(|field| field == "entry_refs") + ); + + let propose = schema_for::(); + let propose_text = serde_json::to_string(&propose).unwrap(); + assert!(!propose_text.contains("subject_id")); + assert!(!propose_text.contains("\"kind\"")); + assert!(propose_text.contains("expected_revision")); + assert!(propose_text.contains("change_reason")); + + let parsed: RememberParams = serde_json::from_str( + r#"{"kind":"lesson","claim":"Remember this","why_useful":"Later recall"}"#, + ) + .unwrap(); + assert!(parsed.entry_refs.is_empty()); + } + + #[test] + fn receipt_identity_is_stable_and_rejects_invalid_values() { + assert_eq!( + receipt_id("session-1", "call-1").unwrap(), + "explicit:session-1:call-1" + ); + assert!(receipt_id("", "call-1").is_err()); + assert!(receipt_id("session-1", "bad\ncall").is_err()); + } + + #[tokio::test] + async fn remember_without_refs_is_pending_until_commit_and_idempotent_per_call() { + let feature = feature(); + let tool = SubjektivExplicitTool { + state: feature.state.clone(), + operation: ExplicitOperation::Remember, + }; + let input = + r#"{"kind":"lesson","claim":"Use fixed revisions","why_useful":"Avoid mixed reads"}"#; + let output = tool + .execute(input, ToolExecutionContext::new("call-1", "batch-1", 0)) + .await + .unwrap(); + let value: serde_json::Value = + 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( + r#"{"kind":"preference","claim":"I prefer X","why_useful":"Personalization"}"#, + ToolExecutionContext::new("call-2", "batch-1", 1), + ) + .await + .unwrap_err(); + assert!(preference.to_string().contains("HumanInput")); + } + + #[tokio::test] + async fn committed_tool_call_stages_pending_candidate_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, + 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"}"#, + ), + crate::session_history::WorkerHistoryProvenance::ModelOutput { + worker: crate::session_history::worker_subject(Default::default()), + }, + )], + usage_history: Vec::new(), + extensions: Vec::new(), + }) + }); + 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); + + PendingCommitHook { + state: state.clone(), + } + .call(&RunCommittedContext { + invocation: crate::hook::HookInvocationContext::default(), + exit: crate::hook::RunCommittedExit::Finished, + committed_history: crate::hook::HookHistoryRange::default(), + }) + .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); + assert_eq!( + staged.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()) + ); + } +} diff --git a/crates/worker/src/session_capture.rs b/crates/worker/src/session_capture.rs index 335af3852..86c9fb911 100644 --- a/crates/worker/src/session_capture.rs +++ b/crates/worker/src/session_capture.rs @@ -600,6 +600,28 @@ impl SessionCapture { ReadResult { entries, truncated } } + /// Resolves the committed tool-call input for a provider call id. This is + /// used by explicit-memory receipts only after the containing run is durable, + /// so a model-authored claim can cite its own committed tool invocation + /// without inventing a SessionEntryRef. + pub(crate) fn evidence_for_tool_call(&self, call_id: &str) -> Option { + let entry_index = self.entries.iter().position(|entry| { + matches!( + &entry.item, + Item::ToolCall { + call_id: stored_call_id, + .. + } if stored_call_id == call_id + ) + })?; + let reference = self.index.iter().find(|entry| { + entry.entry_range[0] == entry_index as u64 + && entry.kind == ReferenceKind::Tool + && entry.tool_part == Some(ToolPart::Input) + })?; + self.evidence_for(reference.id.as_str()) + } + pub(crate) fn evidence_for(&self, id: &str) -> Option { let entry = self.index.iter().find(|entry| entry.id.as_str() == id)?; let excerpt = self @@ -1028,4 +1050,20 @@ mod tests { assert_eq!(source.entry_range, [0, 0]); assert_eq!(source.entry_ref.as_str(), "E00000000"); } + + #[test] + fn committed_tool_call_can_be_resolved_by_provider_call_id() { + let view = SessionCapture::new( + "segment-1", + vec![Item::tool_call( + "call-remember", + "SubjektivMemoryRemember", + r#"{"claim":"remember this"}"#, + )], + ); + let evidence = view.evidence_for_tool_call("call-remember").unwrap(); + assert_eq!(evidence.entry_ref.as_str(), "E00000000"); + assert_eq!(evidence.tool_part, Some(ToolPart::Input)); + assert!(view.evidence_for_tool_call("missing").is_none()); + } } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index b9e42ec4e..b8ad139db 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -626,6 +626,7 @@ pub enum WorkspaceServerOperation { WorkerObservationCapture(server_api::WorkerObservationSubjectRef), SubjektivStageCandidate(server_api::SubjektivStageCandidateRequest), SubjektivRecordSession(server_api::SubjektivRecordSessionRequest), + SubjektivMemory(server_api::SubjektivMemoryBackendRequest), } fn workspace_server_json_request( @@ -710,6 +711,9 @@ fn workspace_server_operation_request( WorkspaceServerOperation::SubjektivRecordSession(request) => { workspace_server_json_request(format!("{base}/subjektiv/sessions"), &request) } + WorkspaceServerOperation::SubjektivMemory(request) => { + workspace_server_json_request(format!("{base}/subjektiv/memory"), &request) + } } } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 34e251b37..ec8d813ab 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -5456,6 +5456,9 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou .merge(server_api::server_api_axum::subjektiv_record_session( service.clone(), )) + .merge(server_api::server_api_axum::subjektiv_memory_backend( + service.clone(), + )) .merge(server_api::server_api_axum::skill_list(service.clone())) .merge(server_api::server_api_axum::skill_lint(service.clone())) .merge(server_api::server_api_axum::skill_get(service.clone())) @@ -7143,6 +7146,26 @@ impl server_api::ServerApi for ServerApiContractService { .map_err(ApiError::into_repository_api_error) } + async fn subjektiv_memory_backend( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + request: server_api::SubjektivMemoryBackendRequest, + ) -> std::result::Result< + server_api::SubjektivMemoryBackendResponse, + server_api::RepositoryApiError, + > { + scoped_subjektiv_memory_backend( + State(self.workspace_api()?.clone()), + AxumPath(ScopedWorkspacePath { workspace_id }), + context, + Json(request), + ) + .await + .map(|Json(response)| response) + .map_err(ApiError::into_repository_api_error) + } + async fn skill_list( &self, workspace_id: String, @@ -16526,21 +16549,19 @@ fn subjektiv_singleton_key(subject_id: &str) -> ApiResult { Ok(key) } -fn subjektiv_session_attribution( +fn subjektiv_subject_scope( api: &WorkspaceApi, workspace_id: &str, context: &server_api::ServerRequestContext, - session_id: &str, -) -> ApiResult { +) -> ApiResult<(String, RuntimeWorkerRef)> { let source = context.runtime_source.as_ref().ok_or_else(|| { Error::WorkspacePermissionDenied( - "subjektiv extraction requires authenticated Runtime-owned source authority" - .to_string(), + "subjektiv operations require authenticated Runtime-owned source authority".to_string(), ) })?; let worker_id = source.worker_id.as_deref().ok_or_else(|| { Error::WorkspacePermissionDenied( - "subjektiv extraction requires Runtime-bound Worker identity".to_string(), + "subjektiv operations require Runtime-bound Worker identity".to_string(), ) })?; let worker = RuntimeWorkerRef::new(&source.runtime_id, worker_id); @@ -16552,7 +16573,17 @@ fn subjektiv_session_attribution( "current Worker has no keyed singleton ownership".to_string(), ) })?; - let subject_id = subjektiv_subject_id_from_singleton_key(&lease.key)?; + let subject_id = subjektiv_subject_id_from_singleton_key(&lease.key)?.to_string(); + Ok((subject_id, worker)) +} + +fn subjektiv_session_attribution( + api: &WorkspaceApi, + workspace_id: &str, + context: &server_api::ServerRequestContext, + session_id: &str, +) -> ApiResult { + let (subject_id, worker) = subjektiv_subject_scope(api, workspace_id, context)?; let session_id = session_id.trim(); if session_id.is_empty() || session_id.chars().any(char::is_control) { return Err(Error::InvalidInput("committed session_id is invalid".to_string()).into()); @@ -16740,6 +16771,749 @@ async fn scoped_record_subjektiv_session( })) } +const SUBJEKTIV_QUERY_DEFAULT_LIMIT: usize = 20; +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; + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct SubjektivQueryCursor { + subject_id: String, + store_revision: u64, + query: Option, + kinds: Vec, + states: Vec, + offset: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct SubjektivRevisionCursor { + subject_id: String, + memory_id: String, + max_revision: u64, + offset: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct SubjektivEvidenceCursor { + subject_id: String, + memory_id: String, + revision: u64, + offset: usize, +} + +async fn scoped_subjektiv_memory_backend( + State(api): State, + AxumPath(path): AxumPath, + context: server_api::ServerRequestContext, + Json(request): Json, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let (subject_id, _) = subjektiv_subject_scope(&api, &path.workspace_id, &context)?; + let store = open_subjektiv_store(&api)?; + let response = match request.operation { + server_api::SubjektivMemoryBackendOperation::Query(input) => { + server_api::SubjektivMemoryBackendResponse::Query(subjektiv_memory_query( + &store, + &subject_id, + input, + )?) + } + server_api::SubjektivMemoryBackendOperation::Read(input) => { + server_api::SubjektivMemoryBackendResponse::Read(subjektiv_memory_read( + &store, + &subject_id, + input, + )?) + } + server_api::SubjektivMemoryBackendOperation::ListRevisions(input) => { + server_api::SubjektivMemoryBackendResponse::ListRevisions( + subjektiv_memory_list_revisions(&store, &subject_id, input)?, + ) + } + server_api::SubjektivMemoryBackendOperation::ValidateProposal(input) => { + server_api::SubjektivMemoryBackendResponse::ProposalValidated( + subjektiv_memory_validate_proposal(&store, &subject_id, input)?, + ) + } + 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, + // or arbitrary source JSON from the model. + let attribution = subjektiv_session_attribution( + &api, + &path.workspace_id, + &context, + &input.session_id, + )?; + server_api::SubjektivMemoryBackendResponse::Staged(subjektiv_memory_stage_explicit( + &store, + input, + attribution, + )?) + } + }; + Ok(Json(response)) +} + +fn subjektiv_memory_query( + store: &crate::subjektiv::SubjektivStore, + subject_id: &str, + input: server_api::SubjektivMemoryQueryRequest, +) -> ApiResult { + let limit = bounded_subjektiv_limit(input.limit, SUBJEKTIV_QUERY_DEFAULT_LIMIT)?; + let query = input + .query + .map(|value| value.trim().to_lowercase()) + .filter(|value| !value.is_empty()); + let kinds = canonical_subjektiv_kinds(input.kinds)?; + let states = canonical_subjektiv_states(input.states)?; + let subject = store + .subject(subject_id) + .map_err(subjektiv_store_error)? + .ok_or_else(|| Error::SubjektivSubjectNotFound(subject_id.to_string()))?; + let mut offset = 0; + if let Some(cursor) = input.cursor { + let cursor: SubjektivQueryCursor = decode_subjektiv_cursor("query", &cursor)?; + if cursor.subject_id != subject_id + || cursor.query != query + || cursor.kinds != kinds + || cursor.states != states + { + return Err(Error::InvalidInput( + "subjektiv query cursor does not match subject or filters".into(), + ) + .into()); + } + if cursor.store_revision != subject.store_revision { + return Err(Error::RepositoryConflict( + "subjektiv query cursor is stale because Memory changed; restart the query".into(), + ) + .into()); + } + offset = cursor.offset; + } + + let kind_filter = kinds.iter().cloned().collect::>(); + let state_filter = states.iter().cloned().collect::>(); + let records = store + .list_memories(subject_id) + .map_err(subjektiv_store_error)? + .into_iter() + .filter(|record| { + kind_filter.contains(record.kind.as_str()) + && state_filter.contains(subjektiv_memory_state_name(record.state)) + && query.as_ref().is_none_or(|query| { + record.claim.to_lowercase().contains(query) + || record.body_md.to_lowercase().contains(query) + }) + }) + .collect::>(); + if offset > records.len() { + return Err(Error::InvalidInput("subjektiv query cursor offset is invalid".into()).into()); + } + let end = offset.saturating_add(limit).min(records.len()); + let has_more = end < records.len(); + let items = records[offset..end] + .iter() + .map(|record| server_api::SubjektivMemoryQueryItem { + id: record.id.clone(), + revision: record.revision, + kind: record.kind.clone(), + state: api_memory_state(record.state), + claim: record.claim.clone(), + excerpt: bounded_memory_excerpt(&record.body_md, 240), + updated_at: record.updated_at.clone(), + }) + .collect(); + let next_cursor = has_more + .then(|| { + encode_subjektiv_cursor( + "query", + &SubjektivQueryCursor { + subject_id: subject_id.to_string(), + store_revision: subject.store_revision, + query, + kinds, + states, + offset: end, + }, + ) + }) + .transpose()?; + Ok(server_api::SubjektivMemoryQueryResponse { + items, + next_cursor, + has_more, + }) +} + +fn subjektiv_memory_read( + store: &crate::subjektiv::SubjektivStore, + subject_id: &str, + input: server_api::SubjektivMemoryReadRequest, +) -> ApiResult { + let current = store + .scoped_memory(subject_id, &input.memory_id) + .map_err(subjektiv_store_error)?; + let record = match input.revision { + Some(0) => { + return Err(Error::InvalidInput("revision must be a positive integer".into()).into()); + } + Some(revision) => store + .scoped_memory_revision(subject_id, &input.memory_id, revision) + .map_err(subjektiv_store_error)? + .ok_or_else(|| { + Error::InvalidInput(format!( + "memory_revision_not_found: Memory `{}` has no revision {revision}", + input.memory_id + )) + })?, + None => current.clone(), + }; + let body_limit = input.limit.unwrap_or(SUBJEKTIV_BODY_DEFAULT_LINES); + if body_limit == 0 || body_limit > SUBJEKTIV_BODY_MAX_LINES { + return Err(Error::InvalidInput(format!( + "body line limit must be within 1..={SUBJEKTIV_BODY_MAX_LINES}" + )) + .into()); + } + let body_offset = input.offset.unwrap_or(0); + 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()); + } + let body_end = body_offset.saturating_add(body_limit).min(lines.len()); + let body_truncated = body_end < lines.len(); + let body_md = lines[body_offset..body_end].concat(); + + let evidence_offset = if let Some(cursor) = input.evidence_cursor { + let cursor: SubjektivEvidenceCursor = decode_subjektiv_cursor("evidence", &cursor)?; + if cursor.subject_id != subject_id + || cursor.memory_id != record.id + || cursor.revision != record.revision + { + return Err(Error::InvalidInput( + "subjektiv evidence cursor does not match subject, Memory, or revision".into(), + ) + .into()); + } + cursor.offset + } else { + 0 + }; + let total_evidence = record + .source_candidate_ids + .len() + .saturating_add(record.derived_from.len()); + if evidence_offset > total_evidence { + return Err( + Error::InvalidInput("subjektiv evidence cursor offset is invalid".into()).into(), + ); + } + let evidence_end = evidence_offset + .saturating_add(SUBJEKTIV_EVIDENCE_PAGE_SIZE) + .min(total_evidence); + let mut source_candidate_ids = Vec::new(); + let mut source_candidates = Vec::new(); + let mut derived_from = Vec::new(); + for index in evidence_offset..evidence_end { + if let Some(candidate_id) = record.source_candidate_ids.get(index) { + let candidate = store + .staging_candidate(subject_id, candidate_id) + .map_err(subjektiv_store_error)? + .ok_or_else(|| { + Error::Store(format!( + "subjektiv candidate provenance `{candidate_id}` is missing" + )) + })?; + source_candidate_ids.push(candidate_id.clone()); + source_candidates.push(server_api::SubjektivMemoryEvidenceCandidate { + candidate_id: candidate.id, + evidence: candidate.evidence, + source_refs: candidate.source_refs, + }); + } else { + let derived_index = index - record.source_candidate_ids.len(); + let reference = &record.derived_from[derived_index]; + derived_from.push(server_api::SubjektivMemoryRevisionRef { + memory_id: reference.memory_id.clone(), + revision: reference.revision, + }); + } + } + let evidence_has_more = evidence_end < total_evidence; + let evidence_next_cursor = evidence_has_more + .then(|| { + encode_subjektiv_cursor( + "evidence", + &SubjektivEvidenceCursor { + subject_id: subject_id.to_string(), + memory_id: record.id.clone(), + revision: record.revision, + offset: evidence_end, + }, + ) + }) + .transpose()?; + + Ok(server_api::SubjektivMemoryReadResponse { + memory_id: record.id, + revision: record.revision, + current_revision: current.revision, + kind: record.kind, + state: api_memory_state(record.state), + claim: record.claim, + body_md, + why_useful: record.why_useful, + staleness: record.staleness, + change_reason: record.change_reason, + created_at: record.created_at, + updated_at: record.updated_at, + body_offset, + body_next_offset: body_truncated.then_some(body_end), + body_truncated, + source_candidate_ids, + source_candidates, + derived_from, + evidence_next_cursor, + evidence_has_more, + }) +} + +fn subjektiv_memory_list_revisions( + store: &crate::subjektiv::SubjektivStore, + subject_id: &str, + input: server_api::SubjektivMemoryListRevisionsRequest, +) -> ApiResult { + let limit = bounded_subjektiv_limit(input.limit, SUBJEKTIV_QUERY_DEFAULT_LIMIT)?; + let current = store + .scoped_memory(subject_id, &input.memory_id) + .map_err(subjektiv_store_error)?; + let (max_revision, offset) = if let Some(cursor) = input.cursor { + let cursor: SubjektivRevisionCursor = decode_subjektiv_cursor("revisions", &cursor)?; + if cursor.subject_id != subject_id || cursor.memory_id != input.memory_id { + return Err(Error::InvalidInput( + "subjektiv revision cursor does not match subject or Memory".into(), + ) + .into()); + } + (cursor.max_revision, cursor.offset) + } else { + (current.revision, 0) + }; + let records = store + .list_memory_revisions(subject_id, &input.memory_id) + .map_err(subjektiv_store_error)? + .into_iter() + .filter(|record| record.revision <= max_revision) + .collect::>(); + if offset > records.len() { + return Err( + Error::InvalidInput("subjektiv revision cursor offset is invalid".into()).into(), + ); + } + let end = offset.saturating_add(limit).min(records.len()); + let has_more = end < records.len(); + let items = records[offset..end] + .iter() + .map(|record| server_api::SubjektivMemoryRevisionItem { + revision: record.revision, + kind: record.kind.clone(), + state: api_memory_state(record.state), + claim: record.claim.clone(), + change_reason: record.change_reason.clone(), + updated_at: record.updated_at.clone(), + }) + .collect(); + let next_cursor = has_more + .then(|| { + encode_subjektiv_cursor( + "revisions", + &SubjektivRevisionCursor { + subject_id: subject_id.to_string(), + memory_id: input.memory_id.clone(), + max_revision, + offset: end, + }, + ) + }) + .transpose()?; + Ok(server_api::SubjektivMemoryListRevisionsResponse { + memory_id: input.memory_id, + current_revision: current.revision, + items, + next_cursor, + has_more, + }) +} + +fn subjektiv_memory_validate_proposal( + store: &crate::subjektiv::SubjektivStore, + subject_id: &str, + input: server_api::SubjektivMemoryValidateProposalRequest, +) -> ApiResult { + if input.expected_revision == 0 { + return Err(Error::InvalidInput("expected_revision must be positive".into()).into()); + } + let current = store + .scoped_memory(subject_id, &input.memory_id) + .map_err(subjektiv_store_error)?; + if current.revision != input.expected_revision { + return Err(Error::RepositoryConflict(format!( + "revision_conflict: Memory `{}` expected {}, current {}", + input.memory_id, input.expected_revision, current.revision + )) + .into()); + } + validate_subjektiv_proposal_transition(current.state, input.intent)?; + Ok(server_api::SubjektivMemoryProposalValidationResponse { + memory_id: input.memory_id, + current_revision: current.revision, + kind: current.kind, + state: api_memory_state(current.state), + intent: input.intent, + }) +} + +fn subjektiv_memory_stage_explicit( + store: &crate::subjektiv::SubjektivStore, + mut input: server_api::SubjektivMemoryStageExplicitRequest, + attribution: crate::subjektiv::SubjectSessionAttribution, +) -> ApiResult { + 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 matches!(input.kind, memory::extract::CandidateKind::Preference) + && input.evidence.iter().any(|evidence| { + !matches!( + evidence.origin.as_ref().map(|origin| &origin.kind), + Some(memory::schema::EvidenceOriginKind::HumanInput) + ) + }) + { + return Err(Error::InvalidInput( + "preference candidates require exclusively HumanInput evidence".into(), + ) + .into()); + } + let mut range_start = u64::MAX; + let mut range_end = 0; + let mut segment_id = None; + for source_ref in &mut input.source_refs { + if source_ref + .session_id + .as_deref() + .is_some_and(|value| value != attribution.session_id) + { + return Err(Error::InvalidInput( + "explicit candidate source belongs to a different Session".into(), + ) + .into()); + } + source_ref.session_id = Some(attribution.session_id.clone()); + let source_segment = source_ref.segment_id.as_deref().ok_or_else(|| { + Error::InvalidInput("explicit candidate source is missing segment_id".into()) + })?; + if segment_id + .as_deref() + .is_some_and(|value| value != source_segment) + { + return Err(Error::InvalidInput( + "explicit candidate evidence spans multiple Session segments".into(), + ) + .into()); + } + segment_id = Some(source_segment.to_string()); + let range = source_ref.entry_range.ok_or_else(|| { + Error::InvalidInput("explicit candidate source is missing entry_range".into()) + })?; + range_start = range_start.min(range[0]); + range_end = range_end.max(range[1]); + } + let source = memory::schema::SourceRef { + segment_id: segment_id.ok_or_else(|| { + Error::InvalidInput("explicit candidate source is missing segment_id".into()) + })?, + range: [range_start, range_end], + }; + let proposal = input + .proposal + .as_ref() + .map(|proposal| { + crate::subjektiv::RevisionProposal::new( + domain_proposal_intent(proposal.intent), + proposal.memory_id.clone(), + proposal.expected_revision, + proposal.change_reason.clone(), + ) + }) + .transpose() + .map_err(subjektiv_store_error)?; + if let Some(proposal) = &proposal { + let current = store + .scoped_memory(&attribution.subject_id, &proposal.memory_id) + .map_err(subjektiv_store_error)?; + if current.kind != input.kind { + return Err(Error::InvalidInput( + "revision proposal kind must be derived from its target Memory".into(), + ) + .into()); + } + } + let candidate = memory::extract::ExtractedCandidate { + kind: input.kind, + claim: input.claim, + why_useful: input.why_useful, + staleness: input.staleness, + evidence_ids: input.evidence.iter().map(|item| item.id.clone()).collect(), + }; + let staging = memory::extract::StagingRecord::from_candidate( + receipt_id, + format!("explicit:{receipt_id}"), + source, + candidate, + input.evidence, + input.source_refs, + ); + let mut staging = + crate::subjektiv::SubjectStagingRecord::attach(&attribution.subject_id, staging); + if let Some(proposal) = proposal.clone() { + staging + .attach_revision_proposal(proposal) + .map_err(subjektiv_store_error)?; + } + let (staged, _) = store + .stage_candidate_with_attribution(staging, attribution) + .map_err(subjektiv_store_error)?; + Ok(server_api::SubjektivMemoryStageExplicitResponse { + candidate_id: staged.id, + receipt_id: receipt_id.to_string(), + status: server_api::SubjektivMemoryReceiptStatus::Staged, + target: proposal + .as_ref() + .map(|proposal| server_api::SubjektivMemoryProposalTarget { + memory_id: proposal.memory_id.clone(), + expected_revision: proposal.expected_revision, + }), + intent: proposal + .as_ref() + .map(|proposal| api_proposal_intent(proposal.intent)), + }) +} + +fn validate_subjektiv_proposal_transition( + state: crate::subjektiv::MemoryState, + intent: server_api::SubjektivMemoryRevisionIntent, +) -> ApiResult<()> { + use crate::subjektiv::MemoryState::{Active, Resolved}; + use server_api::SubjektivMemoryRevisionIntent::{Reopen, Resolve, Retract, Revise}; + let valid = match intent { + Revise => matches!(state, Active | Resolved), + Resolve => state == Active, + Retract => matches!(state, Active | Resolved), + Reopen => state == Resolved, + }; + if valid { + Ok(()) + } else { + Err(Error::InvalidInput(format!( + "invalid_state_transition: {state:?} cannot accept {intent:?}" + )) + .into()) + } +} + +fn canonical_subjektiv_kinds( + kinds: Option>, +) -> ApiResult> { + let kinds = kinds.unwrap_or_else(|| { + use memory::extract::CandidateKind::*; + vec![ + Preference, + WorkingAssumption, + Constraint, + Decision, + OpenQuestion, + Lesson, + ] + }); + if kinds.is_empty() { + return Err(Error::InvalidInput("kinds must not be an empty array".into()).into()); + } + let mut names = kinds + .iter() + .map(|kind| kind.as_str().to_string()) + .collect::>(); + names.sort(); + names.dedup(); + Ok(names) +} + +fn canonical_subjektiv_states( + states: Option>, +) -> ApiResult> { + let states = states.unwrap_or_else(|| vec![server_api::SubjektivMemoryState::Active]); + if states.is_empty() { + return Err(Error::InvalidInput("states must not be an empty array".into()).into()); + } + let mut names = states + .into_iter() + .map(|state| match state { + server_api::SubjektivMemoryState::Active => "active".to_string(), + server_api::SubjektivMemoryState::Resolved => "resolved".to_string(), + server_api::SubjektivMemoryState::Retracted => "retracted".to_string(), + }) + .collect::>(); + names.sort(); + names.dedup(); + Ok(names) +} + +fn bounded_subjektiv_limit(limit: Option, default: usize) -> ApiResult { + let limit = limit.unwrap_or(default); + if limit == 0 || limit > SUBJEKTIV_QUERY_MAX_LIMIT { + return Err(Error::InvalidInput(format!( + "limit must be within 1..={SUBJEKTIV_QUERY_MAX_LIMIT}" + )) + .into()); + } + Ok(limit) +} + +fn api_memory_state(state: crate::subjektiv::MemoryState) -> server_api::SubjektivMemoryState { + match state { + crate::subjektiv::MemoryState::Active => server_api::SubjektivMemoryState::Active, + crate::subjektiv::MemoryState::Resolved => server_api::SubjektivMemoryState::Resolved, + crate::subjektiv::MemoryState::Retracted => server_api::SubjektivMemoryState::Retracted, + } +} + +fn subjektiv_memory_state_name(state: crate::subjektiv::MemoryState) -> &'static str { + match state { + crate::subjektiv::MemoryState::Active => "active", + crate::subjektiv::MemoryState::Resolved => "resolved", + crate::subjektiv::MemoryState::Retracted => "retracted", + } +} + +fn domain_proposal_intent( + intent: server_api::SubjektivMemoryRevisionIntent, +) -> crate::subjektiv::RevisionProposalIntent { + match intent { + server_api::SubjektivMemoryRevisionIntent::Revise => { + crate::subjektiv::RevisionProposalIntent::Revise + } + server_api::SubjektivMemoryRevisionIntent::Resolve => { + crate::subjektiv::RevisionProposalIntent::Resolve + } + server_api::SubjektivMemoryRevisionIntent::Retract => { + crate::subjektiv::RevisionProposalIntent::Retract + } + server_api::SubjektivMemoryRevisionIntent::Reopen => { + crate::subjektiv::RevisionProposalIntent::Reopen + } + } +} + +fn api_proposal_intent( + intent: crate::subjektiv::RevisionProposalIntent, +) -> server_api::SubjektivMemoryRevisionIntent { + match intent { + crate::subjektiv::RevisionProposalIntent::Revise => { + server_api::SubjektivMemoryRevisionIntent::Revise + } + crate::subjektiv::RevisionProposalIntent::Resolve => { + server_api::SubjektivMemoryRevisionIntent::Resolve + } + crate::subjektiv::RevisionProposalIntent::Retract => { + server_api::SubjektivMemoryRevisionIntent::Retract + } + crate::subjektiv::RevisionProposalIntent::Reopen => { + server_api::SubjektivMemoryRevisionIntent::Reopen + } + } +} + +fn memory_body_lines(body: &str) -> Vec<&str> { + if body.is_empty() { + Vec::new() + } else { + body.split_inclusive('\n').collect() + } +} + +fn bounded_memory_excerpt(body: &str, max_chars: usize) -> String { + let normalized = body.split_whitespace().collect::>().join(" "); + let mut chars = normalized.chars(); + let excerpt = chars.by_ref().take(max_chars).collect::(); + if chars.next().is_some() { + format!("{excerpt}…") + } else { + excerpt + } +} + +fn encode_subjektiv_cursor(kind: &str, value: &T) -> ApiResult { + let raw = serde_json::to_vec(value) + .map_err(|error| Error::Store(format!("encode subjektiv cursor: {error}")))?; + let encoded = raw + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + Ok(format!("subjektiv.{kind}.{encoded}")) +} + +fn decode_subjektiv_cursor( + kind: &str, + cursor: &str, +) -> ApiResult { + let prefix = format!("subjektiv.{kind}."); + let encoded = cursor + .strip_prefix(&prefix) + .ok_or_else(|| Error::InvalidInput(format!("invalid subjektiv {kind} cursor type")))?; + if encoded.is_empty() || encoded.len() > 16_384 || encoded.len() % 2 != 0 { + return Err(Error::InvalidInput(format!("invalid subjektiv {kind} cursor")).into()); + } + let mut raw = Vec::with_capacity(encoded.len() / 2); + for index in (0..encoded.len()).step_by(2) { + raw.push( + u8::from_str_radix(&encoded[index..index + 2], 16).map_err(|_| { + Error::InvalidInput(format!("invalid subjektiv {kind} cursor encoding")) + })?, + ); + } + serde_json::from_slice(&raw) + .map_err(|_| Error::InvalidInput(format!("invalid subjektiv {kind} cursor payload")).into()) +} + +fn subjektiv_store_error(error: crate::subjektiv::SubjektivError) -> Error { + match error { + crate::subjektiv::SubjektivError::SubjectScopeMismatch { .. } => { + Error::WorkspacePermissionDenied(format!("subject_scope_mismatch: {error}")) + } + crate::subjektiv::SubjektivError::RevisionConflict { .. } => { + Error::RepositoryConflict(format!("revision_conflict: {error}")) + } + crate::subjektiv::SubjektivError::MemoryNotFound(_) => { + Error::InvalidInput(format!("memory_not_found: {error}")) + } + crate::subjektiv::SubjektivError::Storage(_) => Error::Store(error.to_string()), + _ => Error::InvalidInput(error.to_string()), + } +} + const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation"; const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation"; @@ -34143,7 +34917,7 @@ mod tests { AxumPath(ScopedWorkspacePath { workspace_id: TEST_WORKSPACE_ID.to_string(), }), - context, + context.clone(), Json(server_api::SubjektivStageCandidateRequest { session_id: "session-1".to_string(), operation, @@ -34164,6 +34938,255 @@ mod tests { let attribution = store.session_attribution("session-1").unwrap().unwrap(); assert_eq!(attribution.subject_id, subject.id); assert_eq!(attribution.worker_id, created.worker_id); + + let target = store + .create_memory( + &subject.id, + crate::subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Decision, + "Use the first rule", + "first body", + "Provides a target", + "initial", + ), + ) + .unwrap(); + let explicit_evidence = memory::extract::StagingEvidence { + id: "E00000000".to_string(), + kind: memory::schema::EvidenceKind::new(memory::schema::EvidenceKind::MESSAGE), + entry_range: Some([0, 0]), + origin: Some(memory::schema::EvidenceOrigin { + kind: memory::schema::EvidenceOriginKind::HumanInput, + account_id: None, + workspace_id: None, + runtime_id: None, + worker_id: None, + flow_selector: None, + flow_definition_id: None, + flow_definition_revision: None, + }), + excerpt: Some("Correct the first rule".to_string()), + summary: Some("Human correction".to_string()), + }; + let Json(explicit) = 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::StageExplicit( + server_api::SubjektivMemoryStageExplicitRequest { + receipt_id: "explicit:session-1:call-1".to_string(), + session_id: "session-1".to_string(), + kind: memory::extract::CandidateKind::Decision, + claim: "Use the corrected rule".to_string(), + why_useful: "Records a correction proposal".to_string(), + staleness: None, + evidence: vec![explicit_evidence], + 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: Some(server_api::SubjektivMemoryRevisionProposal { + memory_id: target.id.clone(), + expected_revision: 1, + intent: server_api::SubjektivMemoryRevisionIntent::Revise, + change_reason: "The human corrected the rule".to_string(), + }), + }, + ), + }), + ) + .await + .unwrap(); + let staged = match explicit { + server_api::SubjektivMemoryBackendResponse::Staged(staged) => staged, + other => panic!("unexpected explicit response: {other:?}"), + }; + assert_eq!(staged.candidate_id, "explicit:session-1:call-1"); + let candidate = store + .staging_candidate(&subject.id, &staged.candidate_id) + .unwrap() + .unwrap(); + assert_eq!(candidate.revision_proposal.unwrap().memory_id, target.id); + assert_eq!( + store + .memory(&subject.id, &target.id) + .unwrap() + .unwrap() + .revision, + 1, + "staging a proposal must not create a confirmed revision" + ); + } + + #[tokio::test] + async fn subjektiv_memory_tools_filter_page_and_read_fixed_revisions() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + let store = open_subjektiv_store(&api).unwrap(); + let subject = store + .create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap()) + .unwrap(); + let first = store + .create_memory( + &subject.id, + crate::subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Decision, + "Choose stable cursors", + "line one\nline two\nline three\n", + "Prevents mixed snapshots", + "initial", + ), + ) + .unwrap(); + let second = store + .create_memory( + &subject.id, + crate::subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Lesson, + "Keep immutable history", + "old body\n", + "Supports correction", + "initial", + ), + ) + .unwrap(); + let mut resolved = crate::subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Lesson, + "History was retained", + "new body\n", + "The question ended", + "resolved after verification", + ); + resolved.state = crate::subjektiv::MemoryState::Resolved; + store + .revise_memory(&subject.id, &second.id, 1, resolved) + .unwrap(); + + let default_page = subjektiv_memory_query( + &store, + &subject.id, + server_api::SubjektivMemoryQueryRequest { + query: None, + kinds: None, + states: None, + limit: Some(1), + cursor: None, + }, + ) + .unwrap(); + assert_eq!(default_page.items.len(), 1); + assert_eq!(default_page.items[0].id, first.id); + assert!(!default_page.has_more); + + let resolved_page = subjektiv_memory_query( + &store, + &subject.id, + server_api::SubjektivMemoryQueryRequest { + query: Some("new body".into()), + kinds: None, + states: Some(vec![server_api::SubjektivMemoryState::Resolved]), + limit: None, + cursor: None, + }, + ) + .unwrap(); + assert_eq!(resolved_page.items.len(), 1); + assert_eq!(resolved_page.items[0].id, second.id); + + let snapshot_page = subjektiv_memory_query( + &store, + &subject.id, + server_api::SubjektivMemoryQueryRequest { + query: None, + kinds: None, + states: Some(vec![ + server_api::SubjektivMemoryState::Active, + server_api::SubjektivMemoryState::Resolved, + ]), + limit: Some(1), + cursor: None, + }, + ) + .unwrap(); + assert!(snapshot_page.has_more); + store + .create_memory( + &subject.id, + crate::subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::OpenQuestion, + "Will this stale the cursor?", + "yes", + "Tests snapshot behavior", + "initial", + ), + ) + .unwrap(); + let stale = subjektiv_memory_query( + &store, + &subject.id, + server_api::SubjektivMemoryQueryRequest { + query: None, + kinds: None, + states: Some(vec![ + server_api::SubjektivMemoryState::Active, + server_api::SubjektivMemoryState::Resolved, + ]), + limit: Some(1), + cursor: snapshot_page.next_cursor, + }, + ) + .unwrap_err(); + assert!(matches!(stale.error, Error::RepositoryConflict(_))); + + let read = subjektiv_memory_read( + &store, + &subject.id, + server_api::SubjektivMemoryReadRequest { + memory_id: first.id.clone(), + revision: Some(1), + offset: Some(1), + limit: Some(1), + evidence_cursor: None, + }, + ) + .unwrap(); + assert_eq!(read.body_md, "line two\n"); + assert_eq!(read.body_next_offset, Some(2)); + assert!(read.body_truncated); + assert_eq!(read.revision, 1); + assert_eq!(read.current_revision, 1); + + let revisions = subjektiv_memory_list_revisions( + &store, + &subject.id, + server_api::SubjektivMemoryListRevisionsRequest { + memory_id: second.id.clone(), + limit: Some(1), + cursor: None, + }, + ) + .unwrap(); + assert_eq!(revisions.items[0].revision, 2); + assert!(revisions.has_more); + let older = subjektiv_memory_list_revisions( + &store, + &subject.id, + server_api::SubjektivMemoryListRevisionsRequest { + memory_id: second.id, + limit: Some(1), + cursor: revisions.next_cursor, + }, + ) + .unwrap(); + assert_eq!(older.items[0].revision, 1); + assert_eq!(older.items[0].claim, "Keep immutable history"); + assert!(!older.has_more); } #[tokio::test] diff --git a/crates/workspace-server/src/subjektiv.rs b/crates/workspace-server/src/subjektiv.rs index 639b12a66..17bf71d3c 100644 --- a/crates/workspace-server/src/subjektiv.rs +++ b/crates/workspace-server/src/subjektiv.rs @@ -161,8 +161,74 @@ impl SubjectSessionAttribution { } } -/// The existing extraction schema with only host-owned subject scope and -/// persistence time added. Model input still contains neither field. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RevisionProposalIntent { + Revise, + Resolve, + Retract, + Reopen, +} + +/// Host-owned metadata describing how a staged candidate is expected to change +/// one existing Memory. It is validated against the current scoped Memory when +/// the candidate is staged; it is not model-authored and does not itself apply +/// the revision. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct RevisionProposal { + pub intent: RevisionProposalIntent, + pub memory_id: String, + pub expected_revision: u64, + pub change_reason: String, +} + +impl RevisionProposal { + pub fn new( + intent: RevisionProposalIntent, + memory_id: impl Into, + expected_revision: u64, + change_reason: impl Into, + ) -> Result { + let proposal = Self { + intent, + memory_id: memory_id.into(), + expected_revision, + change_reason: change_reason.into(), + }; + validate_revision_proposal_metadata(&proposal)?; + Ok(proposal) + } +} + +impl<'de> Deserialize<'de> for RevisionProposal { + fn deserialize(deserializer: D) -> std::result::Result + where + D: Deserializer<'de>, + { + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct WireProposal { + intent: RevisionProposalIntent, + memory_id: String, + expected_revision: u64, + change_reason: String, + } + + let proposal = WireProposal::deserialize(deserializer)?; + RevisionProposal::new( + proposal.intent, + proposal.memory_id, + proposal.expected_revision, + proposal.change_reason, + ) + .map_err(serde::de::Error::custom) + } +} + +/// The existing extraction schema with only host-owned subject scope, +/// persistence time, and optional revision proposal metadata added. Model input +/// still contains none of those fields. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct SubjectStagingRecord { @@ -180,6 +246,8 @@ pub struct SubjectStagingRecord { pub evidence: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub source_refs: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub revision_proposal: Option, pub created_at: String, } @@ -203,9 +271,24 @@ impl SubjectStagingRecord { staleness: record.staleness, evidence: record.evidence, source_refs: record.source_refs, + revision_proposal: None, created_at, } } + + /// Attaches host-owned revision proposal metadata to this candidate. + pub fn with_revision_proposal(mut self, proposal: RevisionProposal) -> Result { + validate_revision_proposal_metadata(&proposal)?; + self.revision_proposal = Some(proposal); + Ok(self) + } + + /// Mutably attaches host-owned revision proposal metadata to this candidate. + pub fn attach_revision_proposal(&mut self, proposal: RevisionProposal) -> Result<()> { + validate_revision_proposal_metadata(&proposal)?; + self.revision_proposal = Some(proposal); + Ok(()) + } } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -1119,6 +1202,95 @@ impl SubjektivStore { }) } + /// Reads one current Memory while distinguishing an absent target from a + /// Memory id owned by another subject. + pub fn scoped_memory(&self, subject_id: &str, memory_id: &str) -> Result { + validate_label("subject id", subject_id)?; + validate_label("Memory id", memory_id)?; + self.database.try_with_connection(|connection| { + require_subject_in_connection(connection, subject_id)?; + scoped_memory_in_connection(connection, subject_id, memory_id) + }) + } + + /// Reads one immutable revision after verifying that the Memory belongs to + /// the requested subject. A missing revision of a valid scoped Memory is + /// returned as `None`. + pub fn scoped_memory_revision( + &self, + subject_id: &str, + memory_id: &str, + revision: u64, + ) -> Result> { + validate_label("subject id", subject_id)?; + validate_label("Memory id", memory_id)?; + if revision == 0 { + return Err(SubjektivError::InvalidRecord( + "Memory revision must be greater than zero".into(), + )); + } + self.database.try_with_connection(|connection| { + require_subject_in_connection(connection, subject_id)?; + scoped_memory_in_connection(connection, subject_id, memory_id)?; + let raw = connection + .query_row( + "SELECT record_json FROM memory_revisions + WHERE subject_id = ?1 AND memory_id = ?2 AND revision = ?3", + params![subject_id, memory_id, to_i64(revision)?], + |row| row.get::<_, String>(0), + ) + .optional()?; + raw.map(|raw| parse_memory(&raw)).transpose() + }) + } + + /// Lists current Memory records for one existing subject in deterministic + /// most-recently-updated order. + pub fn list_memories(&self, subject_id: &str) -> Result> { + validate_label("subject id", subject_id)?; + self.database.try_with_connection(|connection| { + require_subject_in_connection(connection, subject_id)?; + let mut statement = connection.prepare( + "SELECT record_json FROM memory_records + WHERE subject_id = ?1 + ORDER BY updated_at DESC, memory_id ASC", + )?; + let rows = statement.query_map([subject_id], |row| row.get::<_, String>(0))?; + let mut records = Vec::new(); + for row in rows { + records.push(parse_memory(&row?)?); + } + Ok(records) + }) + } + + /// Lists every immutable revision of one scoped Memory, newest first. + pub fn list_memory_revisions( + &self, + subject_id: &str, + memory_id: &str, + ) -> Result> { + validate_label("subject id", subject_id)?; + validate_label("Memory id", memory_id)?; + self.database.try_with_connection(|connection| { + require_subject_in_connection(connection, subject_id)?; + scoped_memory_in_connection(connection, subject_id, memory_id)?; + let mut statement = connection.prepare( + "SELECT record_json FROM memory_revisions + WHERE subject_id = ?1 AND memory_id = ?2 + ORDER BY revision DESC", + )?; + let rows = statement.query_map(params![subject_id, memory_id], |row| { + row.get::<_, String>(0) + })?; + let mut records = Vec::new(); + for row in rows { + records.push(parse_memory(&row?)?); + } + Ok(records) + }) + } + pub fn create_surface_snapshot( &self, subject_id: &str, @@ -1382,6 +1554,7 @@ fn write_staging_candidate( } return Err(SubjektivError::CandidateConflict(record.id.clone())); } + validate_staged_revision_proposal(transaction, &record)?; transaction.execute( "INSERT INTO staging_records ( subject_id, candidate_id, kind, record_json, created_at @@ -1571,7 +1744,14 @@ fn require_unresolved_candidate( } fn require_subject(transaction: &Transaction<'_>, subject_id: &str) -> Result { - let raw = transaction + require_subject_in_connection(transaction, subject_id) +} + +fn require_subject_in_connection( + connection: &rusqlite::Connection, + subject_id: &str, +) -> Result { + let raw = connection .query_row( "SELECT record_json FROM subjects WHERE subject_id = ?1", [subject_id], @@ -1774,9 +1954,98 @@ fn validate_staging_record(workspace_id: &str, record: &SubjectStagingRecord) -> }); } } + if let Some(proposal) = &record.revision_proposal { + validate_revision_proposal_metadata(proposal)?; + } Ok(()) } +fn validate_revision_proposal_metadata(proposal: &RevisionProposal) -> Result<()> { + validate_label("revision proposal Memory id", &proposal.memory_id)?; + if proposal.expected_revision == 0 { + return Err(SubjektivError::InvalidRecord( + "revision proposal expected_revision must be greater than zero".into(), + )); + } + validate_nonempty("revision proposal change_reason", &proposal.change_reason) +} + +fn validate_staged_revision_proposal( + transaction: &Transaction<'_>, + record: &SubjectStagingRecord, +) -> Result<()> { + let Some(proposal) = &record.revision_proposal else { + return Ok(()); + }; + let current = + scoped_memory_in_connection(transaction, &record.subject_id, &proposal.memory_id)?; + if current.revision != proposal.expected_revision { + return Err(SubjektivError::RevisionConflict { + memory_id: proposal.memory_id.clone(), + expected: proposal.expected_revision, + actual: current.revision, + }); + } + let to = match proposal.intent { + RevisionProposalIntent::Revise => current.state, + RevisionProposalIntent::Resolve => MemoryState::Resolved, + RevisionProposalIntent::Retract => MemoryState::Retracted, + RevisionProposalIntent::Reopen => MemoryState::Active, + }; + let allowed = match proposal.intent { + RevisionProposalIntent::Revise => { + matches!(current.state, MemoryState::Active | MemoryState::Resolved) + } + RevisionProposalIntent::Resolve => current.state == MemoryState::Active, + RevisionProposalIntent::Retract => { + matches!(current.state, MemoryState::Active | MemoryState::Resolved) + } + RevisionProposalIntent::Reopen => current.state == MemoryState::Resolved, + }; + if allowed { + Ok(()) + } else { + Err(SubjektivError::InvalidStateTransition { + memory_id: proposal.memory_id.clone(), + from: current.state, + to, + }) + } +} + +fn scoped_memory_in_connection( + connection: &rusqlite::Connection, + subject_id: &str, + memory_id: &str, +) -> Result { + let raw = connection + .query_row( + "SELECT record_json FROM memory_records + WHERE subject_id = ?1 AND memory_id = ?2", + params![subject_id, memory_id], + |row| row.get::<_, String>(0), + ) + .optional()?; + if let Some(raw) = raw { + return parse_memory(&raw); + } + let foreign_owner = connection + .query_row( + "SELECT subject_id FROM memory_records WHERE memory_id = ?1 LIMIT 1", + [memory_id], + |row| row.get::<_, String>(0), + ) + .optional()?; + if foreign_owner.is_some() { + Err(SubjektivError::SubjectScopeMismatch { + subject_id: subject_id.to_string(), + reference: memory_id.to_string(), + }) + } else { + Err(SubjektivError::MemoryNotFound(memory_id.to_string())) + } +} + fn validate_memory_draft(draft: &MemoryDraft) -> Result<()> { validate_nonempty("Memory claim", &draft.claim)?; validate_nonempty("Memory why_useful", &draft.why_useful)?; @@ -2011,6 +2280,7 @@ fn resolution_action_name(action: StagingResolutionAction) -> &'static str { mod tests { use std::sync::{Arc, Barrier}; use std::thread; + use std::time::Duration; use memory::extract::{ExtractedCandidate, StagingEvidence}; use memory::schema::{ @@ -2100,6 +2370,15 @@ mod tests { ) } + fn proposal( + intent: RevisionProposalIntent, + memory_id: &str, + expected_revision: u64, + change_reason: &str, + ) -> RevisionProposal { + RevisionProposal::new(intent, memory_id, expected_revision, change_reason).unwrap() + } + fn open_store( root: &std::path::Path, workspace_id: &str, @@ -2122,8 +2401,52 @@ mod tests { decoded.source_refs[0].session_id.as_deref(), Some("session-1") ); + assert!(decoded.revision_proposal.is_none()); + assert!(!raw.contains("revision_proposal")); assert!(raw.contains("human_input")); + let proposed = staging + .clone() + .with_revision_proposal(proposal( + RevisionProposalIntent::Revise, + "memory-1", + 3, + "Incorporate the new bounded evidence", + )) + .unwrap(); + let proposed_raw = serde_json::to_string(&proposed).unwrap(); + let proposed_decoded: SubjectStagingRecord = serde_json::from_str(&proposed_raw).unwrap(); + assert_eq!( + proposed_decoded.revision_proposal, + Some(proposal( + RevisionProposalIntent::Revise, + "memory-1", + 3, + "Incorporate the new bounded evidence", + )) + ); + assert!( + RevisionProposal::new( + RevisionProposalIntent::Revise, + "memory-1", + 0, + "invalid zero revision" + ) + .is_err() + ); + assert!( + RevisionProposal::new(RevisionProposalIntent::Revise, "memory-1", 1, " ").is_err() + ); + assert!( + serde_json::from_value::(serde_json::json!({ + "intent": "revise", + "memory_id": "memory-1", + "expected_revision": 0, + "change_reason": "invalid" + })) + .is_err() + ); + let mut value = serde_json::to_value(&staging).unwrap(); value .as_object_mut() @@ -2393,6 +2716,330 @@ mod tests { ); } + #[test] + fn revision_proposal_roundtrips_through_staging_and_resolution() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(&temp.path().join("storage"), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let memory = store + .create_memory(&subject.id, draft("Existing memory", "initial observation")) + .unwrap(); + let proposal = proposal( + RevisionProposalIntent::Revise, + &memory.id, + 1, + "Refine the wording from new evidence", + ); + let staged = store + .stage_candidate( + candidate(&subject.id, "candidate-proposed", "workspace-a") + .with_revision_proposal(proposal.clone()) + .unwrap(), + ) + .unwrap(); + assert_eq!(staged.revision_proposal, Some(proposal.clone())); + assert_eq!( + store + .staging_candidate(&subject.id, &staged.id) + .unwrap() + .unwrap() + .revision_proposal, + Some(proposal.clone()) + ); + + let resolution = store + .resolve_candidate( + &subject.id, + &staged.id, + StagingResolutionAction::Discarded, + "The current wording remains preferable", + vec![], + ) + .unwrap(); + assert_eq!( + resolution.candidate.revision_proposal, + Some(proposal.clone()) + ); + assert_eq!( + store + .staging_resolution(&subject.id, &staged.id) + .unwrap() + .unwrap() + .candidate + .revision_proposal, + Some(proposal) + ); + } + + #[test] + fn revision_proposals_enforce_scope_revision_and_state_transitions() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(&temp.path().join("storage"), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let other_subject = store.create_subject(role()).unwrap(); + let memory = store + .create_memory( + &subject.id, + draft("Transition target", "initial observation"), + ) + .unwrap(); + + let foreign = candidate(&other_subject.id, "candidate-foreign-target", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Revise, + &memory.id, + 1, + "Must remain subject scoped", + )) + .unwrap(); + assert!(matches!( + store.stage_candidate(foreign), + Err(SubjektivError::SubjectScopeMismatch { + subject_id, + reference, + }) if subject_id == other_subject.id && reference == memory.id + )); + + let stale = candidate(&subject.id, "candidate-stale", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Resolve, + &memory.id, + 2, + "Resolve the active memory", + )) + .unwrap(); + assert!(matches!( + store.stage_candidate(stale), + Err(SubjektivError::RevisionConflict { + memory_id, + expected: 2, + actual: 1, + }) if memory_id == memory.id + )); + + let resolve_proposal = candidate(&subject.id, "candidate-resolve", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Resolve, + &memory.id, + 1, + "The active condition has ended", + )) + .unwrap(); + assert!(store.stage_candidate(resolve_proposal).is_ok()); + let invalid_reopen = candidate(&subject.id, "candidate-invalid-reopen", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Reopen, + &memory.id, + 1, + "An active Memory cannot be reopened", + )) + .unwrap(); + assert!(matches!( + store.stage_candidate(invalid_reopen), + Err(SubjektivError::InvalidStateTransition { + from: MemoryState::Active, + to: MemoryState::Active, + .. + }) + )); + + let mut resolve = draft("Resolved target", "the condition ended"); + resolve.state = MemoryState::Resolved; + let resolved = store + .revise_memory(&subject.id, &memory.id, 1, resolve) + .unwrap(); + let mut attributed_stale = + candidate(&subject.id, "candidate-attributed-stale", "workspace-a"); + attributed_stale.source_refs[0].session_id = None; + attributed_stale + .attach_revision_proposal(proposal( + RevisionProposalIntent::Revise, + &memory.id, + 1, + "This attribution must roll back with the stale proposal", + )) + .unwrap(); + assert!(matches!( + store.stage_candidate_with_attribution( + attributed_stale, + attribution( + &subject.id, + "runtime-proposal", + "worker-proposal", + "session-proposal-stale", + "2026-09-28T12:00:00.000Z", + ), + ), + Err(SubjektivError::RevisionConflict { + expected: 1, + actual: 2, + .. + }) + )); + assert!( + store + .session_attribution("session-proposal-stale") + .unwrap() + .is_none() + ); + let reopen = candidate(&subject.id, "candidate-reopen", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Reopen, + &memory.id, + resolved.revision, + "The condition applies again", + )) + .unwrap(); + assert_eq!( + store + .stage_candidate(reopen) + .unwrap() + .revision_proposal + .unwrap() + .intent, + RevisionProposalIntent::Reopen + ); + let retract_proposal = candidate(&subject.id, "candidate-retract", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Retract, + &memory.id, + resolved.revision, + "The supporting evidence was invalidated", + )) + .unwrap(); + assert!(store.stage_candidate(retract_proposal).is_ok()); + + let mut retract = draft("Retracted target", "the evidence was invalidated"); + retract.state = MemoryState::Retracted; + let retracted = store + .revise_memory(&subject.id, &memory.id, resolved.revision, retract) + .unwrap(); + let terminal = candidate(&subject.id, "candidate-after-retract", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Revise, + &memory.id, + retracted.revision, + "Retraction must remain terminal", + )) + .unwrap(); + assert!(matches!( + store.stage_candidate(terminal), + Err(SubjektivError::InvalidStateTransition { + memory_id, + from: MemoryState::Retracted, + to: MemoryState::Retracted, + }) if memory_id == memory.id + )); + } + + #[test] + fn exact_proposal_retry_remains_idempotent_after_target_advances() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(&temp.path().join("storage"), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let memory = store + .create_memory(&subject.id, draft("Retry target", "initial observation")) + .unwrap(); + let staged = store + .stage_candidate( + candidate(&subject.id, "candidate-retry-proposal", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Revise, + &memory.id, + 1, + "Refine the current revision", + )) + .unwrap(), + ) + .unwrap(); + store + .revise_memory( + &subject.id, + &memory.id, + 1, + draft("Advanced target", "another accepted refinement"), + ) + .unwrap(); + + let mut retry = staged.clone(); + retry.created_at = "2026-09-28T11:00:00.000Z".into(); + let retried = store.stage_candidate(retry).unwrap(); + assert_eq!(retried.created_at, staged.created_at); + assert_eq!(retried.revision_proposal, staged.revision_proposal); + } + + #[test] + fn bounded_memory_reads_are_scoped_ordered_and_immutable() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(&temp.path().join("storage"), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let other_subject = store.create_subject(role()).unwrap(); + assert!(matches!( + store.list_memories("subject-missing"), + Err(SubjektivError::SubjectNotFound(subject_id)) if subject_id == "subject-missing" + )); + + let mut first_draft = draft("First memory", "first reason"); + first_draft.body_md = "Immutable first body".into(); + let first = store.create_memory(&subject.id, first_draft).unwrap(); + thread::sleep(Duration::from_millis(2)); + let second = store + .create_memory(&subject.id, draft("Second memory", "second reason")) + .unwrap(); + let memories = store.list_memories(&subject.id).unwrap(); + assert_eq!( + memories.iter().map(|record| &record.id).collect::>(), + vec![&second.id, &first.id] + ); + + thread::sleep(Duration::from_millis(2)); + let mut revision = draft("First memory revised", "second revision reason"); + revision.body_md = "New second body".into(); + let revised = store + .revise_memory(&subject.id, &first.id, 1, revision) + .unwrap(); + let memories = store.list_memories(&subject.id).unwrap(); + assert_eq!(memories[0].id, first.id); + assert_eq!(memories[0].revision, 2); + + let history = store.list_memory_revisions(&subject.id, &first.id).unwrap(); + assert_eq!( + history + .iter() + .map(|record| record.revision) + .collect::>(), + vec![2, 1] + ); + assert_eq!(history[0].body_md, "New second body"); + assert_eq!(history[0].change_reason, "second revision reason"); + assert_eq!(history[1].body_md, "Immutable first body"); + assert_eq!(history[1].change_reason, "first reason"); + assert_eq!( + store + .scoped_memory_revision(&subject.id, &first.id, 1) + .unwrap() + .unwrap() + .body_md, + "Immutable first body" + ); + let scoped_current = store.scoped_memory(&subject.id, &first.id).unwrap(); + assert_eq!(scoped_current.id, revised.id); + assert_eq!(scoped_current.revision, revised.revision); + assert_eq!(scoped_current.body_md, revised.body_md); + + assert!(matches!( + store.list_memory_revisions(&other_subject.id, &first.id), + Err(SubjektivError::SubjectScopeMismatch { + subject_id, + reference, + }) if subject_id == other_subject.id && reference == first.id + )); + assert!(matches!( + store.scoped_memory(&other_subject.id, &first.id), + Err(SubjektivError::SubjectScopeMismatch { .. }) + )); + } + #[test] fn candidate_apply_revision_history_and_resolution_are_atomic() { let temp = tempfile::tempdir().unwrap(); diff --git a/docs/design/subjektiv-store.md b/docs/design/subjektiv-store.md index 50d9c6357..156713fff 100644 --- a/docs/design/subjektiv-store.md +++ b/docs/design/subjektiv-store.md @@ -159,3 +159,64 @@ subsequent consolidation must compare them using the existing candidate identity/content boundary. The system must not claim that the partial writes were rolled back or silently delete them. Subject recall, consolidation, and legacy Memory migration remain separate work. + +## Recall and explicit proposal tools (T-669) + +An enabled `feature.subjektiv` installs five names that do not overlap the legacy +single-Markdown Memory tools: `SubjektivMemoryQuery`, `SubjektivMemoryRead`, +`SubjektivMemoryListRevisions`, `SubjektivMemoryRemember`, and +`SubjektivMemoryProposeRevision`. Runtime-signed Worker source plus current +`subjektiv:` singleton ownership is re-evaluated for every operation; +none of the model-visible inputs contains a subject, Runtime, Worker, Session, +origin, or raw provenance object. + +Query searches only current projections. Omitted `states` means `active`; explicit +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 +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 +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. + +Revision proposals add optional `revision_proposal` metadata to the existing v2 +`SubjectStagingRecord` envelope; automatic extraction remains proposal-free and +its model schema is unchanged. The metadata is serialized atomically with the +candidate and therefore remains present in candidate reads and immutable +resolution copies: + +```json +{ + "intent": "revise", + "memory_id": "memory-…", + "expected_revision": 3, + "change_reason": "The committed evidence corrects the prior condition" +} +``` + +This is the T-670 handoff fixture. Consolidation must re-read this typed metadata, +re-check subject ownership, `expected_revision`, and the requested state +transition in the same transaction that applies the candidate, and leave a stale +proposal unresolved/conflicted. It must not parse target information from the +claim, retarget to the latest revision, or revive a retracted Memory. Valid +transitions at staging are revise while active/resolved, active→resolved, +active/resolved→retracted, and resolved→active. Retraction remains terminal. diff --git a/openapi/server-api.json b/openapi/server-api.json index b0289a493..69ebe5b89 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -14081,7 +14081,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:f11b2e87b1d89a62f6885000cd88b0753e4f4743abaa8cb1dbd09c682fcd1564" + "x-yoi-source-digest": "sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f" }, "openapi": "3.1.0", "paths": { diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index 402ca4ec2..77a81d918 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f11b2e87b1d89a62f6885000cd88b0753e4f4743abaa8cb1dbd09c682fcd1564) +// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index b7729a45a..256443914 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f11b2e87b1d89a62f6885000cd88b0753e4f4743abaa8cb1dbd09c682fcd1564) +// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index 44018280e..6bc86ddb3 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f11b2e87b1d89a62f6885000cd88b0753e4f4743abaa8cb1dbd09c682fcd1564) +// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/runtime-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index 0e0269569..c6e12177f 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f11b2e87b1d89a62f6885000cd88b0753e4f4743abaa8cb1dbd09c682fcd1564) +// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index 03244afc9..b9d390390 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f11b2e87b1d89a62f6885000cd88b0753e4f4743abaa8cb1dbd09c682fcd1564) +// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check