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

server: compose Workspace config schemas

Browse Source
This commit is contained in:
Hare
2026-08-14 05:38:21 +09:00
parent 7a8bd7717e
commit f8baa1edb7
9 changed files with 869 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/config_source.rs
+333 -67
View File
@@ -1,8 +1,8 @@
use chrono::{SecondsFormat, Utc};
use config_source::{
ConfigContentType, ConfigEntry, ConfigTreeChange, ConfigTreeSnapshot, DECODAL_VERSION,
DEFAULT_IMPORT_POLICY_VERSION, DEFAULT_SCHEMA_VERSION, EvaluationResult, SnapshotEnvironment,
ToolchainContract, VirtualPath,
ConfigContentType, ConfigEntry, ConfigSchemaContribution, ConfigTreeChange, ConfigTreeSnapshot,
DECODAL_VERSION, DEFAULT_IMPORT_POLICY_VERSION, DEFAULT_SCHEMA_VERSION, EvaluationResult,
SnapshotEnvironment, ToolchainContract, VirtualPath, WorkspaceConfigSchemaBundle,
};
use rusqlite::{OptionalExtension, TransactionBehavior, params};
use serde::{Deserialize, Serialize};
@@ -16,14 +16,50 @@ fn main_config_path() -> VirtualPath {
VirtualPath::parse(MAIN_CONFIG_ENTRYPOINT).expect("main config entrypoint is a valid path")
}
fn main_config_contract() -> ToolchainContract {
ToolchainContract::new(
pub trait WorkspaceConfigSchemaProvider: Send + Sync {
fn contribution(&self) -> Result<ConfigSchemaContribution>;
}
#[derive(Clone, Default)]
pub struct WorkspaceConfigSchemaRegistry {
providers: Vec<std::sync::Arc<dyn WorkspaceConfigSchemaProvider>>,
}
impl WorkspaceConfigSchemaRegistry {
pub fn with_provider(
mut self,
provider: std::sync::Arc<dyn WorkspaceConfigSchemaProvider>,
) -> Self {
self.providers.push(provider);
self
}
pub fn compose(&self) -> Result<WorkspaceConfigSchemaBundle> {
WorkspaceConfigSchemaBundle::compose(
self.providers
.iter()
.map(|provider| provider.contribution())
.collect::<Result<Vec<_>>>()?,
)
.map_err(config_error)
}
}
fn main_config_contract_with_schema(
schema_bundle: WorkspaceConfigSchemaBundle,
) -> ToolchainContract {
ToolchainContract::with_schema_bundle(
DEFAULT_SCHEMA_VERSION,
vec![main_config_path()],
DEFAULT_IMPORT_POLICY_VERSION,
schema_bundle,
)
}
fn main_config_contract() -> ToolchainContract {
main_config_contract_with_schema(WorkspaceConfigSchemaBundle::empty())
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, ts_rs::TS)]
#[ts(export)]
pub struct WorkspaceConfigState {
@@ -127,10 +163,11 @@ impl SqliteWorkspaceStore {
})
}
pub fn evaluate_workspace_config_candidate(
pub fn evaluate_workspace_config_candidate_with_schema(
&self,
workspace_id: &str,
request: &ConfigCommitRequest,
schema_bundle: WorkspaceConfigSchemaBundle,
) -> Result<EvaluatedConfigCandidate> {
let current = self
.load_workspace_config(workspace_id)?
@@ -144,14 +181,39 @@ impl SqliteWorkspaceStore {
current.snapshot.revision
)));
}
let expected_contract = main_config_contract();
let expected_contract = main_config_contract_with_schema(schema_bundle.clone());
if expected_contract.fingerprint != request.toolchain_fingerprint {
return Err(config_conflict(format!(
"toolchain fingerprint mismatch; current fingerprint is {}",
expected_contract.fingerprint
)));
}
evaluate_candidate(current, &request.changes)
evaluate_candidate(current, &request.changes, schema_bundle)
}
pub fn evaluate_workspace_config_candidate(
&self,
workspace_id: &str,
request: &ConfigCommitRequest,
) -> Result<EvaluatedConfigCandidate> {
self.evaluate_workspace_config_candidate_with_schema(
workspace_id,
request,
WorkspaceConfigSchemaBundle::empty(),
)
}
pub fn preview_workspace_config_with_schema(
&self,
workspace_id: &str,
request: &ConfigPreviewRequest,
schema_bundle: WorkspaceConfigSchemaBundle,
) -> Result<EvaluatedConfigCandidate> {
validate_entrypoint_request(&request.entrypoints)?;
let current = self
.load_workspace_config(workspace_id)?
.ok_or_else(config_not_materialized)?;
evaluate_candidate(current, &request.changes, schema_bundle)
}
pub fn preview_workspace_config(
@@ -159,11 +221,11 @@ impl SqliteWorkspaceStore {
workspace_id: &str,
request: &ConfigPreviewRequest,
) -> Result<EvaluatedConfigCandidate> {
validate_entrypoint_request(&request.entrypoints)?;
let current = self
.load_workspace_config(workspace_id)?
.ok_or_else(config_not_materialized)?;
evaluate_candidate(current, &request.changes)
self.preview_workspace_config_with_schema(
workspace_id,
request,
WorkspaceConfigSchemaBundle::empty(),
)
}
pub fn commit_evaluated_workspace_config(
@@ -197,9 +259,9 @@ impl SqliteWorkspaceStore {
tx.execute(
r#"INSERT INTO workspace_config_trees (
workspace_id, revision, tree_digest, schema_version, entrypoints_json,
decodal_version, import_policy_version, toolchain_fingerprint,
projection_digest, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
decodal_version, import_policy_version, schema_bundle_json,
toolchain_fingerprint, projection_digest, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
ON CONFLICT(workspace_id) DO UPDATE SET
revision = excluded.revision,
tree_digest = excluded.tree_digest,
@@ -207,6 +269,7 @@ impl SqliteWorkspaceStore {
entrypoints_json = excluded.entrypoints_json,
decodal_version = excluded.decodal_version,
import_policy_version = excluded.import_policy_version,
schema_bundle_json = excluded.schema_bundle_json,
toolchain_fingerprint = excluded.toolchain_fingerprint,
projection_digest = excluded.projection_digest,
updated_at = excluded.updated_at"#,
@@ -219,6 +282,8 @@ impl SqliteWorkspaceStore {
.map_err(|error| Error::Store(error.to_string()))?,
candidate.contract.decodal_version,
candidate.contract.import_policy_version,
serde_json::to_string(&candidate.contract.schema_bundle)
.map_err(|error| Error::Store(error.to_string()))?,
candidate.contract.fingerprint,
candidate.evaluation.projection_digest,
now,
@@ -247,13 +312,15 @@ impl SqliteWorkspaceStore {
tx.execute(
r#"INSERT INTO workspace_config_tree_revisions (
workspace_id, revision, tree_digest, toolchain_fingerprint,
projection_digest, manifest_json, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"#,
schema_bundle_json, projection_digest, manifest_json, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)"#,
params![
workspace_id,
next_revision as i64,
snapshot.digest,
candidate.contract.fingerprint,
serde_json::to_string(&candidate.contract.schema_bundle)
.map_err(|error| Error::Store(error.to_string()))?,
candidate.evaluation.projection_digest,
manifest_json,
now,
@@ -281,11 +348,12 @@ impl SqliteWorkspaceStore {
fn evaluate_candidate(
current: WorkspaceConfigState,
changes: &[ConfigTreeChange],
schema_bundle: WorkspaceConfigSchemaBundle,
) -> Result<EvaluatedConfigCandidate> {
reject_main_entrypoint_mutation(changes)?;
let snapshot = current.snapshot.apply(changes).map_err(config_error)?;
ensure_main_entrypoint(&snapshot)?;
let contract = main_config_contract();
let contract = main_config_contract_with_schema(schema_bundle);
let evaluation = SnapshotEnvironment::new(snapshot.clone())
.evaluate_contract(&contract)
.map_err(|diagnostics| {
@@ -307,10 +375,19 @@ pub(crate) fn load_state(
conn: &rusqlite::Connection,
workspace_id: &str,
) -> Result<Option<WorkspaceConfigState>> {
let header = conn
.query_row(
let has_schema_bundle: bool = conn.query_row(
"SELECT EXISTS(
SELECT 1 FROM pragma_table_info('workspace_config_trees')
WHERE name = 'schema_bundle_json'
)",
[],
|row| row.get(0),
)?;
let header = if has_schema_bundle {
conn.query_row(
r#"SELECT revision, tree_digest, schema_version, entrypoints_json,
decodal_version, import_policy_version, toolchain_fingerprint, projection_digest
decodal_version, import_policy_version, schema_bundle_json,
toolchain_fingerprint, projection_digest
FROM workspace_config_trees WHERE workspace_id = ?1"#,
[workspace_id],
|row| {
@@ -321,12 +398,36 @@ pub(crate) fn load_state(
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, u32>(5)?,
Some(row.get::<_, String>(6)?),
row.get::<_, String>(7)?,
row.get::<_, String>(8)?,
))
},
)
.optional()?
} else {
conn.query_row(
r#"SELECT revision, tree_digest, schema_version, entrypoints_json,
decodal_version, import_policy_version,
toolchain_fingerprint, projection_digest
FROM workspace_config_trees WHERE workspace_id = ?1"#,
[workspace_id],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, u32>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, u32>(5)?,
None,
row.get::<_, String>(6)?,
row.get::<_, String>(7)?,
))
},
)
.optional()?;
.optional()?
};
let Some((
revision,
stored_digest,
@@ -334,6 +435,7 @@ pub(crate) fn load_state(
entrypoints_json,
decodal_version,
import_policy_version,
schema_bundle_json,
fingerprint,
projection_digest,
)) = header
@@ -376,7 +478,17 @@ pub(crate) fn load_state(
}
let entrypoints: Vec<VirtualPath> = serde_json::from_str(&entrypoints_json)
.map_err(|error| Error::RegistryInconsistency(error.to_string()))?;
let contract = ToolchainContract::new(schema_version, entrypoints, import_policy_version);
let schema_bundle = match schema_bundle_json {
Some(schema_bundle_json) => serde_json::from_str(&schema_bundle_json)
.map_err(|error| Error::RegistryInconsistency(error.to_string()))?,
None => WorkspaceConfigSchemaBundle::empty(),
};
let contract = ToolchainContract::with_schema_bundle(
schema_version,
entrypoints,
import_policy_version,
schema_bundle,
);
if decodal_version != DECODAL_VERSION || contract.fingerprint != fingerprint {
return Err(Error::RegistryInconsistency(format!(
"virtual config toolchain metadata mismatch for Workspace {workspace_id}"
@@ -425,35 +537,79 @@ pub(crate) fn insert_materialized_state(
.map_err(|error| Error::Store(error.to_string()))?;
let manifest_json = serde_json::to_string(&state.snapshot.entries)
.map_err(|error| Error::Store(error.to_string()))?;
tx.execute(
"INSERT INTO workspace_config_trees (
workspace_id, revision, tree_digest, schema_version, entrypoints_json,
decodal_version, import_policy_version, toolchain_fingerprint,
projection_digest, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
ON CONFLICT(workspace_id) DO UPDATE SET
revision = excluded.revision,
tree_digest = excluded.tree_digest,
schema_version = excluded.schema_version,
entrypoints_json = excluded.entrypoints_json,
decodal_version = excluded.decodal_version,
import_policy_version = excluded.import_policy_version,
toolchain_fingerprint = excluded.toolchain_fingerprint,
projection_digest = excluded.projection_digest,
updated_at = excluded.updated_at",
rusqlite::params![
workspace_id,
state.snapshot.revision,
state.snapshot.digest,
state.contract.schema_version,
entrypoints_json,
DECODAL_VERSION,
state.contract.import_policy_version,
state.contract.fingerprint,
state.projection_digest,
materialized_at,
],
let has_schema_bundle: bool = tx.query_row(
"SELECT EXISTS(
SELECT 1 FROM pragma_table_info('workspace_config_trees')
WHERE name = 'schema_bundle_json'
)",
[],
|row| row.get(0),
)?;
let schema_bundle_json = serde_json::to_string(&state.contract.schema_bundle)
.map_err(|error| Error::Store(error.to_string()))?;
if has_schema_bundle {
tx.execute(
"INSERT INTO workspace_config_trees (
workspace_id, revision, tree_digest, schema_version, entrypoints_json,
decodal_version, import_policy_version, schema_bundle_json,
toolchain_fingerprint, projection_digest, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
ON CONFLICT(workspace_id) DO UPDATE SET
revision = excluded.revision,
tree_digest = excluded.tree_digest,
schema_version = excluded.schema_version,
entrypoints_json = excluded.entrypoints_json,
decodal_version = excluded.decodal_version,
import_policy_version = excluded.import_policy_version,
schema_bundle_json = excluded.schema_bundle_json,
toolchain_fingerprint = excluded.toolchain_fingerprint,
projection_digest = excluded.projection_digest,
updated_at = excluded.updated_at",
rusqlite::params![
workspace_id,
state.snapshot.revision,
state.snapshot.digest,
state.contract.schema_version,
entrypoints_json,
DECODAL_VERSION,
state.contract.import_policy_version,
schema_bundle_json,
state.contract.fingerprint,
state.projection_digest,
materialized_at,
],
)?;
} else {
tx.execute(
"INSERT INTO workspace_config_trees (
workspace_id, revision, tree_digest, schema_version, entrypoints_json,
decodal_version, import_policy_version, toolchain_fingerprint,
projection_digest, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
ON CONFLICT(workspace_id) DO UPDATE SET
revision = excluded.revision,
tree_digest = excluded.tree_digest,
schema_version = excluded.schema_version,
entrypoints_json = excluded.entrypoints_json,
decodal_version = excluded.decodal_version,
import_policy_version = excluded.import_policy_version,
toolchain_fingerprint = excluded.toolchain_fingerprint,
projection_digest = excluded.projection_digest,
updated_at = excluded.updated_at",
rusqlite::params![
workspace_id,
state.snapshot.revision,
state.snapshot.digest,
state.contract.schema_version,
entrypoints_json,
DECODAL_VERSION,
state.contract.import_policy_version,
state.contract.fingerprint,
state.projection_digest,
materialized_at,
],
)?;
}
for entry in state.snapshot.entries.values() {
tx.execute(
"INSERT INTO workspace_config_entries (
@@ -468,21 +624,40 @@ pub(crate) fn insert_materialized_state(
],
)?;
}
tx.execute(
"INSERT INTO workspace_config_tree_revisions (
workspace_id, revision, tree_digest, toolchain_fingerprint,
projection_digest, manifest_json, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
rusqlite::params![
workspace_id,
state.snapshot.revision,
state.snapshot.digest,
state.contract.fingerprint,
state.projection_digest,
manifest_json,
materialized_at,
],
)?;
if has_schema_bundle {
tx.execute(
"INSERT INTO workspace_config_tree_revisions (
workspace_id, revision, tree_digest, toolchain_fingerprint,
schema_bundle_json, projection_digest, manifest_json, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
rusqlite::params![
workspace_id,
state.snapshot.revision,
state.snapshot.digest,
state.contract.fingerprint,
schema_bundle_json,
state.projection_digest,
manifest_json,
materialized_at,
],
)?;
} else {
tx.execute(
"INSERT INTO workspace_config_tree_revisions (
workspace_id, revision, tree_digest, toolchain_fingerprint,
projection_digest, manifest_json, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
rusqlite::params![
workspace_id,
state.snapshot.revision,
state.snapshot.digest,
state.contract.fingerprint,
state.projection_digest,
manifest_json,
materialized_at,
],
)?;
}
Ok(())
}
@@ -608,6 +783,96 @@ mod tests {
}
}
#[tokio::test]
async fn schema_registry_applies_normal_decodal_composition() {
struct WebSchema;
impl WorkspaceConfigSchemaProvider for WebSchema {
fn contribution(&self) -> Result<ConfigSchemaContribution> {
ConfigSchemaContribution::new(
"builtin:web",
"web",
"1",
"{ web = { enabled = Bool default false; }; }",
)
.map_err(config_error)
}
}
let store = open_store().await;
let current = store.load_workspace_config("w-config").unwrap().unwrap();
let main = current.snapshot.get(&path(MAIN_CONFIG_ENTRYPOINT)).unwrap();
let registry =
WorkspaceConfigSchemaRegistry::default().with_provider(std::sync::Arc::new(WebSchema));
let schema = registry.compose().unwrap();
let expected_contract = main_config_contract_with_schema(schema.clone());
let candidate = store
.evaluate_workspace_config_candidate_with_schema(
"w-config",
&ConfigCommitRequest {
base_revision: current.snapshot.revision,
base_digest: current.snapshot.digest.clone(),
changes: vec![ConfigTreeChange::Update {
path: path(MAIN_CONFIG_ENTRYPOINT),
expected_digest: main.content_digest.clone(),
content: "{ web = {}; }".to_string(),
}],
entrypoints: vec![path(MAIN_CONFIG_ENTRYPOINT)],
toolchain_fingerprint: expected_contract.fingerprint.clone(),
},
schema,
)
.unwrap();
assert_eq!(
candidate.evaluation.projections[0].data_json["web"]["enabled"],
false
);
assert_eq!(
candidate.contract.fingerprint,
expected_contract.fingerprint
);
store
.commit_evaluated_workspace_config("w-config", &candidate)
.unwrap();
assert_eq!(
store
.load_workspace_config("w-config")
.unwrap()
.unwrap()
.contract
.schema_bundle,
expected_contract.schema_bundle
);
}
#[tokio::test]
async fn commit_rejects_stale_schema_bundle_fingerprint() {
let store = open_store().await;
let current = store.load_workspace_config("w-config").unwrap().unwrap();
let schema = WorkspaceConfigSchemaBundle::compose([ConfigSchemaContribution::new(
"builtin:web",
"web",
"1",
"{ web = {}; }",
)
.unwrap()])
.unwrap();
let error = store
.evaluate_workspace_config_candidate_with_schema(
"w-config",
&ConfigCommitRequest {
base_revision: current.snapshot.revision,
base_digest: current.snapshot.digest,
changes: Vec::new(),
entrypoints: vec![path(MAIN_CONFIG_ENTRYPOINT)],
toolchain_fingerprint: current.contract.fingerprint,
},
schema,
)
.unwrap_err();
assert!(error.to_string().contains("toolchain fingerprint mismatch"));
}
#[tokio::test]
async fn workspace_materializes_main_entrypoint() {
let store = open_store().await;
@@ -806,6 +1071,7 @@ mod tests {
[],
)
.unwrap();
crate::store::persist_workspace_config_schema_bundles(&conn).unwrap();
crate::store::materialize_main_config_entrypoint(&conn).unwrap();
let state = load_state(&conn, "legacy").unwrap().unwrap();
assert!(
crates/workspace-server/src/server.rs
+23 -3
View File
@@ -255,6 +255,7 @@ pub struct WorkspaceApi {
pub(crate) config: ServerConfig,
pub(crate) store: Arc<dyn ControlPlaneStore>,
config_store: Arc<crate::SqliteWorkspaceStore>,
config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry,
authority: SqliteWorkspaceAuthority,
runtime: Arc<RuntimeRegistry>,
companion: Arc<CompanionConsole>,
@@ -643,6 +644,14 @@ impl crate::worker_source::VerifiedWorkerRemoveExecutor for WorkspaceWorkerRemov
}
impl WorkspaceApi {
pub fn with_config_schema_provider(
mut self,
provider: Arc<dyn crate::config_source::WorkspaceConfigSchemaProvider>,
) -> Self {
self.config_schema_registry = self.config_schema_registry.with_provider(provider);
self
}
pub async fn new(config: ServerConfig, store: Arc<dyn ControlPlaneStore>) -> Result<Self> {
let resource_broker = BackendResourceBroker::default();
let worker_remove_dispatcher = Arc::new(
@@ -749,6 +758,7 @@ impl WorkspaceApi {
)?);
let api = Self {
config_store,
config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry::default(),
authority: SqliteWorkspaceAuthority::new(
config.database_path.clone(),
config.workspace_id.clone(),
@@ -2528,8 +2538,11 @@ async fn scoped_preview_workspace_config_tree(
) -> ApiResult<Json<crate::config_source::EvaluatedConfigCandidate>> {
validate_workspace_scope(&api, &path.workspace_id)?;
Ok(Json(
api.config_store
.preview_workspace_config(&path.workspace_id, &request)?,
api.config_store.preview_workspace_config_with_schema(
&path.workspace_id,
&request,
api.config_schema_registry.compose()?,
)?,
))
}
@@ -2539,9 +2552,16 @@ async fn scoped_commit_workspace_config_tree(
Json(request): Json<ConfigCommitRequest>,
) -> ApiResult<(StatusCode, Json<WorkspaceConfigTreeResponse>)> {
validate_workspace_scope(&api, &path.workspace_id)?;
let candidate = api
.config_store
.evaluate_workspace_config_candidate_with_schema(
&path.workspace_id,
&request,
api.config_schema_registry.compose()?,
)?;
let state = api
.config_store
.evaluate_and_commit_workspace_config(&path.workspace_id, &request)?;
.commit_evaluated_workspace_config(&path.workspace_id, &candidate)?;
Ok((
StatusCode::CREATED,
Json(WorkspaceConfigTreeResponse {
crates/workspace-server/src/store.rs
+59 -18
View File
@@ -176,6 +176,11 @@ const MIGRATIONS: &[Migration] = &[
name: "materialize required main.dcdl Workspace config entrypoint",
apply: materialize_main_config_entrypoint,
},
Migration {
version: 32,
name: "persist Workspace config schema contribution bundles",
apply: persist_workspace_config_schema_bundles,
},
];
struct Migration {
@@ -4598,6 +4603,42 @@ fn create_workspace_config_source_authority(conn: &Connection) -> Result<()> {
Ok(())
}
pub(crate) fn persist_workspace_config_schema_bundles(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
ALTER TABLE workspace_config_trees
ADD COLUMN schema_bundle_json TEXT NOT NULL DEFAULT '{"contributions":[],"source":"{}","fingerprint":""}';
ALTER TABLE workspace_config_tree_revisions
ADD COLUMN schema_bundle_json TEXT NOT NULL DEFAULT '{"contributions":[],"source":"{}","fingerprint":""}';
"#,
)?;
let bundle = config_source::WorkspaceConfigSchemaBundle::empty();
let bundle_json =
serde_json::to_string(&bundle).map_err(|error| Error::Store(error.to_string()))?;
let contract = config_source::ToolchainContract::with_schema_bundle(
config_source::DEFAULT_SCHEMA_VERSION,
vec![
config_source::VirtualPath::parse(crate::config_source::MAIN_CONFIG_ENTRYPOINT)
.map_err(|error| Error::Store(error.to_string()))?,
],
config_source::DEFAULT_IMPORT_POLICY_VERSION,
bundle,
);
conn.execute(
"UPDATE workspace_config_trees
SET schema_bundle_json = ?1,
toolchain_fingerprint = ?2",
params![bundle_json, contract.fingerprint],
)?;
conn.execute(
"UPDATE workspace_config_tree_revisions
SET schema_bundle_json = ?1,
toolchain_fingerprint = ?2",
params![bundle_json, contract.fingerprint],
)?;
Ok(())
}
fn create_worker_mutation_source_proof_replay_guard(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
@@ -5271,7 +5312,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 31);
assert_eq!(current_schema_version(&conn).unwrap(), 32);
assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
}
@@ -5304,7 +5345,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 31);
assert_eq!(current_schema_version(&conn).unwrap(), 32);
assert!(table_exists(&conn, "flow_sources").unwrap());
assert!(table_exists(&conn, "flow_source_revisions").unwrap());
assert!(!table_exists(&conn, "flow_instances").unwrap());
@@ -5371,7 +5412,7 @@ INSERT INTO worker_workdir_attachment_reservations (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 31);
assert_eq!(current_schema_version(&conn).unwrap(), 32);
let repositories_sql: String = conn
.query_row(
"SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'",
@@ -5551,7 +5592,7 @@ INSERT INTO workdir_registry (
let db = dir.path().join("control-plane.sqlite");
let store = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 31);
assert_eq!(store.schema_version().await.unwrap(), 32);
assert!(
!store
.with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -5568,7 +5609,7 @@ INSERT INTO workdir_registry (
store.upsert_workspace(&record).await.unwrap();
let reopened = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(reopened.schema_version().await.unwrap(), 31);
assert_eq!(reopened.schema_version().await.unwrap(), 32);
assert_eq!(
reopened.get_workspace("local-dev").await.unwrap(),
Some(record)
@@ -5661,8 +5702,8 @@ INSERT INTO workdir_registry (
owner_account_id: None,
display_name: "Workspace A".to_string(),
state: "active".to_string(),
created_at: "2026-07-31T00:00:00Z".to_string(),
updated_at: "2026-07-31T00:00:00Z".to_string(),
created_at: "2026-07-32T00:00:00Z".to_string(),
updated_at: "2026-07-32T00:00:00Z".to_string(),
})
.await
.unwrap();
@@ -5673,7 +5714,7 @@ INSERT INTO workdir_registry (
assignment_id: "assignment-1".to_string(),
worker: RuntimeWorkerRef::new("runtime-1", "worker-1"),
assigned_by: "user-1".to_string(),
assigned_at: "2026-07-31T00:00:01Z".to_string(),
assigned_at: "2026-07-32T00:00:01Z".to_string(),
};
let created = store
.set_current_ticket_worker_assignment(&first, None, "event-1", "operation-1", false)
@@ -5740,7 +5781,7 @@ INSERT INTO workdir_registry (
assignment_id: "assignment-2".to_string(),
worker: RuntimeWorkerRef::new("runtime-2", "worker-2"),
assigned_by: "user-2".to_string(),
assigned_at: "2026-07-31T00:00:02Z".to_string(),
assigned_at: "2026-07-32T00:00:02Z".to_string(),
..first.clone()
};
let replaced = store
@@ -5779,7 +5820,7 @@ INSERT INTO workdir_registry (
"unassign-operation-stale",
"event-stale",
"user-1",
"2026-07-31T00:00:03Z",
"2026-07-32T00:00:03Z",
)
.unwrap_err();
assert!(matches!(stale, Error::TicketAssignmentConflict(_)));
@@ -5792,7 +5833,7 @@ INSERT INTO workdir_registry (
"unassign-operation-2",
"event-3",
"user-2",
"2026-07-31T00:00:03Z",
"2026-07-32T00:00:03Z",
)
.unwrap();
assert_eq!(cleared, Some(second.clone()));
@@ -5804,7 +5845,7 @@ INSERT INTO workdir_registry (
"unassign-operation-2",
"ignored-clear-event",
"user-2",
"2026-07-31T00:00:04Z",
"2026-07-32T00:00:04Z",
)
.unwrap();
assert_eq!(retried_clear, Some(second));
@@ -5816,7 +5857,7 @@ INSERT INTO workdir_registry (
"runtime-3",
None,
"sha256:reserved",
"2026-07-31T00:00:05Z",
"2026-07-32T00:00:05Z",
)
.unwrap();
drop(store);
@@ -5843,7 +5884,7 @@ INSERT INTO workdir_registry (
assignment_id: "assignment-3".to_string(),
worker: RuntimeWorkerRef::new("runtime-3", "worker-3"),
assigned_by: "runtime".to_string(),
assigned_at: "2026-07-31T00:00:06Z".to_string(),
assigned_at: "2026-07-32T00:00:06Z".to_string(),
};
let completed_reservation = store
.set_current_ticket_worker_assignment(
@@ -6115,7 +6156,7 @@ INSERT INTO workdir_registry (
.unwrap();
let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 31);
assert_eq!(store.schema_version().await.unwrap(), 32);
store
.with_conn(|conn| {
@@ -6304,7 +6345,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn repository_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 31);
assert_eq!(store.schema_version().await.unwrap(), 32);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: None,
@@ -6370,7 +6411,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn memory_authority_records_round_trip_and_close_staging() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 31);
assert_eq!(store.schema_version().await.unwrap(), 32);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: None,
@@ -6633,7 +6674,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn account_and_login_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 31);
assert_eq!(store.schema_version().await.unwrap(), 32);
let now = "2026-07-22T00:00:00Z".to_string();
let account = AccountRecord {
account_id: "acct-user-alice".to_string(),