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

feat: connect subjektiv extraction lifecycle

Browse Source
This commit is contained in:
Hare
2026-10-02 06:00:31 +09:00
parent 62b0f87477
commit 6091fafcac
13 changed files with 2082 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server.rs
+521
View File
@@ -1388,6 +1388,7 @@ pub struct WorkspaceApi {
config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry,
prompt_projection_cache: crate::prompt_settings::WorkspacePromptProjectionCache,
feature_storage: crate::WorkspaceFeatureStorage,
subjektiv_registration: crate::RegisteredFeature,
authority: SqliteWorkspaceAuthority,
_worker_projection_shutdown: Arc<WorkerProjectionShutdownGuard>,
runtime: Arc<RuntimeRegistry>,
@@ -3215,6 +3216,10 @@ impl WorkspaceApi {
.map_err(|error| {
Error::Store(format!("failed to initialize Feature storage: {error}"))
})?;
let subjektiv_registration = crate::subjektiv::SubjektivStore::register(&feature_storage)
.map_err(|error| {
Error::Store(format!("failed to register subjektiv storage: {error}"))
})?;
let api = Self {
config_store,
repository_secrets,
@@ -3223,6 +3228,7 @@ impl WorkspaceApi {
prompt_projection_cache:
crate::prompt_settings::WorkspacePromptProjectionCache::default(),
feature_storage,
subjektiv_registration,
authority: SqliteWorkspaceAuthority::new(
config.database_path.clone(),
config.workspace_id.clone(),
@@ -5432,6 +5438,24 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
.merge(server_api::server_api_axum::memory_consolidation(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_subject_create(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_subject_get(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_subject_retire(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_subject_worker_start(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_stage_candidate(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_record_session(
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()))
@@ -7006,6 +7030,119 @@ impl server_api::ServerApi for ServerApiContractService {
.map_err(ApiError::into_repository_api_error)
}
async fn subjektiv_subject_create(
&self,
workspace_id: String,
request: server_api::SubjektivSubjectCreateRequest,
) -> std::result::Result<server_api::SubjektivSubjectResponse, server_api::RepositoryApiError>
{
scoped_create_subjektiv_subject(
State(self.workspace_api()?.clone()),
AxumPath(ScopedWorkspacePath { workspace_id }),
Json(request),
)
.await
.map(|(_, Json(response))| response)
.map_err(ApiError::into_repository_api_error)
}
async fn subjektiv_subject_get(
&self,
workspace_id: String,
subject_id: String,
) -> std::result::Result<server_api::SubjektivSubjectResponse, server_api::RepositoryApiError>
{
scoped_get_subjektiv_subject(
State(self.workspace_api()?.clone()),
AxumPath(ScopedSubjektivSubjectPath {
workspace_id,
subject_id,
}),
)
.await
.map(|Json(response)| response)
.map_err(ApiError::into_repository_api_error)
}
async fn subjektiv_subject_retire(
&self,
workspace_id: String,
subject_id: String,
) -> std::result::Result<server_api::SubjektivSubjectResponse, server_api::RepositoryApiError>
{
scoped_retire_subjektiv_subject(
State(self.workspace_api()?.clone()),
AxumPath(ScopedSubjektivSubjectPath {
workspace_id,
subject_id,
}),
)
.await
.map(|Json(response)| response)
.map_err(ApiError::into_repository_api_error)
}
async fn subjektiv_subject_worker_start(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
subject_id: String,
request: server_api::SubjektivSubjectWorkerStartRequest,
) -> std::result::Result<BrowserCreateWorkerResponse, server_api::RepositoryApiError> {
scoped_start_subjektiv_subject_worker(
State(self.workspace_api()?.clone()),
AxumPath(ScopedSubjektivSubjectPath {
workspace_id,
subject_id,
}),
contract_request_headers(&context)?,
Json(request),
)
.await
.map(|Json(response)| response)
.map_err(ApiError::into_repository_api_error)
}
async fn subjektiv_stage_candidate(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
request: server_api::SubjektivStageCandidateRequest,
) -> std::result::Result<
server_api::SubjektivStageCandidateResponse,
server_api::RepositoryApiError,
> {
scoped_stage_subjektiv_candidate(
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 subjektiv_record_session(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
request: server_api::SubjektivRecordSessionRequest,
) -> std::result::Result<
server_api::SubjektivRecordSessionResponse,
server_api::RepositoryApiError,
> {
scoped_record_subjektiv_session(
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,
@@ -10217,6 +10354,12 @@ struct ScopedWorkspacePath {
workspace_id: String,
}
#[derive(Debug, Clone, Deserialize)]
struct ScopedSubjektivSubjectPath {
workspace_id: String,
subject_id: String,
}
#[derive(Debug, Deserialize)]
struct ScopedFlowPath {
workspace_id: String,
@@ -16352,6 +16495,255 @@ async fn scoped_memory_backend_operation(
Ok(Json(response))
}
const SUBJEKTIV_SINGLETON_PREFIX: &str = "subjektiv:";
fn open_subjektiv_store(api: &WorkspaceApi) -> ApiResult<crate::subjektiv::SubjektivStore> {
crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration)
.map_err(|error| ApiError::from(Error::Store(error.to_string())))
}
fn subjektiv_subject_id_from_singleton_key(key: &str) -> ApiResult<&str> {
key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX)
.filter(|subject_id| {
!subject_id.is_empty()
&& !subject_id.chars().any(char::is_control)
&& subject_id.trim() == *subject_id
})
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"current Worker is not connected to a subjektiv subject".to_string(),
)
.into()
})
}
fn subjektiv_singleton_key(subject_id: &str) -> ApiResult<String> {
if subject_id.trim().is_empty() || subject_id.chars().any(char::is_control) {
return Err(Error::InvalidInput("subjektiv subject id is invalid".to_string()).into());
}
let key = format!("{SUBJEKTIV_SINGLETON_PREFIX}{subject_id}");
crate::store::validate_worker_singleton_key(&key)?;
Ok(key)
}
fn subjektiv_session_attribution(
api: &WorkspaceApi,
workspace_id: &str,
context: &server_api::ServerRequestContext,
session_id: &str,
) -> ApiResult<crate::subjektiv::SubjectSessionAttribution> {
let source = context.runtime_source.as_ref().ok_or_else(|| {
Error::WorkspacePermissionDenied(
"subjektiv extraction requires 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(),
)
})?;
let worker = RuntimeWorkerRef::new(&source.runtime_id, worker_id);
let lease = api
.store
.require_current_worker_singleton_owner(workspace_id, &worker)?
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"current Worker has no keyed singleton ownership".to_string(),
)
})?;
let subject_id = subjektiv_subject_id_from_singleton_key(&lease.key)?;
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());
}
crate::subjektiv::SubjectSessionAttribution::new(
subject_id,
worker.runtime_id,
worker.worker_id,
session_id,
)
.map_err(|error| Error::InvalidInput(error.to_string()).into())
}
fn subjektiv_subject_response(
api: &WorkspaceApi,
subject: crate::subjektiv::SubjectRecord,
) -> server_api::SubjektivSubjectResponse {
let current_worker = subjektiv_singleton_key(&subject.id)
.ok()
.and_then(|key| {
api.store
.current_worker_singleton_owner(&api.config.workspace_id, &key)
.ok()
.flatten()
})
.and_then(|lease| api.runtime.worker(&lease.worker).ok())
.map(worker_launch_worker_summary);
server_api::SubjektivSubjectResponse {
id: subject.id,
role: subject.role.as_str().to_string(),
state: match subject.state {
crate::subjektiv::SubjectState::Active => server_api::SubjektivSubjectState::Active,
crate::subjektiv::SubjectState::Retired => server_api::SubjektivSubjectState::Retired,
},
store_revision: subject.store_revision,
created_at: subject.created_at,
updated_at: subject.updated_at,
current_worker,
}
}
async fn scoped_create_subjektiv_subject(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
Json(request): Json<server_api::SubjektivSubjectCreateRequest>,
) -> ApiResult<(StatusCode, Json<server_api::SubjektivSubjectResponse>)> {
validate_workspace_scope(&api, &path.workspace_id)?;
let role = crate::subjektiv::SubjectRole::new(request.role)
.map_err(|error| Error::InvalidInput(error.to_string()))?;
let subject = open_subjektiv_store(&api)?
.create_subject(role)
.map_err(|error| Error::Store(error.to_string()))?;
Ok((
StatusCode::CREATED,
Json(subjektiv_subject_response(&api, subject)),
))
}
async fn scoped_get_subjektiv_subject(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedSubjektivSubjectPath>,
) -> ApiResult<Json<server_api::SubjektivSubjectResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let subject = open_subjektiv_store(&api)?
.subject(&path.subject_id)
.map_err(|error| Error::Store(error.to_string()))?
.ok_or_else(|| {
Error::InvalidInput(format!("unknown subjektiv subject `{}`", path.subject_id))
})?;
Ok(Json(subjektiv_subject_response(&api, subject)))
}
async fn scoped_retire_subjektiv_subject(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedSubjektivSubjectPath>,
) -> ApiResult<Json<server_api::SubjektivSubjectResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let subject = open_subjektiv_store(&api)?
.retire_subject(&path.subject_id)
.map_err(|error| Error::InvalidInput(error.to_string()))?;
// Retirement deliberately does not stop or remove the current Worker. The
// Worker lease and subject lifecycle are independent authorities.
Ok(Json(subjektiv_subject_response(&api, subject)))
}
async fn scoped_start_subjektiv_subject_worker(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedSubjektivSubjectPath>,
headers: HeaderMap,
Json(request): Json<server_api::SubjektivSubjectWorkerStartRequest>,
) -> ApiResult<Json<BrowserCreateWorkerResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let subject = open_subjektiv_store(&api)?
.subject(&path.subject_id)
.map_err(|error| Error::Store(error.to_string()))?
.ok_or_else(|| {
Error::InvalidInput(format!("unknown subjektiv subject `{}`", path.subject_id))
})?;
if subject.state != crate::subjektiv::SubjectState::Active {
return Err(Error::InvalidInput(format!(
"subjektiv subject `{}` is retired",
path.subject_id
))
.into());
}
let mut worker = request.worker;
if worker.singleton_key.is_some() {
return Err(Error::InvalidInput(
"subjektiv subject Worker start owns singleton selection; singleton_key must be omitted"
.to_string(),
)
.into());
}
worker.singleton_key = Some(subjektiv_singleton_key(&path.subject_id)?);
create_workspace_worker(State(api), headers, Json(worker)).await
}
async fn scoped_stage_subjektiv_candidate(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
context: server_api::ServerRequestContext,
Json(mut request): Json<server_api::SubjektivStageCandidateRequest>,
) -> ApiResult<Json<server_api::SubjektivStageCandidateResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let attribution =
subjektiv_session_attribution(&api, &path.workspace_id, &context, &request.session_id)?;
let subject_id = attribution.subject_id.clone();
let session_id = attribution.session_id.as_str();
for source_ref in &mut request.operation.source_refs {
if source_ref
.session_id
.as_deref()
.is_some_and(|value| value != session_id)
{
return Err(Error::InvalidInput(
"candidate source_ref belongs to a different Session".to_string(),
)
.into());
}
source_ref.session_id = Some(session_id.to_string());
}
if matches!(
request.operation.candidate.kind,
memory::extract::CandidateKind::Preference
) && request.operation.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".to_string(),
)
.into());
}
let candidate_id = Uuid::now_v7().to_string();
let record = memory::extract::StagingRecord::from_candidate(
candidate_id,
request.operation.extract_run_id,
request.operation.source,
request.operation.candidate,
request.operation.evidence,
request.operation.source_refs,
);
let record = crate::subjektiv::SubjectStagingRecord::attach(&subject_id, record);
let (staged, _) = open_subjektiv_store(&api)?
.stage_candidate_with_attribution(record, attribution)
.map_err(|error| Error::InvalidInput(error.to_string()))?;
Ok(Json(server_api::SubjektivStageCandidateResponse {
staging_id: staged.id,
}))
}
async fn scoped_record_subjektiv_session(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
context: server_api::ServerRequestContext,
Json(request): Json<server_api::SubjektivRecordSessionRequest>,
) -> ApiResult<Json<server_api::SubjektivRecordSessionResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let attribution =
subjektiv_session_attribution(&api, &path.workspace_id, &context, &request.session_id)?;
let attribution = open_subjektiv_store(&api)?
.record_session_attribution(attribution)
.map_err(|error| Error::InvalidInput(error.to_string()))?;
Ok(Json(server_api::SubjektivRecordSessionResponse {
subject_id: attribution.subject_id,
session_id: attribution.session_id,
}))
}
const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation";
const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation";
@@ -33523,6 +33915,120 @@ mod tests {
);
}
#[tokio::test]
async fn subjektiv_subject_worker_and_staging_derive_scope_from_singleton_authority() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let subject = open_subjektiv_store(&api)
.unwrap()
.create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap())
.unwrap();
let request = server_api::SubjektivSubjectWorkerStartRequest {
worker: CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
display_name: "Subject Worker".to_string(),
singleton_key: None,
profile: Some("builtin:companion".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
control_operation_id: None,
},
};
let Json(created) = scoped_start_subjektiv_subject_worker(
State(api.clone()),
AxumPath(ScopedSubjektivSubjectPath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
subject_id: subject.id.clone(),
}),
HeaderMap::new(),
Json(request),
)
.await
.unwrap();
assert_eq!(
created.worker.singleton_key.as_deref(),
Some(format!("subjektiv:{}", subject.id).as_str())
);
let context = server_api::ServerRequestContext {
actor: None,
worker_source: None,
runtime_source: Some(server_api::ServerRuntimeSource {
runtime_id: created.runtime_id.clone(),
worker_id: Some(created.worker_id.clone()),
}),
origin: None,
transport_headers: Vec::new(),
};
let operation = memory::backend::MemoryStageCandidateOperation {
source: memory::schema::SourceRef {
segment_id: "segment-1".to_string(),
range: [0, 0],
},
extract_run_id: "extract-1".to_string(),
candidate: memory::extract::ExtractedCandidate {
kind: memory::extract::CandidateKind::Decision,
claim: "Keep subject scope host-derived".to_string(),
why_useful: "Prevents cross-subject writes".to_string(),
staleness: None,
evidence_ids: vec!["E00000000".to_string()],
},
evidence: vec![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("Keep subject scope host-derived".to_string()),
summary: Some("Scope decision".to_string()),
}],
source_refs: vec![memory::schema::SourceEvidenceRef {
segment_id: Some("segment-1".to_string()),
entry_range: Some([0, 0]),
evidence_id: Some("E00000000".to_string()),
evidence_kind: Some(memory::schema::EvidenceKind::new(
memory::schema::EvidenceKind::MESSAGE,
)),
..Default::default()
}],
};
let Json(staged) = scoped_stage_subjektiv_candidate(
State(api.clone()),
AxumPath(ScopedWorkspacePath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
}),
context,
Json(server_api::SubjektivStageCandidateRequest {
session_id: "session-1".to_string(),
operation,
}),
)
.await
.unwrap();
let store = open_subjektiv_store(&api).unwrap();
let candidate = store
.staging_candidate(&subject.id, &staged.staging_id)
.unwrap()
.unwrap();
assert_eq!(candidate.subject_id, subject.id);
assert_eq!(
candidate.source_refs[0].session_id.as_deref(),
Some("session-1")
);
let attribution = store.session_attribution("session-1").unwrap().unwrap();
assert_eq!(attribution.subject_id, subject.id);
assert_eq!(attribution.worker_id, created.worker_id);
}
#[tokio::test]
async fn singleton_retry_disconnect_replacement_and_restore_are_generation_fenced() {
let workspace = tempfile::tempdir().unwrap();
@@ -48089,6 +48595,21 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3);
serde_json::from_slice(&bytes).unwrap()
}
#[test]
fn subjektiv_singleton_scope_is_explicit_and_opaque() {
assert_eq!(
subjektiv_subject_id_from_singleton_key("subjektiv:subject-42").unwrap(),
"subject-42"
);
assert!(subjektiv_subject_id_from_singleton_key("workspace-orchestrator").is_err());
assert!(subjektiv_subject_id_from_singleton_key("subjektiv:").is_err());
assert!(subjektiv_subject_id_from_singleton_key("subjektiv: subject-42").is_err());
assert_eq!(
subjektiv_singleton_key("subject-42").unwrap(),
"subjektiv:subject-42"
);
}
fn write_ticket(
database_path: &Path,
workspace_id: &str,
crates/workspace-server/src/subjektiv.rs
+535 -27
View File
@@ -41,6 +41,16 @@ pub enum SubjektivError {
CandidateConflict(String),
#[error("staging candidate `{0}` is already resolved")]
CandidateResolved(String),
#[error(
"session `{session_id}` is already attributed to subject `{existing_subject_id}` and cannot be attributed to `{requested_subject_id}`"
)]
SessionSubjectConflict {
session_id: String,
existing_subject_id: String,
requested_subject_id: String,
},
#[error("session `{0}` already has different historical attribution")]
SessionAttributionConflict(String),
#[error("memory `{0}` was not found for this subject")]
MemoryNotFound(String),
#[error("memory `{memory_id}` revision conflict: expected {expected}, current {actual}")]
@@ -117,6 +127,40 @@ pub struct SubjectRecord {
pub updated_at: String,
}
/// Immutable host-recorded history that attributes one durable Session to the
/// subject and Worker that produced it. This is archival provenance only: it is
/// deliberately not a current-Worker link or execution ownership record.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SubjectSessionAttribution {
pub schema_version: u32,
pub subject_id: String,
pub runtime_id: String,
pub worker_id: String,
pub session_id: String,
pub attributed_at: String,
}
impl SubjectSessionAttribution {
pub fn new(
subject_id: impl Into<String>,
runtime_id: impl Into<String>,
worker_id: impl Into<String>,
session_id: impl Into<String>,
) -> Result<Self> {
let record = Self {
schema_version: SUBJEKTIV_SCHEMA_VERSION,
subject_id: subject_id.into(),
runtime_id: runtime_id.into(),
worker_id: worker_id.into(),
session_id: session_id.into(),
attributed_at: now(),
};
validate_session_attribution(&record)?;
Ok(record)
}
}
/// The existing extraction schema with only host-owned subject scope and
/// persistence time added. Model input still contains neither field.
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -602,11 +646,45 @@ END;
Ok(())
}
static MIGRATIONS: &[FeatureMigration] = &[FeatureMigration::new(
1,
"create subjektiv subject memory store",
create_schema,
)];
fn add_subject_session_attribution(
transaction: &Transaction<'_>,
) -> crate::feature_storage::Result<()> {
transaction.execute_batch(
r#"
CREATE TABLE subject_session_attributions (
session_id TEXT PRIMARY KEY,
subject_id TEXT NOT NULL,
runtime_id TEXT NOT NULL,
worker_id TEXT NOT NULL,
record_json TEXT NOT NULL,
attributed_at TEXT NOT NULL,
FOREIGN KEY (subject_id) REFERENCES subjects(subject_id) ON DELETE RESTRICT
);
CREATE INDEX subject_session_attributions_by_subject
ON subject_session_attributions(subject_id, attributed_at, session_id);
CREATE TRIGGER subject_session_attributions_no_update
BEFORE UPDATE ON subject_session_attributions BEGIN
SELECT RAISE(ABORT, 'subjektiv session attribution is immutable');
END;
CREATE TRIGGER subject_session_attributions_no_delete
BEFORE DELETE ON subject_session_attributions BEGIN
SELECT RAISE(ABORT, 'subjektiv session attribution is retained');
END;
"#,
)?;
Ok(())
}
static MIGRATIONS: &[FeatureMigration] = &[
FeatureMigration::new(1, "create subjektiv subject memory store", create_schema),
FeatureMigration::new(
2,
"add immutable subject session attribution",
add_subject_session_attribution,
),
];
pub const REGISTRATION: FeatureRegistration =
FeatureRegistration::new(SUBJEKTIV_FEATURE_ID, MIGRATIONS);
@@ -740,33 +818,89 @@ impl SubjektivStore {
let raw = serde_json::to_string(&record)?;
self.database.try_transaction(|transaction| {
require_active_subject(transaction, &record.subject_id)?;
let existing = transaction
write_staging_candidate(transaction, self.workspace_id(), record, raw)
})
}
/// Atomically records immutable host attribution for the committed Session
/// and stages its extracted candidate. Exact retries return the first stored
/// records, including their original host timestamps.
pub fn stage_candidate_with_attribution(
&self,
mut record: SubjectStagingRecord,
attribution: SubjectSessionAttribution,
) -> Result<(SubjectStagingRecord, SubjectSessionAttribution)> {
validate_session_attribution(&attribution)?;
if record.subject_id != attribution.subject_id {
return Err(SubjektivError::SubjectScopeMismatch {
subject_id: record.subject_id,
reference: format!(
"session {} is attributed to subject {}",
attribution.session_id, attribution.subject_id
),
});
}
attach_session_to_candidate(&mut record, &attribution.session_id)?;
validate_staging_record(self.workspace_id(), &record)?;
let candidate_raw = serde_json::to_string(&record)?;
let attribution_raw = serde_json::to_string(&attribution)?;
self.database.try_transaction(|transaction| {
require_active_subject(transaction, &record.subject_id)?;
let attribution = write_session_attribution(transaction, attribution, attribution_raw)?;
let candidate =
write_staging_candidate(transaction, self.workspace_id(), record, candidate_raw)?;
Ok((candidate, attribution))
})
}
pub fn record_session_attribution(
&self,
attribution: SubjectSessionAttribution,
) -> Result<SubjectSessionAttribution> {
validate_session_attribution(&attribution)?;
let raw = serde_json::to_string(&attribution)?;
self.database.try_transaction(|transaction| {
require_active_subject(transaction, &attribution.subject_id)?;
write_session_attribution(transaction, attribution, raw)
})
}
pub fn session_attribution(
&self,
session_id: &str,
) -> Result<Option<SubjectSessionAttribution>> {
validate_label("session id", session_id)?;
self.database.try_with_connection(|connection| {
let raw = connection
.query_row(
"SELECT record_json FROM staging_records
WHERE subject_id = ?1 AND candidate_id = ?2",
params![record.subject_id, record.id],
"SELECT record_json FROM subject_session_attributions
WHERE session_id = ?1",
[session_id],
|row| row.get::<_, String>(0),
)
.optional()?;
if let Some(existing) = existing {
if equivalent_staging_payload(&existing, &raw)? {
return parse_staging(self.workspace_id(), &existing);
}
return Err(SubjektivError::CandidateConflict(record.id.clone()));
}
transaction.execute(
"INSERT INTO staging_records (
subject_id, candidate_id, kind, record_json, created_at
) VALUES (?1, ?2, ?3, ?4, ?5)",
params![
record.subject_id,
record.id,
candidate_kind_name(&record.kind),
raw,
record.created_at
],
raw.map(|raw| parse_session_attribution(&raw)).transpose()
})
}
pub fn subject_session_attributions(
&self,
subject_id: &str,
) -> Result<Vec<SubjectSessionAttribution>> {
validate_label("subject id", subject_id)?;
self.database.try_with_connection(|connection| {
let mut statement = connection.prepare(
"SELECT record_json FROM subject_session_attributions
WHERE subject_id = ?1
ORDER BY attributed_at ASC, session_id ASC",
)?;
Ok(record)
let rows = statement.query_map([subject_id], |row| row.get::<_, String>(0))?;
let mut records = Vec::new();
for row in rows {
records.push(parse_session_attribution(&row?)?);
}
Ok(records)
})
}
@@ -1228,6 +1362,85 @@ fn write_memory_revision(
Ok(record)
}
fn write_staging_candidate(
transaction: &Transaction<'_>,
workspace_id: &str,
record: SubjectStagingRecord,
raw: String,
) -> Result<SubjectStagingRecord> {
let existing = transaction
.query_row(
"SELECT record_json FROM staging_records
WHERE subject_id = ?1 AND candidate_id = ?2",
params![record.subject_id, record.id],
|row| row.get::<_, String>(0),
)
.optional()?;
if let Some(existing) = existing {
if equivalent_staging_payload(&existing, &raw)? {
return parse_staging(workspace_id, &existing);
}
return Err(SubjektivError::CandidateConflict(record.id.clone()));
}
transaction.execute(
"INSERT INTO staging_records (
subject_id, candidate_id, kind, record_json, created_at
) VALUES (?1, ?2, ?3, ?4, ?5)",
params![
record.subject_id,
record.id,
candidate_kind_name(&record.kind),
raw,
record.created_at
],
)?;
Ok(record)
}
fn write_session_attribution(
transaction: &Transaction<'_>,
record: SubjectSessionAttribution,
raw: String,
) -> Result<SubjectSessionAttribution> {
let existing = transaction
.query_row(
"SELECT record_json FROM subject_session_attributions WHERE session_id = ?1",
[&record.session_id],
|row| row.get::<_, String>(0),
)
.optional()?;
if let Some(existing) = existing {
let existing = parse_session_attribution(&existing)?;
if existing.subject_id != record.subject_id {
return Err(SubjektivError::SessionSubjectConflict {
session_id: record.session_id,
existing_subject_id: existing.subject_id,
requested_subject_id: record.subject_id,
});
}
if equivalent_session_attribution(&existing, &record) {
return Ok(existing);
}
return Err(SubjektivError::SessionAttributionConflict(
record.session_id,
));
}
transaction.execute(
"INSERT INTO subject_session_attributions (
session_id, subject_id, runtime_id, worker_id, record_json, attributed_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![
record.session_id,
record.subject_id,
record.runtime_id,
record.worker_id,
raw,
record.attributed_at
],
)?;
Ok(record)
}
fn insert_resolution(
transaction: &Transaction<'_>,
subject_id: &str,
@@ -1436,6 +1649,48 @@ fn validate_memory_refs(
Ok(())
}
fn validate_session_attribution(record: &SubjectSessionAttribution) -> Result<()> {
validate_schema_version(record.schema_version, "subject Session attribution")?;
validate_label("subject id", &record.subject_id)?;
validate_label("runtime id", &record.runtime_id)?;
validate_label("worker id", &record.worker_id)?;
validate_label("session id", &record.session_id)?;
validate_label("Session attribution timestamp", &record.attributed_at)?;
chrono::DateTime::parse_from_rfc3339(&record.attributed_at).map_err(|error| {
SubjektivError::InvalidRecord(format!(
"Session attribution timestamp is not RFC 3339: {error}"
))
})?;
Ok(())
}
fn attach_session_to_candidate(record: &mut SubjectStagingRecord, session_id: &str) -> Result<()> {
if record.source_refs.is_empty() {
record.source_refs.push(SourceEvidenceRef {
session_id: Some(session_id.to_string()),
segment_id: Some(record.source.segment_id.clone()),
entry_range: Some(record.source.range),
..SourceEvidenceRef::default()
});
return Ok(());
}
for source in &mut record.source_refs {
match source.session_id.as_deref() {
Some(existing) if existing != session_id => {
return Err(SubjektivError::SubjectScopeMismatch {
subject_id: record.subject_id.clone(),
reference: format!(
"candidate source Session `{existing}` does not match attributed Session `{session_id}`"
),
});
}
Some(_) => {}
None => source.session_id = Some(session_id.to_string()),
}
}
Ok(())
}
fn validate_staging_record(workspace_id: &str, record: &SubjectStagingRecord) -> Result<()> {
if record.schema_version != STAGING_SCHEMA_VERSION {
return Err(SubjektivError::InvalidRecord(format!(
@@ -1450,6 +1705,18 @@ fn validate_staging_record(workspace_id: &str, record: &SubjectStagingRecord) ->
validate_nonempty("candidate claim", &record.claim)?;
validate_nonempty("candidate why_useful", &record.why_useful)?;
validate_range("candidate source", record.source.range)?;
if matches!(record.kind, CandidateKind::Preference)
&& record.evidence.iter().any(|evidence| {
!matches!(
evidence.origin.as_ref().map(|origin| &origin.kind),
Some(memory::schema::EvidenceOriginKind::HumanInput)
)
})
{
return Err(SubjektivError::InvalidRecord(
"preference candidates require exclusively HumanInput evidence".to_string(),
));
}
let mut evidence_ids = HashSet::new();
for evidence in &record.evidence {
@@ -1552,6 +1819,17 @@ fn parse_subject(raw: &str) -> Result<SubjectRecord> {
Ok(record)
}
fn equivalent_session_attribution(
left: &SubjectSessionAttribution,
right: &SubjectSessionAttribution,
) -> bool {
left.schema_version == right.schema_version
&& left.subject_id == right.subject_id
&& left.runtime_id == right.runtime_id
&& left.worker_id == right.worker_id
&& left.session_id == right.session_id
}
fn equivalent_staging_payload(left: &str, right: &str) -> Result<bool> {
let mut left = serde_json::from_str::<serde_json::Value>(left)?;
let mut right = serde_json::from_str::<serde_json::Value>(right)?;
@@ -1569,6 +1847,12 @@ fn equivalent_staging_payload(left: &str, right: &str) -> Result<bool> {
Ok(left == right)
}
fn parse_session_attribution(raw: &str) -> Result<SubjectSessionAttribution> {
let record: SubjectSessionAttribution = serde_json::from_str(raw)?;
validate_session_attribution(&record)?;
Ok(record)
}
fn parse_staging(workspace_id: &str, raw: &str) -> Result<SubjectStagingRecord> {
let record = parse_staging_without_workspace(raw)?;
validate_staging_record(workspace_id, &record)?;
@@ -1736,6 +2020,12 @@ mod tests {
use super::*;
use crate::FeatureStorage;
static V1_MIGRATIONS: &[FeatureMigration] = &[FeatureMigration::new(
1,
"create subjektiv subject memory store",
create_schema,
)];
fn role() -> SubjectRole {
SubjectRole::new("workspace_companion").unwrap()
}
@@ -1787,6 +2077,19 @@ mod tests {
SubjectStagingRecord::attach_at(subject_id, record, "2026-09-28T10:00:00.000Z".into())
}
fn attribution(
subject_id: &str,
runtime_id: &str,
worker_id: &str,
session_id: &str,
attributed_at: &str,
) -> SubjectSessionAttribution {
let mut record =
SubjectSessionAttribution::new(subject_id, runtime_id, worker_id, session_id).unwrap();
record.attributed_at = attributed_at.to_string();
record
}
fn draft(claim: &str, reason: &str) -> MemoryDraft {
MemoryDraft::active(
CandidateKind::Constraint,
@@ -1851,12 +2154,217 @@ mod tests {
assert!(serde_json::from_str::<MemoryRecord>(&raw.replace("active", "unknown")).is_err());
}
#[test]
fn v1_store_migrates_before_attributed_staging() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("storage");
let subject_id = {
let manager = FeatureStorage::new(&root);
let workspace = manager.workspace("workspace-a").unwrap();
let registration = workspace
.register(FeatureRegistration::new(
SUBJEKTIV_FEATURE_ID,
V1_MIGRATIONS,
))
.unwrap();
let store = SubjektivStore::open(&workspace, &registration).unwrap();
store.create_subject(role()).unwrap().id
};
let (_manager, _workspace, store) = open_store(&root, "workspace-a");
let (staged, stored_attribution) = store
.stage_candidate_with_attribution(
candidate(&subject_id, "candidate-after-migration", "workspace-a"),
attribution(
&subject_id,
"runtime-1",
"worker-1",
"session-1",
"2026-09-28T10:00:00.000Z",
),
)
.unwrap();
assert_eq!(staged.subject_id, subject_id);
assert_eq!(stored_attribution.session_id, "session-1");
assert_eq!(
store.session_attribution("session-1").unwrap().unwrap(),
stored_attribution
);
}
#[test]
fn attributed_staging_is_atomic_and_retry_idempotent() {
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 first_attribution = attribution(
&subject.id,
"runtime-1",
"worker-1",
"session-1",
"2026-09-28T10:00:00.000Z",
);
let (first_candidate, first_attribution) = store
.stage_candidate_with_attribution(
candidate(&subject.id, "candidate-1", "workspace-a"),
first_attribution,
)
.unwrap();
let mut retried_candidate = first_candidate.clone();
retried_candidate.created_at = "2026-09-28T11:00:00.000Z".into();
let retried_attribution = attribution(
&subject.id,
"runtime-1",
"worker-1",
"session-1",
"2026-09-28T11:00:00.000Z",
);
let (retried_candidate, retried_attribution) = store
.stage_candidate_with_attribution(retried_candidate, retried_attribution)
.unwrap();
assert_eq!(retried_candidate.created_at, first_candidate.created_at);
assert_eq!(retried_attribution, first_attribution);
let mut conflicting_candidate = candidate(&subject.id, "candidate-1", "workspace-a");
conflicting_candidate.claim = "different content".into();
conflicting_candidate.source_refs[0].session_id = Some("session-2".into());
let result = store.stage_candidate_with_attribution(
conflicting_candidate,
attribution(
&subject.id,
"runtime-1",
"worker-1",
"session-2",
"2026-09-28T12:00:00.000Z",
),
);
assert!(matches!(
result,
Err(SubjektivError::CandidateConflict(candidate_id))
if candidate_id == "candidate-1"
));
assert!(store.session_attribution("session-2").unwrap().is_none());
}
#[test]
fn worker_replacement_preserves_subject_history_and_rejects_cross_subject_session() {
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();
store
.stage_candidate_with_attribution(
candidate(&subject.id, "candidate-1", "workspace-a"),
attribution(
&subject.id,
"runtime-1",
"worker-1",
"session-1",
"2026-09-28T10:00:00.000Z",
),
)
.unwrap();
let mut replacement_candidate = candidate(&subject.id, "candidate-2", "workspace-a");
replacement_candidate.source_refs[0].session_id = None;
let (replacement_candidate, _) = store
.stage_candidate_with_attribution(
replacement_candidate,
attribution(
&subject.id,
"runtime-2",
"worker-2",
"session-2",
"2026-09-28T11:00:00.000Z",
),
)
.unwrap();
assert_eq!(
replacement_candidate.source_refs[0].session_id.as_deref(),
Some("session-2")
);
let history = store.subject_session_attributions(&subject.id).unwrap();
assert_eq!(history.len(), 2);
assert_eq!(history[0].worker_id, "worker-1");
assert_eq!(history[1].worker_id, "worker-2");
assert!(history.iter().all(|entry| entry.subject_id == subject.id));
let update = store
.database
.try_with_connection::<_, SubjektivError>(|connection| {
connection.execute(
"UPDATE subject_session_attributions SET worker_id = 'worker-other'
WHERE session_id = 'session-1'",
[],
)?;
Ok(())
});
assert!(update.is_err());
let delete = store
.database
.try_with_connection::<_, SubjektivError>(|connection| {
connection.execute(
"DELETE FROM subject_session_attributions WHERE session_id = 'session-1'",
[],
)?;
Ok(())
});
assert!(delete.is_err());
let mut foreign_candidate =
candidate(&other_subject.id, "candidate-foreign", "workspace-a");
foreign_candidate.source_refs[0].session_id = Some("session-1".into());
let result = store.stage_candidate_with_attribution(
foreign_candidate,
attribution(
&other_subject.id,
"runtime-1",
"worker-1",
"session-1",
"2026-09-28T12:00:00.000Z",
),
);
assert!(matches!(
result,
Err(SubjektivError::SessionSubjectConflict {
session_id,
existing_subject_id,
requested_subject_id,
}) if session_id == "session-1"
&& existing_subject_id == subject.id
&& requested_subject_id == other_subject.id
));
assert!(
store
.staging_candidate(&other_subject.id, "candidate-foreign")
.unwrap()
.is_none()
);
}
#[test]
fn malformed_nested_provenance_is_rejected_without_staging() {
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 mut non_human_preference =
candidate(&subject.id, "candidate-preference", "workspace-a");
non_human_preference.kind = CandidateKind::Preference;
non_human_preference.evidence[0]
.origin
.as_mut()
.unwrap()
.kind = EvidenceOriginKind::ModelOutput;
assert!(matches!(
store.stage_candidate(non_human_preference),
Err(SubjektivError::InvalidRecord(message))
if message.contains("HumanInput")
));
let mut reversed = candidate(&subject.id, "candidate-reversed", "workspace-a");
reversed.evidence[0].entry_range = Some([12, 10]);
assert!(matches!(