fix: merge develop and reconcile schema histories

Preserve both published schema-103 histories and converge their Passkey results and Job input retention features transactionally at schema 104. Retain local Reviewer, integration-event, auth and sidebar fixes while adopting develop restore and E2E updates; regenerate API contracts.

Validation: schema convergence regressions, full Server/Worker/Runtime suites, server-api and merge-request tests, Web unit/component tests, root cargo check, E2E harness tests and restore/auth target compilation, generated artifacts, fmt and naming checks. Product-process E2E not rerun; live services and databases untouched.
This commit is contained in:
2026-10-28 04:01:34 +09:00
218 changed files with 48985 additions and 11155 deletions
@@ -477,7 +477,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
(100, "Provisional users and scoped passkey activation".into()),
(101, "Durable explicit Subject body release".into()),
(102, "Durable failed Subject creation recovery".into()),
(103, "Passkey metadata and atomic addition results".into())
(103, "Passkey metadata and atomic addition results".into()),
(104, "Converge Passkey results and Job input retention".into())
]
);
for (table, rows) in retained.iter().zip(original.iter()) {
@@ -463,10 +463,52 @@ impl BackendJobRequest {
}
}
/// Availability of the immutable input body, independent of execution outcome.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum JobInputDisposition {
Available,
Reclaimed,
Erased,
Invalidated,
}
impl JobInputDisposition {
pub fn as_str(self) -> &'static str {
match self {
Self::Available => "available",
Self::Reclaimed => "reclaimed",
Self::Erased => "erased",
Self::Invalidated => "invalidated",
}
}
}
/// Host-authorized retirement; erasure and legacy invalidation do not change outcomes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobInputRetirementReason {
Completed,
ExplicitErasure,
LegacyInvalidation,
}
impl JobInputRetirementReason {
pub fn disposition(self) -> JobInputDisposition {
match self {
Self::Completed => JobInputDisposition::Reclaimed,
Self::ExplicitErasure => JobInputDisposition::Erased,
Self::LegacyInvalidation => JobInputDisposition::Invalidated,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct BackendJobRecord {
pub workspace_id: String,
/// Receipt metadata. After retirement `input` is null; use the verified loader to execute.
pub request: BackendJobRequest,
pub input_digest: String,
pub input_disposition: JobInputDisposition,
pub intent_fingerprint: String,
pub state: BackendJobState,
pub current_attempt: u8,
@@ -479,6 +521,28 @@ pub struct BackendJobRecord {
pub completed_at: Option<String>,
}
impl BackendJobRecord {
/// Verify the original immutable body and envelope before any execution or retry.
/// No Session read or replay request may repair unavailable/corrupt input.
pub fn load_executable_request(&self) -> Result<&BackendJobRequest> {
if self.input_disposition != JobInputDisposition::Available {
return Err(Error::InvalidInput(format!(
"Backend Job input is {}",
self.input_disposition.as_str()
)));
}
self.request.validate()?;
if self.request.input_digest()? != self.input_digest
|| self.request.fingerprint()? != self.intent_fingerprint
{
return Err(Error::Store(
"Backend Job immutable input or intent fingerprint mismatch".into(),
));
}
Ok(&self.request)
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum BackendJobWorkerCleanupState {
@@ -665,6 +729,8 @@ mod tests {
"workspace_id": "workspace-a",
"request": request(),
"intent_fingerprint": request().fingerprint().unwrap(),
"input_digest": request().input_digest().unwrap(),
"input_disposition": "available",
"state": "pending",
"current_attempt": 2,
"created_at": "created",
@@ -1124,3 +1190,17 @@ mod tests {
}
}
}
pub(crate) const INPUT_RETENTION_TRIGGERS: &str = r#"
CREATE TRIGGER backend_job_identity_immutable BEFORE UPDATE OF
workspace_id, job_id, purpose, input_digest, input_ref, resource_key, intent_fingerprint ON backend_jobs
BEGIN SELECT RAISE(ABORT, 'immutable Backend Job identity'); END;
CREATE TRIGGER backend_job_input_retirement BEFORE UPDATE OF request_json, input_disposition ON backend_jobs
WHEN (
((OLD.input_disposition = 'available' AND NEW.input_disposition IN ('reclaimed','erased','invalidated'))
OR (OLD.input_disposition IN ('reclaimed','invalidated') AND NEW.input_disposition = 'erased'))
AND json_remove(NEW.request_json,'$.input') = json_remove(OLD.request_json,'$.input')
AND json_type(NEW.request_json,'$.input') = 'null'
) IS NOT TRUE
BEGIN SELECT RAISE(ABORT, 'immutable Backend Job input retirement'); END;
"#;
@@ -71,6 +71,11 @@ fn legacy_content_database(path: &Path) -> (Connection, String) {
conn.execute("INSERT INTO workspace_config_tree_revisions VALUES('space',?1,?2,'toolchain','projection',?3,?4,'{}')",
params![ordinal,snapshot.digest,manifest,created]).unwrap();
}
seed_legacy_content_job(&conn);
(conn, snapshot.digest)
}
fn seed_legacy_content_job(conn: &Connection) {
let request = BackendJobRequest {
job_id: "legacy-job".into(),
purpose: "migration_test".into(),
@@ -93,7 +98,19 @@ fn legacy_content_database(path: &Path) -> (Connection, String) {
VALUES('space','legacy-job','legacy-job:attempt:1',1,'old-number','dispatched','runtime','worker','2026-08-14T00:00:00Z','2026-08-13T00:00:00Z','2026-08-13T00:00:00Z');
INSERT INTO backend_job_deliveries(workspace_id,delivery_id,job_id,attempt_id,target_runtime_id,target_worker_id,state,created_at,updated_at)
VALUES('space','delivery','legacy-job','legacy-job:attempt:1','runtime','target','pending','created','updated');") .unwrap();
(conn, snapshot.digest)
}
// Replay/resource checks use the current Store only after the full retained chain.
// Keep the focused partial DDL fixture above for the cutover's schema assertions.
fn legacy_content_job_database(path: &Path) -> Connection {
super::tests::prepare_retained_schema(path, 85);
let conn = Connection::open(path).unwrap();
configure_sqlite(&conn).unwrap();
conn.execute_batch("INSERT INTO accounts VALUES('owner','user','owner','Owner','created','updated');
INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at)
VALUES('space','owner','Space','active','created','updated');").unwrap();
seed_legacy_content_job(&conn);
conn
}
#[test]
@@ -432,7 +449,7 @@ fn content_cutover_reserved_only_releases_resource_but_uncertain_dispatch_keeps_
] {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("content.db");
let (conn, _) = legacy_content_database(&path);
let conn = legacy_content_job_database(&path);
conn.execute(
"UPDATE backend_job_attempts SET state=?1,
runtime_id=CASE WHEN ?1='reserved' THEN NULL ELSE runtime_id END,
@@ -441,11 +458,22 @@ fn content_cutover_reserved_only_releases_resource_but_uncertain_dispatch_keeps_
)
.unwrap();
migrate_content_state_v85_to_v86(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 86);
let cutover: (String, String, Option<String>) = conn.query_row(
"SELECT j.state,a.state,a.worker_cleanup_state FROM backend_jobs j JOIN backend_job_attempts a USING(workspace_id,job_id)",
[], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
).unwrap();
assert_eq!(
cutover,
if state == "reserved" {
("failed".into(), "failed".into(), Some("completed".into()))
} else {
("unknown".into(), "unknown".into(), Some("pending".into()))
}
);
apply_migrations(&conn).unwrap();
drop(conn);
let store = SqliteWorkspaceStore {
db_executor: crate::db_executor::DbExecutor::shared(),
conn: Arc::new(Mutex::new(Connection::open(path).unwrap())),
};
let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let job = store
.get_backend_job("space", "legacy-job")
.unwrap()
@@ -478,7 +506,7 @@ fn content_cutover_reserved_only_releases_resource_but_uncertain_dispatch_keeps_
fn content_cutover_completed_result_cleanup_and_replay_remain_unchanged() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("content.db");
let (conn, _) = legacy_content_database(&path);
let conn = legacy_content_job_database(&path);
let result = serde_json::json!({"finding":"preserved"});
let (digest, encoded) = job::result_digest(&result, 4096).unwrap();
conn.execute("UPDATE backend_jobs SET state='completed',result_json=?1,result_digest=?2,completed_at='finished'",params![encoded,digest]).unwrap();
@@ -490,11 +518,23 @@ fn content_cutover_completed_result_cleanup_and_replay_remain_unchanged() {
)
.unwrap();
migrate_content_state_v85_to_v86(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 86);
let cutover: (String, String, String, String) = conn.query_row(
"SELECT j.state,j.result_json,a.worker_cleanup_state,a.worker_cleanup_completed_at FROM backend_jobs j JOIN backend_job_attempts a USING(workspace_id,job_id)",
[], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
).unwrap();
assert_eq!(
cutover,
(
"completed".into(),
encoded,
"completed".into(),
"cleaned".into()
)
);
apply_migrations(&conn).unwrap();
drop(conn);
let store = SqliteWorkspaceStore {
db_executor: crate::db_executor::DbExecutor::shared(),
conn: Arc::new(Mutex::new(Connection::open(path).unwrap())),
};
let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let job = store
.get_backend_job("space", "legacy-job")
.unwrap()
+78 -3
View File
@@ -1326,6 +1326,13 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
Err("Session source permanent erasure is unavailable on this Runtime".into())
}
fn session_recording_capture(
&self,
_request: runtime_api::SessionRecordingCaptureRequest,
) -> Result<runtime_api::SessionRecordingCaptureAvailability, String> {
Err("committed recording capture is unavailable on this Runtime".into())
}
fn session_material_catalog(
&self,
_request: runtime_api::SessionMaterialCatalogRequest,
@@ -2336,6 +2343,21 @@ impl RuntimeRegistry {
})
}
pub fn session_recording_capture(
&self,
runtime_id: &str,
request: runtime_api::SessionRecordingCaptureRequest,
) -> Result<runtime_api::SessionRecordingCaptureAvailability, RuntimeRegistryError> {
validate_backend_identifier("runtime_id", runtime_id)?;
self.runtime(runtime_id)?
.session_recording_capture(request)
.map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: runtime_id.into(),
code: "session_recording_capture_failed".into(),
message,
})
}
pub fn session_material_catalog(
&self,
runtime_id: &str,
@@ -3441,6 +3463,19 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
})
}
fn session_recording_capture(
&self,
request: runtime_api::SessionRecordingCaptureRequest,
) -> Result<runtime_api::SessionRecordingCaptureAvailability, String> {
if request.workspace_id != self.workspace_id {
return Err("recording capture Workspace mismatch".into());
}
let scope = RuntimeWorkspaceScope::new(&request.workspace_id, "embedded-backend");
Ok(self
.runtime
.session_recording_capture_scoped(&scope, &request))
}
fn session_material_catalog(
&self,
request: runtime_api::SessionMaterialCatalogRequest,
@@ -3995,12 +4030,27 @@ impl WorkdirHttpAuthorization for RemoteWorkdirAuthorization {
return authorizer
.issue_async(method, path_and_query, operation, None, body)
.await
.map_err(|error| WorkdirError::Unavailable(error.message));
.map_err(|error| {
use workdir::WorkdirFailureReason as Reason;
let reason = match error.code.as_str() {
"workspace_runtime_authorization_stale" => Reason::AuthorizationStale,
"workspace_signing_identity_unavailable" => {
Reason::SigningIdentityUnavailable
}
"workspace_runtime_authorization_sign_failed" => {
Reason::AuthorizationSignFailed
}
_ => Reason::AuthorizationUnavailable,
};
WorkdirError::Unavailable("Workdir authorization is unavailable".into())
.with_failure_reason(reason)
});
}
authorization.fallback_bearer_token.ok_or_else(|| {
WorkdirError::Unavailable(
"remote Runtime does not have bearer authorization configured".to_string(),
)
.with_failure_reason(workdir::WorkdirFailureReason::AuthorizationMissing)
})
})
}
@@ -5063,6 +5113,9 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
runtime_api::WorkerRestoreState::RolledBack => {
server_api::WorkerRestoreState::RolledBack
}
runtime_api::WorkerRestoreState::Superseded => {
server_api::WorkerRestoreState::Superseded
}
runtime_api::WorkerRestoreState::ReconciliationRequired => {
server_api::WorkerRestoreState::ReconciliationRequired
}
@@ -5662,6 +5715,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
.map_err(|diagnostic| diagnostic.message)
}
fn session_recording_capture(
&self,
request: runtime_api::SessionRecordingCaptureRequest,
) -> Result<runtime_api::SessionRecordingCaptureAvailability, String> {
self.run_runtime_api(
self.request_timeout,
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move { client.session_recording_capture(request).await },
)
.map_err(|diagnostic| diagnostic.message)
}
fn session_material_catalog(
&self,
request: runtime_api::SessionMaterialCatalogRequest,
@@ -6839,9 +6904,16 @@ mod tests {
.await
.unwrap_err();
assert!(matches!(
error,
WorkdirError::Unavailable(message) if message == Error::DbCapacityExceeded.to_string()
error.classification_source(),
WorkdirError::Unavailable(_)
));
assert_eq!(
error.failure_reason(),
Some(workdir::WorkdirFailureReason::AuthorizationUnavailable)
);
let transport = workdir::http::WorkdirTransportError::from_workdir_error(&error);
assert_eq!(transport.message, "Workdir session is unavailable");
assert!(!format!("{error:?}").contains(&Error::DbCapacityExceeded.to_string()));
drop(release);
executor.drain().await.unwrap();
}
@@ -8805,6 +8877,9 @@ mod tests {
session_id: workdir::http::WorkdirSessionId::new("session-1").unwrap(),
workdir_id: Workdir::new("wd-1").id().clone(),
capabilities: workdir::WorkdirSessionCapabilities::ALL,
environment: workdir::WorkdirSessionEnvironment::root(
workdir::WorkdirSessionCapabilities::ALL,
),
};
let stat = workdir::http::WorkdirSessionOperationResult::Stat(StatResult {
path: WorkdirPath::new("hello.txt").unwrap(),
@@ -315,3 +315,71 @@ async fn capability_claims_must_match_current_workspace_identity_before_signing(
.unwrap();
verify(&fixture.verifier(), &token).unwrap();
}
#[tokio::test]
async fn remote_workdir_authorization_failures_keep_safe_reason_without_dispatch_or_fallback() {
use workdir::WorkdirFailureReason as Reason;
use workdir::http::{OpenWorkdirSessionRequest, RemoteWorkdirSession, WorkdirTransportError};
for case in ["stale", "missing_material", "missing_authorization"] {
let fixture = Fixture::new().await;
if case == "stale" {
fixture
.store
.revoke_workspace_runtime_binding(WORKSPACE, RUNTIME, "revoked")
.unwrap();
} else if case == "missing_material" {
fixture
.materials
.delete(&fixture.identity.private_material_ref)
.unwrap();
}
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let authorization = RemoteWorkdirAuthorization {
workspace_authorization: (case != "missing_authorization")
.then(|| fixture.authorizer.clone()),
fallback_bearer_token: (case != "missing_authorization")
.then(|| "bearer-secret".into()),
};
let error = RemoteWorkdirSession::open_with_authorization(
reqwest::Client::builder().no_proxy().build().unwrap(),
format!("http://{}", listener.local_addr().unwrap())
.parse()
.unwrap(),
Arc::new(authorization),
workdir::Workdir::new("workdir-secret").id().clone(),
OpenWorkdirSessionRequest::default(),
)
.await
.unwrap_err();
let reason = match case {
"stale" => Reason::AuthorizationStale,
"missing_material" => Reason::AuthorizationUnavailable,
_ => Reason::AuthorizationMissing,
};
assert_eq!(error.failure_reason(), Some(reason));
assert_eq!(reason.stage().as_str(), "authorization_preparation");
let transport = WorkdirTransportError::from_workdir_error(&error);
assert_eq!(transport.code.http_status(), 503);
assert_eq!(transport.message, "Workdir session is unavailable");
assert!(
!serde_json::to_string(&transport)
.unwrap()
.contains("secret")
);
let debug = format!("{error:?}");
for secret in [
"secret",
"workspace-signing/",
"private material",
"Authorization:",
"Bearer ",
] {
assert!(!debug.contains(secret), "{case}: {debug}");
}
assert!(
matches!(listener.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock),
"authorization failure sent HTTP or fell back"
);
}
}
@@ -384,7 +384,10 @@ async fn implementation_result_v97_to_v98_preserves_legacy_bytes_receipts_and_no
DROP TRIGGER passkey_metadata_insert;
DROP TABLE passkey_addition_receipts;
DROP TABLE passkey_metadata;
UPDATE __yoi_schema_migrations SET version=102 WHERE version=103;
DROP TRIGGER backend_job_input_retirement;
DROP TRIGGER backend_job_identity_immutable;
ALTER TABLE backend_jobs DROP COLUMN input_disposition;
UPDATE __yoi_schema_migrations SET version=102 WHERE version=104;
DROP TABLE subject_create_recoveries;
UPDATE __yoi_schema_migrations SET version=101 WHERE version=102;
DROP TABLE subject_worker_releases;
@@ -524,7 +527,10 @@ async fn mr_notification_upgrade_recovers_events_not_comments_and_retains_unknow
DROP TRIGGER passkey_metadata_insert;
DROP TABLE passkey_addition_receipts;
DROP TABLE passkey_metadata;
UPDATE __yoi_schema_migrations SET version=102 WHERE version=103;
DROP TRIGGER backend_job_input_retirement;
DROP TRIGGER backend_job_identity_immutable;
ALTER TABLE backend_jobs DROP COLUMN input_disposition;
UPDATE __yoi_schema_migrations SET version=102 WHERE version=104;
DROP TABLE subject_create_recoveries;
UPDATE __yoi_schema_migrations SET version=101 WHERE version=102;
DROP TABLE subject_worker_releases;
@@ -690,7 +696,10 @@ async fn mr_event_actors_preserve_self_suppression_in_capture_and_backfill() {
DROP TRIGGER passkey_metadata_insert;
DROP TABLE passkey_addition_receipts;
DROP TABLE passkey_metadata;
UPDATE __yoi_schema_migrations SET version=102 WHERE version=103;
DROP TRIGGER backend_job_input_retirement;
DROP TRIGGER backend_job_identity_immutable;
ALTER TABLE backend_jobs DROP COLUMN input_disposition;
UPDATE __yoi_schema_migrations SET version=102 WHERE version=104;
DROP TABLE subject_create_recoveries;
UPDATE __yoi_schema_migrations SET version=101 WHERE version=102;
DROP TABLE subject_worker_releases;
@@ -2,8 +2,8 @@
//! lifecycle, are the durable source. Fixed Worker recipients never follow leases.
use super::*;
use crate::backend_job::{
BackendJobAttemptState, BackendJobDeliveryRecord, BackendJobDeliveryState,
BackendJobResultAcceptance, BackendJobState,
BackendJobAttemptState, BackendJobDeliveryRecord, BackendJobDeliveryState, BackendJobRequest,
BackendJobState,
};
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -103,6 +103,32 @@ pub(in crate::store) fn install(conn: &Connection) -> Result<()> {
Ok(())
}
struct NotificationResultSnapshot {
request: BackendJobRequest,
result: Option<serde_json::Value>,
result_digest: Option<String>,
}
// The schema-96 import runs before later Job lifecycle migrations. Read only
// the retained result envelope, not the executable Job's current-schema columns.
fn read_result_snapshot(
conn: &Connection,
workspace: &str,
job: &str,
) -> Result<NotificationResultSnapshot> {
let (request, result, result_digest): (String, Option<String>, Option<String>) = conn
.query_row(
"SELECT request_json, result_json, result_digest FROM backend_jobs WHERE workspace_id=?1 AND job_id=?2",
params![workspace, job],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)?;
Ok(NotificationResultSnapshot {
request: decode(&request)?,
result: result.map(|value| decode(&value)).transpose()?,
result_digest,
})
}
pub(in crate::store) fn capture(
conn: &Connection,
workspace: &str,
@@ -115,8 +141,7 @@ pub(in crate::store) fn capture(
attempt_id: attempt.into(),
delivery_id: id.clone(),
};
let job = read_backend_job(conn, workspace, job)?
.ok_or_else(|| Error::Store("Job notification source missing".into()))?;
let job = read_result_snapshot(conn, workspace, job)?;
let target = &job.request.notification_target;
let facts = serde_json::json!({"principal_kind":if target.is_some(){Some("worker")}else{None}, "principal_id":null,"runtime_id":target.as_ref().map(|w|&w.runtime_id),"worker_id":target.as_ref().map(|w|&w.worker_id),"status":null,"case_state":"open","released":false,"backend_job":source,"recipient_kind":"fixed_worker","job_purpose":job.request.purpose,"result_digest":job.result_digest});
conn.execute("INSERT OR IGNORE INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) VALUES(?1,NULL,?2,'backend_job','','backend_job_result',?3)",params![workspace,id,json(&facts)?])?;
@@ -290,22 +315,13 @@ fn import_legacy(conn: &Connection, workspace: &str, job_id: &str, attempt_id: &
| "notification_too_large",
) => ("defer", NotificationReason::SourceUnavailable),
_ => {
let job = read_backend_job(conn, workspace, job_id)?
.ok_or_else(|| Error::Store("Legacy Job result missing".into()))?;
let attempt = read_backend_job_attempt(conn, workspace, job_id, attempt_id)?
.ok_or_else(|| Error::Store("Legacy Job attempt missing".into()))?;
let acceptance = BackendJobResultAcceptance {
job,
attempt,
replayed: true,
};
let facts = if acceptance.job.request.notification_target.as_ref() != Some(&old.target)
{
let snapshot = read_result_snapshot(conn, workspace, job_id)?;
let facts = if snapshot.request.notification_target.as_ref() != Some(&old.target) {
JobNotificationFacts::Failed(NotificationFailureCode::PreparationFailed)
} else {
legacy_result_facts(&acceptance)
legacy_result_facts(&snapshot, job_id, attempt_id)
};
let decision = policy(acceptance.job.request.notification_target.is_some(), &facts);
let decision = policy(snapshot.request.notification_target.is_some(), &facts);
match facts {
JobNotificationFacts::Result(body) => content = Some(body),
JobNotificationFacts::Failed(code) => failure = Some(code),
@@ -338,28 +354,26 @@ fn import_legacy(conn: &Connection, workspace: &str, job_id: &str, attempt_id: &
Ok(())
}
fn legacy_result_facts(acceptance: &BackendJobResultAcceptance) -> JobNotificationFacts {
let content = if acceptance.job.request.purpose == crate::ticket_item_checker::PURPOSE {
let Some(result) = &acceptance.job.result else {
fn legacy_result_facts(
snapshot: &NotificationResultSnapshot,
job_id: &str,
attempt_id: &str,
) -> JobNotificationFacts {
let content = if snapshot.request.purpose == crate::ticket_item_checker::PURPOSE {
let Some(result) = &snapshot.result else {
return JobNotificationFacts::Failed(NotificationFailureCode::InvalidResult);
};
match crate::ticket_item_checker::validate_result(&acceptance.job.request.input, result) {
match crate::ticket_item_checker::validate_result(&snapshot.request.input, result) {
Ok(result) if result.findings.is_empty() => return JobNotificationFacts::NoFindings,
Ok(_) => {}
Err(_) => return JobNotificationFacts::Failed(NotificationFailureCode::InvalidResult),
}
crate::ticket_item_checker::render_result(acceptance)
crate::ticket_item_checker::render_result(&snapshot.request.input, result)
} else {
Ok(format!(
"Backend Job `{}` completed attempt `{}` for `{}`. Durable result digest: {}.",
acceptance.job.request.job_id,
acceptance.attempt.attempt_id,
acceptance.job.request.purpose,
acceptance
.job
.result_digest
.as_deref()
.unwrap_or("unavailable")
"Backend Job `{job_id}` completed attempt `{attempt_id}` for `{}`. Durable result digest: {}.",
snapshot.request.purpose,
snapshot.result_digest.as_deref().unwrap_or("unavailable")
))
};
match content {
@@ -789,6 +789,7 @@ CREATE TABLE "worker_registry" (
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
);
CREATE TABLE backend_jobs (
input_disposition TEXT NOT NULL DEFAULT 'available' CHECK (input_disposition IN ('available','reclaimed','erased','invalidated')),
workspace_id TEXT NOT NULL,
job_id TEXT NOT NULL,
purpose TEXT NOT NULL,
@@ -1642,6 +1643,19 @@ CREATE TABLE IF NOT EXISTS ticket_assignment_work_releases (
FOREIGN KEY (workspace_id, assignment_id)
REFERENCES ticket_worker_assignments(workspace_id, assignment_id) ON DELETE CASCADE
);
CREATE TRIGGER backend_job_identity_immutable BEFORE UPDATE OF
workspace_id, job_id, purpose, input_digest, input_ref, resource_key, intent_fingerprint ON backend_jobs
BEGIN SELECT RAISE(ABORT, 'immutable Backend Job identity'); END;
CREATE TRIGGER backend_job_input_retirement BEFORE UPDATE OF request_json, input_disposition ON backend_jobs
WHEN (
((OLD.input_disposition = 'available' AND NEW.input_disposition IN ('reclaimed','erased','invalidated'))
OR (OLD.input_disposition IN ('reclaimed','invalidated') AND NEW.input_disposition = 'erased'))
AND json_remove(NEW.request_json,'$.input') = json_remove(OLD.request_json,'$.input')
AND json_type(NEW.request_json,'$.input') = 'null'
) IS NOT TRUE
BEGIN SELECT RAISE(ABORT, 'immutable Backend Job input retirement'); END;
-- Ticket case assignment work constraints (schema 91).
DROP VIEW IF EXISTS ticket_active_worker_assignments;
CREATE VIEW ticket_active_worker_assignments AS
+3
View File
@@ -19,6 +19,9 @@ pub mod repositories;
pub mod repository_access;
pub mod repository_source;
pub mod resource_broker;
#[cfg(test)]
#[path = "../../../tests/support/restore_environment.rs"]
mod restore_test_environment;
pub mod retention;
pub mod runtime_settings;
pub mod runtime_subscription;
@@ -0,0 +1,240 @@
// Included in store::tests. Exercise the production migration entrypoints on
// isolated SQLite files; both published branches are built from retained DDL.
mod schema_convergence {
use super::*;
use std::path::PathBuf;
fn variant_fixture(job_variant: bool, baseline_only: bool) -> (tempfile::TempDir, PathBuf) {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("schema-103.sqlite");
prepare_retained_schema(&path, 102);
let conn = Connection::open(&path).unwrap();
configure_sqlite(&conn).unwrap();
if job_variant {
(JOB_INPUT_RETENTION_SCHEMA_103.apply)(&conn).unwrap();
} else {
migrate_passkey_results_v102_to_v103(&conn).unwrap();
}
if baseline_only {
conn.execute_batch("DELETE FROM __yoi_schema_migrations;
INSERT INTO __yoi_schema_migrations(version,name) VALUES(103,'workspace schema baseline');").unwrap();
}
conn.execute_batch(r#"
INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at)
VALUES('owner','user','owner','Owner','1','2');
INSERT INTO users(user_id,account_id,handle,display_name,created_at,updated_at)
VALUES('user','owner','owner','Owner','1','2');
INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at)
VALUES('workspace','owner','Workspace','active','1','2');
INSERT INTO passkey_credentials(credential_id,user_id,public_key_cose,transports_json,sign_count,created_at,last_used_at)
VALUES('key','user','preserved-public-key','["internal"]',9,'1','2');
"#).unwrap();
if !job_variant {
conn.execute_batch("UPDATE passkey_metadata SET key_reference='stable-reference',normal_login_verified_at='verified';
INSERT INTO passkey_addition_receipts(challenge_id,credential_id,user_id,authentication_reference)
VALUES('challenge','key','user','saved-authentication-reference');").unwrap();
}
for disposition in ["available", "reclaimed", "erased", "invalidated"] {
let request = resource_job_request(disposition, disposition, "candidate");
conn.execute("INSERT INTO backend_jobs(workspace_id,job_id,purpose,input_digest,input_ref,request_json,intent_fingerprint,state,current_attempt,result_json,result_digest,created_at,updated_at,completed_at)
VALUES('workspace',?1,?2,?3,?4,?5,?6,'completed',1,'{\"saved\":true}','result-digest','1','2','2')",
params![request.job_id, request.purpose, request.input_digest().unwrap(), request.input_ref, serde_json::to_string(&request).unwrap(), request.fingerprint().unwrap()]).unwrap();
conn.execute("INSERT INTO backend_job_attempts(workspace_id,job_id,attempt_id,attempt,input_digest,state,deadline_at,result_json,result_digest,created_at,updated_at,completed_at)
VALUES('workspace',?1,?1,1,?2,'completed','deadline','{\"saved\":true}','result-digest','1','2','2')",
params![request.job_id, request.input_digest().unwrap()]).unwrap();
if job_variant && disposition != "available" {
conn.execute("UPDATE backend_jobs SET request_json=json_set(request_json,'$.input',json('null')),input_disposition=?1 WHERE job_id=?1", [disposition]).unwrap();
}
}
verify_schema_history(&conn, 103).unwrap();
(dir, path)
}
fn feature_rows(conn: &Connection, job_variant: bool) -> Vec<Vec<Vec<rusqlite::types::Value>>> {
let mut tables = vec!["passkey_credentials", "backend_job_attempts"];
if !job_variant {
tables.extend(["passkey_metadata", "passkey_addition_receipts"]);
}
let mut rows: Vec<_> = tables
.into_iter()
.map(|table| stored_fixture_rows(conn, table))
.collect();
// A Passkey-branch Job gains only input_disposition; all old columns must
// remain byte-for-byte equal, including body, digest and result receipts.
rows.push(fixture_query_rows(conn, "SELECT workspace_id,job_id,purpose,input_digest,input_ref,resource_key,request_json,intent_fingerprint,state,current_attempt,result_json,result_digest,failure_category,failure_detail,created_at,updated_at,completed_at FROM backend_jobs ORDER BY job_id"));
if job_variant {
rows.push(fixture_query_rows(
conn,
"SELECT job_id,input_disposition FROM backend_jobs ORDER BY job_id",
));
}
rows
}
fn schema_rows(conn: &Connection) -> Vec<Vec<rusqlite::types::Value>> {
fixture_query_rows(
conn,
"SELECT type,name,tbl_name,sql FROM sqlite_schema ORDER BY type,name",
)
}
#[test]
fn schema_103_variants_converge_preserving_history_values_and_constraints() {
for job_variant in [false, true] {
for baseline_only in [false, true] {
let (_dir, path) = variant_fixture(job_variant, baseline_only);
let conn = Connection::open(&path).unwrap();
let history = workspace_schema_migration_history(&conn).unwrap();
let rows = feature_rows(&conn, job_variant);
let schema = schema_rows(&conn);
let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
assert_eq!(plan.current_schema_version, 103);
assert_eq!(
plan.migrations,
vec![WorkspaceSchemaMigrationStep {
version: 104,
name: "Converge Passkey results and Job input retention".into(),
}]
);
assert_eq!(
schema_rows(&conn),
schema,
"preflight must not mutate the source"
);
assert_eq!(feature_rows(&conn, job_variant), rows);
drop(conn);
let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap();
upgraded.with_conn(|conn| {
assert_eq!(feature_rows(conn, job_variant), rows);
let mut expected = history.clone();
expected.push((104, "Converge Passkey results and Job input retention".into()));
assert_eq!(workspace_schema_migration_history(conn)?, expected);
for table in ["passkey_credentials", "passkey_metadata", "passkey_addition_receipts", "backend_jobs", "backend_job_attempts"] {
assert_eq!(state_contract_schema_signature(conn, table), fresh.with_conn(|c| Ok(state_contract_schema_signature(c, table)))?, "{table}");
}
assert_eq!(conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |r| r.get::<_, i64>(0))?, 0);
assert!(conn.execute("UPDATE backend_jobs SET input_digest='changed'", []).is_err());
assert!(conn.execute("UPDATE backend_jobs SET input_disposition='unsupported'", []).is_err());
assert!(conn.execute("UPDATE backend_jobs SET request_json=json_set(request_json,'$.input','replacement')", []).is_err());
assert!(conn.execute("INSERT INTO passkey_metadata(credential_id,key_reference) VALUES('missing','new')", []).is_err());
assert!(conn.execute("INSERT INTO passkey_addition_receipts VALUES('other','missing','user','reference')", []).is_err());
let reference: (String, Option<String>) = conn.query_row("SELECT key_reference,normal_login_verified_at FROM passkey_metadata WHERE credential_id='key'", [], |r| Ok((r.get(0)?,r.get(1)?)))?;
if job_variant {
assert_eq!(reference.0.len(), 32);
assert_eq!(reference.1, None);
} else {
assert_eq!(reference, ("stable-reference".into(), Some("verified".into())));
assert_eq!(conn.query_row("SELECT count(*) FROM backend_jobs WHERE input_disposition='available'", [], |r| r.get::<_, i64>(0))?, 4);
}
Ok(())
}).unwrap();
let summaries = upgraded.list_passkey_summaries("user").unwrap();
drop(upgraded);
let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap();
assert_eq!(reopened.list_passkey_summaries("user").unwrap(), summaries);
reopened
.with_conn(|conn| {
assert_eq!(feature_rows(conn, job_variant), rows);
assert_eq!(
workspace_schema_migration_history(conn)?.len(),
history.len() + 1
);
Ok(())
})
.unwrap();
}
}
}
#[test]
fn schema_103_convergence_failure_rolls_back_both_variants_and_retries() {
for job_variant in [false, true] {
for baseline_only in [false, true] {
let (_dir, path) = variant_fixture(job_variant, baseline_only);
let conn = Connection::open(&path).unwrap();
conn.execute_batch("CREATE TRIGGER interrupt_convergence BEFORE INSERT ON __yoi_schema_migrations WHEN NEW.version=104 BEGIN SELECT RAISE(ABORT,'injected convergence failure'); END;").unwrap();
let schema = schema_rows(&conn);
let history = workspace_schema_migration_history(&conn).unwrap();
let rows = feature_rows(&conn, job_variant);
drop(conn);
let error = SqliteWorkspaceStore::open_for_test(&path)
.err()
.expect("migration must fail");
assert!(
error.to_string().contains("injected convergence failure"),
"{error}"
);
let conn = Connection::open(&path).unwrap();
assert_eq!(schema_rows(&conn), schema);
assert_eq!(workspace_schema_migration_history(&conn).unwrap(), history);
assert_eq!(feature_rows(&conn, job_variant), rows);
conn.execute_batch("DROP TRIGGER interrupt_convergence")
.unwrap();
drop(conn);
let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap();
upgraded
.with_conn(|conn| {
assert_eq!(feature_rows(conn, job_variant), rows);
assert_eq!(current_schema_version(conn)?, 104);
Ok(())
})
.unwrap();
}
}
}
#[test]
fn schema_103_unknown_migration_identity_is_rejected_without_changes() {
for job_variant in [false, true] {
let (_dir, path) = variant_fixture(job_variant, false);
let conn = Connection::open(&path).unwrap();
conn.execute("UPDATE __yoi_schema_migrations SET name='unrecognized schema 103' WHERE version=103", []).unwrap();
let schema = schema_rows(&conn);
let history = workspace_schema_migration_history(&conn).unwrap();
let rows = feature_rows(&conn, job_variant);
assert!(SqliteWorkspaceStore::migration_plan(&path).is_err());
assert!(SqliteWorkspaceStore::open_for_test(&path).is_err());
assert_eq!(schema_rows(&conn), schema);
assert_eq!(workspace_schema_migration_history(&conn).unwrap(), history);
assert_eq!(feature_rows(&conn, job_variant), rows);
}
}
#[test]
fn schema_103_mismatched_identity_or_incomplete_features_are_rejected_atomically() {
for job_variant in [false, true] {
for mismatched_identity in [false, true] {
let (_dir, path) = variant_fixture(job_variant, false);
let conn = Connection::open(&path).unwrap();
if mismatched_identity {
let other = if job_variant {
"Passkey metadata and atomic addition results"
} else {
"Job input retirement independent of immutable receipts"
};
conn.execute(
"UPDATE __yoi_schema_migrations SET name=?1 WHERE version=103",
[other],
)
.unwrap();
} else {
let trigger = if job_variant {
"backend_job_identity_immutable"
} else {
"passkey_metadata_insert"
};
conn.execute_batch(&format!("DROP TRIGGER {trigger}"))
.unwrap();
}
let schema = schema_rows(&conn);
let history = workspace_schema_migration_history(&conn).unwrap();
let rows = feature_rows(&conn, job_variant);
assert!(SqliteWorkspaceStore::open_for_test(&path).is_err());
assert_eq!(schema_rows(&conn), schema);
assert_eq!(workspace_schema_migration_history(&conn).unwrap(), history);
assert_eq!(feature_rows(&conn, job_variant), rows);
}
}
}
}
@@ -1,40 +1,41 @@
// Included in store::tests so both paths use the real retained migration chain.
fn state_contract_schema_signature(conn: &Connection, table: &str) -> Vec<Vec<String>> {
let queries = [
"SELECT name, type, CAST(\"notnull\" AS TEXT), COALESCE(dflt_value, '<NULL>'), CAST(pk AS TEXT) FROM pragma_table_info(?1) ORDER BY name",
"SELECT \"table\", \"from\", \"to\", on_update, on_delete, match, CAST(seq AS TEXT) FROM pragma_foreign_key_list(?1) ORDER BY \"table\", \"from\", \"to\", seq",
"SELECT CAST(i.\"unique\" AS TEXT), CAST(i.partial AS TEXT), (SELECT group_concat(name, ',') FROM (SELECT name FROM pragma_index_info(i.name) ORDER BY seqno)) FROM pragma_index_list(?1) AS i ORDER BY 1, 2, 3",
"SELECT name, sql FROM sqlite_schema WHERE type='trigger' AND tbl_name=?1 ORDER BY name",
];
queries
.into_iter()
.flat_map(|sql| {
let mut statement = conn.prepare(sql).unwrap();
let count = statement.column_count();
statement
.query_map([table], |row| {
(0..count)
.map(|i| row.get::<_, String>(i))
.collect::<rusqlite::Result<Vec<_>>>()
})
.unwrap()
.map(|row| {
row.unwrap()
.into_iter()
.map(|value| {
value
.chars()
.filter(|c| !c.is_whitespace() && *c != '"')
.collect()
})
.collect()
})
.collect::<Vec<_>>()
})
.collect()
}
#[test]
fn fresh_and_upgraded_state_contract_schemas_enforce_the_same_constraints() {
fn signature(conn: &Connection, table: &str) -> Vec<Vec<String>> {
let queries = [
"SELECT name, type, CAST(\"notnull\" AS TEXT), COALESCE(dflt_value, '<NULL>'), CAST(pk AS TEXT) FROM pragma_table_info(?1) ORDER BY name",
"SELECT \"table\", \"from\", \"to\", on_update, on_delete, match, CAST(seq AS TEXT) FROM pragma_foreign_key_list(?1) ORDER BY \"table\", \"from\", \"to\", seq",
"SELECT CAST(i.\"unique\" AS TEXT), CAST(i.partial AS TEXT), (SELECT group_concat(name, ',') FROM (SELECT name FROM pragma_index_info(i.name) ORDER BY seqno)) FROM pragma_index_list(?1) AS i ORDER BY 1, 2, 3",
"SELECT name, sql FROM sqlite_schema WHERE type='trigger' AND tbl_name=?1 ORDER BY name",
];
queries
.into_iter()
.flat_map(|sql| {
let mut statement = conn.prepare(sql).unwrap();
let count = statement.column_count();
statement
.query_map([table], |row| {
(0..count)
.map(|i| row.get::<_, String>(i))
.collect::<rusqlite::Result<Vec<_>>>()
})
.unwrap()
.map(|row| {
row.unwrap()
.into_iter()
.map(|value| {
value
.chars()
.filter(|c| !c.is_whitespace() && *c != '"')
.collect()
})
.collect()
})
.collect::<Vec<_>>()
})
.collect()
}
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("upgraded.db");
prepare_retained_schema(&path, 85);
@@ -58,6 +59,9 @@ fn fresh_and_upgraded_state_contract_schemas_enforce_the_same_constraints() {
"workspace_memory_settings",
"worker_create_reservations",
"subject_worker_releases",
"passkey_credentials",
"passkey_metadata",
"passkey_addition_receipts",
"backend_jobs",
"backend_job_attempts",
"flow_sources",
@@ -92,9 +96,11 @@ fn fresh_and_upgraded_state_contract_schemas_enforce_the_same_constraints() {
"workdir_create_legacy_ssh_archives",
"repository_secret_legacy_receipts",
] {
let expected = fresh.with_conn(|conn| Ok(signature(conn, table))).unwrap();
let expected = fresh
.with_conn(|conn| Ok(state_contract_schema_signature(conn, table)))
.unwrap();
let actual = upgraded
.with_conn(|conn| Ok(signature(conn, table)))
.with_conn(|conn| Ok(state_contract_schema_signature(conn, table)))
.unwrap();
assert!(!expected.is_empty(), "missing canonical table {table}");
if actual != expected {
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,36 @@
//! Per-request audit notifications, independent of provider result classification.
//!
//! The scope follows the operation future, not a session-wide last-result slot: concurrent
//! requests cannot consume one another's diagnostics. Every failure is also emitted immediately
//! for direct session callers and requests cancelled before a response. This is notification,
//! not a replacement durable audit ledger or crash-recovery/effect-deduplication guarantee.
use std::{cell::RefCell, future::Future};
use workdir::http::WorkdirAuditFailure;
tokio::task_local! {
static FAILURES: RefCell<Vec<WorkdirAuditFailure>>;
}
pub(super) async fn capture<T>(future: impl Future<Output = T>) -> (T, Vec<WorkdirAuditFailure>) {
FAILURES
.scope(RefCell::new(Vec::new()), async {
let result = future.await;
let failures = FAILURES.with(|failures| failures.take());
(result, failures)
})
.await
}
pub(super) fn notify(failure: WorkdirAuditFailure, operation: &'static str) {
tracing::error!(
event = "external_workdir_audit_failed",
operation,
operation_id = %failure.operation_id,
audit_stage = ?failure.stage,
audit_reason = ?failure.reason,
operation_outcome = ?failure.operation_outcome,
"External Workdir audit persistence failed; operation result is independent; do not resend"
);
let _ = FAILURES.try_with(|failures| failures.borrow_mut().push(failure));
}
@@ -4,7 +4,7 @@ use super::*;
use crate::backend_job::{
BackendJobGrants, BackendJobRecord, SubjektivJournalGrant, journal_serialization_key,
};
use subjektiv::journal::recording::{JournalMaterial, select_materials_bounded};
use subjektiv::journal::recording::JournalInput;
use subjektiv::journal::transport::*;
use subjektiv::journal::{self, *};
@@ -92,20 +92,82 @@ pub(super) fn prepare(api: &WorkspaceApi, reservation: &BackendJobReservation) -
};
let store = store(api)?;
let host = host(api, &store, grant)?;
store
.prepare_journal_generation(
let input = saved_input(&reservation.job.request, grant)?;
let g = store
.prepare_journal_input(
&host,
&grant.journal_id,
&input,
subjektiv::api::JobAttemptBinding {
job_id: reservation.job.request.job_id.clone(),
attempt_id: reservation.attempt.attempt_id.clone(),
},
grant.selected_ranges.clone(),
)
.map_err(domain)?;
input.validate_generation(&g).map_err(domain)?;
Ok(())
}
fn legacy_descriptor(request: &BackendJobRequest) -> bool {
let Some(grant) = request.grants.subjektiv_journal.as_ref() else {
return false;
};
let Some(object) = request.input.as_object() else {
return false;
};
object.len() == 2
&& object.get("journal_id").and_then(|v| v.as_str()) == Some(&grant.journal_id)
&& object.get("selected_ranges")
== serde_json::to_value(&grant.selected_ranges).ok().as_ref()
}
/// Retire only the explicitly recognized descriptor wire. Malformed body snapshots are
/// failures, not migration requests and never a reason to reselect from current Sessions.
pub(super) fn migrate_reserved_legacy(
api: &WorkspaceApi,
reservation: &BackendJobReservation,
) -> Result<bool> {
if !legacy_descriptor(&reservation.job.request) {
return Ok(false);
}
reservation.job.load_executable_request()?;
close(api, &reservation.job.request, &reservation.attempt)?;
api.store.finish_backend_job_attempt(
api.workspace_id(),
&reservation.job.request.job_id,
&reservation.attempt.attempt_id,
BackendJobAttemptState::Failed,
"journal_descriptor_invalidated",
"Descriptor-only Journal input retired; saved checkpoints retained",
&Utc::now().to_rfc3339(),
)?;
api.store.reclaim_backend_job_input(
api.workspace_id(),
&reservation.job.request.job_id,
&reservation.job.input_digest,
crate::backend_job::JobInputRetirementReason::LegacyInvalidation,
)?;
let queued = api.clone();
let request = reservation.job.request.clone();
api.host_executors.host()?.dispatch(move || {
drain_after_completion(&queued, &request);
})?;
Ok(true)
}
fn saved_input(request: &BackendJobRequest, grant: &SubjektivJournalGrant) -> Result<JournalInput> {
let input = JournalInput::decode(&request.input).map_err(domain)?;
if input.subject_id != grant.subject_id
|| input.journal_id != grant.journal_id
|| input.capture.session_id != grant.session_id
|| input.selected_ranges != grant.selected_ranges
{
return Err(Error::WorkspacePermissionDenied(
"Journal saved input/capability mismatch".into(),
));
}
Ok(input)
}
fn live(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
@@ -125,6 +187,7 @@ fn live(
.store
.get_backend_job(api.workspace_id(), &binding.job_id)?
.ok_or_else(|| Error::WorkspacePermissionDenied("Journal Job missing".into()))?;
job.load_executable_request()?;
let attempt = api
.store
.get_backend_job_attempt(api.workspace_id(), &binding.job_id, &binding.attempt_id)?
@@ -180,15 +243,44 @@ pub(super) fn execute(
worker: &RuntimeWorkerRef,
request: JournalJobRequest,
) -> Result<JournalJobResponse> {
let (grant, _, _) = live(api, worker)?;
let gate = lifecycle_lock(api, &grant.subject_id)?;
let _guard = gate
.lock()
.map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?;
// Dispatch holds this same gate until matching Runtime acceptance is attested.
// Revalidate after waiting: erasure, cancellation or replacement may have won.
let (grant, job, attempt) = live(api, worker)?;
let store = store(api)?;
let host = host(api, &store, &grant)?;
let g = generation(&store, &host, &grant, &job, &attempt)?;
let input = saved_input(&job.request, &grant)?;
input.validate_generation(&g).map_err(domain)?;
let ctx = JournalJobContext::validated(&store, &host, &g.id, g.job.clone()).map_err(domain)?;
journal::service::execute_bounded(&store, &ctx, request, |source, start, limit| {
read_source(api, &grant, source, start, limit)
.map_err(|e| JournalError::Conflict(e.to_string()))
})
journal::service::execute_snapshot(&store, &ctx, &attempt.input_digest, request).map_err(domain)
}
/// Called only after matching tracked Runtime acceptance of the exact rendered saved input.
/// The durable request and Worker binding alone do not establish body delivery.
pub(super) fn attest_delivery(
api: &WorkspaceApi,
reservation: &BackendJobReservation,
) -> Result<()> {
let Some(grant) = &reservation.job.request.grants.subjektiv_journal else {
return Ok(());
};
let input = saved_input(&reservation.job.request, grant)?;
let store = store(api)?;
let host = host(api, &store, grant)?;
let g = generation(&store, &host, grant, &reservation.job, &reservation.attempt)?;
input.validate_generation(&g).map_err(domain)?;
let ctx = JournalJobContext::validated(&store, &host, &g.id, g.job.clone()).map_err(domain)?;
journal::service::attest_input_delivery(
&store,
&ctx,
&reservation.attempt.input_digest,
&input.selected_ranges,
)
.map_err(domain)
}
@@ -196,21 +288,29 @@ pub(super) fn finish(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
) -> Result<JournalJobCompletion> {
let (grant, _, _) = live(api, worker)?;
let gate = lifecycle_lock(api, &grant.subject_id)?;
let _guard = gate
.lock()
.map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?;
// Dispatch holds this same gate until matching Runtime acceptance is attested.
// Revalidate after waiting: erasure, cancellation or replacement may have won.
let (grant, job, attempt) = live(api, worker)?;
let store = store(api)?;
let host = host(api, &store, &grant)?;
let input = saved_input(&job.request, &grant)?;
// prepare is an exact binding lookup too, including the saved completion of a closed generation.
let g = store
.prepare_journal_generation(
.prepare_journal_input(
&host,
&grant.journal_id,
&input,
subjektiv::api::JobAttemptBinding {
job_id: job.request.job_id,
attempt_id: attempt.attempt_id,
},
grant.selected_ranges.clone(),
)
.map_err(domain)?;
input.validate_generation(&g).map_err(domain)?;
let ctx = JournalJobContext::validated_recovery(&store, &host, &g.id, g.job.clone())
.map_err(domain)?;
journal::service::finish(&store, &ctx).map_err(domain)
@@ -277,21 +377,6 @@ pub(super) fn close(
Ok(())
}
fn read_source(
api: &WorkspaceApi,
grant: &SubjektivJournalGrant,
source: &JournalSource,
start: u64,
limit: usize,
) -> Result<JournalMaterial> {
let store = store(api)?;
let attr = store
.session_attribution(&grant.session_id)
.map_err(|e| Error::Store(e.to_string()))?
.ok_or_else(|| Error::WorkspacePermissionDenied("Session attribution absent".into()))?;
read_canonical(&material_reader(api, &attr)?, source, start, limit)
}
pub(super) fn bind_host_session(
api: &WorkspaceApi,
attr: crate::subjektiv::SubjectSessionAttribution,
@@ -524,22 +609,63 @@ fn dispatch(
.close_journal_generation(&host, &g.id, &g.job)
.map_err(domain)?;
}
let reader = material_reader(api, &attribution)?;
let materials = select_materials_bounded(
&store,
&host,
journal_id,
96 * 1024,
|source, start, limit| {
read_canonical(&reader, source, start, limit)
.map_err(|e| JournalError::Conflict(e.to_string()))
},
)
let mut after = None;
loop {
let jobs = api.store.list_backend_journal_jobs(
api.workspace_id(),
&attribution.session_id,
after.as_deref(),
100,
)?;
for job in &jobs {
if job
.request
.grants
.subjektiv_journal
.as_ref()
.is_none_or(|g| g.journal_id != journal_id)
|| job.input_disposition != crate::backend_job::JobInputDisposition::Available
{
continue;
}
if job.state == BackendJobState::Pending || job.state == BackendJobState::Unknown {
return Ok(());
}
if job.state == BackendJobState::Failed && !legacy_descriptor(&job.request) {
job.load_executable_request()?;
saved_input(
&job.request,
job.request
.grants
.subjektiv_journal
.as_ref()
.expect("checked Journal grant"),
)?;
let reservation = api.store.reserve_backend_job_retry(
api.workspace_id(),
&job.request.job_id,
&Utc::now().to_rfc3339(),
)?;
return queue_reservation(api, reservation);
}
}
if jobs.len() < 100 {
break;
}
after = jobs.last().map(|j| j.request.job_id.clone());
}
let input = JournalInput::build(&store, &host, journal_id, |order, offset, budget| {
capture_input(api, &attribution, order, offset, budget)
.map_err(|e| JournalError::Conflict(e.to_string()))
})
.map_err(domain)?;
if materials.is_empty() {
if input.selected_ranges.is_empty() {
return Ok(());
}
let request = recording_request(&host, attribution, journal_id, materials)?;
store
.accept_journal_sources(&host, "Session activity", &input.selected_ranges)
.map_err(domain)?;
let request = recording_request(&host, attribution, journal_id, input)?;
// The intent is saved before generation binding or dispatch. If binding/dispatch fails,
// Reserved recovery repeats this exact intent, never creates a replacement blindly.
api.resolve_backend_job_profile(&request)?;
@@ -548,6 +674,10 @@ fn dispatch(
&request,
&Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
)?;
queue_reservation(api, reservation)
}
fn queue_reservation(api: &WorkspaceApi, reservation: BackendJobReservation) -> Result<()> {
let queued = api.clone();
if let Err(error)=api.host_executors.host()?.dispatch(move || {
if let Err(error)=queued.dispatch_backend_job_reservation(reservation) {
@@ -578,33 +708,21 @@ fn admit_sources(
));
}
let mut id = store.journal_for_session(&host).map_err(domain)?;
let reader = material_reader(api, &attribution)?;
if reader.not_materialized {
if store.journal_retention_required(&host).map_err(domain)? {
return Err(Error::RepositoryConflict(
"Admitted Journal original Session is not materialized".into(),
));
}
retain_source(
api.runtime.as_ref(),
api.workspace_id(),
&attribution,
material_source(api, &attribution)?,
true,
)?;
return Ok(JournalRecordingResponse::NoNewSources);
}
let (order, offset) = if let Some(id) = id.as_deref() {
let journal = store.read_journal(&host, id).map_err(domain)?;
(journal.process_order, journal.process_byte)
} else {
(0, 0)
};
let capture = capture_input(api, &attribution, order, offset, 48 * 1024)?;
retain_source(
api.runtime.as_ref(),
api.workspace_id(),
&attribution,
material_source(api, &attribution)?,
false,
capture.segment_id == "not-materialized",
)?;
let sources: Vec<_> = reader
.descriptors()
.map(|d| source_from_descriptor(&attribution.session_id, d))
.collect();
let sources = journal::recording::capture_ranges(&capture).map_err(domain)?;
for batch in sources.chunks(MAX_JOURNAL_RANGES) {
id = Some(
store
@@ -995,7 +1113,7 @@ pub(super) fn release_subject(
{
// Legacy saved-before-admission Sessions are not optional. Prove canonical
// source identity first, then bind only this original Host-attributed Session.
let _canonical = material_reader(api, &attr)?;
let _canonical = capture_input(api, &attr, 0, 0, 48 * 1024)?;
let host =
JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?;
if host
@@ -1281,80 +1399,6 @@ fn material_source(
},
)
}
fn material_reader<'a>(
api: &'a WorkspaceApi,
attribution: &crate::subjektiv::SubjectSessionAttribution,
) -> Result<MaterialReader<'a>> {
MaterialReader::open(
api.runtime.as_ref(),
api.workspace_id(),
&attribution.runtime_id,
material_source(api, attribution)?,
&attribution.session_id,
)
}
fn source_from_descriptor(
session: &str,
d: &session_store::public_material::PublicMaterialDescriptor,
) -> JournalSourceRange {
JournalSourceRange {
source: JournalSource {
session_id: session.into(),
pointer: JournalSourcePointer {
segment_id: d.identity.segment_id,
entry_ref: d.identity.entry_ref.clone(),
},
order: d.canonical_order,
content_sha256: d.sha256.clone(),
total_bytes: d.total_bytes,
},
bytes: JournalByteRange {
start: 0,
end: d.total_bytes,
},
}
}
fn read_canonical(
reader: &MaterialReader<'_>,
source: &JournalSource,
start: u64,
limit: usize,
) -> Result<JournalMaterial> {
if reader.session_id != source.session_id {
return Err(Error::WorkspacePermissionDenied(
"Journal source Session mismatch".into(),
));
}
let descriptor = reader
.descriptors()
.find(|d| source_from_descriptor(&source.session_id, d).source == *source)
.ok_or_else(|| {
Error::Store("Journal original public source changed or unavailable".into())
})?;
// The reader streams the complete canonical digest while returning only this bounded window.
let page = reader
.read_content(descriptor, start, limit)
.map_err(|e| Error::Store(e.to_string()))?;
if page.sha256 != source.content_sha256
|| page.total_bytes != source.total_bytes
|| page.next_offset != start.saturating_add(page.content.len() as u64)
{
return Err(Error::RepositoryConflict(
"Journal canonical source identity changed".into(),
));
}
Ok(JournalMaterial {
range: JournalSourceRange {
source: source.clone(),
bytes: JournalByteRange {
start,
end: page.next_offset,
},
},
text: page.content,
})
}
/// Close the first-run save/notification gap: Runtime can commit an initial turn before
/// the create handler has finalized the Backend registry required by recording callbacks.
/// Reconcile only after creation succeeds, using the ordinary current-owner/Feature checks.
@@ -1475,7 +1519,7 @@ pub(super) fn recording_request(
host: &JournalHostContext,
attribution: crate::subjektiv::SubjectSessionAttribution,
journal_id: &str,
materials: Vec<journal::recording::JournalMaterial>,
input: JournalInput,
) -> Result<BackendJobRequest> {
let binding_id = host
.writer_binding_id()
@@ -1488,18 +1532,88 @@ pub(super) fn recording_request(
subject_id: attribution.subject_id,
session_id: attribution.session_id,
journal_id: journal_id.into(),
selected_ranges: materials.iter().map(|m| m.range.clone()).collect(),
selected_ranges: input.selected_ranges.clone(),
};
Ok(BackendJobRequest {
job_id:format!("journal-recording:{}",Uuid::new_v4()),purpose:"subjektiv_journal".into(),
input_ref:format!("subjektiv://{}/journal/{journal_id}",grant.subject_id),
input:serde_json::json!({"journal_id":journal_id,"selected_ranges":grant.selected_ranges}),
instruction:"Record ONLY this frozen original public selected_ranges prefix. The input contains descriptors, NOT source bodies. First ReadJournalState and use bounded ReadJournalText Source until every byte of every selected range is returned; record operations are refused until then. ReadJournalText also exposes initial/current position/open text. Sequentially save CreateJournalEntry/RewriteJournalEntry/FixJournalEntry, update current position and CheckpointJournal using retained record receipts. Never infer event completion from turn/task/job/compaction boundaries. Preserve exact operation ID/full input after unknown outcomes. SubmitBackendJobResult only after the full selected prefix has a saved checkpoint; no aggregate JournalRecordResult.".into(),
profile:"builtin:subjektiv-journal-recording".into(),
grants:BackendJobGrants {subjektiv_journal:Some(grant),..Default::default()},
serialization_key:Some(journal_serialization_key(journal_id)),source_worker:None,notification_target:None,
limits:crate::backend_job::BackendJobLimits {timeout_seconds:600,..Default::default()},
})
let request = BackendJobRequest {
job_id: format!("journal-recording:{}",Uuid::new_v4()), purpose: "subjektiv_journal".into(),
input_ref: format!("subjektiv://{}/journal/{journal_id}",grant.subject_id),
input: serde_json::to_value(input).map_err(|e| Error::Store(e.to_string()))?,
instruction: "Record ONLY the immutable body-bearing capture supplied in this input. These exact canonical source bytes, identities and ranges are the whole input for this Job; do not fetch original Sessions or use ReadJournalText Source. First ReadJournalState; initial/current Journal position/open text remain separately paged by ReadJournalText. Sequentially save CreateJournalEntry/RewriteJournalEntry/FixJournalEntry, current position and receipt-backed CheckpointJournal. Input delivery is not understanding or reflection. Never infer event completion from turn/task/job/compaction boundaries. Preserve exact operation ID/full input after Unknown. SubmitBackendJobResult only after the complete selected prefix has a saved checkpoint; no aggregate JournalRecordResult.".into(),
profile: "builtin:subjektiv-journal-recording".into(),
grants: BackendJobGrants { subjektiv_journal: Some(grant), ..Default::default() },
serialization_key: Some(journal_serialization_key(journal_id)), source_worker: None, notification_target: None,
limits: crate::backend_job::BackendJobLimits { timeout_seconds: 600, ..Default::default() },
};
saved_input(
&request,
request
.grants
.subjektiv_journal
.as_ref()
.expect("constructed grant"),
)?;
request.validate()?;
Ok(request)
}
pub(super) fn capture_input(
api: &WorkspaceApi,
attr: &crate::subjektiv::SubjectSessionAttribution,
order: u64,
offset: u64,
budget: usize,
) -> Result<session_store::capture::CaptureDto> {
match api
.runtime
.session_recording_capture(
&attr.runtime_id,
runtime_api::SessionRecordingCaptureRequest {
workspace_id: api.workspace_id().into(),
source: material_source(api, attr)?,
expected_session_id: attr.session_id.clone(),
start: session_store::capture::RecordingPosition {
canonical_order: order,
byte_offset: offset,
},
max_entries: MAX_JOURNAL_RANGES,
max_serialized_bytes: budget,
},
)
.map_err(RuntimeRegistryError::into_error)?
{
runtime_api::SessionRecordingCaptureAvailability::Captured { slice } => {
if slice.capture.session_id != attr.session_id
|| serde_json::to_vec(&slice)
.map_err(|e| Error::Store(e.to_string()))?
.len()
> budget
{
return Err(Error::RepositoryConflict(
"recording capture identity/budget mismatch".into(),
));
}
Ok(slice.capture)
}
runtime_api::SessionRecordingCaptureAvailability::NotMaterialized { session_id }
if session_id == attr.session_id && order == 0 && offset == 0 =>
{
Ok(session_store::capture::CaptureDto {
version: 1,
session_id,
segment_id: "not-materialized".into(),
view_entry_count: 0,
entries: vec![],
})
}
runtime_api::SessionRecordingCaptureAvailability::Unavailable { reason, .. } => {
Err(Error::Store(format!(
"committed Journal capture unavailable ({reason:?}); no latest-source fallback"
)))
}
_ => Err(Error::Store(
"committed Journal capture identity mismatch; no latest-source fallback".into(),
)),
}
}
fn removal_attribution(
@@ -1575,6 +1689,16 @@ fn finish_erased_release(
"explicit erase tombstone missing".into(),
));
}
for session in store
.subject_session_attributions(&attr.subject_id)
.map_err(|e| Error::Store(e.to_string()))?
.into_iter()
.filter(|session| {
session.runtime_id == attr.runtime_id && session.worker_id == attr.worker_id
})
{
require_erased_job_copies(api, &session.session_id)?;
}
// Pre-binding bodies have no write token. This stable receipt label is not a writer
// grant: the domain erase operation records an inactive lifetime without binding Sessions.
let erased_body_id = format!("erased-body:{}", attr.worker_id);
@@ -1835,183 +1959,6 @@ pub(super) async fn finish_route(
}
/// A bounded Runtime transport view, not a second history index or a local-path fallback.
/// Reopening it observes a new catalog; each generation still compares its frozen source identity.
struct MaterialReader<'a> {
runtime: &'a RuntimeRegistry,
workspace_id: &'a str,
runtime_id: String,
source: runtime_api::SessionPublicSource,
session_id: String,
not_materialized: bool,
descriptors: Vec<session_store::public_material::PublicMaterialDescriptor>,
}
fn material_descriptor(
d: runtime_api::SessionMaterialDescriptor,
) -> Result<session_store::public_material::PublicMaterialDescriptor> {
Ok(session_store::public_material::PublicMaterialDescriptor {
identity: session_store::public_material::PublicMaterialIdentity {
segment_id: Uuid::parse_str(&d.segment_id).map_err(|e| Error::Store(e.to_string()))?,
entry_ref: d.entry_ref,
},
canonical_order: d.canonical_order,
origin: serde_json::from_str(&d.origin_json).map_err(|e| Error::Store(e.to_string()))?,
sha256: d.sha256,
total_bytes: d.total_bytes,
})
}
impl<'a> MaterialReader<'a> {
fn open(
runtime: &'a RuntimeRegistry,
workspace_id: &'a str,
runtime_id: &str,
source: runtime_api::SessionPublicSource,
session_id: &str,
) -> Result<Self> {
let mut descriptors = Vec::new();
let mut offset = 0;
let mut condition = None;
loop {
let page = match runtime
.session_material_catalog(
runtime_id,
runtime_api::SessionMaterialCatalogRequest {
workspace_id: workspace_id.into(),
source: source.clone(),
expected_session_id: session_id.into(),
expected_catalog_sha256: condition.clone(),
offset,
limit: 100,
},
)
.map_err(|e| e.into_error())?
{
runtime_api::SessionMaterialCatalogAvailability::NotMaterialized {
session_id: observed,
} => {
if observed != session_id || offset != 0 || condition.is_some() {
return Err(Error::RepositoryConflict(
"Journal Session materialization proof changed during catalog read"
.into(),
));
}
return Ok(Self {
runtime,
workspace_id,
runtime_id: runtime_id.into(),
source,
session_id: session_id.into(),
not_materialized: true,
descriptors,
});
}
runtime_api::SessionMaterialCatalogAvailability::Page { page } => page,
runtime_api::SessionMaterialCatalogAvailability::Unavailable {
reason,
message,
} => {
return Err(Error::Store(format!(
"Journal original material unavailable ({reason:?}): {message}"
)));
}
};
if page.session_id != session_id
|| condition
.as_ref()
.is_some_and(|c| c != &page.catalog_sha256)
|| page.descriptors.len() > 100
|| descriptors.len().saturating_add(page.descriptors.len()) > 1_000_000
{
return Err(Error::RepositoryConflict(
"Journal canonical Session catalog changed or exceeded its bound".into(),
));
}
condition = Some(page.catalog_sha256);
for d in page.descriptors {
let d = material_descriptor(d)?;
if d.canonical_order != descriptors.len() as u64 {
return Err(Error::RepositoryConflict(
"Journal canonical catalog order is discontinuous".into(),
));
}
descriptors.push(d);
}
match page.next_offset {
Some(next) if next > offset && next == descriptors.len() as u64 => offset = next,
None => break,
_ => {
return Err(Error::RepositoryConflict(
"Journal catalog continuation did not advance".into(),
));
}
}
}
Ok(Self {
runtime,
workspace_id,
runtime_id: runtime_id.into(),
source,
session_id: session_id.into(),
not_materialized: false,
descriptors,
})
}
fn descriptors(
&self,
) -> impl Iterator<Item = &session_store::public_material::PublicMaterialDescriptor> {
self.descriptors.iter()
}
fn read_content(
&self,
descriptor: &session_store::public_material::PublicMaterialDescriptor,
offset: u64,
limit: usize,
) -> Result<session_store::public_material::PublicMaterialContent> {
let page = match self
.runtime
.session_material_read(
&self.runtime_id,
runtime_api::SessionMaterialReadRequest {
workspace_id: self.workspace_id.into(),
source: self.source.clone(),
expected_session_id: self.session_id.clone(),
expected_descriptor: runtime_api::SessionMaterialDescriptor {
segment_id: descriptor.identity.segment_id.to_string(),
entry_ref: descriptor.identity.entry_ref.clone(),
canonical_order: descriptor.canonical_order,
origin_json: serde_json::to_string(&descriptor.origin)
.map_err(|e| Error::Store(e.to_string()))?,
sha256: descriptor.sha256.clone(),
total_bytes: descriptor.total_bytes,
},
byte_offset: offset,
max_content_bytes: limit.clamp(4, 16 * 1024),
},
)
.map_err(|e| e.into_error())?
{
runtime_api::SessionMaterialReadAvailability::Page { page } => page,
runtime_api::SessionMaterialReadAvailability::Unavailable { reason, message } => {
return Err(Error::Store(format!(
"Journal original material read unavailable ({reason:?}): {message}"
)));
}
};
if material_descriptor(page.descriptor)? != *descriptor
|| page.content.len() > limit.clamp(4, 16 * 1024)
{
return Err(Error::RepositoryConflict(
"Journal canonical material identity changed".into(),
));
}
Ok(session_store::public_material::PublicMaterialContent {
content: page.content,
next_offset: page.next_offset,
total_bytes: descriptor.total_bytes,
sha256: descriptor.sha256.clone(),
})
}
}
fn erase_source(
runtime: &RuntimeRegistry,
workspace: &str,
@@ -2145,17 +2092,29 @@ pub(super) fn reconcile_unknown(
let host = host(api, &store, grant)?;
let erased = store.journal_session_erased(&host).map_err(domain)?;
if !erased {
let g = store
.prepare_journal_generation(
&host,
&grant.journal_id,
subjektiv::api::JobAttemptBinding {
job_id: job.request.job_id.clone(),
attempt_id: attempt.attempt_id.clone(),
},
grant.selected_ranges.clone(),
)
.map_err(domain)?;
job.load_executable_request()?;
let binding = subjektiv::api::JobAttemptBinding {
job_id: job.request.job_id.clone(),
attempt_id: attempt.attempt_id.clone(),
};
let g = if legacy_descriptor(&job.request) {
// Only reconcile already-saved legacy receipts; never redispatch descriptors.
store
.prepare_journal_generation(
&host,
&grant.journal_id,
binding,
grant.selected_ranges.clone(),
)
.map_err(domain)?
} else {
let input = saved_input(&job.request, grant)?;
let generation = store
.prepare_journal_input(&host, &input, binding)
.map_err(domain)?;
input.validate_generation(&generation).map_err(domain)?;
generation
};
let ctx = JournalJobContext::validated_recovery(&store, &host, &g.id, g.job.clone())
.map_err(domain)?;
match journal::service::finish(&store, &ctx) {
@@ -2187,6 +2146,14 @@ pub(super) fn reconcile_unknown(
);
if current.worker_cleanup_state == Some(BackendJobWorkerCleanupState::Completed) {
drain_after_completion(api, &job.request);
if job.input_disposition == crate::backend_job::JobInputDisposition::Available {
api.store.reclaim_backend_job_input(
api.workspace_id(),
&job.request.job_id,
&job.input_digest,
crate::backend_job::JobInputRetirementReason::Completed,
)?;
}
} else {
api.schedule_backend_job_worker_cleanup(&current);
}
@@ -2226,8 +2193,8 @@ pub(super) fn reconcile_unknown(
&Utc::now().to_rfc3339(),
)?;
if !erased {
// The old immutable input may include reflected bytes. Select the remaining
// prefix afresh only AFTER definitive lifecycle/domain fences; never replace Unknown.
// Resume the same saved body input only after canonical execution/domain fences.
// dispatch selects its retained intent before it can capture any newer source.
let attr = store
.session_attribution(&grant.session_id)
.map_err(|e| Error::Store(e.to_string()))?
@@ -2256,6 +2223,30 @@ fn reconcile_session_jobs(
.store
.get_backend_job_attempt(api.workspace_id(), &job.request.job_id, &id)?
.ok_or_else(|| Error::Store("Journal current attempt missing".into()))?;
if !erased
&& legacy_descriptor(&job.request)
&& job.input_disposition == crate::backend_job::JobInputDisposition::Available
{
job.load_executable_request()?;
close(api, &job.request, &attempt)?;
if !attempt.state.terminal() {
attempt = api.store.finish_backend_job_attempt(
api.workspace_id(),
&job.request.job_id,
&id,
BackendJobAttemptState::Failed,
"journal_descriptor_invalidated",
"Descriptor-only Journal input retired; saved checkpoints retained",
&Utc::now().to_rfc3339(),
)?;
}
api.store.reclaim_backend_job_input(
api.workspace_id(),
&job.request.job_id,
&job.input_digest,
crate::backend_job::JobInputRetirementReason::LegacyInvalidation,
)?;
}
if erased && !attempt.state.terminal() {
// The domain tombstone precedes this cancellation and also fences any
// crash window before it. No erased Job can prepare or use tools again.
@@ -2269,6 +2260,30 @@ fn reconcile_session_jobs(
&Utc::now().to_rfc3339(),
)?;
}
if erased {
api.store.reclaim_backend_job_input(
api.workspace_id(),
&job.request.job_id,
&job.input_digest,
crate::backend_job::JobInputRetirementReason::ExplicitErasure,
)?;
// Every historical execution can own a delivered copy, not just the current
// attempt or an unacknowledged result. Cleanup is canonical Purge and retryable.
for number in 1..=job.current_attempt {
let historical_id = crate::backend_job::attempt_id(&job.request.job_id, number);
let historical = api
.store
.get_backend_job_attempt(
api.workspace_id(),
&job.request.job_id,
&historical_id,
)?
.ok_or_else(|| Error::Store("historical Journal attempt missing".into()))?;
if historical.state.terminal() {
api.schedule_backend_job_worker_cleanup(&historical);
}
}
}
if attempt.state == BackendJobAttemptState::Unknown {
reconcile_unknown(api, job, &attempt)?;
}
@@ -2284,6 +2299,41 @@ fn reconcile_session_jobs(
Ok(())
}
fn require_erased_job_copies(api: &WorkspaceApi, session: &str) -> Result<()> {
let mut after = None;
loop {
let jobs = api.store.list_backend_journal_jobs(
api.workspace_id(),
session,
after.as_deref(),
100,
)?;
for job in &jobs {
if job.input_disposition != crate::backend_job::JobInputDisposition::Erased {
return Err(Error::RepositoryConflict(
"Journal input erasure remains incomplete".into(),
));
}
for number in 1..=job.current_attempt {
let id = crate::backend_job::attempt_id(&job.request.job_id, number);
let attempt = api
.store
.get_backend_job_attempt(api.workspace_id(), &job.request.job_id, &id)?
.ok_or_else(|| Error::Store("erased Journal attempt missing".into()))?;
if attempt.worker.is_some()
&& attempt.worker_cleanup_state != Some(BackendJobWorkerCleanupState::Completed)
{
return Err(Error::RepositoryConflict("Journal delivered-copy Purge remains pending; retry the same erasure after cleanup".into()));
}
}
}
if jobs.len() < 100 {
return Ok(());
}
after = jobs.last().map(|j| j.request.job_id.clone());
}
}
impl WorkspaceApi {
/// Host-only erasure saves the domain fence, cancels Jobs, then erases originals.
/// Runtime requires a stopped source Worker. Physical failure leaves the domain
@@ -2301,6 +2351,10 @@ impl WorkspaceApi {
.ok_or_else(|| {
Error::WorkspacePermissionDenied("Session attribution missing".into())
})?;
let gate = lifecycle_lock(self, &attr.subject_id)?;
let _guard = gate
.lock()
.map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?;
let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?;
store
.erase_journal_session(&host, request_id)
@@ -2311,7 +2365,8 @@ impl WorkspaceApi {
self.workspace_id(),
&attr,
material_source(self, &attr)?,
)
)?;
require_erased_job_copies(self, &attr.session_id)
}
}
@@ -605,6 +605,10 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() {
workdir::WorkdirFailureReason::SessionClosed,
workdir::WorkdirFailureReason::CommandTerminalMissing,
workdir::WorkdirFailureReason::HttpTimeout,
workdir::WorkdirFailureReason::OperationResultMismatch,
workdir::WorkdirFailureReason::SessionOpenTaskFailed,
workdir::WorkdirFailureReason::AuthorizationStale,
workdir::WorkdirFailureReason::RuntimeStopped,
] {
let request = runtime_source_request(
&identity,
@@ -653,6 +657,7 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() {
);
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["failure_reason"], reason.as_str());
assert_eq!(events[0]["failure_stage"], reason.stage().as_str());
assert_eq!(events[0]["operation"], "command_output");
assert_eq!(events[0]["workdir_stage"], "provider_dispatch");
let log = serde_json::to_string(&events).unwrap();
@@ -1249,3 +1254,54 @@ async fn non_api_workspace_paths_are_not_logged_as_ticket_api_errors() {
assert_eq!(status, StatusCode::BAD_REQUEST);
assert!(events.is_empty());
}
#[tokio::test]
async fn session_open_failure_causes_survive_registry_api_conversion_without_raw_diagnostics() {
use workdir::WorkdirFailureReason as Reason;
for reason in [
Reason::SessionOpenFailed,
Reason::SessionOpenTaskFailed,
Reason::SessionIdentityMismatch,
Reason::AuthorizationStale,
Reason::AuthorizationUnavailable,
Reason::RuntimeStopped,
Reason::WorkingDirectoryRecordInvalid,
Reason::OperationResultMismatch,
] {
let mut fixture = manual_worker_assignment_fixture().await;
fixture.runtime.workdir_session_factory = Some(Arc::new(move |_| {
Err(workdir::WorkdirError::Unavailable(
"/host/private body-secret proof-secret key-secret credential-secret".into(),
)
.with_failure_reason(reason))
}));
fixture
.api
.runtime
.register_or_replace(fixture.runtime.clone());
let identity =
worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id)
.unwrap();
configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations");
let request = runtime_source_request(&identity, Some(&fixture.worker.worker_id), "POST", &path, serde_json::to_vec(&json!({
"target_workdir": "checkout",
"operation": {"operation": "read", "request": {"path":"README.md", "offset":0, "limit":10, "max_bytes":1024}}
})).unwrap());
let (status, body, events) =
request_with_logs(build_router(fixture.api.clone()), request).await;
assert_eq!(status, StatusCode::BAD_GATEWAY, "{body}");
assert_eq!(
body["message"],
"workdir_session_open_failed: Workdir session is unavailable"
);
assert!(body.get("failure_reason").is_none(), "{body}");
assert!(!body.to_string().contains("secret"));
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["workdir_stage"], "session_open");
assert_eq!(events[0]["failure_reason"], reason.as_str());
assert_eq!(events[0]["failure_stage"], reason.stage().as_str());
assert_eq!(events[0]["message"], "Workdir operation failed");
assert_eq!(events[0]["diagnostics"], "[]");
}
}
@@ -39,6 +39,8 @@ fn page(cursor: u64, bytes: &[u8], terminal: bool) -> workdir::CommandOutput {
},
next_cursor: cursor + bytes.len() as u64,
terminal: terminal.then_some(workdir::CommandTerminal {
process_failure: None,
output_collection: workdir::CommandOutputCollection::Complete,
status: workdir::CommandStatus::Cancelled,
exit_code: None,
stdout_bytes: cursor + bytes.len() as u64,
@@ -11,6 +11,20 @@ use worker_runtime::worker_backend::{
WorkerRuntimeExecutionBackend,
};
pub(super) fn connected_restore_tempdir() -> tempfile::TempDir {
crate::restore_test_environment::require_restore_test_environment();
let checkout = Path::new(env!("CARGO_MANIFEST_DIR"))
.parent()
.unwrap()
.parent()
.unwrap();
tempfile::Builder::new()
.prefix("c")
.rand_bytes(3)
.tempdir_in(checkout.join("target"))
.unwrap()
}
const WATCHDOG: Duration = Duration::from_secs(45);
#[derive(Clone)]
@@ -82,24 +96,91 @@ impl RuntimeWorkerFactory for CapturingFactory {
) -> std::result::Result<(), String> {
self.inner.preflight_restore(request).await
}
async fn restore_controller(
async fn recover_restore_source(
&self,
request: &worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<bool, worker_runtime::worker_backend::RuntimeRestoreFailure> {
self.inner.recover_restore_source(request).await
}
async fn prepare_restore(
&self,
request: worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<RuntimeWorkerController, String> {
) -> std::result::Result<
Box<dyn worker_runtime::worker_backend::RuntimeRestoreCandidate>,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
let worker_ref = request.worker_ref.clone();
let controller = self.inner.restore_controller(request).await?;
let inner = self.inner.prepare_restore(request).await?;
Ok(Box::new(CapturingCandidate {
inner,
worker_ref,
captures: self.captures.clone(),
}))
}
async fn cleanup_failed_restore(
&self,
request: &worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreCleanupFailure> {
self.inner.cleanup_failed_restore(request).await
}
async fn reconcile_stopped_worker(
&self,
worker_ref: &worker_runtime::identity::WorkerRef,
) -> std::result::Result<(), String> {
self.inner.reconcile_stopped_worker(worker_ref).await
}
fn activate_restored_controller(
&self,
worker_ref: &worker_runtime::identity::WorkerRef,
workspace_id: Option<&str>,
handle: &worker::WorkerHandle,
) -> std::result::Result<(), String> {
self.inner
.activate_restored_controller(worker_ref, workspace_id, handle)
}
}
struct CapturingCandidate {
inner: Box<dyn worker_runtime::worker_backend::RuntimeRestoreCandidate>,
worker_ref: worker_runtime::identity::WorkerRef,
captures: Arc<Captures>,
}
#[async_trait::async_trait]
impl worker_runtime::worker_backend::RuntimeRestoreCandidate for CapturingCandidate {
fn initial_state(&self) -> protocol::WorkerStateSnapshot {
self.inner.initial_state()
}
async fn activate(
&mut self,
) -> std::result::Result<
RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
let controller = self.inner.activate().await?;
self.captures
.handles
.lock()
.unwrap()
.push((worker_ref.clone(), controller.handle.clone()));
.push((self.worker_ref.clone(), controller.handle.clone()));
self.captures
.restored_clients
.lock()
.unwrap()
.push((worker_ref, controller.workspace_client.clone()));
.push((self.worker_ref.clone(), controller.workspace_client.clone()));
Ok(controller)
}
async fn discard(
&mut self,
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreCleanupFailure> {
self.inner.discard().await
}
}
struct Resources {
@@ -259,7 +340,7 @@ async fn call(tools: &BTreeMap<String, Arc<dyn Tool>>, name: &str, args: Value)
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn restored_runtime_parent_lists_and_inputs_existing_child() {
let root = tempfile::tempdir().unwrap();
let root = connected_restore_tempdir();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let mut config = test_server_config(root.path());
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -200,6 +200,7 @@ async fn backend_delegation_reaches_remote_repository_resolution() {
session_id: WorkdirSessionId::new("delegation-session").unwrap(),
workdir_id,
capabilities: Caps::ALL,
environment: workdir::WorkdirSessionEnvironment::root(Caps::ALL),
})
}
}),
@@ -1,3 +1,5 @@
mod audit_tests;
use super::*;
use sha2::Digest;
use std::sync::atomic::AtomicI64;
@@ -459,7 +461,7 @@ async fn provider_error_and_disconnect_release_dispatch_admission() {
assert_json(
result,
if disconnect {
StatusCode::SERVICE_UNAVAILABLE
StatusCode::INTERNAL_SERVER_ERROR
} else {
StatusCode::FORBIDDEN
},
@@ -575,8 +577,8 @@ async fn provider_timeout_releases_admission_without_wall_clock_wait() {
let (long, _reply) = f.start_read("1", &mut a).await;
tokio::time::pause();
tokio::time::advance(std::time::Duration::from_secs(31)).await;
let body = assert_json(long.await.unwrap(), StatusCode::SERVICE_UNAVAILABLE).await;
assert_eq!(body["code"], "unavailable");
let body = assert_json(long.await.unwrap(), StatusCode::INTERNAL_SERVER_ERROR).await;
assert_eq!(body["code"], "outcome_unknown");
tokio::time::resume();
let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await;
drop(
@@ -0,0 +1,885 @@
//! Integration coverage for signed loopback HTTP -> production admission/auth -> external
//! Workdir session -> real SQLite audit persistence. Only the external provider boundary
//! is substituted with a controlled command receiver. Temporary DBs, listeners and bounded
//! requests are isolated per case; these run in the normal workspace-server library suite.
use super::*;
use std::io::Write;
use tracing::instrument::WithSubscriber;
const AUDIT_SECRET: &str = "audit-store-password-secret";
const WRITE_SECRET: &[u8] = b"write-content-secret";
const PATH_SECRET: &str = "private-request-path-secret.txt";
const PROVIDER_SECRET: &str = "/host/private/provider-secret";
type ProviderReply =
oneshot::Sender<std::result::Result<WorkdirSessionOperationResult, WorkdirTransportError>>;
#[derive(Clone, Default)]
struct CapturedAuditLog(Arc<std::sync::Mutex<Vec<u8>>>);
impl Write for CapturedAuditLog {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
struct PendingHttpWrite {
task: tokio::task::JoinHandle<(StatusCode, Value)>,
proof: String,
}
impl Drop for PendingHttpWrite {
fn drop(&mut self) {
self.task.abort();
}
}
struct ProviderProbe {
receiver: mpsc::Receiver<ExternalProviderCommand>,
dispatches: usize,
}
impl ProviderProbe {
async fn receive_operation(
&mut self,
pending: &mut PendingHttpWrite,
) -> (String, WorkdirSessionOperation, ProviderReply) {
let command = tokio::time::timeout(BOUND, async {
tokio::select! {
command = self.receiver.recv() => command.expect("provider channel closed"),
response = &mut pending.task => {
panic!("HTTP completed before write dispatch: {:?}", response.unwrap());
}
}
})
.await
.expect("signed HTTP write never reached provider");
let ExternalProviderCommand::Operation {
operation_id,
operation,
response,
} = command
else {
panic!("expected write dispatch, received cancellation")
};
self.dispatches += 1;
(operation_id.as_str().to_owned(), operation, response)
}
async fn receive_write(&mut self, pending: &mut PendingHttpWrite) -> (String, ProviderReply) {
let (operation_id, operation, response) = self.receive_operation(pending).await;
let WorkdirSessionOperation::Write(request) = operation else {
panic!("expected mutation, got {operation:?}")
};
assert_eq!(
request.path,
workdir::WorkdirPath::new(PATH_SECRET).unwrap()
);
assert_eq!(request.content, WRITE_SECRET);
assert_eq!(request.expected_hash, None);
(operation_id, response)
}
fn assert_dispatches(&mut self, expected: usize) {
while let Ok(command) = self.receiver.try_recv() {
if matches!(command, ExternalProviderCommand::Operation { .. }) {
self.dispatches += 1;
}
}
assert_eq!(
self.dispatches, expected,
"unexpected mutation dispatch/retry"
);
}
}
impl Fixture {
fn attach_audited_writer(&self) -> ProviderProbe {
let receiver = self.attach("1");
// Setup changes only: real signed requests still traverse production authorization.
// All three capability sources must agree before the session is first resolved.
let connection = rusqlite::Connection::open(&self.api.config.database_path).unwrap();
assert_eq!(
connection
.execute(
"UPDATE external_workdir_grants SET permissions='read_write' \
WHERE workspace_id=?1 AND grant_id='external-1'",
[TEST_WORKSPACE_ID],
)
.unwrap(),
1
);
assert_eq!(
connection
.execute(
"UPDATE worker_workdir_links SET capabilities='read_write' \
WHERE workspace_id=?1 AND runtime_id='workdirless-runtime' \
AND worker_id='1' AND alias='attachment' AND unlinked_at IS NULL",
[TEST_WORKSPACE_ID],
)
.unwrap(),
1
);
let mut registry = self.api.external_workdir_providers.lock().unwrap();
let provider = registry.connections.get_mut("external-1").unwrap();
Arc::get_mut(provider).unwrap().capabilities =
workdir::WorkdirSessionCapabilities::READ_WRITE;
ProviderProbe {
receiver,
dispatches: 0,
}
}
}
struct AuditHttpFixture {
fixture: Fixture,
logs: CapturedAuditLog,
client: reqwest::Client,
base_url: String,
server: tokio::task::JoinHandle<()>,
}
impl Drop for AuditHttpFixture {
fn drop(&mut self) {
self.server.abort();
}
}
impl AuditHttpFixture {
async fn new() -> (Self, ProviderProbe) {
let fixture = Fixture::new().await;
let provider = fixture.attach_audited_writer();
let logs = CapturedAuditLog::default();
let writer = logs.clone();
let subscriber = tracing_subscriber::fmt()
.without_time()
.with_ansi(false)
.json()
.flatten_event(true)
.with_writer(move || writer.clone())
.finish();
let dispatch = tracing::Dispatch::new(subscriber);
// Axum spawns connection tasks: scope the subscriber around each request rather
// than relying on the listener task's thread-local subscriber being inherited.
let app = fixture.app.clone().layer(axum::middleware::from_fn(
move |request: Request<Body>, next: axum::middleware::Next| {
let dispatch = dispatch.clone();
async move { next.run(request).with_subscriber(dispatch).await }
},
));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let client = reqwest::Client::builder()
.no_proxy()
.timeout(BOUND)
.build()
.unwrap();
(
Self {
fixture,
logs,
client,
base_url,
server,
},
provider,
)
}
fn fail_audit(&self, before_dispatch: bool) {
let filter = if before_dispatch { "=" } else { "<>" };
rusqlite::Connection::open(&self.fixture.api.config.database_path)
.unwrap()
.execute_batch(&format!(
"CREATE TRIGGER fail_operation_audit BEFORE INSERT ON audit_events \
WHEN NEW.action='external_workdir_operation' AND NEW.outcome {filter} 'requested' \
BEGIN SELECT RAISE(FAIL, '{AUDIT_SECRET}'); END;"
))
.unwrap();
}
fn start_write(&self) -> PendingHttpWrite {
self.start_operation(WorkdirSessionOperation::Write(workdir::WriteRequest {
path: workdir::WorkdirPath::new(PATH_SECRET).unwrap(),
content: WRITE_SECRET.to_vec(),
expected_hash: None,
}))
}
fn start_operation(&self, operation: WorkdirSessionOperation) -> PendingHttpWrite {
let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations");
let body = serde_json::to_vec(&WorkspaceWorkdirSessionOperationRequest {
target_workdir: "attachment".into(),
operation,
})
.unwrap();
let request = self.fixture.request(
Some("1"),
&path,
body.clone(),
START,
worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION,
);
let proof = request.headers()[worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER]
.to_str()
.unwrap()
.to_owned();
let url = format!("{}{path}", self.base_url);
let client = self.client.clone();
let headers = request.headers().clone();
let task = tokio::spawn(async move {
let response = client
.post(url)
.headers(headers)
.body(body)
.send()
.await
.unwrap();
let status = response.status();
assert_eq!(response.headers()[CONTENT_TYPE], "application/json");
let bytes = response.bytes().await.unwrap();
(status, serde_json::from_slice(&bytes).unwrap())
});
PendingHttpWrite { task, proof }
}
async fn finish(
&self,
mut pending: PendingHttpWrite,
expected_status: StatusCode,
) -> (Value, Vec<Value>) {
let (status, body) = tokio::time::timeout(BOUND, &mut pending.task)
.await
.expect("HTTP did not finish; possible automatic retry")
.unwrap();
assert_eq!(status, expected_status, "{body}");
let text = String::from_utf8(self.logs.0.lock().unwrap().clone()).unwrap();
let encoded_content = serde_json::to_string(WRITE_SECRET).unwrap();
for output in [body.to_string(), text.clone()] {
for secret in [
AUDIT_SECRET,
std::str::from_utf8(WRITE_SECRET).unwrap(),
encoded_content.as_str(),
PATH_SECRET,
PROVIDER_SECRET,
PROVIDER_SECRET.trim_start_matches('/'),
pending.proof.as_str(),
worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
self.fixture.identity.private_key.as_str(),
self.fixture.identity.public_key.as_str(),
self.fixture._root.path().to_str().unwrap(),
] {
assert!(!output.contains(secret), "sensitive data escaped: {output}");
}
}
let events = text
.lines()
.map(|line| serde_json::from_str(line).unwrap())
.collect();
(body, events)
}
fn assert_audits(&self, expected: &[(&str, &str)]) {
self.assert_operation_audits(expected, "write");
}
fn assert_operation_audits(&self, expected: &[(&str, &str)], operation: &str) {
let connection =
rusqlite::Connection::open(&self.fixture.api.config.database_path).unwrap();
let mut statement = connection
.prepare(
"SELECT request_id, outcome, summary FROM audit_events \
WHERE workspace_id=?1 AND action='external_workdir_operation' \
AND target_id='external-1' ORDER BY rowid",
)
.unwrap();
let rows: Vec<(String, String, String)> = statement
.query_map([TEST_WORKSPACE_ID], |row| {
Ok((row.get(0)?, row.get(1)?, row.get(2)?))
})
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap();
let actual: Vec<(&str, &str)> = rows
.iter()
.map(|(operation_id, outcome, _)| (operation_id.as_str(), outcome.as_str()))
.collect();
assert_eq!(actual, expected, "persisted audits: {rows:?}");
for (_, _, summary) in rows {
assert!(
summary.contains(&format!("operation_kind={operation}")),
"{summary}"
);
for secret in [
AUDIT_SECRET,
PATH_SECRET,
PROVIDER_SECRET,
"write-content-secret",
] {
assert!(
!summary.contains(secret),
"sensitive audit summary: {summary}"
);
}
}
}
fn apply_provider_write(&self) {
std::fs::write(self.effect_path(), WRITE_SECRET).unwrap();
}
fn effect_path(&self) -> std::path::PathBuf {
self.fixture._root.path().join("provider-effect")
}
}
fn write_result() -> WorkdirSessionOperationResult {
WorkdirSessionOperationResult::Write(workdir::WriteResult {
bytes_written: WRITE_SECRET.len(),
created: true,
})
}
fn assert_audit_failure(
body: &Value,
events: &[Value],
stage: &str,
debug_stage: &str,
outcome: Value,
debug_outcome: &str,
) -> String {
let failures = body["audit_failures"]
.as_array()
.expect("independent audit receipt");
assert_eq!(failures.len(), 1, "{body}");
let operation_id = failures[0]["operation_id"].as_str().unwrap();
assert!(operation_id.starts_with("op-"), "{body}");
assert_eq!(
failures[0],
json!({
"operation_id": operation_id,
"stage": stage,
"reason": "audit_store_unavailable",
"operation_outcome": outcome,
})
);
let notifications: Vec<&Value> = events
.iter()
.filter(|event| event["event"] == "external_workdir_audit_failed")
.collect();
assert_eq!(notifications.len(), 1, "{events:?}");
let event = notifications[0];
assert_eq!(event["operation_id"], operation_id);
assert_eq!(event["operation"], "write");
assert_eq!(event["audit_stage"], debug_stage);
assert_eq!(event["audit_reason"], "AuditStoreUnavailable");
assert_eq!(event["operation_outcome"], debug_outcome);
operation_id.to_owned()
}
fn assert_failure_reason(events: &[Value], reason: &str, stage: &str) {
let errors: Vec<&Value> = events
.iter()
.filter(|event| event["event"] == "api_error")
.collect();
assert_eq!(errors.len(), 1, "{events:?}");
assert_eq!(errors[0]["failure_reason"], reason, "{events:?}");
assert_eq!(errors[0]["failure_stage"], stage, "{events:?}");
}
fn assert_operational_failure(
body: &Value,
events: &[Value],
reason: &str,
stage: &str,
code: &str,
message: &str,
) {
assert_failure_reason(events, reason, stage);
assert_eq!(body["code"], code, "{body}");
assert_eq!(body["message"], message, "{body}");
for field in ["failure_reason", "failure_stage"] {
assert!(
body.get(field).is_none(),
"internal diagnostic escaped: {body}"
);
}
}
#[tokio::test]
async fn successful_write_after_audit_failure_preserves_result_and_does_not_repeat_mutation() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
h.apply_provider_write();
reply.send(Ok(write_result())).unwrap();
let (mut body, events) = h.finish(pending, StatusCode::OK).await;
let receipt_id = assert_audit_failure(
&body,
&events,
"response_received",
"ResponseReceived",
json!({"kind":"succeeded"}),
"Succeeded",
);
assert_eq!(receipt_id, dispatched_id);
body.as_object_mut().unwrap().remove("audit_failures");
assert_eq!(body, serde_json::to_value(write_result()).unwrap());
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[tokio::test]
async fn denied_write_after_audit_failure_preserves_provider_error_and_separate_receipt() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
let mut original = WorkdirTransportError::from_workdir_error(&workdir::WorkdirError::Denied(
PROVIDER_SECRET.into(),
));
let expected_message = original.message.clone();
original.message = PROVIDER_SECRET.into();
reply.send(Err(original)).unwrap();
let (body, events) = h.finish(pending, StatusCode::FORBIDDEN).await;
assert_eq!(body["code"], "denied");
assert_eq!(body["message"], expected_message);
let receipt_id = assert_audit_failure(
&body,
&events,
"response_received",
"ResponseReceived",
json!({"kind":"failed", "code":"denied"}),
"Failed { code: Denied }",
);
assert_eq!(receipt_id, dispatched_id);
assert!(!h.effect_path().exists());
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[tokio::test]
async fn lost_write_response_after_audit_failure_stays_unknown_without_repeating_effect() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
h.apply_provider_write();
// The provider applied the effect, but the server cannot infer that from response loss.
drop(reply);
let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await;
assert_operational_failure(
&body,
&events,
"provider_response_lost",
"provider_response",
"outcome_unknown",
"Workdir operation may have effects; inspect before retrying",
);
let receipt_id = assert_audit_failure(
&body,
&events,
"response_unknown",
"ResponseUnknown",
json!({"kind":"unknown"}),
"Unknown",
);
assert_eq!(receipt_id, dispatched_id);
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[tokio::test]
async fn before_write_audit_failure_is_not_dispatched_and_has_no_provider_effect() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(true);
let (body, events) = h
.finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE)
.await;
assert_eq!(body["code"], "not_dispatched");
assert_audit_failure(
&body,
&events,
"before_dispatch",
"BeforeDispatch",
json!({"kind":"not_dispatched"}),
"NotDispatched",
);
assert_failure_reason(&events, "audit_store_unavailable", "audit");
assert!(
body.get("failure_reason").is_none(),
"internal reason leaked: {body}"
);
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[]);
}
#[tokio::test]
async fn rejected_write_send_with_audit_failure_is_not_dispatched_not_unknown() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
provider.receiver.close();
let (body, events) = h
.finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE)
.await;
assert_eq!(body["code"], "not_dispatched");
let receipt_id = assert_audit_failure(
&body,
&events,
"dispatch_rejected",
"DispatchRejected",
json!({"kind":"not_dispatched"}),
"NotDispatched",
);
assert_operational_failure(
&body,
&events,
"provider_dispatch_rejected",
"provider_dispatch",
"not_dispatched",
"Workdir operation was not dispatched",
);
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[(&receipt_id, "requested")]);
}
#[tokio::test]
async fn expired_connection_grant_is_not_dispatched_and_logs_admission_reason() {
let (h, mut provider) = AuditHttpFixture::new().await;
{
let mut registry = h.fixture.api.external_workdir_providers.lock().unwrap();
let connection = registry.connections.get_mut("external-1").unwrap();
// Keep the persisted grant online so the real handler reaches operate's
// connection-expiry fence rather than the separate session-open fence.
Arc::get_mut(connection).unwrap().expires_at =
Some(Utc::now() - chrono::Duration::seconds(1));
}
h.fail_audit(false);
let (body, events) = h
.finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE)
.await;
assert_operational_failure(
&body,
&events,
"provider_grant_expired",
"provider_admission",
"unavailable",
"Workdir session is unavailable",
);
assert!(body.get("audit_failures").is_none(), "{body}");
assert!(
events
.iter()
.all(|event| event["event"] != "external_workdir_audit_failed")
);
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[]);
}
#[tokio::test]
async fn closed_provider_admission_is_not_dispatched_and_logs_admission_reason() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fixture
.api
.external_workdir_providers
.lock()
.unwrap()
.connection("external-1")
.unwrap()
.admission
.close();
h.fail_audit(false);
let (body, events) = h
.finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE)
.await;
assert_operational_failure(
&body,
&events,
"provider_admission_closed",
"provider_admission",
"unavailable",
"Workdir session is unavailable",
);
assert!(body.get("audit_failures").is_none(), "{body}");
assert!(
events
.iter()
.all(|event| event["event"] != "external_workdir_audit_failed")
);
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[]);
}
#[tokio::test]
async fn timed_out_write_after_audit_failure_stays_unknown_and_logs_timeout_reason() {
let (mut h, mut provider) = AuditHttpFixture::new().await;
// Keep the HTTP client's independent deadline beyond the unchanged 30s
// provider timeout. Pause only after the signed HTTP request was dispatched,
// avoiding automatic virtual-time advancement during socket/SQLite setup.
h.client = reqwest::Client::builder()
.no_proxy()
.timeout(std::time::Duration::from_secs(40))
.build()
.unwrap();
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
h.apply_provider_write();
tokio::time::pause();
tokio::time::advance(std::time::Duration::from_secs(31)).await;
tokio::time::resume();
let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await;
assert_operational_failure(
&body,
&events,
"provider_operation_timeout",
"provider_response",
"outcome_unknown",
"Workdir operation may have effects; inspect before retrying",
);
let receipt_id = assert_audit_failure(
&body,
&events,
"response_unknown",
"ResponseUnknown",
json!({"kind":"unknown"}),
"Unknown",
);
assert_eq!(receipt_id, dispatched_id);
assert!(reply.is_closed(), "timeout retained its response receiver");
assert!(matches!(
provider.receiver.try_recv().unwrap(),
ExternalProviderCommand::Cancel { operation_id }
if operation_id.as_str() == dispatched_id.as_str()
));
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[tokio::test]
async fn successful_write_persists_requested_and_completed_with_same_operation_id() {
let (h, mut provider) = AuditHttpFixture::new().await;
let mut pending = h.start_write();
let (operation_id, reply) = provider.receive_write(&mut pending).await;
h.apply_provider_write();
reply.send(Ok(write_result())).unwrap();
let (body, events) = h.finish(pending, StatusCode::OK).await;
assert_eq!(body, serde_json::to_value(write_result()).unwrap());
assert!(
events
.iter()
.all(|event| event["event"] != "external_workdir_audit_failed")
);
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
h.assert_audits(&[(&operation_id, "requested"), (&operation_id, "completed")]);
}
#[tokio::test]
async fn provider_reported_unknown_preserves_received_response_and_unknown_effects() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
reply
.send(Err(WorkdirTransportError::from_workdir_error(
&workdir::WorkdirError::OutcomeUnknown(PROVIDER_SECRET.into()),
)))
.unwrap();
let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await;
assert_eq!(body["code"], "outcome_unknown");
let receipt_id = assert_audit_failure(
&body,
&events,
"response_received",
"ResponseReceived",
json!({"kind":"unknown"}),
"Unknown",
);
assert_eq!(receipt_id, dispatched_id);
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[derive(Clone, Copy, Debug)]
enum CheckoutReplyFault {
WrongOperation,
WrongPath,
EmptyValidator,
WrongOutput,
ProviderUnavailable,
ProviderTransport,
}
async fn assert_checkout_unknown_audit(fault: CheckoutReplyFault, fail_completion: bool) {
let (h, mut provider) = AuditHttpFixture::new().await;
if fail_completion {
h.fail_audit(false);
}
let target = workdir::WorkdirPath::new(PATH_SECRET).unwrap();
let mut pending = h.start_operation(WorkdirSessionOperation::CheckoutExecute(
workdir::CheckoutRequest {
target: target.clone(),
validator: vec![1],
operation: workdir::CheckoutOperation::Write {
content: WRITE_SECRET.to_vec(),
expected_hash: [0; 32],
},
},
));
let (operation_id, operation, reply) = provider.receive_operation(&mut pending).await;
let WorkdirSessionOperation::CheckoutExecute(request) = operation else {
panic!("expected checkout mutation, got {operation:?}");
};
assert_eq!(request.target, target);
assert_eq!(request.validator, vec![1]);
assert!(
matches!(request.operation, workdir::CheckoutOperation::Write { ref content, .. } if content == WRITE_SECRET)
);
h.apply_provider_write();
let result = match fault {
CheckoutReplyFault::WrongOperation => Ok(write_result()),
CheckoutReplyFault::ProviderUnavailable => Err(WorkdirTransportError::from_workdir_error(
&workdir::WorkdirError::Unavailable(PROVIDER_SECRET.into()),
)),
CheckoutReplyFault::ProviderTransport => Err(WorkdirTransportError::from_workdir_error(
&workdir::WorkdirError::Transport(PROVIDER_SECRET.into()),
)),
_ => {
let mut result = workdir::CheckoutResult {
observation: workdir::CheckoutObservation {
path: target,
validator: vec![2],
kind: workdir::EntryKind::File,
size: WRITE_SECRET.len() as u64,
capabilities: workdir::WorkdirSessionCapabilities::READ_WRITE,
},
output: workdir::CheckoutOutput::Write(workdir::WriteResult {
bytes_written: WRITE_SECRET.len(),
created: false,
}),
};
match fault {
CheckoutReplyFault::WrongPath => {
result.observation.path =
workdir::WorkdirPath::new(PROVIDER_SECRET.trim_start_matches('/')).unwrap()
}
CheckoutReplyFault::EmptyValidator => result.observation.validator.clear(),
CheckoutReplyFault::WrongOutput => {
result.output = workdir::CheckoutOutput::Edit(workdir::EditResult {
replacements: 1,
content_hash: [0; 32],
bytes_written: 1,
})
}
_ => unreachable!(),
}
Ok(WorkdirSessionOperationResult::CheckoutExecute(result))
}
};
reply.send(result).unwrap();
let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await;
assert_eq!(body["code"], "outcome_unknown", "{fault:?}: {body}");
assert_eq!(
body["message"],
"Workdir operation may have effects; inspect before retrying"
);
if !matches!(
fault,
CheckoutReplyFault::ProviderUnavailable | CheckoutReplyFault::ProviderTransport
) {
assert_failure_reason(&events, "operation_result_mismatch", "response_validation");
}
if fail_completion {
assert_eq!(
body["audit_failures"],
json!([{
"operation_id": operation_id,
"stage": "response_received",
"reason": "audit_store_unavailable",
"operation_outcome": {"kind":"unknown"},
}]),
"{fault:?}: {body}"
);
let audit_logs: Vec<_> = events
.iter()
.filter(|e| e["event"] == "external_workdir_audit_failed")
.collect();
assert_eq!(audit_logs.len(), 1, "{events:?}");
let event = audit_logs[0];
assert_eq!(event["operation_id"], operation_id);
assert_eq!(event["operation"], "checkout_execute");
assert_eq!(event["audit_stage"], "ResponseReceived");
assert_eq!(event["operation_outcome"], "Unknown");
h.assert_operation_audits(&[(&operation_id, "requested")], "checkout_execute");
} else {
assert!(body.get("audit_failures").is_none(), "{body}");
assert!(
events
.iter()
.all(|e| e["event"] != "external_workdir_audit_failed")
);
h.assert_operation_audits(
&[
(&operation_id, "requested"),
(&operation_id, "outcome_unknown"),
],
"checkout_execute",
);
}
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
}
#[tokio::test]
async fn checkout_invalid_reply_records_unknown_in_healthy_completion_audit() {
for fault in [
CheckoutReplyFault::WrongOperation,
CheckoutReplyFault::WrongPath,
CheckoutReplyFault::EmptyValidator,
CheckoutReplyFault::WrongOutput,
] {
assert_checkout_unknown_audit(fault, false).await;
}
}
#[tokio::test]
async fn checkout_invalid_reply_records_unknown_in_failed_completion_receipt_and_log() {
for fault in [
CheckoutReplyFault::WrongOperation,
CheckoutReplyFault::WrongPath,
CheckoutReplyFault::EmptyValidator,
CheckoutReplyFault::WrongOutput,
] {
assert_checkout_unknown_audit(fault, true).await;
}
}
#[tokio::test]
async fn checkout_uncertain_reply_normalization_matches_audit_outcome() {
for fault in [
CheckoutReplyFault::ProviderUnavailable,
CheckoutReplyFault::ProviderTransport,
] {
for fail_completion in [false, true] {
assert_checkout_unknown_audit(fault, fail_completion).await;
}
}
}
@@ -467,7 +467,6 @@ impl AttentionWorkerFactory {
worker::WorkerBootstrapLayout::RuntimeManagedRun {
run_dir: root.join(uuid::Uuid::now_v7().to_string()),
},
worker::WorkerControllerTransport::InProcess,
)
.start()
.await
@@ -477,6 +476,7 @@ impl AttentionWorkerFactory {
shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))),
controller_task: started.controller_task,
workspace_client,
restore_streams: None,
})
}
}
@@ -490,12 +490,61 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for AttentionWorkerFac
self.controller(request.worker_ref.worker_id.to_string(), false)
.await
}
async fn restore_controller(
async fn prepare_restore(
&self,
request: worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<worker_runtime::worker_backend::RuntimeWorkerController, String> {
self.controller(request.worker_ref.worker_id.to_string(), true)
.await
) -> std::result::Result<
Box<dyn worker_runtime::worker_backend::RuntimeRestoreCandidate>,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
Ok(Box::new(AttentionRestoreCandidate {
inputs: Some((
AttentionWorkerFactory {
root: self.root.clone(),
},
request.worker_ref.worker_id.to_string(),
)),
}))
}
}
// Preparation owns only fixture inputs; reconstruction and Controller startup
// remain lazy so discard never needs to shut down a started Worker.
struct AttentionRestoreCandidate {
inputs: Option<(AttentionWorkerFactory, String)>,
}
#[async_trait::async_trait]
impl worker_runtime::worker_backend::RuntimeRestoreCandidate for AttentionRestoreCandidate {
fn initial_state(&self) -> protocol::WorkerStateSnapshot {
protocol::WorkerStateSnapshot::initial()
}
async fn activate(
&mut self,
) -> std::result::Result<
worker_runtime::worker_backend::RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
let (factory, worker_id) = self.inputs.take().ok_or_else(|| {
worker_runtime::worker_backend::RuntimeRestoreFailure::refused(
worker_runtime::worker_backend::RuntimeRestoreStage::Controller,
"attention fixture candidate already consumed",
)
})?;
factory.controller(worker_id, true).await.map_err(|_| {
worker_runtime::worker_backend::RuntimeRestoreFailure::unproven(
worker_runtime::worker_backend::RuntimeRestoreStage::Reconstruction,
"attention fixture reconstruction failed",
)
})
}
async fn discard(
&mut self,
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreCleanupFailure> {
self.inputs.take();
Ok(())
}
}
@@ -602,7 +651,7 @@ async fn orchestrator_attention_real_worker_lost_response_retry_commits_once() {
async fn real_attention_worker_start_restore(lose_response: bool) {
use session_store::{LogEntry, Store, WorkerSessionStore};
let dir = tempfile::tempdir().unwrap();
let dir = restored_parent_control_tests::connected_restore_tempdir();
init_clean_git_workspace(dir.path());
let config = test_server_config(dir.path());
let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap());
@@ -14,17 +14,37 @@ async fn notification_diagnostics_http_auth_scope_and_readonly_evidence() {
seed_test_api_token(api.store.as_ref(), TEST_WORKSPACE_ID);
let ticket = delivery_ticket(&api, "Notification diagnostics auth fixture");
delivery_assign(&api, ticket.id.as_str(), "diagnostics-assignment").await;
let path = format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?ticket={}&limit=1",ticket.id.as_str());
let page = delivery_http(&api,"GET",&path,None,StatusCode::OK).await;
assert_eq!(page["items"].as_array().unwrap().len(),1);
assert_eq!(page["items"][0]["model_processing_confirmed"],false);
let path = format!(
"/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?ticket={}&limit=1",
ticket.id.as_str()
);
let page = delivery_http(&api, "GET", &path, None, StatusCode::OK).await;
assert_eq!(page["items"].as_array().unwrap().len(), 1);
assert_eq!(page["items"][0]["model_processing_confirmed"], false);
assert!(page["items"][0].get("content").is_none());
let again=delivery_http(&api,"GET",&path,None,StatusCode::OK).await;
assert_eq!(again,page);
let response=build_router(api.clone()).oneshot(Request::builder().uri(&path).body(Body::empty()).unwrap()).await.unwrap();
assert_eq!(response.status(),StatusCode::UNAUTHORIZED);
delivery_http(&api,"GET","/api/w/other-workspace/notification-diagnostics",None,StatusCode::NOT_FOUND).await;
delivery_http(&api,"GET",&format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?limit=101"),None,StatusCode::BAD_REQUEST).await;
let again = delivery_http(&api, "GET", &path, None, StatusCode::OK).await;
assert_eq!(again, page);
let response = build_router(api.clone())
.oneshot(Request::builder().uri(&path).body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
delivery_http(
&api,
"GET",
"/api/w/other-workspace/notification-diagnostics",
None,
StatusCode::NOT_FOUND,
)
.await;
delivery_http(
&api,
"GET",
&format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?limit=101"),
None,
StatusCode::BAD_REQUEST,
)
.await;
}
fn delivery_assignment_path(ticket: &str) -> String {
@@ -800,7 +820,6 @@ impl DeliveryWorkerFactory {
worker::WorkerBootstrapLayout::RuntimeManagedRun {
run_dir: root.join(uuid::Uuid::now_v7().to_string()),
},
worker::WorkerControllerTransport::InProcess,
)
.with_model_client(self.client.clone());
let prepared = if restore {
@@ -815,6 +834,7 @@ impl DeliveryWorkerFactory {
shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))),
controller_task: started.controller_task,
workspace_client,
restore_streams: None,
})
}
}
@@ -828,12 +848,62 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for DeliveryWorkerFact
self.controller(request.worker_ref.worker_id.to_string(), false)
.await
}
async fn restore_controller(
async fn prepare_restore(
&self,
request: delivery_execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<worker_runtime::worker_backend::RuntimeWorkerController, String> {
self.controller(request.worker_ref.worker_id.to_string(), true)
.await
) -> std::result::Result<
Box<dyn worker_runtime::worker_backend::RuntimeRestoreCandidate>,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
Ok(Box::new(DeliveryRestoreCandidate {
inputs: Some((
DeliveryWorkerFactory {
root: self.root.clone(),
client: self.client.clone(),
},
request.worker_ref.worker_id.to_string(),
)),
}))
}
}
// Retain the same scripted client/receipt store without opening either store or
// starting a Controller until activation. Discard drops only these owned inputs.
struct DeliveryRestoreCandidate {
inputs: Option<(DeliveryWorkerFactory, String)>,
}
#[async_trait::async_trait]
impl worker_runtime::worker_backend::RuntimeRestoreCandidate for DeliveryRestoreCandidate {
fn initial_state(&self) -> protocol::WorkerStateSnapshot {
protocol::WorkerStateSnapshot::initial()
}
async fn activate(
&mut self,
) -> std::result::Result<
worker_runtime::worker_backend::RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
let (factory, worker_id) = self.inputs.take().ok_or_else(|| {
worker_runtime::worker_backend::RuntimeRestoreFailure::refused(
worker_runtime::worker_backend::RuntimeRestoreStage::Controller,
"delivery fixture candidate already consumed",
)
})?;
factory.controller(worker_id, true).await.map_err(|_| {
worker_runtime::worker_backend::RuntimeRestoreFailure::unproven(
worker_runtime::worker_backend::RuntimeRestoreStage::Reconstruction,
"delivery fixture reconstruction failed",
)
})
}
async fn discard(
&mut self,
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreCleanupFailure> {
self.inputs.take();
Ok(())
}
}
@@ -932,6 +1002,7 @@ async fn delivery_real_fixture(
Arc<DeliveryResponseLossBackend>,
DeliveryScriptedClient,
) {
let _environment = restored_parent_control_tests::connected_restore_tempdir();
let config = test_server_config(root);
let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap());
seed_test_registered_workspace(store.as_ref(), &config)
@@ -1137,9 +1208,22 @@ async fn assignment_real_worker_response_loss_restore_and_concurrent_scans_commi
"uncertain delivery changed request id or payload: {attempts:?}"
);
assert_eq!(delivery_receipts(&api, &worker).len(), 1);
let unknown = api.store.notification_diagnostics(TEST_WORKSPACE_ID, &server_api::NotificationDiagnosticsQuery {
assignment_id: Some(assignment.into()), ..Default::default()
}).unwrap().items.into_iter().find(|item| item.stage == server_api::NotificationDiagnosticStage::AcceptanceUnknown).expect("saved unknown acceptance must remain visible while Runtime already has its receipt");
let unknown = api
.store
.notification_diagnostics(
TEST_WORKSPACE_ID,
&server_api::NotificationDiagnosticsQuery {
assignment_id: Some(assignment.into()),
..Default::default()
},
)
.unwrap()
.items
.into_iter()
.find(|item| item.stage == server_api::NotificationDiagnosticStage::AcceptanceUnknown)
.expect(
"saved unknown acceptance must remain visible while Runtime already has its receipt",
);
assert!(!unknown.model_processing_confirmed);
// Change mutable rendering context while the acknowledgement is unknown.
// A new self-report is not a new input and cannot change the frozen retry.
@@ -1200,11 +1284,24 @@ async fn assignment_real_worker_response_loss_restore_and_concurrent_scans_commi
}
delivery_wait_idle(&reopened, &worker, 2).await;
assert_eq!(delivery_receipts(&reopened, &worker).len(), 2);
let evidence = delivery_http(&reopened,"GET",&format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?source_id={}",unknown.source_id),None,StatusCode::OK).await;
assert_eq!(evidence["items"][0]["stage"],"runtime_accepted");
assert_eq!(evidence["items"][0]["delivery_id"],json!(unknown.delivery_id));
assert_eq!(evidence["items"][0]["model_processing_confirmed"],false);
assert_eq!(evidence["items"][0]["retry_candidate"],false);
let evidence = delivery_http(
&reopened,
"GET",
&format!(
"/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?source_id={}",
unknown.source_id
),
None,
StatusCode::OK,
)
.await;
assert_eq!(evidence["items"][0]["stage"], "runtime_accepted");
assert_eq!(
evidence["items"][0]["delivery_id"],
json!(unknown.delivery_id)
);
assert_eq!(evidence["items"][0]["model_processing_confirmed"], false);
assert_eq!(evidence["items"][0]["retry_candidate"], false);
assert_eq!(
delivery_commits(&reopened, &worker)
.iter()
@@ -187,7 +187,14 @@ async fn config_call(
runtime
.invoke(
path.into(),
if matches!(operation,"edit"|"write") {wip_protocol::InterfaceReference {scope:"/".into(),name:format!("yoi.file/{operation}-text/v1")}} else {config_interface(path)},
if matches!(operation, "edit" | "write") {
wip_protocol::InterfaceReference {
scope: "/".into(),
name: format!("yoi.file/{operation}-text/v1"),
}
} else {
config_interface(path)
},
operation.into(),
args,
agen::tool::ToolExecutionContext::direct(),
@@ -581,11 +588,88 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for ConfigCaptureFacto
*self.controller_abort.lock().unwrap() = Some(controller.controller_task.abort_handle());
Ok(controller)
}
async fn restore_controller(
async fn preflight_restore(
&self,
request: &worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<(), String> {
self.inner.preflight_restore(request).await
}
async fn recover_restore_source(
&self,
request: &worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<bool, worker_runtime::worker_backend::RuntimeRestoreFailure> {
self.inner.recover_restore_source(request).await
}
async fn prepare_restore(
&self,
request: worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<worker_runtime::worker_backend::RuntimeWorkerController, String> {
self.inner.restore_controller(request).await
) -> std::result::Result<
Box<dyn worker_runtime::worker_backend::RuntimeRestoreCandidate>,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
let inner = self.inner.prepare_restore(request).await?;
Ok(Box::new(ConfigCaptureCandidate {
inner,
client: self.client.clone(),
controller_abort: self.controller_abort.clone(),
}))
}
async fn cleanup_failed_restore(
&self,
request: &worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreCleanupFailure> {
self.inner.cleanup_failed_restore(request).await
}
async fn reconcile_stopped_worker(
&self,
worker_ref: &worker_runtime::identity::WorkerRef,
) -> std::result::Result<(), String> {
self.inner.reconcile_stopped_worker(worker_ref).await
}
fn activate_restored_controller(
&self,
worker_ref: &worker_runtime::identity::WorkerRef,
workspace_id: Option<&str>,
handle: &worker::WorkerHandle,
) -> std::result::Result<(), String> {
self.inner
.activate_restored_controller(worker_ref, workspace_id, handle)
}
}
struct ConfigCaptureCandidate {
inner: Box<dyn worker_runtime::worker_backend::RuntimeRestoreCandidate>,
client: Arc<Mutex<Option<Arc<dyn worker::WorkspaceClient>>>>,
controller_abort: Arc<Mutex<Option<tokio::task::AbortHandle>>>,
}
#[async_trait]
impl worker_runtime::worker_backend::RuntimeRestoreCandidate for ConfigCaptureCandidate {
fn initial_state(&self) -> protocol::WorkerStateSnapshot {
self.inner.initial_state()
}
async fn activate(
&mut self,
) -> std::result::Result<
worker_runtime::worker_backend::RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
let controller = self.inner.activate().await?;
*self.client.lock().unwrap() = Some(controller.workspace_client.clone());
*self.controller_abort.lock().unwrap() = Some(controller.controller_task.abort_handle());
Ok(controller)
}
async fn discard(
&mut self,
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreCleanupFailure> {
self.inner.discard().await
}
}
@@ -714,6 +798,7 @@ async fn workspace_config_actual_remote_and_embedded_clients_use_authenticated_w
RUNTIME_REQUEST_SOURCE_PROOF_HEADER, WORKSPACE_REQUEST_PERMISSION,
decode_runtime_request_source_claims,
};
let root = restored_parent_control_tests::connected_restore_tempdir();
// Remote Runtime-owned HTTP adapter, including its real request signer.
let mut fixture = manual_worker_assignment_fixture().await;
let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
@@ -793,7 +878,6 @@ async fn workspace_config_actual_remote_and_embedded_clients_use_authenticated_w
// Actual embedded Runtime factory -> real Worker controller -> captured
// production WorkspaceClient -> signed HTTP -> production Backend middleware.
let root = tempfile::tempdir().unwrap();
init_clean_git_workspace(root.path());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
+529 -13
View File
@@ -119,7 +119,7 @@ pub use ticket::{
};
const OLDEST_SCHEMA_VERSION: i64 = 50;
const LATEST_SCHEMA_VERSION: i64 = 103;
const LATEST_SCHEMA_VERSION: i64 = 104;
const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
@@ -442,8 +442,21 @@ const MIGRATIONS: &[Migration] = &[
name: "Passkey metadata and atomic addition results",
apply: migrate_passkey_results_v102_to_v103,
},
Migration {
version: 104,
name: "Converge Passkey results and Job input retention",
apply: migrate_schema_103_variants_to_v104,
},
];
// Both schema-103 identities shipped. Keep the alternate implementation and accept
// its exact receipt without renaming or replacing any applied history.
const JOB_INPUT_RETENTION_SCHEMA_103: Migration = Migration {
version: 103,
name: "Job input retirement independent of immutable receipts",
apply: migrate_job_input_retention_v102_to_v103,
};
#[derive(Clone, Copy)]
struct Migration {
version: i64,
@@ -1806,6 +1819,15 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
request: &BackendJobRequest,
now: &str,
) -> Result<BackendJobReservation>;
/// Host-only retirement; fence domain mutations and delivered snapshots first for erasure.
/// Completed reclamation additionally requires canonical Worker cleanup for all attempts.
fn reclaim_backend_job_input(
&self,
workspace_id: &str,
job_id: &str,
expected_input_digest: &str,
reason: crate::backend_job::JobInputRetirementReason,
) -> Result<BackendJobRecord>;
fn reserve_backend_job_retry(
&self,
workspace_id: &str,
@@ -7488,6 +7510,48 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn reclaim_backend_job_input(
&self,
workspace_id: &str,
job_id: &str,
expected_input_digest: &str,
reason: crate::backend_job::JobInputRetirementReason,
) -> Result<BackendJobRecord> {
use crate::backend_job::{JobInputDisposition, JobInputRetirementReason};
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let job = read_backend_job(&tx, workspace_id, job_id)?
.ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job `{job_id}`")))?;
if job.input_digest != expected_input_digest {
return Err(Error::InvalidInput("Backend Job input digest mismatch".into()));
}
let desired = reason.disposition();
if job.input_disposition == desired {
tx.commit()?;
reclaim_backend_job_input_storage(conn)?;
return Ok(job);
}
if reason == JobInputRetirementReason::Completed {
job.load_executable_request()?;
let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, &attempt_id(job_id, job.current_attempt))?
.ok_or_else(|| Error::Store("Backend Job current attempt missing".into()))?;
if job.state != BackendJobState::Completed || attempt.state != BackendJobAttemptState::Completed
|| job.result.is_none() || backend_job_has_unfinished_cleanup(&tx, workspace_id, job_id)?
{ return Err(Error::InvalidInput("input reclaim requires known completed result and completed Worker cleanup".into())); }
}
if job.input_disposition != JobInputDisposition::Available
&& reason != JobInputRetirementReason::ExplicitErasure
{ return Err(Error::InvalidInput("retired Backend Job input cannot be restored or reclassified".into())); }
let mut metadata = job.request.clone();
metadata.input = serde_json::Value::Null;
tx.execute("UPDATE backend_jobs SET request_json=?3, input_disposition=?4 WHERE workspace_id=?1 AND job_id=?2", params![workspace_id,job_id,serde_json::to_string(&metadata).map_err(|e| Error::Store(e.to_string()))?,desired.as_str()])?;
let job = read_backend_job(&tx, workspace_id, job_id)?.ok_or_else(|| Error::Store("retired Backend Job disappeared".into()))?;
tx.commit()?;
reclaim_backend_job_input_storage(conn)?;
Ok(job)
})
}
fn reserve_backend_job_retry(
&self,
workspace_id: &str,
@@ -7498,6 +7562,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let mut job = read_backend_job(&tx, workspace_id, job_id)?
.ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job `{job_id}`")))?;
job.load_executable_request()?;
let previous_id = attempt_id(job_id, job.current_attempt);
let previous = read_backend_job_attempt(&tx, workspace_id, job_id, &previous_id)?
.ok_or_else(|| Error::Store("Backend Job current attempt is missing".to_string()))?;
@@ -7537,7 +7602,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
tx.execute(
"INSERT INTO backend_job_attempts (workspace_id, job_id, attempt_id, attempt, input_digest, state, deadline_at, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, 'reserved', ?6, ?7, ?7)",
params![workspace_id, job_id, next_id, next as i64, job.request.input_digest()?, deadline, now],
params![workspace_id, job_id, next_id, next as i64, job.input_digest, deadline, now],
)?;
job = read_backend_job(&tx, workspace_id, job_id)?
.ok_or_else(|| Error::Store("retried Backend Job is missing".to_string()))?;
@@ -7576,7 +7641,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
"SELECT job.job_id FROM backend_jobs job JOIN backend_job_attempts attempt
ON attempt.workspace_id = job.workspace_id AND attempt.job_id = job.job_id
AND attempt.attempt = job.current_attempt
WHERE job.workspace_id = ?1 AND job.state = 'pending'
WHERE job.workspace_id = ?1 AND job.state = 'pending' AND job.input_disposition = 'available'
AND attempt.state = 'dispatched' AND attempt.runtime_id = ?2 AND attempt.worker_id = ?3
AND attempt.runtime_run_id IS NOT NULL",
params![workspace_id, worker.runtime_id.as_str(), worker.worker_id.as_str()],
@@ -7636,6 +7701,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let job = read_backend_job(&tx, workspace_id, job_id)?
.ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job `{job_id}`")))?;
job.load_executable_request()?;
let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)?
.ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`")))?;
if attempt.state != BackendJobAttemptState::Reserved {
@@ -7673,7 +7739,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Vec<BackendJobAttemptRecord>> {
self.with_conn(|conn| {
let mut statement = conn.prepare(
"SELECT job_id, attempt_id FROM backend_job_attempts WHERE workspace_id = ?1 AND state = 'reserved' ORDER BY created_at, attempt_id LIMIT ?2",
"SELECT a.job_id, a.attempt_id FROM backend_job_attempts a JOIN backend_jobs j ON j.workspace_id=a.workspace_id AND j.job_id=a.job_id WHERE a.workspace_id=?1 AND a.state='reserved' AND j.input_disposition='available' ORDER BY a.created_at, a.attempt_id LIMIT ?2",
)?;
let identities = statement.query_map(params![workspace_id, limit as i64], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
@@ -7698,9 +7764,10 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let job = read_backend_job(&tx, workspace_id, job_id)?
.ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job `{job_id}`")))?;
job.load_executable_request()?;
let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)?
.ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`")))?;
if attempt.attempt != job.current_attempt || attempt.input_digest != job.request.input_digest()? {
if attempt.attempt != job.current_attempt || attempt.input_digest != job.input_digest {
return Err(Error::InvalidInput(
"Backend Job attempt is stale for the current input digest".to_string(),
));
@@ -7759,6 +7826,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let job = read_backend_job(&tx, workspace_id, job_id)?.ok_or_else(|| Error::InvalidInput("Backend Job missing".into()))?;
job.load_executable_request()?;
let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)?
.ok_or_else(|| {
Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`"))
@@ -7812,7 +7881,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
let attempt = read_backend_job_attempt(&tx, workspace_id, &submission.job_id, &submission.attempt_id)?
.ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job attempt `{}`", submission.attempt_id)))?;
let (digest, result_json) = result_digest(&submission.result, job.request.limits.max_result_bytes)?;
if submission.input_digest != job.request.input_digest()?
if submission.input_digest != job.input_digest
|| attempt.input_digest != submission.input_digest
|| attempt.attempt != job.current_attempt
{
@@ -7839,6 +7908,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
tx.commit()?;
return Ok(BackendJobResultAcceptance { job, attempt, replayed: true });
}
job.load_executable_request()?;
if attempt.state != BackendJobAttemptState::Dispatched || job.state != BackendJobState::Pending {
return Err(Error::InvalidInput(
"Backend Job attempt no longer accepts results".to_string(),
@@ -7906,7 +7976,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
// Full snapshot CAS includes Worker, tracked Run, input and cleanup evidence.
if &attempt != expected || job.current_attempt != attempt.attempt
|| job.request.grants.subjektiv_journal.is_none()
|| attempt.input_digest != job.request.input_digest()?
|| attempt.input_digest != job.input_digest
|| !matches!(job.state, BackendJobState::Unknown | BackendJobState::Completed | BackendJobState::Failed) {
return Err(Error::InvalidInput("Journal reconciliation binding changed".into()));
}
@@ -12248,6 +12318,20 @@ fn read_memory_staging_resolution_record(
})
}
// The receipt/disposition commit precedes physical cleanup. An error here is
// safely retried with the same expected digest/reason, not a new execution.
fn reclaim_backend_job_input_storage(conn: &Connection) -> Result<()> {
conn.execute_batch("PRAGMA secure_delete=ON; VACUUM;")?;
let busy: i64 = conn.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |r| r.get(0))?;
if busy != 0 {
return Err(Error::Store(
"Backend Job input retired; SQLite body cleanup blocked by a reader, retry retirement"
.into(),
));
}
Ok(())
}
fn backend_job_deadline(now: &str, timeout_seconds: u32) -> Result<String> {
let now = chrono::DateTime::parse_from_rfc3339(now)
.map_err(|error| Error::InvalidInput(format!("invalid Backend Job timestamp: {error}")))?;
@@ -12311,7 +12395,7 @@ pub(crate) fn terminal_journal_job_worker(
Ok(job.request.grants.subjektiv_journal.is_some()
&& attempt.state.terminal()
&& attempt.worker.as_ref() == Some(worker)
&& attempt.input_digest == job.request.input_digest()?)
&& attempt.input_digest == job.input_digest)
}
fn read_backend_job(
@@ -12320,13 +12404,13 @@ fn read_backend_job(
job_id: &str,
) -> Result<Option<BackendJobRecord>> {
let row = conn.query_row(
"SELECT request_json, intent_fingerprint, state, current_attempt, result_json, result_digest, failure_category, failure_detail, created_at, updated_at, completed_at FROM backend_jobs WHERE workspace_id = ?1 AND job_id = ?2",
"SELECT request_json, intent_fingerprint, state, current_attempt, result_json, result_digest, failure_category, failure_detail, created_at, updated_at, completed_at, input_digest, input_disposition FROM backend_jobs WHERE workspace_id = ?1 AND job_id = ?2",
params![workspace_id, job_id],
|row| Ok((
row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?,
row.get::<_, i64>(3)?, row.get::<_, Option<String>>(4)?, row.get::<_, Option<String>>(5)?,
row.get::<_, Option<String>>(6)?, row.get::<_, Option<String>>(7)?, row.get::<_, String>(8)?,
row.get::<_, String>(9)?, row.get::<_, Option<String>>(10)?,
row.get::<_, String>(9)?, row.get::<_, Option<String>>(10)?, row.get::<_, String>(11)?, row.get::<_, String>(12)?,
)),
).optional()?;
let Some((
@@ -12341,11 +12425,20 @@ fn read_backend_job(
created_at,
updated_at,
completed_at,
input_digest,
input_disposition,
)) = row
else {
return Ok(None);
};
let request: BackendJobRequest = serde_json::from_str(&request_json)
let request_value: serde_json::Value = serde_json::from_str(&request_json)
.map_err(|error| Error::Store(format!("invalid stored Backend Job request: {error}")))?;
if request_value.get("input").is_none() {
return Err(Error::Store(
"stored Backend Job input field is missing".into(),
));
}
let request: BackendJobRequest = serde_json::from_value(request_value)
.map_err(|error| Error::Store(format!("invalid stored Backend Job request: {error}")))?;
let result = result_json
.map(|value| serde_json::from_str(&value))
@@ -12357,6 +12450,9 @@ fn read_backend_job(
workspace_id: workspace_id.to_string(),
request,
intent_fingerprint,
input_digest,
input_disposition: serde_json::from_value(serde_json::Value::String(input_disposition))
.map_err(|e| Error::Store(format!("invalid Job input disposition: {e}")))?,
state: BackendJobState::parse(&state)?,
current_attempt,
result,
@@ -15285,6 +15381,91 @@ fn migrate_passkey_results_v102_to_v103(conn: &Connection) -> Result<()> {
Ok(())
}
fn migrate_job_input_retention_v102_to_v103(conn: &Connection) -> Result<()> {
if current_schema_version(conn)? != 102 {
return Err(Error::Store(
"Job input retention migration requires schema 102".into(),
));
}
let tx = conn.unchecked_transaction()?;
tx.execute_batch("ALTER TABLE backend_jobs ADD COLUMN input_disposition TEXT NOT NULL DEFAULT 'available' CHECK (input_disposition IN ('available','reclaimed','erased','invalidated'));")?;
tx.execute_batch(crate::backend_job::INPUT_RETENTION_TRIGGERS)?;
tx.execute(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(103,?1)",
["Job input retirement independent of immutable receipts"],
)?;
verify_workspace_resource_constraints(&tx)?;
tx.commit()?;
Ok(())
}
fn migrate_schema_103_variants_to_v104(conn: &Connection) -> Result<()> {
if current_schema_version(conn)? != 103 {
return Err(Error::Store(
"Schema convergence migration requires schema 103".into(),
));
}
verify_schema_history(conn, 103)?;
let tx = conn.unchecked_transaction()?;
// Fresh databases on either branch recorded only a schema-103 baseline, so
// the receipt alone cannot identify which feature needs to be installed.
// Never rebuild existing metadata or receipts: key references and retired
// Job input dispositions are already externally observable saved values.
let passkey_results = table_exists(&tx, "passkey_metadata")?;
let job_retention = column_exists(&tx, "backend_jobs", "input_disposition")?;
let identity: String = tx.query_row(
"SELECT name FROM __yoi_schema_migrations WHERE version=103",
[],
|row| row.get(0),
)?;
let matches_identity = match identity.as_str() {
SCHEMA_BASELINE_NAME => passkey_results != job_retention,
"Passkey metadata and atomic addition results" => passkey_results && !job_retention,
"Job input retirement independent of immutable receipts" => {
job_retention && !passkey_results
}
_ => false,
};
if !matches_identity {
return Err(Error::Store(
"Schema-103 features do not match the applied migration identity".into(),
));
}
if !passkey_results {
tx.execute_batch(include_str!("passkey_results_schema.sql"))?;
}
if !job_retention {
tx.execute_batch("ALTER TABLE backend_jobs ADD COLUMN input_disposition TEXT NOT NULL DEFAULT 'available' CHECK (input_disposition IN ('available','reclaimed','erased','invalidated'));")?;
tx.execute_batch(crate::backend_job::INPUT_RETENTION_TRIGGERS)?;
}
let objects: i64 = tx.query_row(
"SELECT count(*) FROM sqlite_schema WHERE
(type='table' AND name IN ('passkey_metadata','passkey_addition_receipts')) OR
(type='trigger' AND name IN ('passkey_metadata_insert','backend_job_identity_immutable','backend_job_input_retirement'))",
[],
|row| row.get(0),
)?;
if objects != 5 {
return Err(Error::Store("Incomplete schema-103 feature schema".into()));
}
verify_workspace_resource_constraints(&tx)?;
let violations: i64 =
tx.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| {
row.get(0)
})?;
if violations != 0 {
return Err(Error::Store(
"Schema convergence foreign-key violation".into(),
));
}
tx.execute(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(104,?1)",
["Converge Passkey results and Job input retention"],
)?;
tx.commit()?;
Ok(())
}
fn migrate_subject_create_recovery_v101_to_v102(conn: &Connection) -> Result<()> {
if current_schema_version(conn)? != 101 {
return Err(Error::Store(
@@ -16248,6 +16429,17 @@ fn verify_schema_history(conn: &Connection, current: i64) -> Result<()> {
.iter()
.filter(|migration| migration.version > *baseline_version && migration.version <= current)
{
// The two published schema-103 migrations are alternatives, not aliases.
// Match the exact saved identity while retaining every other history check.
let migration = if migration.version == JOB_INPUT_RETENTION_SCHEMA_103.version
&& rows.iter().any(|(version, name)| {
*version == JOB_INPUT_RETENTION_SCHEMA_103.version
&& name == JOB_INPUT_RETENTION_SCHEMA_103.name
}) {
&JOB_INPUT_RETENTION_SCHEMA_103
} else {
migration
};
expected.push((migration.version, migration.name.to_string()));
}
if rows != expected || expected.last().map(|row| row.0) != Some(current) {
@@ -18658,6 +18850,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
include!("activation_tests.rs");
include!("schema_equivalence_tests.rs");
include!("schema_convergence_tests.rs");
include!("ticket_tag_upgrade_tests.rs");
include!("ticket_post_upgrade_tests.rs");
include!("migration_entrypoint_tests.rs");
@@ -18690,7 +18883,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
assert!(conn.execute("INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at) VALUES('invalid','missing','Invalid','active','1','1')", []).is_err());
}
fn prepare_retained_schema(path: &Path, version: i64) {
pub(super) fn prepare_retained_schema(path: &Path, version: i64) {
// Cache only empty preconditions, built from frozen DDL and real retained
// migrations. The test's seeded rows and migration under test are never shared.
static TEMPLATES: std::sync::OnceLock<Mutex<std::collections::BTreeMap<i64, Connection>>> =
@@ -19071,6 +19264,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
WorkspaceSchemaMigrationStep {version:101, name:"Durable explicit Subject body release".into()},
WorkspaceSchemaMigrationStep {version:102, name:"Durable failed Subject creation recovery".into()},
WorkspaceSchemaMigrationStep {version:103, name:"Passkey metadata and atomic addition results".into()},
WorkspaceSchemaMigrationStep {version:104, name:"Converge Passkey results and Job input retention".into()},
]
);
@@ -19184,6 +19378,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
(101, "Durable explicit Subject body release".to_string()),
(102, "Durable failed Subject creation recovery".to_string()),
(103, "Passkey metadata and atomic addition results".to_string()),
(104, "Converge Passkey results and Job input retention".to_string()),
]
);
assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -22468,7 +22663,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
(100, "Provisional users and scoped passkey activation".into()),
(101, "Durable explicit Subject body release".into()),
(102, "Durable failed Subject creation recovery".into()),
(103, "Passkey metadata and atomic addition results".into())]);
(103, "Passkey metadata and atomic addition results".into()),
(104, "Converge Passkey results and Job input retention".into())]);
for (table, rows) in unchanged.iter().zip(before.iter()) {
assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
}
@@ -26386,6 +26582,326 @@ INSERT INTO worker_registry (
.unwrap();
}
// Storage component tests: seed already-saved result/cleanup receipts, without
// constructing an execution Worker or claiming a model/dispatch integration.
fn seed_completed_job_receipts(store: &SqliteWorkspaceStore, id: &str, cleanup: &str) {
let result = serde_json::json!({"stored":true});
let (digest, json) = result_digest(&result, job::ABSOLUTE_MAX_RESULT_BYTES).unwrap();
store.with_conn(|conn| {
conn.execute("UPDATE backend_job_attempts SET state='completed',result_json=?2,result_digest=?3,worker_cleanup_state=?4,worker_cleanup_completed_at=?5 WHERE job_id=?1", params![id,json,digest,cleanup,RESOURCE_JOB_NOW])?;
conn.execute("UPDATE backend_jobs SET state='completed',result_json=?2,result_digest=?3,completed_at=?4 WHERE job_id=?1", params![id,json,digest,RESOURCE_JOB_NOW])?;
Ok(())
}).unwrap();
}
#[tokio::test]
async fn backend_job_input_reclaim_preserves_receipts_reopen_and_exact_replay() {
use crate::backend_job::{JobInputDisposition, JobInputRetirementReason};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("input.sqlite");
let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap();
seed_resource_job_workspace(&store).await;
let original = resource_job_request("reclaim", "subject-a", "candidate");
let reservation = store
.reserve_backend_job("resource-workspace", &original, RESOURCE_JOB_NOW)
.unwrap();
assert!(
store
.reclaim_backend_job_input(
"resource-workspace",
"reclaim",
&reservation.job.input_digest,
JobInputRetirementReason::Completed
)
.is_err()
);
seed_completed_job_receipts(&store, "reclaim", "pending");
assert!(
store
.reclaim_backend_job_input(
"resource-workspace",
"reclaim",
&reservation.job.input_digest,
JobInputRetirementReason::Completed
)
.is_err()
);
store.with_conn(|conn| { conn.execute("UPDATE backend_job_attempts SET worker_cleanup_state='completed' WHERE job_id='reclaim'",[])?; Ok(()) }).unwrap();
let before = store
.get_backend_job("resource-workspace", "reclaim")
.unwrap()
.unwrap();
let attempt = store
.get_backend_job_attempt(
"resource-workspace",
"reclaim",
&reservation.attempt.attempt_id,
)
.unwrap()
.unwrap();
assert!(
store
.reclaim_backend_job_input(
"resource-workspace",
"reclaim",
"wrong",
JobInputRetirementReason::Completed
)
.is_err()
);
let retired = store
.reclaim_backend_job_input(
"resource-workspace",
"reclaim",
&before.input_digest,
JobInputRetirementReason::Completed,
)
.unwrap();
assert_eq!(retired.input_disposition, JobInputDisposition::Reclaimed);
assert!(retired.request.input.is_null());
assert!(retired.load_executable_request().is_err());
assert_eq!(retired.input_digest, original.input_digest().unwrap());
assert_eq!(retired.intent_fingerprint, before.intent_fingerprint);
assert_eq!(retired.request.grants, original.grants);
assert_eq!(retired.result, before.result);
assert_eq!(
store
.get_backend_job_attempt("resource-workspace", "reclaim", &attempt.attempt_id)
.unwrap()
.unwrap(),
attempt
);
drop(store);
let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let replay = reopened
.reserve_backend_job("resource-workspace", &original, RESOURCE_JOB_NOW)
.unwrap();
assert!(replay.replayed);
assert_eq!(replay.job, retired);
assert_eq!(
replay.outcome().input_digest,
original.input_digest().unwrap()
);
let mut changed = original.clone();
changed.input = serde_json::json!({"changed":true});
assert!(
reopened
.reserve_backend_job("resource-workspace", &changed, RESOURCE_JOB_NOW)
.is_err()
);
assert!(
reopened
.reserve_backend_job_retry("resource-workspace", "reclaim", RESOURCE_JOB_NOW)
.is_err()
);
assert!(
reopened
.claim_backend_job_attempt_dispatch(
"resource-workspace",
"reclaim",
&attempt.attempt_id,
RESOURCE_JOB_NOW
)
.is_err()
);
assert!(reopened.with_conn(|conn| { conn.execute("UPDATE backend_jobs SET input_disposition='available',request_json=?1 WHERE job_id='reclaim'", [serde_json::to_string(&original).unwrap()])?; Ok(()) }).is_err());
let erased = reopened
.reclaim_backend_job_input(
"resource-workspace",
"reclaim",
&retired.input_digest,
JobInputRetirementReason::ExplicitErasure,
)
.unwrap();
assert_eq!(erased.input_disposition, JobInputDisposition::Erased);
assert_eq!(erased.state, BackendJobState::Completed);
assert_eq!(erased.result, before.result);
}
#[tokio::test]
async fn backend_job_input_unknown_retention_and_legacy_invalidation_do_not_reinterpret_outcome()
{
use crate::backend_job::{JobInputDisposition, JobInputRetirementReason};
let store = SqliteWorkspaceStore::in_memory_fixture().unwrap();
seed_resource_job_workspace(&store).await;
for (id, state) in [
("unfinished", None),
("unknown-input", Some(BackendJobAttemptState::Unknown)),
("failed-input", Some(BackendJobAttemptState::Failed)),
] {
let original = resource_job_request(id, id, "candidate");
let reservation = store
.reserve_backend_job("resource-workspace", &original, RESOURCE_JOB_NOW)
.unwrap();
if let Some(state) = state {
store
.finish_backend_job_attempt(
"resource-workspace",
id,
&reservation.attempt.attempt_id,
state,
"fixture",
"known stored outcome",
RESOURCE_JOB_NOW,
)
.unwrap();
}
let before = store
.get_backend_job("resource-workspace", id)
.unwrap()
.unwrap();
assert!(
store
.reclaim_backend_job_input(
"resource-workspace",
id,
&before.input_digest,
JobInputRetirementReason::Completed
)
.is_err()
);
assert_eq!(before.load_executable_request().unwrap(), &original);
assert_eq!(
store
.get_backend_job("resource-workspace", id)
.unwrap()
.unwrap(),
before
);
let retired = store
.reclaim_backend_job_input(
"resource-workspace",
id,
&before.input_digest,
JobInputRetirementReason::LegacyInvalidation,
)
.unwrap();
assert_eq!(retired.input_disposition, JobInputDisposition::Invalidated);
assert_eq!(retired.state, before.state);
assert_eq!(retired.failure_category, before.failure_category);
assert_eq!(retired.completed_at, before.completed_at);
assert!(retired.load_executable_request().is_err());
assert!(
store
.reserve_backend_job_retry("resource-workspace", id, RESOURCE_JOB_NOW)
.is_err()
);
assert!(
store
.claim_backend_job_attempt_dispatch(
"resource-workspace",
id,
&reservation.attempt.attempt_id,
RESOURCE_JOB_NOW
)
.is_err()
);
assert_eq!(
store
.reserve_backend_job("resource-workspace", &original, RESOURCE_JOB_NOW)
.unwrap()
.job,
retired
);
}
}
#[tokio::test]
async fn backend_job_input_available_loader_and_dispatch_refuse_missing_tampered_or_changed_intent()
{
for mutation in [
"json_set(request_json,'$.input',json('{\"tampered\":true}'))",
"json_remove(request_json,'$.input')",
"json_set(request_json,'$.instruction','changed')",
] {
let store = SqliteWorkspaceStore::in_memory_fixture().unwrap();
seed_resource_job_workspace(&store).await;
let original = resource_job_request("tamper", "subject", "candidate");
let reservation = store
.reserve_backend_job("resource-workspace", &original, RESOURCE_JOB_NOW)
.unwrap();
store
.with_conn(|conn| {
conn.execute_batch("DROP TRIGGER backend_job_input_retirement")?;
conn.execute_batch(&format!(
"UPDATE backend_jobs SET request_json={mutation}"
))?;
Ok(())
})
.unwrap();
if let Ok(Some(job)) = store.get_backend_job("resource-workspace", "tamper") {
assert!(job.load_executable_request().is_err());
}
assert!(
store
.claim_backend_job_attempt_dispatch(
"resource-workspace",
"tamper",
&reservation.attempt.attempt_id,
RESOURCE_JOB_NOW
)
.is_err()
);
}
}
#[tokio::test]
async fn backend_job_input_migration_preserves_original_identity_and_rolls_back_failure() {
use crate::backend_job::{JobInputDisposition, JobInputRetirementReason};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("migration.sqlite");
let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap();
seed_resource_job_workspace(&store).await;
let original = resource_job_request("upgrade", "subject", "candidate");
let before = store
.reserve_backend_job("resource-workspace", &original, RESOURCE_JOB_NOW)
.unwrap();
store.with_conn(|conn| {
conn.execute_batch("DROP TRIGGER backend_job_input_retirement; DROP TRIGGER backend_job_identity_immutable;
ALTER TABLE backend_jobs DROP COLUMN input_disposition;
DROP TRIGGER passkey_metadata_insert;
DROP TABLE passkey_addition_receipts;
DROP TABLE passkey_metadata;
UPDATE __yoi_schema_migrations SET version=102 WHERE version=104;
CREATE TRIGGER backend_job_input_retirement BEFORE UPDATE ON backend_job_attempts BEGIN SELECT RAISE(ABORT,'collision'); END;")?; Ok(())
}).unwrap();
drop(store);
let conn = Connection::open(&path).unwrap();
configure_sqlite(&conn).unwrap();
assert!(migrate_job_input_retention_v102_to_v103(&conn).is_err());
assert_eq!(current_schema_version(&conn).unwrap(), 102);
assert!(!column_exists(&conn, "backend_jobs", "input_disposition").unwrap());
conn.execute_batch("DROP TRIGGER backend_job_input_retirement")
.unwrap();
migrate_job_input_retention_v102_to_v103(&conn).unwrap();
drop(conn);
let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap();
assert_eq!(
reopened
.get_backend_job("resource-workspace", "upgrade")
.unwrap()
.unwrap(),
before.job
);
let erased = reopened
.reclaim_backend_job_input(
"resource-workspace",
"upgrade",
&before.job.input_digest,
JobInputRetirementReason::ExplicitErasure,
)
.unwrap();
assert_eq!(erased.input_disposition, JobInputDisposition::Erased);
assert_eq!(erased.state, BackendJobState::Pending);
assert_eq!(
reopened
.reserve_backend_job("resource-workspace", &original, RESOURCE_JOB_NOW)
.unwrap()
.job,
erased
);
}
#[tokio::test]
async fn backend_job_resource_reserve_converges_across_connections_without_changing_input() {
let dir = tempfile::tempdir().unwrap();
@@ -231,12 +231,10 @@ pub fn result_facts(
current_ticket: &ticket::Ticket,
) -> Result<TicketItemCheckFacts> {
let input = parse_input(&acceptance.job.request.input)?;
let result = validate_result(
&acceptance.job.request.input,
acceptance.job.result.as_ref().ok_or_else(|| {
Error::InvalidInput("completed Ticket item checker Job has no result".to_string())
})?,
)?;
let result_value = acceptance.job.result.as_ref().ok_or_else(|| {
Error::InvalidInput("completed Ticket item checker Job has no result".to_string())
})?;
let result = validate_result(&acceptance.job.request.input, result_value)?;
if input.content_digest != content_digest(current_ticket)
|| input.ticket_id != current_ticket.meta.id
{
@@ -246,21 +244,17 @@ pub fn result_facts(
return Ok(TicketItemCheckFacts::NoFindings);
}
Ok(TicketItemCheckFacts::Findings(render_result(acceptance)?))
Ok(TicketItemCheckFacts::Findings(render_result(
&acceptance.job.request.input,
result_value,
)?))
}
/// Stable rendering from the original checked snapshot. Used to pin pre-upgrade
/// unacknowledged requests without reinterpreting them against newer Ticket text.
pub(crate) fn render_result(acceptance: &BackendJobResultAcceptance) -> Result<String> {
let input = parse_input(&acceptance.job.request.input)?;
let result = validate_result(
&acceptance.job.request.input,
acceptance
.job
.result
.as_ref()
.ok_or_else(|| Error::InvalidInput("completed checker has no result".into()))?,
)?;
pub(crate) fn render_result(input: &Value, result: &Value) -> Result<String> {
let checked = parse_input(input)?;
let result = validate_result(input, result)?;
if result.findings.is_empty() {
return Err(Error::InvalidInput(
"clean checker result has no advisory body".into(),
@@ -268,7 +262,7 @@ pub(crate) fn render_result(acceptance: &BackendJobResultAcceptance) -> Result<S
}
let mut content = format!(
"[Ticket item checker advisory]\nTicket: {}\nContent digest: {}\n\nThis is a post-save advisory based only on the inspected Ticket text. It is not a new user request, approval gate, workflow blocker, or implementation instruction.\n",
input.ticket_resource_key, input.content_digest
checked.ticket_resource_key, checked.content_digest
);
for (index, finding) in result.findings.iter().enumerate() {
let quoted = serde_json::to_string(&finding.quote).map_err(|error| {
@@ -467,6 +461,8 @@ mod tests {
job: BackendJobRecord {
workspace_id: "checker-unit".into(),
intent_fingerprint: request.fingerprint().unwrap(),
input_digest: request.input_digest().unwrap(),
input_disposition: crate::backend_job::JobInputDisposition::Available,
request,
state: BackendJobState::Completed,
current_attempt: 1,