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

feat: replace revision counters with domain state contracts

Browse Source 
Use content digests, key fingerprints, binding and trust identities, immutable mutation receipts, and observation tokens instead of state-update ordinals across API, storage, Runtime, and Web contracts.

Preserve historical replay and secret material through frozen migrations; preflight uncertain removals before cutover, resume provisioning receipts, validate encrypted migration candidates, and align fresh/upgraded schemas. Include reviewed regression fixes, generated artifacts, and validation evidence.
This commit is contained in:
Hare
2026-10-10 15:55:20 +09:00
parent 72ce817de4
commit c78f7f7d80
358 changed files with 25965 additions and 12069 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/standalone/src/job_store.rs
+36 -24
View File
@@ -13,7 +13,10 @@ use serde::{Deserialize, Serialize};
use serde_json::Value;
use thiserror::Error;
const SCHEMA_VERSION: i64 = 2;
#[path = "job_store/legacy_migrations.rs"]
mod legacy_migrations;
const SCHEMA_VERSION: i64 = 3;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
@@ -69,7 +72,10 @@ impl JobSnapshot {
pub fn outcome(&self) -> job::JobOutcome {
job::JobOutcome {
job_id: self.request.job_id.clone(),
input_revision: self.request.input_revision.clone(),
input_digest: self
.request
.input_digest()
.expect("validated immutable Job input"),
attempt_id: self.attempt.attempt_id.clone(),
attempt: self.attempt.number,
state: match self.state {
@@ -135,8 +141,8 @@ pub enum JobStoreError {
ConcurrentLimit,
#[error("result or lifecycle operation is not bound to the current Job attempt")]
AttemptMismatch,
#[error("result input revision does not match the immutable request")]
RevisionMismatch,
#[error("result input digest does not match the immutable request")]
DigestMismatch,
#[error("accepted Job result cannot be changed")]
ResultConflict,
#[error("invalid Job transition: {0}")]
@@ -176,8 +182,11 @@ impl JobStore {
tx.execute_batch(GRANT_SCHEMA)?;
tx.pragma_update(None, "user_version", SCHEMA_VERSION)?;
}
1 => {
tx.execute_batch(GRANT_SCHEMA)?;
1 | 2 => {
if version == 1 {
tx.execute_batch(GRANT_SCHEMA)?;
}
legacy_migrations::migrate_legacy_jobs(&tx)?;
tx.pragma_update(None, "user_version", SCHEMA_VERSION)?;
}
SCHEMA_VERSION => {}
@@ -370,8 +379,8 @@ impl JobStore {
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let snapshot = require_snapshot(&tx, &bound.job_id)?;
check_attempt(&snapshot, &bound.attempt_id)?;
if snapshot.request.input_revision != bound.input_revision {
return Err(JobStoreError::RevisionMismatch);
if snapshot.request.input_digest()? != bound.input_digest {
return Err(JobStoreError::DigestMismatch);
}
let (digest, encoded) =
job::result_digest(&bound.result, snapshot.request.limits.max_result_bytes)?;
@@ -527,13 +536,13 @@ fn insert_attempt(
number: u8,
) -> Result<(), JobStoreError> {
conn.execute(
"INSERT INTO job_attempts (job_id, number, attempt_id, input_revision, state)
"INSERT INTO job_attempts (job_id, number, attempt_id, input_digest, state)
VALUES (?1, ?2, ?3, ?4, 'reserved')",
params![
request.job_id,
number,
job::attempt_id(&request.job_id, number),
request.input_revision
request.input_digest()?
],
)?;
Ok(())
@@ -669,7 +678,7 @@ CREATE TABLE job_attempts (
job_id TEXT NOT NULL REFERENCES job_intents(job_id),
number INTEGER NOT NULL CHECK (number BETWEEN 1 AND 255),
attempt_id TEXT NOT NULL UNIQUE,
input_revision TEXT NOT NULL,
input_digest TEXT NOT NULL,
state TEXT NOT NULL CHECK (state IN ('reserved', 'dispatched', 'completed', 'failed', 'unknown', 'cancelled')),
result_json TEXT,
result_digest TEXT,
@@ -685,7 +694,7 @@ CREATE TRIGGER job_intent_immutable BEFORE UPDATE OF
job_id, request_json, fingerprint, serialization_key, max_concurrent ON job_intents
BEGIN SELECT RAISE(ABORT, 'immutable Job intent'); END;
CREATE TRIGGER job_attempt_immutable BEFORE UPDATE OF
job_id, number, attempt_id, input_revision ON job_attempts
job_id, number, attempt_id, input_digest ON job_attempts
BEGIN SELECT RAISE(ABORT, 'immutable Job attempt identity'); END;
CREATE TRIGGER job_result_immutable BEFORE UPDATE OF result_json, result_digest ON job_attempts
WHEN OLD.result_json IS NOT NULL AND
@@ -705,7 +714,6 @@ mod tests {
JobRequest {
job_id: id.into(),
purpose: "fixture".into(),
input_revision: "revision-1".into(),
input_ref: "fixture:input".into(),
input: json!({"immutable": [1, 2, 3]}),
instruction: "Produce a structured result".into(),
@@ -725,7 +733,10 @@ mod tests {
JobResultSubmission {
job_id: snapshot.request.job_id.clone(),
attempt_id: snapshot.attempt.attempt_id.clone(),
input_revision: snapshot.request.input_revision.clone(),
input_digest: snapshot
.request
.input_digest()
.expect("validated immutable Job input"),
result,
}
}
@@ -762,10 +773,10 @@ mod tests {
for sql in [
"UPDATE job_intents SET request_json = '{}' WHERE job_id = 'identity'",
"UPDATE job_intents SET fingerprint = 'changed' WHERE job_id = 'identity'",
"UPDATE job_attempts SET input_revision = 'changed' WHERE job_id = 'identity'",
"UPDATE job_attempts SET input_digest = 'changed' WHERE job_id = 'identity'",
"UPDATE job_attempts SET attempt_id = 'changed' WHERE job_id = 'identity'",
"UPDATE job_intents SET current_attempt = 2 WHERE job_id = 'identity'",
"INSERT INTO job_attempts (job_id, number, attempt_id, input_revision, state)
"INSERT INTO job_attempts (job_id, number, attempt_id, input_digest, state)
VALUES ('absent', 1, 'absent:attempt:1', 'r', 'reserved')",
] {
assert!(
@@ -818,9 +829,6 @@ mod tests {
assert_eq!(store.reserve(original.clone()).unwrap(), snapshot);
let mut variants = Vec::new();
let mut changed = original.clone();
changed.input_revision = "revision-2".into();
variants.push(changed);
let mut changed = original.clone();
changed.input = json!({"changed": true});
variants.push(changed);
let mut changed = original.clone();
@@ -917,12 +925,16 @@ mod tests {
Err(JobStoreError::AttemptMismatch)
));
let prior: (String, String, String) = store.conn.query_row(
"SELECT state, failure, input_revision FROM job_attempts WHERE job_id = 'retry' AND number = 1",
"SELECT state, failure, input_digest FROM job_attempts WHERE job_id = 'retry' AND number = 1",
[], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
).unwrap();
assert_eq!(
prior,
("failed".into(), "timeout".into(), "revision-1".into())
(
"failed".into(),
"timeout".into(),
original.input_digest().unwrap()
)
);
store
.finish(
@@ -960,10 +972,10 @@ mod tests {
Err(JobStoreError::AttemptMismatch)
));
let mut invalid = valid.clone();
invalid.input_revision = "revision-2".into();
invalid.input_digest = "wrong-digest".into();
assert!(matches!(
store.accept(&invalid),
Err(JobStoreError::RevisionMismatch)
Err(JobStoreError::DigestMismatch)
));
let mut invalid = valid.clone();
invalid.result = json!("x".repeat(32));
@@ -1537,7 +1549,7 @@ mod tests {
let absent = JobResultSubmission {
job_id: "missing".into(),
attempt_id: "attempt".into(),
input_revision: "r".into(),
input_digest: request("missing").input_digest().unwrap(),
result: json!({}),
};
assert!(matches!(
crates/standalone/src/job_store/legacy_migrations.rs
+295
View File
@@ -0,0 +1,295 @@
use super::*;
/// Frozen schema-1/2 adapter. Preserve original intent and attempt bindings
/// before replacing counters with identities derived from immutable input.
pub(super) fn migrate_legacy_jobs(conn: &Connection) -> Result<(), JobStoreError> {
conn.execute_batch(
"CREATE TABLE legacy_job_archive (
job_id TEXT PRIMARY KEY NOT NULL,
request_json TEXT NOT NULL,
fingerprint TEXT NOT NULL,
attempts_json TEXT NOT NULL
);
DROP TRIGGER job_intent_immutable;
DROP TRIGGER job_attempt_immutable;",
)?;
let intents = conn
.prepare("SELECT job_id, request_json, fingerprint FROM job_intents")?
.query_map([], |r| {
Ok((
r.get::<_, String>(0)?,
r.get::<_, String>(1)?,
r.get::<_, String>(2)?,
))
})?
.collect::<Result<Vec<_>, _>>()?;
for (id, original, fingerprint) in intents {
let attempts = conn.prepare(
"SELECT number, attempt_id, input_revision, state, result_json, result_digest, failure
FROM job_attempts WHERE job_id = ?1 ORDER BY number",
)?.query_map([&id], |r| Ok(serde_json::json!({
"number":r.get::<_, u8>(0)?, "attempt_id":r.get::<_, String>(1)?,
"input_revision":r.get::<_, String>(2)?, "state":r.get::<_, String>(3)?,
"result_json":r.get::<_, Option<String>>(4)?, "result_digest":r.get::<_, Option<String>>(5)?,
"failure":r.get::<_, Option<String>>(6)?
})))?.collect::<Result<Vec<_>, _>>()?;
conn.execute(
"INSERT INTO legacy_job_archive VALUES (?1, ?2, ?3, ?4)",
params![id, original, fingerprint, serde_json::to_string(&attempts)?],
)?;
let mut value: Value = serde_json::from_str(&original)?;
value
.as_object_mut()
.ok_or_else(|| JobStoreError::Corrupt("legacy request is not an object".into()))?
.remove("input_revision");
let request: JobRequest = serde_json::from_value(value)?;
if request.job_id != id {
return Err(JobStoreError::Corrupt(
"legacy request Job id mismatch".into(),
));
}
conn.execute(
"UPDATE job_intents SET request_json = ?2, fingerprint = ?3 WHERE job_id = ?1",
params![id, serde_json::to_string(&request)?, request.fingerprint()?],
)?;
conn.execute(
"UPDATE job_attempts SET input_revision = ?2 WHERE job_id = ?1",
params![id, request.input_digest()?],
)?;
}
conn.execute_batch(
"ALTER TABLE job_attempts RENAME COLUMN input_revision TO input_digest;
CREATE TRIGGER job_intent_immutable BEFORE UPDATE OF
job_id, request_json, fingerprint, serialization_key, max_concurrent ON job_intents
BEGIN SELECT RAISE(ABORT, 'immutable Job intent'); END;
CREATE TRIGGER job_attempt_immutable BEFORE UPDATE OF
job_id, number, attempt_id, input_digest ON job_attempts
BEGIN SELECT RAISE(ABORT, 'immutable Job attempt identity'); END;
CREATE TRIGGER legacy_job_archive_immutable BEFORE UPDATE ON legacy_job_archive
BEGIN SELECT RAISE(ABORT, 'immutable legacy Job archive'); END;
CREATE TRIGGER legacy_job_archive_retained BEFORE DELETE ON legacy_job_archive
BEGIN SELECT RAISE(ABORT, 'retained legacy Job archive'); END;",
)?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn request(id: &str) -> JobRequest {
JobRequest {
job_id: id.into(),
purpose: "migration".into(),
input_ref: "fixture:input".into(),
input: json!({"immutable":[1,2,3]}),
instruction: "Return a result".into(),
profile: "builtin:job".into(),
serialization_key: None,
limits: Default::default(),
}
}
fn seed(path: &Path, version: i64) -> (String, String, String) {
let mut conn = Connection::open(path).unwrap();
feature_storage::configure_connection(&conn).unwrap();
let tx = conn.transaction().unwrap();
tx.execute_batch(&SCHEMA.replace("input_digest", "input_revision"))
.unwrap();
if version == 2 {
tx.execute_batch(GRANT_SCHEMA).unwrap();
}
let mut original = serde_json::to_value(request("done")).unwrap();
original["input_revision"] = json!("old-token");
let original = serde_json::to_string_pretty(&original).unwrap();
let receipt = "{\"accepted\":true,\"input_revision\":\"receipt-domain-data\"}".to_string();
let digest = job::result_digest(
&serde_json::from_str(&receipt).unwrap(),
job::ABSOLUTE_MAX_RESULT_BYTES,
)
.unwrap()
.0;
tx.execute("INSERT INTO job_intents (job_id,request_json,fingerprint,max_concurrent,state,current_attempt,acknowledged)
VALUES ('done',?1,'legacy-fingerprint',8,'completed',2,1)", [&original]).unwrap();
tx.execute(
"INSERT INTO job_attempts (job_id,number,attempt_id,input_revision,state,failure)
VALUES ('done',1,'done:attempt:1','old-token','failed','timeout')",
[],
)
.unwrap();
tx.execute("INSERT INTO job_attempts (job_id,number,attempt_id,input_revision,state,result_json,result_digest)
VALUES ('done',2,'done:attempt:2','old-token','completed',?1,?2)", params![receipt,digest]).unwrap();
let grant = "{\"subject_id\":\"local-subject\",\"candidate_ids\":[\"c1\"]}".to_string();
if version == 2 {
tx.execute("INSERT INTO job_domain_grants VALUES ('done',?1)", [&grant])
.unwrap();
}
for (id, state) in [("pending", "reserved"), ("running", "dispatched")] {
let mut value = serde_json::to_value(request(id)).unwrap();
value["input_revision"] = json!("old-token");
tx.execute("INSERT INTO job_intents (job_id,request_json,fingerprint,max_concurrent,state,current_attempt)
VALUES (?1,?2,'legacy-fingerprint',8,'pending',1)", params![id,serde_json::to_string(&value).unwrap()]).unwrap();
tx.execute(
"INSERT INTO job_attempts (job_id,number,attempt_id,input_revision,state)
VALUES (?1,1,?2,'old-token',?3)",
params![id, job::attempt_id(id, 1), state],
)
.unwrap();
}
tx.pragma_update(None, "user_version", version).unwrap();
tx.commit().unwrap();
(original, receipt, grant)
}
#[test]
fn legacy_versions_preserve_intent_receipts_attempts_grants_and_replay() {
for version in [1, 2] {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("jobs.sqlite3");
let (original, receipt, grant) = seed(&path, version);
let mut store = JobStore::open(&path).unwrap();
let grant_value =
(version == 2).then(|| serde_json::from_str::<Value>(&grant).unwrap());
let snapshot = store
.reserve_granted(request("done"), grant_value.clone())
.unwrap();
assert!(snapshot.acknowledged);
assert_eq!(snapshot.state, JobState::Completed);
assert_eq!(snapshot.attempt.number, 2);
assert_eq!(
snapshot.result,
Some(serde_json::from_str(&receipt).unwrap())
);
let replay = JobResultSubmission {
job_id: "done".into(),
attempt_id: snapshot.attempt.attempt_id.clone(),
input_digest: request("done").input_digest().unwrap(),
result: snapshot.result.clone().unwrap(),
};
assert_eq!(store.accept(&replay).unwrap(), snapshot);
let mut stale = replay.clone();
stale.attempt_id = "done:attempt:1".into();
assert!(matches!(
store.accept(&stale),
Err(JobStoreError::AttemptMismatch)
));
let mut changed = request("done");
changed.input = json!({"different":true});
assert!(matches!(
store.reserve_granted(changed, grant_value.clone()),
Err(JobStoreError::IntentConflict(_))
));
let archived: (String, String, String) = store.conn.query_row(
"SELECT request_json,fingerprint,attempts_json FROM legacy_job_archive WHERE job_id='done'", [],
|r| Ok((r.get(0)?,r.get(1)?,r.get(2)?))).unwrap();
assert_eq!(archived.0, original);
assert_eq!(archived.1, "legacy-fingerprint");
let attempts: Value = serde_json::from_str(&archived.2).unwrap();
assert_eq!(attempts[0]["failure"], "timeout");
assert_eq!(attempts[1]["input_revision"], "old-token");
assert_eq!(attempts[1]["result_json"], receipt);
if version == 2 {
let persisted: String = store
.conn
.query_row("SELECT grant_json FROM job_domain_grants", [], |r| r.get(0))
.unwrap();
assert_eq!(persisted, grant);
}
let digests: Vec<String> = store
.conn
.prepare(
"SELECT input_digest FROM job_attempts WHERE job_id='done' ORDER BY number",
)
.unwrap()
.query_map([], |r| r.get(0))
.unwrap()
.collect::<Result<_, _>>()
.unwrap();
assert_eq!(digests, vec![request("done").input_digest().unwrap(); 2]);
assert_eq!(store.get("running").unwrap().state, JobState::Unknown);
assert_eq!(
store.reserve(request("pending")).unwrap().attempt.state,
JobAttemptState::Reserved
);
assert!(
store
.conn
.execute_batch("DELETE FROM legacy_job_archive")
.is_err()
);
assert!(
store
.conn
.execute_batch("UPDATE legacy_job_archive SET fingerprint='changed'")
.is_err()
);
assert!(
store
.conn
.execute_batch("UPDATE job_intents SET fingerprint='changed'")
.is_err()
);
assert!(
store
.conn
.execute_batch("UPDATE job_attempts SET input_digest='changed'")
.is_err()
);
assert!(
!store
.conn
.prepare("PRAGMA foreign_key_check")
.unwrap()
.exists([])
.unwrap()
);
drop(store);
let mut store = JobStore::open(&path).unwrap();
assert_eq!(
store.reserve_granted(request("done"), grant_value).unwrap(),
snapshot
);
}
}
#[test]
fn failed_migration_rolls_back_schema_archive_and_immutable_triggers() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("jobs.sqlite3");
seed(&path, 2);
let conn = Connection::open(&path).unwrap();
conn.execute_batch(
"DROP TRIGGER job_intent_immutable;
UPDATE job_intents SET request_json='{}' WHERE job_id='running';
CREATE TRIGGER job_intent_immutable BEFORE UPDATE OF request_json ON job_intents
BEGIN SELECT RAISE(ABORT,'immutable'); END;",
)
.unwrap();
drop(conn);
assert!(JobStore::open(&path).is_err());
let conn = Connection::open(&path).unwrap();
assert_eq!(
conn.pragma_query_value(None, "user_version", |r| r.get::<_, i64>(0))
.unwrap(),
2
);
assert!(conn.prepare("SELECT * FROM legacy_job_archive").is_err());
let binding: String = conn
.query_row(
"SELECT input_revision FROM job_attempts WHERE job_id='done' AND number=2",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(binding, "old-token");
assert!(
conn.execute_batch("UPDATE job_intents SET request_json='{}' WHERE job_id='done'")
.is_err()
);
assert!(
conn.execute_batch("UPDATE job_attempts SET input_revision='changed'")
.is_err()
);
}
}
crates/standalone/src/jobs.rs
+12 -6
View File
@@ -219,7 +219,10 @@ impl StandaloneJobs {
store: self.store.clone(),
job_id: job_id.into(),
attempt_id: snapshot.attempt.attempt_id.clone(),
input_revision: snapshot.request.input_revision.clone(),
input_digest: snapshot
.request
.input_digest()
.map_err(JobStoreError::from)?,
deadline,
});
let grant = match (
@@ -295,7 +298,10 @@ impl StandaloneJobs {
store: store.clone(),
job_id: id.clone(),
attempt_id: attempt_id.clone(),
input_revision: snapshot.request.input_revision.clone(),
input_digest: snapshot
.request
.input_digest()
.map_err(JobStoreError::from)?,
deadline,
});
let (cancel, mut cancellation) = watch::channel(false);
@@ -451,14 +457,14 @@ struct BoundResultSink {
store: Arc<Mutex<Option<JobStore>>>,
job_id: String,
attempt_id: String,
input_revision: String,
input_digest: String,
deadline: tokio::time::Instant,
}
impl worker::job::JobResultSink for BoundResultSink {
fn submit(&self, submission: JobResultSubmission) -> Result<(), String> {
if submission.job_id != self.job_id
|| submission.attempt_id != self.attempt_id
|| submission.input_revision != self.input_revision
|| submission.input_digest != self.input_digest
{
return Err("Job result capability binding mismatch".into());
}
@@ -482,7 +488,7 @@ struct BoundAttemptFence {
store: Arc<Mutex<Option<JobStore>>>,
job_id: String,
attempt_id: String,
input_revision: String,
input_digest: String,
deadline: tokio::time::Instant,
}
impl JobAttemptFence for BoundAttemptFence {
@@ -499,7 +505,7 @@ impl JobAttemptFence for BoundAttemptFence {
if snapshot.state != JobState::Pending
|| snapshot.attempt.state != JobAttemptState::Dispatched
|| snapshot.attempt.attempt_id != self.attempt_id
|| snapshot.request.input_revision != self.input_revision
|| snapshot.request.input_digest().map_err(|e| e.to_string())? != self.input_digest
{
return Err("Job domain grant is not bound to the current live attempt".into());
}
crates/standalone/src/jobs/tests.rs
+9 -10
View File
@@ -11,7 +11,6 @@ fn request(id: &str) -> JobRequest {
JobRequest {
job_id: id.into(),
purpose: "test".into(),
input_revision: "r1".into(),
input_ref: "fixture:input".into(),
input: serde_json::json!({"data":42}),
instruction: "Return a structured result".into(),
@@ -25,7 +24,7 @@ fn sink(jobs: &StandaloneJobs, id: &str, attempt: &str) -> BoundResultSink {
store: jobs.store.clone(),
job_id: id.into(),
attempt_id: attempt.into(),
input_revision: "r1".into(),
input_digest: request(id).input_digest().unwrap(),
deadline: tokio::time::Instant::now() + Duration::from_secs(10),
}
}
@@ -33,7 +32,7 @@ fn result(id: &str, attempt: &str) -> JobResultSubmission {
JobResultSubmission {
job_id: id.into(),
attempt_id: attempt.into(),
input_revision: "r1".into(),
input_digest: request(id).input_digest().unwrap(),
result: serde_json::json!({"valid":true}),
}
}
@@ -99,7 +98,7 @@ type CapturedDomainAttempt = (JobSnapshot, Arc<dyn JobAttemptFence>);
#[derive(Default)]
struct RecordingDomainProvider(Mutex<Vec<CapturedDomainAttempt>>);
fn domain_grant(id: &str) -> Value {
json!({"consumer":"test-domain", "job_id":id, "input_revision":"r1"})
json!({"consumer":"test-domain", "job_id":id, "input_digest":request(id).input_digest().unwrap()})
}
fn domain_request(id: &str) -> JobRequest {
let mut intent = request(id);
@@ -244,7 +243,7 @@ async fn persisted_grant_requires_provider_and_missing_grant_cannot_be_derived_f
}
#[tokio::test]
async fn foreign_job_revision_and_consumer_grants_are_refused_before_model_dispatch() {
async fn foreign_job_digest_and_consumer_grants_are_refused_before_model_dispatch() {
let temp = tempfile::tempdir().unwrap();
let jobs =
StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().to_path_buf()).unwrap();
@@ -253,12 +252,12 @@ async fn foreign_job_revision_and_consumer_grants_are_refused_before_model_dispa
for (id, grant) in [
("foreign-job", domain_grant("another-job")),
(
"foreign-revision",
json!({"consumer":"test-domain","job_id":"foreign-revision","input_revision":"r0"}),
"foreign-digest",
json!({"consumer":"test-domain","job_id":"foreign-digest","input_digest":"r0"}),
),
(
"foreign-consumer",
json!({"consumer":"another-domain","job_id":"foreign-consumer","input_revision":"r1"}),
json!({"consumer":"another-domain","job_id":"foreign-consumer","input_digest":request("foreign-consumer").input_digest().unwrap()}),
),
] {
let reserved = jobs.request_granted(domain_request(id), grant).unwrap();
@@ -378,7 +377,7 @@ async fn bound_domain_does_not_change_root_generic_result_only_execution() {
}
#[tokio::test]
async fn host_result_capability_cannot_cross_job_attempt_revision_or_deadline() {
async fn host_result_capability_cannot_cross_job_attempt_digest_or_deadline() {
let temp = tempfile::tempdir().unwrap();
let jobs =
StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().to_path_buf()).unwrap();
@@ -398,7 +397,7 @@ async fn host_result_capability_cannot_cross_job_attempt_revision_or_deadline()
);
assert!(bound.submit(result("first", "first:attempt:2")).is_err());
let mut stale = result("first", &first.attempt.attempt_id);
stale.input_revision = "r0".into();
stale.input_digest = "r0".into();
assert!(bound.submit(stale).is_err());
let mut expired = sink(&jobs, "first", &first.attempt.attempt_id);
expired.deadline = tokio::time::Instant::now() - Duration::from_secs(1);
crates/standalone/src/store.rs
+146 -26
View File
@@ -17,7 +17,7 @@ const LEASE_FILE: &str = "lease.json";
const LEASE_LOCK_FILE: &str = "lease.lock";
const SESSIONS_DIR: &str = "sessions";
const WORKER_DIR: &str = "worker";
const SCHEMA_VERSION: u32 = 1;
const SCHEMA_VERSION: u32 = 2;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StandaloneCwdIdentity {
@@ -79,7 +79,6 @@ pub enum StandaloneShutdownReason {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StandaloneWorkerRecord {
pub schema_version: u32,
pub revision: u64,
pub worker_id: WorkerId,
/// User-facing Worker name resolved from the profile.
pub worker_name: String,
@@ -194,7 +193,6 @@ impl StandaloneWorkerStore {
let now = now_unix_ms()?;
let record = StandaloneWorkerRecord {
schema_version: SCHEMA_VERSION,
revision: 1,
worker_id: allocation.worker_id,
worker_name: manifest.worker.name.clone(),
storage_key,
@@ -326,13 +324,12 @@ impl StandaloneWorkerStore {
active_segment_id: Option<SegmentId>,
) -> Result<StandaloneWorkerRecord, StandaloneStoreError> {
let mut next = record.clone();
next.revision = next.revision.saturating_add(1);
next.updated_at_unix_ms = now_unix_ms()?;
next.active_session_id = active_session_id;
next.active_segment_id = active_segment_id;
next.status = StandaloneWorkerStatus::Active;
next.shutdown_reason = None;
self.commit_record(Some(record.revision), &next)?;
self.commit_record(Some(record), &next)?;
Ok(next)
}
@@ -344,13 +341,12 @@ impl StandaloneWorkerStore {
reason: StandaloneShutdownReason,
) -> Result<StandaloneWorkerRecord, StandaloneStoreError> {
let mut next = record.clone();
next.revision = next.revision.saturating_add(1);
next.updated_at_unix_ms = now_unix_ms()?;
next.active_session_id = active_session_id;
next.active_segment_id = active_segment_id;
next.status = StandaloneWorkerStatus::Stopped;
next.shutdown_reason = Some(reason);
self.commit_record(Some(record.revision), &next)?;
self.commit_record(Some(record), &next)?;
Ok(next)
}
@@ -405,7 +401,7 @@ impl StandaloneWorkerStore {
fn commit_record(
&self,
expected_revision: Option<u64>,
expected_record: Option<&StandaloneWorkerRecord>,
next: &StandaloneWorkerRecord,
) -> Result<(), StandaloneStoreError> {
let dir = self.worker_dir(next.worker_id);
@@ -421,19 +417,21 @@ impl StandaloneWorkerStore {
StandaloneStoreError::Io(error)
}
})?;
writeln!(marker_file, "{}", next.revision).map_err(StandaloneStoreError::Io)?;
writeln!(marker_file, "{}", next.worker_id).map_err(StandaloneStoreError::Io)?;
marker_file.sync_all().map_err(StandaloneStoreError::Io)?;
sync_directory(&dir)?;
if let Some(expected) = expected_revision {
if let Some(expected) = expected_record {
let current = self.load_record_while_committing(next.worker_id)?;
if current.revision != expected {
let _ = fs::remove_file(&marker);
return Err(StandaloneStoreError::RevisionConflict {
id: next.worker_id,
expected,
found: current.revision,
});
// The create-new commit marker excludes other writers across this
// comparison and replacement. Compare the observed metadata itself,
// including its manifest, pointer and lifecycle state, not a counter.
if serde_json::to_value(&current).map_err(StandaloneStoreError::Json)?
!= serde_json::to_value(expected).map_err(StandaloneStoreError::Json)?
{
fs::remove_file(&marker).map_err(StandaloneStoreError::Io)?;
sync_directory(&dir)?;
return Err(StandaloneStoreError::MetadataChanged(next.worker_id));
}
}
@@ -681,6 +679,19 @@ fn decode_worker_record(
"standalone Worker record must be an object",
))
})?;
// Frozen schema-1 adapter: the counter was not domain data. Preserve
// every metadata field and use content comparison for future writes.
if object
.get("schema_version")
.and_then(serde_json::Value::as_u64)
== Some(1)
{
object.remove("revision");
object.insert(
"schema_version".to_string(),
serde_json::json!(SCHEMA_VERSION),
);
}
let persisted_manifest = object.remove("manifest").ok_or_else(|| {
serde_json::Error::io(io::Error::new(
io::ErrorKind::InvalidData,
@@ -770,14 +781,8 @@ pub enum StandaloneStoreError {
LeaseOwnershipLost,
#[error("standalone Worker {0} must be stopped before deletion")]
DeleteActive(WorkerId),
#[error(
"standalone Worker {id} metadata revision changed (expected {expected}, found {found})"
)]
RevisionConflict {
id: WorkerId,
expected: u64,
found: u64,
},
#[error("standalone Worker {0} metadata changed since it was read")]
MetadataChanged(WorkerId),
#[error("system clock is before the Unix epoch or out of range")]
Clock,
#[error("standalone metadata serialization failed")]
@@ -816,7 +821,6 @@ permission = "write"
let manifest = test_manifest();
let record = StandaloneWorkerRecord {
schema_version: SCHEMA_VERSION,
revision: 6,
worker_id,
worker_name: manifest.worker.name.clone(),
storage_key: "standalone-test".to_string(),
@@ -859,6 +863,122 @@ permission = "write"
);
}
#[test]
fn metadata_write_compares_observed_content_even_with_equal_timestamps() {
let root = tempfile::tempdir().unwrap();
let store = StandaloneWorkerStore::open(root.path().join("workers")).unwrap();
let allocation = store
.allocate(root.path(), StaleLeasePolicy::Reject)
.unwrap();
let record = store
.commit_created(
&allocation,
test_manifest(),
"metadata-test".into(),
session_store::new_session_id(),
None,
)
.unwrap();
// Simulate another committed pointer change in the same clock tick.
let mut changed = record.clone();
changed.active_session_id = session_store::new_session_id();
store.commit_record(Some(&record), &changed).unwrap();
assert!(matches!(
store.mark_stopped(
&record,
record.active_session_id,
None,
StandaloneShutdownReason::UserExit
),
Err(StandaloneStoreError::MetadataChanged(_))
));
assert_eq!(
store.load(record.worker_id).unwrap().active_session_id,
changed.active_session_id
);
assert!(
!store
.worker_dir(record.worker_id)
.join(COMMIT_MARKER)
.exists()
);
let stopped = store
.mark_stopped(
&changed,
changed.active_session_id,
None,
StandaloneShutdownReason::UserExit,
)
.unwrap();
assert_eq!(stopped.status, StandaloneWorkerStatus::Stopped);
}
#[test]
fn schema_one_metadata_preserves_content_without_an_update_counter() {
let root = tempfile::tempdir().unwrap();
let store = StandaloneWorkerStore::open(root.path().join("workers")).unwrap();
let allocation = store
.allocate(root.path(), StaleLeasePolicy::Reject)
.unwrap();
let record = store
.commit_created_connected(
&allocation,
test_manifest(),
"migration-test".into(),
session_store::new_session_id(),
None,
Some(crate::subjektiv::StandaloneSubjectBinding {
scope_id: "local-scope".into(),
subject_id: "local-subject".into(),
}),
)
.unwrap();
let record = store
.mark_stopped(
&record,
record.active_session_id,
Some(session_store::new_segment_id()),
StandaloneShutdownReason::UserExit,
)
.unwrap();
let mut persisted = Vec::new();
write_worker_record(&mut persisted, &record).unwrap();
let mut legacy: serde_json::Value = serde_json::from_slice(&persisted).unwrap();
legacy["schema_version"] = serde_json::json!(1);
legacy["revision"] = serde_json::json!(42);
let legacy_bytes = serde_json::to_vec(&legacy).unwrap();
fs::write(
store.worker_dir(record.worker_id).join(RECORD_FILE),
&legacy_bytes,
)
.unwrap();
let migrated = store.load(record.worker_id).unwrap();
assert_eq!(
serde_json::to_value(&migrated).unwrap(),
serde_json::to_value(&record).unwrap()
);
let mut encoded = Vec::new();
write_worker_record(&mut encoded, &migrated).unwrap();
let saved: serde_json::Value = serde_json::from_slice(&encoded).unwrap();
assert!(saved.get("revision").is_none());
assert_eq!(saved["schema_version"], 2);
let resumed = store
.update_active_pointer(
&migrated,
migrated.active_session_id,
migrated.active_segment_id,
)
.unwrap();
let disk: serde_json::Value = serde_json::from_slice(
&fs::read(store.worker_dir(record.worker_id).join(RECORD_FILE)).unwrap(),
)
.unwrap();
assert!(disk.get("revision").is_none());
assert_eq!(disk["schema_version"], 2);
assert_eq!(resumed.subject, record.subject);
assert_eq!(resumed.active_segment_id, record.active_segment_id);
}
#[test]
fn lease_liveness_requires_positive_live_or_stale_evidence() {
assert_eq!(
crates/standalone/src/subjektiv.rs
+4 -5
View File
@@ -352,7 +352,7 @@ fn operation_error(error: subjektiv::api::OperationError) -> SubjektivHostError
} else if message.contains("candidate_decision_conflict") {
"candidate_decision_conflict"
} else {
"revision_conflict"
"change_conflict"
}
.into(),
message,
@@ -655,7 +655,7 @@ impl LocalSubjectHost {
worker_id: Some(self.worker_id.to_string()),
flow_selector: None,
flow_definition_id: None,
flow_definition_revision: None,
flow_definition_fingerprint: None,
};
anchor.origin = Some(origin.clone());
source.origin = Some(origin);
@@ -835,8 +835,7 @@ impl SubjektivHost for LocalSubjectHost {
job::DEFAULT_MAX_RESULT_BYTES,
)
.map_err(operation_error)?;
let subject = self
.catalog
self.catalog
.store
.subject(&self.binding.subject_id)
.map_err(domain_error)?
@@ -847,7 +846,7 @@ impl SubjektivHost for LocalSubjectHost {
};
let request=job::JobRequest {
job_id:format!("subjektiv-consolidation:{}",uuid::Uuid::now_v7()),purpose:"subjektiv_consolidation".into(),
input_revision:subject.store_revision.to_string(),input_ref:format!("subjektiv://{}/consolidation",self.binding.subject_id),
input_ref:format!("subjektiv://{}/consolidation",self.binding.subject_id),
input:serde_json::json!({"subject_id":self.binding.subject_id,"candidate_ids":grant.candidate_ids}),
instruction:"Resolve ONLY candidate_ids with MemoryStagingList/Read, recall and MemoryApplyCandidate. Already resolved IDs are durable receipts; do not repeat them. Leave new candidates for a later Job. When the entire immutable batch has a disposition, call SubmitJobResult with result: {subject_id, candidate_ids}. The Host awaits clean-context surface generation and verifies actual receipts/outcome before accepting success; never supply surface claims.".into(),
profile:"builtin:standalone-subjektiv-consolidation".into(),serialization_key:Some(format!("subjektiv:{}",self.binding.subject_id)),
crates/standalone/tests/host.rs
+1 -1
View File
@@ -487,7 +487,7 @@ async fn standalone_restore_recovers_only_a_proven_stale_lease() -> TestResult {
source: manifest::ProfileRegistrySource::User,
name: "user-standalone".to_string(),
path: None,
provenance: Some("user-config-revision-7".to_string()),
provenance: Some("user-config-digest".to_string()),
},
profile: Some(manifest::ProfileMetadata {
name: Some("User standalone".to_string()),
crates/standalone/tests/jobs.rs
-1
View File
@@ -91,7 +91,6 @@ fn request(id: &str) -> JobRequest {
JobRequest {
job_id: id.into(),
purpose: "generic_test".into(),
input_revision: "r1".into(),
input_ref: "test:immutable".into(),
input: serde_json::json!({"test_input":42}),
instruction: "Return a structured validity result".into(),
crates/standalone/tests/subjektiv.rs
+7 -7
View File
@@ -273,7 +273,7 @@ impl ScriptedClient {
"SubmitMemorySurface",
json!({"points":[{
"body_md":format!("- {}", material["body_md"].as_str().unwrap()),
"memory_refs":[{"memory_id":material["memory_id"], "revision":material["revision"]}]
"memory_refs":[{"memory_id":material["memory_id"], "change_id":material["change_id"]}]
}]}),
)
}
@@ -1155,7 +1155,7 @@ async fn automatic_generic_consolidation_applies_candidate_publishes_surface_and
);
let memory = store.list_memories(&subject.id).unwrap().pop().unwrap();
assert_eq!(memory.claim, claim);
assert_eq!(memory.revision, 1);
assert!(!memory.change_id.is_empty());
assert_eq!(memory.source_candidate_ids, vec![ids[0].as_str().unwrap()]);
let surface = store.resident_surface(&subject.id).unwrap();
assert_eq!(surface.availability, subjektiv::SurfaceAvailability::Ready);
@@ -1167,14 +1167,14 @@ async fn automatic_generic_consolidation_applies_candidate_publishes_surface_and
assert!(snapshot.body_md.contains(claim));
assert_eq!(
snapshot.memory_refs,
vec![subjektiv::MemoryRevisionRef {
vec![subjektiv::MemoryChangeRef {
memory_id: memory.id.clone(),
revision: 1
change_id: memory.change_id.clone()
}]
);
assert_eq!(
snapshot.built_from_store_revision,
accepted["surface"]["store_revision"].as_u64().unwrap()
snapshot.built_from_memory_fingerprint,
accepted["surface"]["memory_fingerprint"].as_str().unwrap()
);
memory
});
@@ -1260,7 +1260,7 @@ async fn transport_loss_after_candidate_apply_preserves_memory_and_later_job_onl
let memory = store.list_memories(&subject.id).unwrap();
assert_eq!(memory.len(), 1);
assert_eq!(memory[0].claim, claim);
assert_eq!(memory[0].revision, 1);
assert!(!memory[0].change_id.is_empty());
assert_ne!(
store.resident_surface(&subject.id).unwrap().availability,
subjektiv::SurfaceAvailability::Ready