feat: add subject-scoped memory tools

This commit is contained in:
2026-10-02 07:39:04 +09:00
parent 1f26a7f2bb
commit b391392246
17 changed files with 3048 additions and 19 deletions
+279
View File
@@ -1176,6 +1176,20 @@ pub trait ServerApi {
#[path] workspace_id: String,
#[body] request: SubjektivRecordSessionRequest,
) -> Result<SubjektivRecordSessionResponse, RepositoryApiError>;
#[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<SubjektivMemoryBackendResponse, RepositoryApiError>;
#[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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub kinds: Option<Vec<memory::extract::CandidateKind>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub states: Option<Vec<SubjektivMemoryState>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(range(min = 1, max = 100))]
pub limit: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cursor: Option<String>,
}
#[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<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
pub offset: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(range(min = 1, max = 1000))]
pub limit: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub evidence_cursor: Option<String>,
}
#[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<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cursor: Option<String>,
}
#[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<String>,
#[serde(default)]
pub evidence: Vec<memory::extract::StagingEvidence>,
#[serde(default)]
pub source_refs: Vec<memory::schema::SourceEvidenceRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub proposal: Option<SubjektivMemoryRevisionProposal>,
}
#[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<SubjektivMemoryQueryItem>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_cursor: Option<String>,
pub has_more: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct SubjektivMemoryEvidenceCandidate {
pub candidate_id: String,
pub evidence: Vec<memory::extract::StagingEvidence>,
pub source_refs: Vec<memory::schema::SourceEvidenceRef>,
}
#[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<String>,
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<usize>,
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<String>,
pub source_candidates: Vec<SubjektivMemoryEvidenceCandidate>,
/// Derivation references on this evidence page.
pub derived_from: Vec<SubjektivMemoryRevisionRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub evidence_next_cursor: Option<String>,
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<SubjektivMemoryRevisionItem>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_cursor: Option<String>,
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<SubjektivMemoryProposalTarget>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub intent: Option<SubjektivMemoryRevisionIntent>,
}
#[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",
];
@@ -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,
)
}
}
})
})
+9
View File
@@ -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(
+1
View File
@@ -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;
@@ -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),
@@ -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<dyn WorkspaceClient>,
capture: CommittedSessionCaptureHandle,
pending: Arc<Mutex<HashMap<String, PendingExplicit>>>,
}
#[derive(Clone)]
struct PendingExplicit {
receipt_id: String,
call_id: String,
kind: CandidateKind,
claim: String,
why_useful: String,
staleness: Option<String>,
proposal: Option<SubjektivMemoryRevisionProposal>,
}
impl SubjektivMemoryFeature {
pub(crate) fn from_resolved_config(
config: &manifest::ResolvedSubjektivFeatureConfig,
capture: CommittedSessionCaptureHandle,
client: Arc<dyn WorkspaceClient>,
) -> std::io::Result<Option<Self>> {
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::<SubjektivMemoryQueryRequest>(),
ReadOperation::Read => schema_for::<SubjektivMemoryReadRequest>(),
ReadOperation::ListRevisions => schema_for::<SubjektivMemoryListRevisionsRequest>(),
};
let tool: Arc<dyn Tool> = 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<ToolOutput, ToolError> {
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::<RememberParams>(),
ExplicitOperation::Propose => schema_for::<ProposeRevisionParams>(),
};
let tool: Arc<dyn Tool> = 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<String>,
#[serde(default)]
entry_refs: Vec<String>,
}
#[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<String>,
#[serde(default)]
entry_refs: Vec<String>,
}
struct SubjektivExplicitTool {
state: SubjektivMemoryState,
operation: ExplicitOperation,
}
#[async_trait]
impl Tool for SubjektivExplicitTool {
async fn execute(
&self,
input_json: &str,
context: ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
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<SubjektivMemoryBackendResponse, ToolError> {
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<String>,
entries: Vec<SessionEntryEvidence>,
proposal: Option<SubjektivMemoryRevisionProposal>,
) -> Result<SubjektivMemoryStageExplicitResponse, ToolError> {
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<RunCommitted> for PendingCommitHook {
async fn call(&self, _input: &RunCommittedContext) -> Result<(), HookError> {
let capture =
self.state.capture.capture().map_err(|error| {
HookError::new(HookErrorCategory::Dependency, error.to_string())
})?;
let view = SessionCapture::from_history_entries(
capture.segment_id.clone(),
capture.history.clone(),
);
let pending = self
.state
.pending
.lock()
.map_err(|_| HookError::new(HookErrorCategory::Internal, "pending queue poisoned"))?
.values()
.cloned()
.collect::<Vec<_>>();
let mut completed = Vec::new();
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<Vec<SessionEntryEvidence>, 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<String, ToolError> {
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<T: serde::de::DeserializeOwned>(input: &str, tool: &str) -> Result<T, ToolError> {
serde_json::from_str(input)
.map_err(|error| ToolError::InvalidArgument(format!("invalid {tool} input: {error}")))
}
fn schema_for<T: JsonSchema>() -> serde_json::Value {
serde_json::to_value(schemars::schema_for!(T)).unwrap_or_else(|_| serde_json::json!({}))
}
fn staged_output(response: SubjektivMemoryStageExplicitResponse) -> Result<ToolOutput, ToolError> {
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<String>,
value: &serde_json::Value,
) -> Result<ToolOutput, ToolError> {
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<WorkspaceResponse, WorkspaceClientError> {
Err(WorkspaceClientError::Request(
"test client has no backend".into(),
))
}
}
#[derive(Debug, Default)]
struct RecordingWorkspaceClient {
requests: Mutex<Vec<WorkspaceRequest>>,
}
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<WorkspaceResponse, WorkspaceClientError> {
self.requests.lock().unwrap().push(request);
Ok(WorkspaceResponse {
status: 200,
body: serde_json::to_string(&SubjektivMemoryBackendResponse::Staged(
SubjektivMemoryStageExplicitResponse {
candidate_id: "explicit:session-1:call-commit".into(),
receipt_id: "explicit:session-1:call-commit".into(),
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::<Vec<_>>();
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::<RememberParams>();
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::<ProposeRevisionParams>();
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())
);
}
}
+38
View File
@@ -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<SessionEntryEvidence> {
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<SessionEntryEvidence> {
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());
}
}
+4
View File
@@ -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<T: serde::Serialize>(
@@ -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)
}
}
}
File diff suppressed because it is too large Load Diff
+650 -3
View File
@@ -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<String>,
expected_revision: u64,
change_reason: impl Into<String>,
) -> Result<Self> {
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<D>(deserializer: D) -> std::result::Result<Self, D::Error>
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<StagingEvidence>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub source_refs: Vec<SourceEvidenceRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub revision_proposal: Option<RevisionProposal>,
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<Self> {
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<MemoryRecord> {
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<Option<MemoryRecord>> {
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<Vec<MemoryRecord>> {
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<Vec<MemoryRecord>> {
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<SubjectRecord> {
let raw = transaction
require_subject_in_connection(transaction, subject_id)
}
fn require_subject_in_connection(
connection: &rusqlite::Connection,
subject_id: &str,
) -> Result<SubjectRecord> {
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<MemoryRecord> {
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::<RevisionProposal>(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<_>>(),
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<_>>(),
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();