server: persist evaluated virtual config trees

2026-08-13 22:26:11 +09:00
parent 7aa4d3067e
commit c8b57a6a5d
6 changed files with 631 additions and 3 deletions
@@ -6227,6 +6227,7 @@ dependencies = [
 "async-trait",
 "axum",
 "chrono",
+ "config-source",
 "flow",
 "futures",
 "manifest",
@@ -19,6 +19,7 @@ async-trait.workspace = true
 axum = { workspace = true, features = ["ws"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 futures.workspace = true
+config-source.workspace = true
 flow = { path = "../flow" }
 manifest.workspace = true
 protocol = { workspace = true }
@@ -0,0 +1,463 @@
+use chrono::{SecondsFormat, Utc};
+use config_source::{
+    ConfigContentType, ConfigEntry, ConfigTreeChange, ConfigTreeSnapshot, DECODAL_VERSION,
+    DEFAULT_IMPORT_POLICY_VERSION, DEFAULT_SCHEMA_VERSION, EvaluationResult, SnapshotEnvironment,
+    ToolchainContract, VirtualPath,
+};
+use rusqlite::{OptionalExtension, TransactionBehavior, params};
+use serde::{Deserialize, Serialize};
+
+use crate::{Error, Result, SqliteWorkspaceStore};
+
+pub const DEFAULT_CONFIG_ENTRYPOINT: &str = "workspace.dcdl";
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct WorkspaceConfigState {
+    pub snapshot: ConfigTreeSnapshot,
+    pub contract: ToolchainContract,
+    pub projection_digest: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct EvaluatedConfigCandidate {
+    pub base_revision: u64,
+    pub base_digest: String,
+    pub snapshot: ConfigTreeSnapshot,
+    pub contract: ToolchainContract,
+    pub evaluation: EvaluationResult,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ConfigCommitRequest {
+    pub base_revision: u64,
+    pub base_digest: String,
+    pub changes: Vec<ConfigTreeChange>,
+    pub entrypoints: Vec<VirtualPath>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ConfigPreviewRequest {
+    pub changes: Vec<ConfigTreeChange>,
+    pub entrypoints: Vec<VirtualPath>,
+}
+
+impl SqliteWorkspaceStore {
+    pub fn load_workspace_config(
+        &self,
+        workspace_id: &str,
+    ) -> Result<Option<WorkspaceConfigState>> {
+        self.with_conn(|conn| load_state(conn, workspace_id))
+    }
+
+    pub fn evaluate_workspace_config_candidate(
+        &self,
+        workspace_id: &str,
+        request: &ConfigCommitRequest,
+    ) -> Result<EvaluatedConfigCandidate> {
+        let current = self
+            .load_workspace_config(workspace_id)?
+            .unwrap_or_else(empty_state);
+        if current.snapshot.revision != request.base_revision
+            || current.snapshot.digest != request.base_digest
+        {
+            return Err(config_conflict(format!(
+                "base revision/digest mismatch; current revision is {}",
+                current.snapshot.revision
+            )));
+        }
+        evaluate_candidate(current, &request.changes, request.entrypoints.clone())
+    }
+
+    pub fn preview_workspace_config(
+        &self,
+        workspace_id: &str,
+        request: &ConfigPreviewRequest,
+    ) -> Result<EvaluatedConfigCandidate> {
+        let current = self
+            .load_workspace_config(workspace_id)?
+            .unwrap_or_else(empty_state);
+        evaluate_candidate(current, &request.changes, request.entrypoints.clone())
+    }
+
+    pub fn commit_evaluated_workspace_config(
+        &self,
+        workspace_id: &str,
+        candidate: &EvaluatedConfigCandidate,
+    ) -> Result<WorkspaceConfigState> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let workspace_exists: bool = tx.query_row(
+                "SELECT EXISTS(SELECT 1 FROM workspaces WHERE workspace_id = ?1)",
+                [workspace_id],
+                |row| row.get(0),
+            )?;
+            if !workspace_exists {
+                return Err(Error::WorkspaceIdMismatch);
+            }
+            let current = load_state(&tx, workspace_id)?.unwrap_or_else(empty_state);
+            if current.snapshot.revision != candidate.base_revision
+                || current.snapshot.digest != candidate.base_digest
+            {
+                return Err(config_conflict(format!(
+                    "base revision/digest mismatch; current revision is {}",
+                    current.snapshot.revision
+                )));
+            }
+            let next_revision = current.snapshot.revision + 1;
+            let mut snapshot = candidate.snapshot.clone();
+            snapshot.revision = next_revision;
+            let now = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true);
+            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)
+                ON CONFLICT(workspace_id) DO UPDATE SET
+                    revision = excluded.revision,
+                    tree_digest = excluded.tree_digest,
+                    schema_version = excluded.schema_version,
+                    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"#,
+                params![
+                    workspace_id,
+                    next_revision as i64,
+                    snapshot.digest,
+                    candidate.contract.schema_version,
+                    serde_json::to_string(&candidate.contract.entrypoints)
+                        .map_err(|error| Error::Store(error.to_string()))?,
+                    candidate.contract.decodal_version,
+                    candidate.contract.import_policy_version,
+                    candidate.contract.fingerprint,
+                    candidate.evaluation.projection_digest,
+                    now,
+                ],
+            )?;
+            tx.execute(
+                "DELETE FROM workspace_config_entries WHERE workspace_id = ?1",
+                [workspace_id],
+            )?;
+            for entry in snapshot.entries.values() {
+                tx.execute(
+                    r#"INSERT INTO workspace_config_entries (
+                        workspace_id, path, content_type, content, content_digest
+                    ) VALUES (?1, ?2, ?3, ?4, ?5)"#,
+                    params![
+                        workspace_id,
+                        entry.path.as_str(),
+                        content_type_label(entry.content_type),
+                        entry.content,
+                        entry.content_digest,
+                    ],
+                )?;
+            }
+            let manifest_json = serde_json::to_string(&snapshot.entries)
+                .map_err(|error| Error::Store(error.to_string()))?;
+            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)"#,
+                params![
+                    workspace_id,
+                    next_revision as i64,
+                    snapshot.digest,
+                    candidate.contract.fingerprint,
+                    candidate.evaluation.projection_digest,
+                    manifest_json,
+                    now,
+                ],
+            )?;
+            tx.commit()?;
+            Ok(WorkspaceConfigState {
+                snapshot,
+                contract: candidate.contract.clone(),
+                projection_digest: candidate.evaluation.projection_digest.clone(),
+            })
+        })
+    }
+
+    pub fn evaluate_and_commit_workspace_config(
+        &self,
+        workspace_id: &str,
+        request: &ConfigCommitRequest,
+    ) -> Result<WorkspaceConfigState> {
+        let candidate = self.evaluate_workspace_config_candidate(workspace_id, request)?;
+        self.commit_evaluated_workspace_config(workspace_id, &candidate)
+    }
+}
+
+fn evaluate_candidate(
+    current: WorkspaceConfigState,
+    changes: &[ConfigTreeChange],
+    entrypoints: Vec<VirtualPath>,
+) -> Result<EvaluatedConfigCandidate> {
+    let snapshot = current.snapshot.apply(changes).map_err(config_error)?;
+    let contract = ToolchainContract::new(
+        DEFAULT_SCHEMA_VERSION,
+        entrypoints,
+        DEFAULT_IMPORT_POLICY_VERSION,
+    );
+    let evaluation = SnapshotEnvironment::new(snapshot.clone())
+        .evaluate_contract(&contract)
+        .map_err(|diagnostics| {
+            Error::InvalidInput(
+                serde_json::to_string(&diagnostics)
+                    .unwrap_or_else(|_| "virtual config evaluation failed".to_string()),
+            )
+        })?;
+    Ok(EvaluatedConfigCandidate {
+        base_revision: current.snapshot.revision,
+        base_digest: current.snapshot.digest,
+        snapshot,
+        contract,
+        evaluation,
+    })
+}
+
+fn load_state(
+    conn: &rusqlite::Connection,
+    workspace_id: &str,
+) -> Result<Option<WorkspaceConfigState>> {
+    let header = 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)?,
+                    row.get::<_, String>(6)?,
+                    row.get::<_, String>(7)?,
+                ))
+            },
+        )
+        .optional()?;
+    let Some((
+        revision,
+        stored_digest,
+        schema_version,
+        entrypoints_json,
+        decodal_version,
+        import_policy_version,
+        fingerprint,
+        projection_digest,
+    )) = header
+    else {
+        return Ok(None);
+    };
+    let mut statement = conn.prepare(
+        r#"SELECT path, content_type, content, content_digest
+           FROM workspace_config_entries WHERE workspace_id = ?1 ORDER BY path"#,
+    )?;
+    let entries = statement
+        .query_map([workspace_id], |row| {
+            let path = row.get::<_, String>(0)?;
+            let content_type = row.get::<_, String>(1)?;
+            let content = row.get::<_, String>(2)?;
+            let stored_entry_digest = row.get::<_, String>(3)?;
+            Ok((path, content_type, content, stored_entry_digest))
+        })?
+        .collect::<std::result::Result<Vec<_>, _>>()?
+        .into_iter()
+        .map(|(path, content_type, content, stored_entry_digest)| {
+            let path = VirtualPath::parse(path).map_err(config_error)?;
+            let entry = ConfigEntry::new(path, parse_content_type(&content_type)?, content)
+                .map_err(config_error)?;
+            if entry.content_digest != stored_entry_digest {
+                return Err(Error::RegistryInconsistency(format!(
+                    "virtual config entry digest mismatch for {}",
+                    entry.path
+                )));
+            }
+            Ok(entry)
+        })
+        .collect::<Result<Vec<_>>>()?;
+    let snapshot =
+        ConfigTreeSnapshot::from_entries(revision as u64, entries).map_err(config_error)?;
+    if snapshot.digest != stored_digest {
+        return Err(Error::RegistryInconsistency(format!(
+            "virtual config tree digest mismatch for Workspace {workspace_id}"
+        )));
+    }
+    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);
+    if decodal_version != DECODAL_VERSION || contract.fingerprint != fingerprint {
+        return Err(Error::RegistryInconsistency(format!(
+            "virtual config toolchain metadata mismatch for Workspace {workspace_id}"
+        )));
+    }
+    Ok(Some(WorkspaceConfigState {
+        snapshot,
+        contract,
+        projection_digest,
+    }))
+}
+
+fn empty_state() -> WorkspaceConfigState {
+    WorkspaceConfigState {
+        snapshot: ConfigTreeSnapshot::empty(),
+        contract: ToolchainContract::new(
+            DEFAULT_SCHEMA_VERSION,
+            Vec::new(),
+            DEFAULT_IMPORT_POLICY_VERSION,
+        ),
+        projection_digest: config_source::digest_bytes(b"[]"),
+    }
+}
+
+fn content_type_label(value: ConfigContentType) -> &'static str {
+    match value {
+        ConfigContentType::Decodal => "decodal",
+        ConfigContentType::Text => "text",
+    }
+}
+
+fn parse_content_type(value: &str) -> Result<ConfigContentType> {
+    match value {
+        "decodal" => Ok(ConfigContentType::Decodal),
+        "text" => Ok(ConfigContentType::Text),
+        _ => Err(Error::RegistryInconsistency(format!(
+            "unknown virtual config content type {value:?}"
+        ))),
+    }
+}
+
+fn config_error(error: impl std::fmt::Display) -> Error {
+    Error::InvalidInput(error.to_string())
+}
+
+fn config_conflict(message: impl Into<String>) -> Error {
+    Error::WorkspaceConfigConflict(message.into())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{ControlPlaneStore, WorkspaceRecord};
+
+    fn workspace() -> WorkspaceRecord {
+        WorkspaceRecord {
+            workspace_id: "w-config".into(),
+            owner_account_id: None,
+            display_name: "Config".into(),
+            state: "active".into(),
+            created_at: "2026-08-13T00:00:00Z".into(),
+            updated_at: "2026-08-13T00:00:00Z".into(),
+        }
+    }
+
+    fn path(value: &str) -> VirtualPath {
+        VirtualPath::parse(value).unwrap()
+    }
+
+    #[tokio::test]
+    async fn invalid_candidate_is_never_persisted() {
+        let store = SqliteWorkspaceStore::in_memory().unwrap();
+        store.upsert_workspace(&workspace()).await.unwrap();
+        let current = ConfigTreeSnapshot::empty();
+        let error = store
+            .evaluate_and_commit_workspace_config(
+                "w-config",
+                &ConfigCommitRequest {
+                    base_revision: 0,
+                    base_digest: current.digest,
+                    changes: vec![ConfigTreeChange::Create {
+                        path: path(DEFAULT_CONFIG_ENTRYPOINT),
+                        content_type: ConfigContentType::Decodal,
+                        content: "{ broken = ; }".into(),
+                    }],
+                    entrypoints: vec![path(DEFAULT_CONFIG_ENTRYPOINT)],
+                },
+            )
+            .unwrap_err();
+        assert!(matches!(error, Error::InvalidInput(_)));
+        assert!(store.load_workspace_config("w-config").unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn valid_candidate_commits_snapshot_revision_and_provenance_atomically() {
+        let store = SqliteWorkspaceStore::in_memory().unwrap();
+        store.upsert_workspace(&workspace()).await.unwrap();
+        let empty = ConfigTreeSnapshot::empty();
+        let committed = store
+            .evaluate_and_commit_workspace_config(
+                "w-config",
+                &ConfigCommitRequest {
+                    base_revision: 0,
+                    base_digest: empty.digest,
+                    changes: vec![ConfigTreeChange::Create {
+                        path: path(DEFAULT_CONFIG_ENTRYPOINT),
+                        content_type: ConfigContentType::Decodal,
+                        content: "{ answer = 42; }".into(),
+                    }],
+                    entrypoints: vec![path(DEFAULT_CONFIG_ENTRYPOINT)],
+                },
+            )
+            .unwrap();
+        assert_eq!(committed.snapshot.revision, 1);
+        assert_eq!(committed.contract.decodal_version, DECODAL_VERSION);
+        assert!(!committed.projection_digest.is_empty());
+        let reread = store.load_workspace_config("w-config").unwrap().unwrap();
+        assert_eq!(reread.snapshot, committed.snapshot);
+    }
+
+    #[tokio::test]
+    async fn stale_cas_cannot_overwrite_newer_tree() {
+        let store = SqliteWorkspaceStore::in_memory().unwrap();
+        store.upsert_workspace(&workspace()).await.unwrap();
+        let empty = ConfigTreeSnapshot::empty();
+        let request = ConfigCommitRequest {
+            base_revision: 0,
+            base_digest: empty.digest,
+            changes: vec![ConfigTreeChange::Create {
+                path: path(DEFAULT_CONFIG_ENTRYPOINT),
+                content_type: ConfigContentType::Decodal,
+                content: "{ answer = 42; }".into(),
+            }],
+            entrypoints: vec![path(DEFAULT_CONFIG_ENTRYPOINT)],
+        };
+        let candidate = store
+            .evaluate_workspace_config_candidate("w-config", &request)
+            .unwrap();
+        store
+            .commit_evaluated_workspace_config("w-config", &candidate)
+            .unwrap();
+        let error = store
+            .commit_evaluated_workspace_config("w-config", &candidate)
+            .unwrap_err();
+        assert!(matches!(error, Error::WorkspaceConfigConflict(_)));
+    }
+
+    #[test]
+    fn migration_creates_config_authority_without_changing_applied_migrations() {
+        let store = SqliteWorkspaceStore::in_memory().unwrap();
+        store
+            .with_conn(|conn| {
+                for table in [
+                    "workspace_config_trees",
+                    "workspace_config_entries",
+                    "workspace_config_tree_revisions",
+                ] {
+                    let exists: bool = conn.query_row(
+                        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)",
+                        [table],
+                        |row| row.get(0),
+                    )?;
+                    assert!(exists, "missing {table}");
+                }
+                Ok(())
+            })
+            .unwrap();
+    }
+}
@@ -8,6 +8,7 @@ pub mod auth;
 pub mod authority;
 pub mod companion;
 pub mod config;
+pub mod config_source;
 pub mod hosts;
 pub mod identity;
 pub mod memory_backend;
@@ -106,6 +107,8 @@ pub enum Error {
    TicketAssignmentConflict(String),
    #[error("Workdir attachment conflict: {0}")]
    WorkdirAttachmentConflict(String),
+    #[error("Workspace config update conflict: {0}")]
+    WorkspaceConfigConflict(String),
    #[error("Registry inconsistency: {0}")]
    RegistryInconsistency(String),
    #[error("Worker source identity is invalid: {0}")]
@@ -12,6 +12,7 @@ use axum::response::{IntoResponse, Response};
 use axum::routing::{delete, get, patch, post, put};
 use axum::{Json, Router};
 use chrono::{Duration, SecondsFormat, Utc};
+use config_source::ConfigTreeSnapshot;
 use flow::{FlowSourceKind, FlowSourceResolveRequest, ResolvedFlowSource};
 use futures::{SinkExt, StreamExt};
 use memory::backend::{
@@ -61,6 +62,7 @@ use crate::companion::{
    CompanionStatusResponse, CompanionTranscriptProjection,
 };
 use crate::config::{BackendRuntimesConfigFile, RemoteRuntimeConfigFile, resolve_remote_runtime};
+use crate::config_source::{ConfigCommitRequest, ConfigPreviewRequest};
 use crate::hosts::{
    ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EMBEDDED_RUNTIME_ID,
    EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime,
@@ -252,6 +254,7 @@ const ORCHESTRATOR_ATTENTION_PROMPT: &str = include_str!(concat!(
 pub struct WorkspaceApi {
    pub(crate) config: ServerConfig,
    pub(crate) store: Arc<dyn ControlPlaneStore>,
+    config_store: Arc<crate::SqliteWorkspaceStore>,
    authority: SqliteWorkspaceAuthority,
    runtime: Arc<RuntimeRegistry>,
    companion: Arc<CompanionConsole>,
@@ -741,7 +744,11 @@ impl WorkspaceApi {
        let runtime = Arc::new(runtime);
        let companion = Arc::new(CompanionConsole::disabled());
        let observation_proxy = BackendObservationProxy::new(config.runtime_event_sources.clone());
+        let config_store = Arc::new(crate::SqliteWorkspaceStore::open(
+            config.database_path.clone(),
+        )?);
        let api = Self {
+            config_store,
            authority: SqliteWorkspaceAuthority::new(
                config.database_path.clone(),
                config.workspace_id.clone(),
@@ -1132,6 +1139,22 @@ pub fn build_router(api: WorkspaceApi) -> Router {
            "/api/w/{workspace_id}/settings/workspace",
            get(scoped_get_workspace_settings).put(scoped_update_workspace_settings),
        )
+        .route(
+            "/api/w/{workspace_id}/config/source-tree",
+            get(scoped_get_workspace_config_tree),
+        )
+        .route(
+            "/api/w/{workspace_id}/config/source-tree/preview",
+            post(scoped_preview_workspace_config_tree),
+        )
+        .route(
+            "/api/w/{workspace_id}/config/source-tree/commit",
+            post(scoped_commit_workspace_config_tree),
+        )
+        .route(
+            "/api/w/{workspace_id}/config/source-tree/entries/{*path}",
+            get(scoped_get_workspace_config_entry),
+        )
        .route(
            "/api/w/{workspace_id}/settings/profiles",
            get(scoped_get_profile_settings).post(scoped_create_profile_source),
@@ -2418,6 +2441,95 @@ async fn scoped_update_workspace_settings(
    ))
 }

+#[derive(Debug, Deserialize)]
+struct WorkspaceConfigEntryPath {
+    workspace_id: String,
+    path: String,
+}
+
+#[derive(Debug, Serialize)]
+struct WorkspaceConfigTreeResponse {
+    snapshot: ConfigTreeSnapshot,
+    contract: config_source::ToolchainContract,
+    projection_digest: String,
+}
+
+async fn scoped_get_workspace_config_tree(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+) -> ApiResult<Json<WorkspaceConfigTreeResponse>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let state = api
+        .config_store
+        .load_workspace_config(&path.workspace_id)?
+        .unwrap_or_else(|| crate::config_source::WorkspaceConfigState {
+            snapshot: ConfigTreeSnapshot::empty(),
+            contract: config_source::ToolchainContract::new(
+                config_source::DEFAULT_SCHEMA_VERSION,
+                Vec::new(),
+                config_source::DEFAULT_IMPORT_POLICY_VERSION,
+            ),
+            projection_digest: config_source::digest_bytes(b"[]"),
+        });
+    Ok(Json(WorkspaceConfigTreeResponse {
+        snapshot: state.snapshot,
+        contract: state.contract,
+        projection_digest: state.projection_digest,
+    }))
+}
+
+async fn scoped_get_workspace_config_entry(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<WorkspaceConfigEntryPath>,
+) -> ApiResult<Json<config_source::ConfigEntry>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let virtual_path = config_source::VirtualPath::parse(&path.path)
+        .map_err(|error| ApiError::from(Error::InvalidInput(error.to_string())))?;
+    let state = api
+        .config_store
+        .load_workspace_config(&path.workspace_id)?
+        .ok_or_else(|| {
+            ApiError::from(Error::InvalidRecordId("virtual config source tree".into()))
+        })?;
+    let entry = state
+        .snapshot
+        .get(&virtual_path)
+        .cloned()
+        .ok_or_else(|| ApiError::from(Error::InvalidRecordId(path.path)))?;
+    Ok(Json(entry))
+}
+
+async fn scoped_preview_workspace_config_tree(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+    Json(request): Json<ConfigPreviewRequest>,
+) -> 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)?,
+    ))
+}
+
+async fn scoped_commit_workspace_config_tree(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+    Json(request): Json<ConfigCommitRequest>,
+) -> ApiResult<(StatusCode, Json<WorkspaceConfigTreeResponse>)> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let state = api
+        .config_store
+        .evaluate_and_commit_workspace_config(&path.workspace_id, &request)?;
+    Ok((
+        StatusCode::CREATED,
+        Json(WorkspaceConfigTreeResponse {
+            snapshot: state.snapshot,
+            contract: state.contract,
+            projection_digest: state.projection_digest,
+        }),
+    ))
+}
+
 async fn scoped_get_profile_settings(
    State(api): State<WorkspaceApi>,
    AxumPath(path): AxumPath<ScopedWorkspacePath>,
@@ -11255,9 +11367,9 @@ impl IntoResponse for ApiError {
            Error::BrowserMergeConfirmationRequired | Error::BrowserReopenConfirmationRequired => {
                StatusCode::FORBIDDEN
            }
-            Error::TicketAssignmentConflict(_) | Error::WorkdirAttachmentConflict(_) => {
-                StatusCode::CONFLICT
-            }
+            Error::TicketAssignmentConflict(_)
+            | Error::WorkdirAttachmentConflict(_)
+            | Error::WorkspaceConfigConflict(_) => StatusCode::CONFLICT,
            Error::WorkerSourceIdentity(_) | Error::InvalidInput(_) => StatusCode::BAD_REQUEST,
            Error::InvalidRuntimeIdentifier { .. } | Error::ReservedWorkerName(_) => {
                StatusCode::BAD_REQUEST
@@ -166,6 +166,11 @@ const MIGRATIONS: &[Migration] = &[
        name: "create Worker mutation source proof replay guard",
        apply: create_worker_mutation_source_proof_replay_guard,
    },
+    Migration {
+        version: 30,
+        name: "create Workspace virtual config source authority",
+        apply: create_workspace_config_source_authority,
+    },
 ];

 struct Migration {
@@ -4527,6 +4532,49 @@ fn current_schema_version(conn: &Connection) -> Result<i64> {
    .map_err(Error::from)
 }

+fn create_workspace_config_source_authority(conn: &Connection) -> Result<()> {
+    conn.execute_batch(
+        r#"
+        CREATE TABLE workspace_config_trees (
+            workspace_id TEXT PRIMARY KEY,
+            revision INTEGER NOT NULL CHECK (revision >= 0),
+            tree_digest TEXT NOT NULL,
+            schema_version INTEGER NOT NULL,
+            entrypoints_json TEXT NOT NULL,
+            decodal_version TEXT NOT NULL,
+            import_policy_version INTEGER NOT NULL,
+            toolchain_fingerprint TEXT NOT NULL,
+            projection_digest TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
+        );
+        CREATE TABLE workspace_config_entries (
+            workspace_id TEXT NOT NULL,
+            path TEXT NOT NULL,
+            content_type TEXT NOT NULL,
+            content TEXT NOT NULL,
+            content_digest TEXT NOT NULL,
+            PRIMARY KEY (workspace_id, path),
+            FOREIGN KEY (workspace_id) REFERENCES workspace_config_trees(workspace_id) ON DELETE CASCADE
+        );
+        CREATE INDEX idx_workspace_config_entries_prefix
+            ON workspace_config_entries(workspace_id, path);
+        CREATE TABLE workspace_config_tree_revisions (
+            workspace_id TEXT NOT NULL,
+            revision INTEGER NOT NULL,
+            tree_digest TEXT NOT NULL,
+            toolchain_fingerprint TEXT NOT NULL,
+            projection_digest TEXT NOT NULL,
+            manifest_json TEXT NOT NULL,
+            created_at TEXT NOT NULL,
+            PRIMARY KEY (workspace_id, revision),
+            FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
+        );
+        "#,
+    )?;
+    Ok(())
+}
+
 fn create_worker_mutation_source_proof_replay_guard(conn: &Connection) -> Result<()> {
    conn.execute_batch(
        r#"