fix: verify signed Workspace query targets

# Conflicts:
#	crates/workspace-server/src/store.rs

Cargo.lock

@@ -6131,9 +6131,11 @@ dependencies = [
  "tokio-tungstenite 0.29.0",
  "toml",
  "tower",
+ "url",
  "uuid",
  "workdir",
  "worker",
+ "workspace-api",
 ]
 
 [[package]]

crates/client/src/backend_workspace.rs

@@ -1,5 +1,6 @@
 use serde::{Deserialize, Serialize};
 use std::fmt;
+use workspace_api::{RepositoryObservedStatus, RepositorySource};
 
 const DEFAULT_WORKSPACE_LIMIT: usize = 200;
 
@@ -44,8 +45,13 @@ pub struct CreateBackendWorkspaceRepositoryRecord {
     pub repository_id: String,
     pub name: String,
     pub kind: String,
-    pub uri: String,
+    pub provider: Option<String>,
+    pub source: RepositorySource,
     pub default_ref: Option<String>,
+    pub source_revision: u64,
+    pub source_fingerprint: String,
+    pub observed_status: RepositoryObservedStatus,
+    pub observed_at: Option<String>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]

crates/workdir/src/workspace.rs

@@ -211,6 +211,7 @@ pub struct WorkingDirectoryListResponse {
 #[serde(deny_unknown_fields)]
 pub struct WorkingDirectoryDetailResponse {
     pub workspace_id: String,
+    pub runtime_id: String,
     pub item: WorkingDirectorySummary,
     pub diagnostics: Vec<WorkingDirectoryDiagnostic>,
 }
@@ -306,6 +307,7 @@ mod tests {
 
         let detail = WorkingDirectoryDetailResponse {
             workspace_id: decoded.workspace_id.clone(),
+            runtime_id: "arcadia".to_string(),
             item: decoded.items[0].clone(),
             diagnostics: decoded.diagnostics.clone(),
         };

crates/worker-runtime/Cargo.toml

@@ -41,9 +41,11 @@ tar.workspace = true
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["net", "rt", "sync", "time"] }
 toml.workspace = true
+url.workspace = true
 uuid = { workspace = true, features = ["v7"] }
 tower = { workspace = true, features = ["util"], optional = true }
 worker.workspace = true
+workspace-api = { path = "../workspace-api" }
 workdir.workspace = true
 
 [dev-dependencies]

crates/worker-runtime/src/auth.rs

@@ -3,6 +3,7 @@ use base64::engine::general_purpose::URL_SAFE_NO_PAD;
 use ring::rand::{SecureRandom, SystemRandom};
 use ring::signature::{ED25519, Ed25519KeyPair, KeyPair, UnparsedPublicKey};
 use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
 use std::fmt;
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -14,6 +15,11 @@ pub const WORKER_MUTATION_SOURCE_PROOF_HEADER: &str = "x-yoi-worker-mutation-pro
 const WORKER_MUTATION_SOURCE_PROOF_PREFIX: &str = "yoi-worker-source-v1";
 const WORKER_MUTATION_SOURCE_SIGNING_INPUT_PREFIX: &str = "yoi-worker-source-v1.";
 pub const WORKER_REMOVE_PERMISSION: &str = "workspace:worker-remove";
+pub const RUNTIME_REQUEST_SOURCE_PROOF_HEADER: &str = "x-yoi-runtime-request-proof";
+pub const WORKSPACE_REQUEST_PERMISSION: &str = "workspace:request";
+pub const BACKEND_RESOURCE_FETCH_PERMISSION: &str = "workspace:resource-fetch";
+const RUNTIME_REQUEST_SOURCE_PROOF_PREFIX: &str = "yoi-runtime-request-v1";
+const RUNTIME_REQUEST_SOURCE_SIGNING_INPUT_PREFIX: &str = "yoi-runtime-request-v1.";
 
 #[derive(Debug, thiserror::Error)]
 pub enum RuntimeAuthError {
@@ -33,6 +39,10 @@ pub enum RuntimeAuthError {
     InvalidTokenFormat,
     #[error("malformed capability token claims: {0}")]
     MalformedClaims(#[from] serde_json::Error),
+    #[error("runtime request proof contains an invalid `{0}` claim")]
+    InvalidClaim(&'static str),
+    #[error("runtime request proof does not match the HTTP request")]
+    ClaimMismatch,
     #[error("unknown token issuer `{0}`")]
     UnknownIssuer(String),
     #[error("invalid token signature")]
@@ -224,6 +234,162 @@ pub fn verify_capability_token(
     })
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RuntimeRequestSourceClaims {
+    pub iss: String,
+    pub aud: String,
+    pub workspace_id: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub worker_id: Option<String>,
+    pub permission: String,
+    pub method: String,
+    pub path: String,
+    pub body_digest: String,
+    pub iat: i64,
+    pub exp: i64,
+    pub jti: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct RuntimeRequestSourceSigner {
+    identity_id: String,
+    private_key: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct RuntimeRequestSourceExpectation<'a> {
+    pub identity_id: &'a str,
+    pub audience: &'a str,
+    pub workspace_id: &'a str,
+    pub worker_id: Option<&'a str>,
+    pub permission: &'a str,
+    pub method: &'a str,
+    pub path: &'a str,
+    pub body_digest: &'a str,
+    pub now_unix: i64,
+}
+
+pub fn request_body_digest(body: &[u8]) -> String {
+    URL_SAFE_NO_PAD.encode(Sha256::digest(body))
+}
+
+impl RuntimeRequestSourceSigner {
+    pub fn from_identity(identity: &RuntimeIdentityMaterial) -> Self {
+        Self {
+            identity_id: identity.identity_id.clone(),
+            private_key: identity.private_key.clone(),
+        }
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub fn issue(
+        &self,
+        audience: &str,
+        workspace_id: &str,
+        worker_id: Option<&str>,
+        permission: &str,
+        method: &str,
+        path: &str,
+        body: &[u8],
+        now_unix: i64,
+        ttl_seconds: u64,
+    ) -> Result<String, RuntimeAuthError> {
+        for (name, value) in [
+            ("audience", audience),
+            ("workspace_id", workspace_id),
+            ("permission", permission),
+            ("method", method),
+            ("path", path),
+        ] {
+            if value.trim().is_empty() {
+                return Err(RuntimeAuthError::InvalidClaim(name));
+            }
+        }
+        if worker_id.is_some_and(str::is_empty) {
+            return Err(RuntimeAuthError::InvalidClaim("worker_id"));
+        }
+        let ttl_seconds = i64::try_from(ttl_seconds).unwrap_or(i64::MAX);
+        let claims = RuntimeRequestSourceClaims {
+            iss: self.identity_id.clone(),
+            aud: audience.to_owned(),
+            workspace_id: workspace_id.to_owned(),
+            worker_id: worker_id.map(str::to_owned),
+            permission: permission.to_owned(),
+            method: method.to_owned(),
+            path: path.to_owned(),
+            body_digest: request_body_digest(body),
+            iat: now_unix,
+            exp: now_unix.saturating_add(ttl_seconds),
+            jti: new_token_id()?,
+        };
+        let payload = serde_json::to_vec(&claims)?;
+        let payload = URL_SAFE_NO_PAD.encode(payload);
+        let signing_input = format!("{RUNTIME_REQUEST_SOURCE_SIGNING_INPUT_PREFIX}{payload}");
+        let private = decode_private_key(&self.private_key)?;
+        let key_pair = Ed25519KeyPair::from_pkcs8(&private)
+            .map_err(|_| RuntimeAuthError::InvalidPrivateKey)?;
+        let signature = URL_SAFE_NO_PAD.encode(key_pair.sign(signing_input.as_bytes()).as_ref());
+        Ok(format!(
+            "{RUNTIME_REQUEST_SOURCE_PROOF_PREFIX}.{payload}.{signature}"
+        ))
+    }
+}
+
+pub fn decode_runtime_request_source_claims(
+    proof: &str,
+) -> Result<RuntimeRequestSourceClaims, RuntimeAuthError> {
+    let (prefix, payload, _signature) = split_runtime_request_source_proof(proof)?;
+    if prefix != RUNTIME_REQUEST_SOURCE_PROOF_PREFIX {
+        return Err(RuntimeAuthError::InvalidTokenFormat);
+    }
+    let payload = URL_SAFE_NO_PAD.decode(payload)?;
+    serde_json::from_slice(&payload).map_err(RuntimeAuthError::from)
+}
+
+pub fn verify_runtime_request_source(
+    proof: &str,
+    public_key: &str,
+    expected: &RuntimeRequestSourceExpectation<'_>,
+) -> Result<RuntimeRequestSourceClaims, RuntimeAuthError> {
+    let (prefix, payload, signature) = split_runtime_request_source_proof(proof)?;
+    if prefix != RUNTIME_REQUEST_SOURCE_PROOF_PREFIX {
+        return Err(RuntimeAuthError::InvalidTokenFormat);
+    }
+    let signature = URL_SAFE_NO_PAD.decode(signature)?;
+    let signing_input = format!("{RUNTIME_REQUEST_SOURCE_SIGNING_INPUT_PREFIX}{payload}");
+    let public_key = decode_public_key(public_key)?;
+    UnparsedPublicKey::new(&ED25519, public_key)
+        .verify(signing_input.as_bytes(), &signature)
+        .map_err(|_| RuntimeAuthError::InvalidSignature)?;
+    let claims = decode_runtime_request_source_claims(proof)?;
+    if claims.iss != expected.identity_id
+        || claims.aud != expected.audience
+        || claims.workspace_id != expected.workspace_id
+        || claims.worker_id.as_deref() != expected.worker_id
+        || claims.permission != expected.permission
+        || claims.method != expected.method
+        || claims.path != expected.path
+        || claims.body_digest != expected.body_digest
+    {
+        return Err(RuntimeAuthError::ClaimMismatch);
+    }
+    if claims.iat > expected.now_unix || claims.exp < expected.now_unix {
+        return Err(RuntimeAuthError::Expired);
+    }
+    Ok(claims)
+}
+
+fn split_runtime_request_source_proof(proof: &str) -> Result<(&str, &str, &str), RuntimeAuthError> {
+    let mut parts = proof.split('.');
+    let prefix = parts.next().unwrap_or_default();
+    let payload = parts.next().unwrap_or_default();
+    let signature = parts.next().unwrap_or_default();
+    if prefix.is_empty() || payload.is_empty() || signature.is_empty() || parts.next().is_some() {
+        return Err(RuntimeAuthError::InvalidTokenFormat);
+    }
+    Ok((prefix, payload, signature))
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct WorkerMutationSourceClaims {
     pub iss: String,
@@ -592,6 +758,99 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn runtime_request_source_proof_binds_request_and_rejects_spoofed_signature() {
+        let trusted = RuntimeIdentityMaterial::generate("runtime-main").unwrap();
+        let signer = RuntimeRequestSourceSigner::from_identity(&trusted);
+        let body = br#"{"ticket":"T-1"}"#;
+        let proof = signer
+            .issue(
+                "server-main",
+                "workspace-a",
+                Some("worker-7"),
+                WORKSPACE_REQUEST_PERMISSION,
+                "POST",
+                "/api/w/workspace-a/tickets/comment",
+                body,
+                90,
+                10,
+            )
+            .unwrap();
+        let expected = RuntimeRequestSourceExpectation {
+            identity_id: "runtime-main",
+            audience: "server-main",
+            workspace_id: "workspace-a",
+            worker_id: Some("worker-7"),
+            permission: WORKSPACE_REQUEST_PERMISSION,
+            method: "POST",
+            path: "/api/w/workspace-a/tickets/comment",
+            body_digest: &request_body_digest(body),
+            now_unix: 99,
+        };
+        let claims = verify_runtime_request_source(&proof, &trusted.public_key, &expected).unwrap();
+        assert_eq!(claims.iss, "runtime-main");
+        let changed_body = RuntimeRequestSourceExpectation {
+            body_digest: &request_body_digest(br#"{"ticket":"T-2"}"#),
+            ..expected.clone()
+        };
+        assert!(matches!(
+            verify_runtime_request_source(&proof, &trusted.public_key, &changed_body),
+            Err(RuntimeAuthError::ClaimMismatch)
+        ));
+        let spoofed = RuntimeIdentityMaterial::generate("runtime-main").unwrap();
+        assert!(matches!(
+            verify_runtime_request_source(&proof, &spoofed.public_key, &expected),
+            Err(RuntimeAuthError::InvalidSignature)
+        ));
+    }
+
+    #[test]
+    fn runtime_request_source_proof_rejects_wrong_scope_and_expiry() {
+        let runtime = RuntimeIdentityMaterial::generate("runtime-main").unwrap();
+        let proof = RuntimeRequestSourceSigner::from_identity(&runtime)
+            .issue(
+                "server-main",
+                "workspace-a",
+                None,
+                BACKEND_RESOURCE_FETCH_PERMISSION,
+                "POST",
+                "/api/runtime/v1/workspaces/workspace-a/resources/fetch",
+                b"{}",
+                90,
+                10,
+            )
+            .unwrap();
+        let digest = request_body_digest(b"{}");
+        let expected = RuntimeRequestSourceExpectation {
+            identity_id: "runtime-main",
+            audience: "server-main",
+            workspace_id: "workspace-a",
+            worker_id: None,
+            permission: BACKEND_RESOURCE_FETCH_PERMISSION,
+            method: "POST",
+            path: "/api/runtime/v1/workspaces/workspace-a/resources/fetch",
+            body_digest: &digest,
+            now_unix: 99,
+        };
+        assert!(verify_runtime_request_source(&proof, &runtime.public_key, &expected).is_ok());
+        let wrong_workspace = RuntimeRequestSourceExpectation {
+            workspace_id: "workspace-b",
+            ..expected.clone()
+        };
+        assert!(matches!(
+            verify_runtime_request_source(&proof, &runtime.public_key, &wrong_workspace),
+            Err(RuntimeAuthError::ClaimMismatch)
+        ));
+        let expired = RuntimeRequestSourceExpectation {
+            now_unix: 101,
+            ..expected
+        };
+        assert!(matches!(
+            verify_runtime_request_source(&proof, &runtime.public_key, &expired),
+            Err(RuntimeAuthError::Expired)
+        ));
+    }
+
     #[test]
     fn capability_token_verifies_signature_audience_expiry_and_permission() {
         let server = RuntimeIdentityMaterial::generate("server-main").unwrap();

crates/worker-runtime/src/catalog.rs

@@ -2,7 +2,6 @@ use crate::identity::{RuntimeWorkerRef, WorkerId, WorkerRef};
 use crate::interaction::WorkerInput;
 use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef};
 use serde::{Deserialize, Serialize};
-use std::path::PathBuf;
 
 fn is_false(value: &bool) -> bool {
     !*value
@@ -85,9 +84,9 @@ impl std::ops::Deref for RepositorySelector {
 pub struct WorkingDirectoryRepository {
     pub id: String,
     pub provider: String,
-    pub uri: String,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub local_path: Option<PathBuf>,
+    pub source: workspace_api::RepositorySource,
+    pub source_revision: u64,
+    pub source_fingerprint: String,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub selector: Option<RepositorySelector>,
 }

crates/worker-runtime/src/main.rs

@@ -160,15 +160,33 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
     };
     let mut factory = ProfileRuntimeWorkerFactory::new(fs_paths.worker_dir.join("worker-root"))
         .with_runtime_store_dir(runtime_store_dir);
-    if let Some(identity) = read_runtime_auth_file(&runtime_auth_path(config))?.identity {
-        factory = factory.with_remote_worker_mutation_identity(identity);
+    let runtime_auth = read_runtime_auth_file(&runtime_auth_path(config))?;
+    if let Some(identity) = runtime_auth.identity.clone() {
+        if let [trusted_server] = runtime_auth.trusted_servers.as_slice() {
+            factory =
+                factory.with_runtime_request_identity(identity, trusted_server.server_id.clone());
+        } else {
+            factory = factory.with_remote_worker_mutation_identity(identity);
+        }
     }
     if let Some(endpoint) = config.backend_resource_endpoint.clone() {
+        let identity = runtime_auth.identity.as_ref().ok_or_else(|| {
+            ProcessError::Auth(
+                "--backend-resource-endpoint requires a configured Runtime identity".to_owned(),
+            )
+        })?;
+        let [trusted_server] = runtime_auth.trusted_servers.as_slice() else {
+            return Err(ProcessError::Auth(
+                "--backend-resource-endpoint requires exactly one trusted Server identity"
+                    .to_owned(),
+            ));
+        };
         factory = factory.with_resource_client(Arc::new(
             worker_runtime::resource::HttpBackendResourceClient::new(
                 endpoint,
                 config.backend_resource_token.clone(),
-            ),
+            )
+            .with_runtime_request_source(identity, trusted_server.server_id.clone()),
         ));
     }
     let backend = Arc::new(

crates/worker-runtime/src/resource.rs

@@ -1,3 +1,7 @@
+use crate::auth::{
+    BACKEND_RESOURCE_FETCH_PERMISSION, RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
+    RuntimeIdentityMaterial, RuntimeRequestSourceSigner, unix_now_seconds,
+};
 use crate::identity::WorkerId;
 use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef, sha256_hex};
 use async_trait::async_trait;
@@ -108,6 +112,8 @@ pub trait BackendResourceClient: Send + Sync + 'static {
 pub struct HttpBackendResourceClient {
     endpoint: String,
     bearer_token: Option<String>,
+    request_source_signer: Option<RuntimeRequestSourceSigner>,
+    request_source_audience: Option<String>,
     client: reqwest::Client,
 }
 
@@ -117,9 +123,21 @@ impl HttpBackendResourceClient {
         Self {
             endpoint: endpoint.into(),
             bearer_token,
+            request_source_signer: None,
+            request_source_audience: None,
             client: reqwest::Client::new(),
         }
     }
+
+    pub fn with_runtime_request_source(
+        mut self,
+        identity: &RuntimeIdentityMaterial,
+        audience: impl Into<String>,
+    ) -> Self {
+        self.request_source_signer = Some(RuntimeRequestSourceSigner::from_identity(identity));
+        self.request_source_audience = Some(audience.into());
+        self
+    }
 }
 
 #[cfg(feature = "http-server")]
@@ -129,7 +147,44 @@ impl BackendResourceClient for HttpBackendResourceClient {
         &self,
         request: BackendResourceFetchRequest,
     ) -> Result<BackendResourceFetchResponse, BackendResourceError> {
-        let builder = self.client.post(&self.endpoint).json(&request);
+        let body = serde_json::to_vec(&request).map_err(|error| {
+            BackendResourceError::InvalidResponse {
+                message: error.to_string(),
+            }
+        })?;
+        let endpoint = reqwest::Url::parse(&self.endpoint).map_err(|error| {
+            BackendResourceError::Transport {
+                message: error.to_string(),
+            }
+        })?;
+        let mut builder = self
+            .client
+            .post(endpoint.clone())
+            .header(reqwest::header::CONTENT_TYPE, "application/json")
+            .body(body.clone());
+        if let Some(signer) = self.request_source_signer.as_ref() {
+            let audience = self.request_source_audience.as_deref().ok_or_else(|| {
+                BackendResourceError::Unauthorized {
+                    message: "Runtime request proof audience is unavailable".to_owned(),
+                }
+            })?;
+            let proof = signer
+                .issue(
+                    audience,
+                    &request.handle.workspace_id,
+                    None,
+                    BACKEND_RESOURCE_FETCH_PERMISSION,
+                    "POST",
+                    endpoint.path(),
+                    &body,
+                    i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX),
+                    30,
+                )
+                .map_err(|error| BackendResourceError::Unauthorized {
+                    message: error.to_string(),
+                })?;
+            builder = builder.header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, proof);
+        }
         let builder = if let Some(token) = self.bearer_token.as_deref() {
             builder.bearer_auth(token)
         } else {

crates/worker-runtime/src/worker_backend.rs

@@ -14,7 +14,10 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, mpsc};
 use std::time::Duration;
 
-use crate::auth::RuntimeIdentityMaterial;
+use crate::auth::{
+    BACKEND_RESOURCE_FETCH_PERMISSION, RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
+    RuntimeIdentityMaterial, RuntimeRequestSourceSigner, unix_now_seconds,
+};
 use crate::catalog::{
     CreateWorkerRequest, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource,
     WorkingDirectoryRequest, WorkingDirectoryStatus,
@@ -295,6 +298,7 @@ pub struct ProfileRuntimeWorkerFactory {
     prompt_projection_cache: Arc<WorkspacePromptProjectionCache>,
     runtime_id: Option<String>,
     worker_mutation_identity: Option<RuntimeIdentityMaterial>,
+    runtime_request_audience: Option<String>,
     embedded_worker_mutation_dispatcher: Option<Arc<dyn EmbeddedWorkerMutationDispatcher>>,
     controller_transport: WorkerControllerTransport,
 }
@@ -311,6 +315,7 @@ impl ProfileRuntimeWorkerFactory {
             prompt_projection_cache: Arc::new(WorkspacePromptProjectionCache::default()),
             runtime_id: None,
             worker_mutation_identity: None,
+            runtime_request_audience: None,
             embedded_worker_mutation_dispatcher: None,
             controller_transport: WorkerControllerTransport::UnixSocket,
         }
@@ -331,6 +336,17 @@ impl ProfileRuntimeWorkerFactory {
         self
     }
 
+    pub fn with_runtime_request_identity(
+        mut self,
+        identity: RuntimeIdentityMaterial,
+        audience: impl Into<String>,
+    ) -> Self {
+        self.runtime_id = Some(identity.identity_id.clone());
+        self.worker_mutation_identity = Some(identity);
+        self.runtime_request_audience = Some(audience.into());
+        self
+    }
+
     pub fn with_embedded_worker_mutation_dispatcher(
         mut self,
         runtime_id: impl Into<String>,
@@ -457,13 +473,15 @@ impl ProfileRuntimeWorkerFactory {
     async fn resolve_profile_source_archive(
         &self,
         source: &ProfileSourceArchiveSource,
+        request_audience: Option<&str>,
     ) -> Result<crate::profile_archive::VerifiedProfileSourceArchive, String> {
         match source {
             ProfileSourceArchiveSource::Embedded { archive } => archive
                 .verify()
                 .map_err(|err| format!("failed to verify embedded profile source archive: {err}")),
             ProfileSourceArchiveSource::Http { location } => {
-                self.fetch_profile_source_archive(location).await
+                self.fetch_profile_source_archive(location, request_audience)
+                    .await
             }
         }
     }
@@ -471,10 +489,18 @@ impl ProfileRuntimeWorkerFactory {
     async fn fetch_profile_source_archive(
         &self,
         location: &ProfileSourceArchiveHttpRef,
+        request_audience: Option<&str>,
     ) -> Result<crate::profile_archive::VerifiedProfileSourceArchive, String> {
         if let Some(cached) = self.profile_archive_cache.get(&location.archive.digest) {
-            let response =
-                fetch_profile_source_archive_http(location, Some(&location.archive.digest)).await?;
+            let response = fetch_profile_source_archive_http(
+                location,
+                Some(&location.archive.digest),
+                self.worker_mutation_identity.as_ref(),
+                self.runtime_request_audience
+                    .as_deref()
+                    .or(request_audience),
+            )
+            .await?;
             if let Some(fetched) = response {
                 self.profile_archive_cache.insert(fetched.clone());
                 fetched.verify().map_err(|err| {
@@ -486,12 +512,19 @@ impl ProfileRuntimeWorkerFactory {
                     .map_err(|err| format!("failed to verify cached profile source archive: {err}"))
             }
         } else {
-            let archive = fetch_profile_source_archive_http(location, None)
-                .await?
-                .ok_or_else(|| {
-                    "profile source archive HTTP revalidation returned 304 without a cached archive"
-                        .to_string()
-                })?;
+            let archive = fetch_profile_source_archive_http(
+                location,
+                None,
+                self.worker_mutation_identity.as_ref(),
+                self.runtime_request_audience
+                    .as_deref()
+                    .or(request_audience),
+            )
+            .await?
+            .ok_or_else(|| {
+                "profile source archive HTTP revalidation returned 304 without a cached archive"
+                    .to_string()
+            })?;
             self.profile_archive_cache.insert(archive.clone());
             archive
                 .verify()
@@ -527,6 +560,7 @@ impl RuntimeWorkspaceBackendRef {
         worker_ref: &WorkerRef,
         workspace_scope: Option<&crate::runtime::RuntimeWorkspaceScope>,
         mutation_identity: Option<&RuntimeIdentityMaterial>,
+        runtime_request_audience: Option<&str>,
         embedded_dispatcher: Option<&Arc<dyn EmbeddedWorkerMutationDispatcher>>,
         prompt_projection_cache: Option<Arc<WorkspacePromptProjectionCache>>,
     ) -> WorkerWorkspaceContext {
@@ -546,6 +580,13 @@ impl RuntimeWorkspaceBackendRef {
                 if let Some(cache) = prompt_projection_cache {
                     client = client.with_prompt_projection_cache(cache);
                 }
+                if let Some(identity) = mutation_identity {
+                    let audience = runtime_request_audience
+                        .or_else(|| workspace_scope.map(|scope| scope.server_id.as_str()));
+                    if let Some(audience) = audience {
+                        client = client.with_runtime_request_source(identity, audience.to_owned());
+                    }
+                }
                 if let (Some(scope), Some(identity)) = (workspace_scope, mutation_identity) {
                     client = client.with_worker_remove(RuntimeWorkerMutationForwarder::remote(
                         identity,
@@ -576,9 +617,40 @@ impl RuntimeWorkspaceBackendRef {
 async fn fetch_profile_source_archive_http(
     location: &ProfileSourceArchiveHttpRef,
     cached_digest: Option<&str>,
+    identity: Option<&RuntimeIdentityMaterial>,
+    audience: Option<&str>,
 ) -> Result<Option<crate::profile_archive::ProfileSourceArchive>, String> {
     let client = reqwest::Client::new();
-    let mut request = client.get(&location.url);
+    let url = reqwest::Url::parse(&location.url)
+        .map_err(|error| format!("profile source archive URL is invalid: {error}"))?;
+    let path = url.path().to_owned();
+    let workspace_id = path
+        .split('/')
+        .collect::<Vec<_>>()
+        .windows(2)
+        .find_map(|parts| (parts[0] == "w").then_some(parts[1]))
+        .filter(|value| !value.is_empty())
+        .ok_or_else(|| "profile source archive URL is not workspace-scoped".to_owned())?;
+    let mut request = client.get(url);
+    if let Some(identity) = identity {
+        let audience = audience.ok_or_else(|| {
+            "profile source archive request proof audience is unavailable".to_owned()
+        })?;
+        let proof = RuntimeRequestSourceSigner::from_identity(identity)
+            .issue(
+                audience,
+                workspace_id,
+                None,
+                BACKEND_RESOURCE_FETCH_PERMISSION,
+                "GET",
+                &path,
+                b"",
+                i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX),
+                30,
+            )
+            .map_err(|error| error.to_string())?;
+        request = request.header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, proof);
+    }
     if cached_digest == Some(location.archive.digest.as_str()) {
         if let Some(etag) = location.etag.as_deref() {
             request = request.header(reqwest::header::IF_NONE_MATCH, etag);
@@ -620,6 +692,8 @@ async fn fetch_profile_source_archive_http(
 async fn fetch_profile_source_archive_http(
     _location: &ProfileSourceArchiveHttpRef,
     _cached_digest: Option<&str>,
+    _identity: Option<&RuntimeIdentityMaterial>,
+    _audience: Option<&str>,
 ) -> Result<Option<crate::profile_archive::ProfileSourceArchive>, String> {
     Err(
         "HTTP profile source archive fetch requires the worker-runtime http-server feature"
@@ -743,12 +817,19 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
             &request.worker_ref,
             request.workspace_scope.as_ref(),
             self.worker_mutation_identity.as_ref(),
+            self.runtime_request_audience.as_deref(),
             self.embedded_worker_mutation_dispatcher.as_ref(),
             Some(self.prompt_projection_cache.clone()),
         );
         let selector = profile.as_ref();
         let archive = self
-            .resolve_profile_source_archive(&request.request.profile_source)
+            .resolve_profile_source_archive(
+                &request.request.profile_source,
+                request
+                    .workspace_scope
+                    .as_ref()
+                    .map(|scope| scope.server_id.as_str()),
+            )
             .await?;
         let (mut manifest, mut loader) = {
             let manifest = archive
@@ -909,6 +990,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
             &request.worker_ref,
             request.workspace_scope.as_ref(),
             self.worker_mutation_identity.as_ref(),
+            self.runtime_request_audience.as_deref(),
             self.embedded_worker_mutation_dispatcher.as_ref(),
             Some(self.prompt_projection_cache.clone()),
         );
@@ -2187,12 +2269,18 @@ mod tests {
         let scope = crate::runtime::RuntimeWorkspaceScope::new("workspace-a", "server-main");
 
         let before_restart =
-            backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None, None);
+            backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None, None, None);
         let adapter = WorkerRuntimeExecutionBackend::new(FailingFactory).unwrap();
         let (after_restore_kind, after_restore_workspace_id) = adapter
             .run_on_adapter_runtime(async move {
-                let after_restore =
-                    backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None, None);
+                let after_restore = backend.worker_context(
+                    &worker_ref,
+                    Some(&scope),
+                    Some(&identity),
+                    None,
+                    None,
+                    None,
+                );
                 let client = after_restore.client_handle();
                 Ok((
                     client.kind().to_string(),
@@ -2383,6 +2471,7 @@ mod tests {
                 None,
                 None,
                 None,
+                None,
             );
             let workspace_client = workspace_context.client_handle();
             self.observed_workspace_clients.lock().unwrap().push((
@@ -2631,8 +2720,12 @@ mod tests {
             repository: WorkingDirectoryRepository {
                 id: "repo-main".to_string(),
                 provider: "git".to_string(),
-                uri: ".".to_string(),
-                local_path: Some(repo.to_path_buf()),
+                source: workspace_api::RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::LocalPath,
+                    uri: repo.display().to_string(),
+                },
+                source_revision: 1,
+                source_fingerprint: "sha256:test".to_string(),
                 selector: Some(RepositorySelector::from("HEAD")),
             },
             materializer: MaterializerKind::LocalGitWorktree,
@@ -2781,7 +2874,7 @@ mod tests {
             archive: bundle.profile_source_archive.clone().unwrap(),
         };
         factory
-            .resolve_profile_source_archive(&source)
+            .resolve_profile_source_archive(&source, None)
             .await
             .expect("embedded archive should resolve without Backend resource client");
     }

crates/worker-runtime/src/worker_source.rs

@@ -7,8 +7,9 @@ use worker::{
 };
 
 use crate::auth::{
-    RuntimeAuthError, RuntimeIdentityMaterial, RuntimeWorkerMutationSourceSigner,
-    WORKER_REMOVE_PERMISSION, WorkerMutationActorKind, WorkerMutationOperation,
+    RUNTIME_REQUEST_SOURCE_PROOF_HEADER, RuntimeAuthError, RuntimeIdentityMaterial,
+    RuntimeRequestSourceSigner, RuntimeWorkerMutationSourceSigner, WORKER_REMOVE_PERMISSION,
+    WORKSPACE_REQUEST_PERMISSION, WorkerMutationActorKind, WorkerMutationOperation,
     WorkerMutationSourceClaims, new_token_id,
 };
 use crate::runtime::RuntimeWorkspaceScope;
@@ -289,6 +290,8 @@ pub struct RuntimeOwnedWorkspaceClient {
     worker_id: String,
     request_timeout: Option<Duration>,
     worker_remove: Option<RuntimeWorkerMutationForwarder>,
+    request_source_signer: Option<RuntimeRequestSourceSigner>,
+    request_source_audience: Option<String>,
     prompt_projection_cache: Option<Arc<WorkspacePromptProjectionCache>>,
 }
 
@@ -306,6 +309,8 @@ impl RuntimeOwnedWorkspaceClient {
             worker_id: worker_id.into(),
             request_timeout: None,
             worker_remove: None,
+            request_source_signer: None,
+            request_source_audience: None,
             prompt_projection_cache: None,
         }
     }
@@ -315,6 +320,16 @@ impl RuntimeOwnedWorkspaceClient {
         self
     }
 
+    pub fn with_runtime_request_source(
+        mut self,
+        identity: &RuntimeIdentityMaterial,
+        audience: impl Into<String>,
+    ) -> Self {
+        self.request_source_signer = Some(RuntimeRequestSourceSigner::from_identity(identity));
+        self.request_source_audience = Some(audience.into());
+        self
+    }
+
     pub(crate) fn with_prompt_projection_cache(
         mut self,
         cache: Arc<WorkspacePromptProjectionCache>,
@@ -363,15 +378,21 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
         request: WorkspaceRequest,
     ) -> Result<WorkspaceResponse, WorkspaceClientError> {
         let base_url = self.base_url.clone();
+        let workspace_id = self.workspace_id.clone();
         let runtime_id = self.runtime_id.clone();
         let worker_id = self.worker_id.clone();
+        let request_source_signer = self.request_source_signer.clone();
+        let request_source_audience = self.request_source_audience.clone();
         let request_timeout = self.request_timeout;
         if tokio::runtime::Handle::try_current().is_ok() {
             std::thread::spawn(move || {
                 execute_runtime_owned_workspace_http(
                     &base_url,
+                    &workspace_id,
                     &runtime_id,
                     &worker_id,
+                    request_source_signer.as_ref(),
+                    request_source_audience.as_deref(),
                     request_timeout,
                     request,
                 )
@@ -383,8 +404,11 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
         } else {
             execute_runtime_owned_workspace_http(
                 &self.base_url,
+                &self.workspace_id,
                 &self.runtime_id,
                 &self.worker_id,
+                self.request_source_signer.as_ref(),
+                self.request_source_audience.as_deref(),
                 self.request_timeout,
                 request,
             )
@@ -484,8 +508,11 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
 
 fn execute_runtime_owned_workspace_http(
     base_url: &str,
+    workspace_id: &str,
     runtime_id: &str,
     worker_id: &str,
+    request_source_signer: Option<&RuntimeRequestSourceSigner>,
+    request_source_audience: Option<&str>,
     request_timeout: Option<Duration>,
     request: WorkspaceRequest,
 ) -> Result<WorkspaceResponse, WorkspaceClientError> {
@@ -510,11 +537,33 @@ fn execute_runtime_owned_workspace_http(
             ))
         })?;
     let request_label = format!("{method} {}", request.path);
+    let body = request.body.unwrap_or_default();
     let mut request_builder = client
-        .request(method, url)
+        .request(method.clone(), url)
         .header("x-yoi-runtime-id", runtime_id)
         .header("x-yoi-worker-id", worker_id);
-    if let Some(body) = request.body {
+    if let Some(signer) = request_source_signer {
+        let audience = request_source_audience.ok_or_else(|| {
+            WorkspaceClientError::Request(
+                "runtime request proof audience is unavailable".to_owned(),
+            )
+        })?;
+        let proof = signer
+            .issue(
+                audience,
+                workspace_id,
+                Some(worker_id),
+                WORKSPACE_REQUEST_PERMISSION,
+                method.as_str(),
+                &request.path,
+                body.as_bytes(),
+                i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX),
+                30,
+            )
+            .map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
+        request_builder = request_builder.header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, proof);
+    }
+    if !body.is_empty() {
         request_builder = request_builder
             .header(reqwest::header::CONTENT_TYPE, "application/json")
             .body(body);
@@ -590,8 +639,8 @@ fn unix_now_seconds() -> u64 {
 mod tests {
     use super::*;
     use crate::auth::{
-        WorkerMutationSourceExpectation, decode_worker_mutation_source_claims,
-        verify_worker_mutation_source_proof,
+        WorkerMutationSourceExpectation, decode_runtime_request_source_claims,
+        decode_worker_mutation_source_claims, verify_worker_mutation_source_proof,
     };
 
     #[test]
@@ -797,7 +846,7 @@ mod tests {
     }
 
     #[test]
-    fn ordinary_workspace_forwarding_stamps_legacy_source_only_inside_runtime() {
+    fn ordinary_workspace_forwarding_stamps_runtime_identity_and_signs_path_and_query() {
         use std::io::{Read, Write};
         use std::net::TcpListener;
         use std::sync::Mutex;
@@ -817,21 +866,41 @@ mod tests {
                 .unwrap();
         });
 
+        let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap();
         let client = RuntimeOwnedWorkspaceClient::new(
             "workspace-a",
             format!("http://{address}"),
             "runtime-a",
             "worker-a",
-        );
+        )
+        .with_runtime_request_source(&identity, "server-a");
         let response = client
-            .execute(WorkspaceRequest::get("/api/w/workspace-a/tickets/search"))
+            .execute(WorkspaceRequest::get(
+                "/api/w/workspace-a/tickets/search?state=planning&limit=20",
+            ))
             .unwrap();
         assert_eq!(response.status, 200);
         server.join().unwrap();
-        let request = received.lock().unwrap().to_ascii_lowercase();
-        assert!(request.contains("x-yoi-runtime-id: runtime-a"));
-        assert!(request.contains("x-yoi-worker-id: worker-a"));
-        assert!(!request.contains("authorization:"));
+        let request = received.lock().unwrap().clone();
+        let lowercase_request = request.to_ascii_lowercase();
+        assert!(lowercase_request.contains("x-yoi-runtime-id: runtime-a"));
+        assert!(lowercase_request.contains("x-yoi-worker-id: worker-a"));
+        assert!(lowercase_request.contains("x-yoi-runtime-request-proof: yoi-runtime-request-v1."));
+        assert!(!lowercase_request.contains("authorization:"));
+        let token = request
+            .lines()
+            .find_map(|line| {
+                line.split_once(':').and_then(|(name, value)| {
+                    name.eq_ignore_ascii_case(RUNTIME_REQUEST_SOURCE_PROOF_HEADER)
+                        .then(|| value.trim())
+                })
+            })
+            .expect("runtime proof header");
+        let claims = decode_runtime_request_source_claims(token).unwrap();
+        assert_eq!(
+            claims.path,
+            "/api/w/workspace-a/tickets/search?state=planning&limit=20"
+        );
     }
 
     #[test]

crates/worker-runtime/src/working_directory.rs

@@ -318,18 +318,36 @@ impl LocalGitWorktreeMaterializer {
                 ),
             ));
         }
-        if is_remote_uri(&request.repository.uri) {
-            return Err(WorkingDirectoryDiagnostic::new(
-                "working_directory_remote_repository_unsupported",
-                "remote repository URI materialization is not implemented in v0",
-            ));
-        }
-
-        let source_path = request
-            .repository
-            .local_path
-            .clone()
-            .unwrap_or_else(|| PathBuf::from(&request.repository.uri));
+        let source_path = match request.repository.source.kind {
+            workspace_api::RepositorySourceKind::LocalPath => {
+                PathBuf::from(&request.repository.source.uri)
+            }
+            workspace_api::RepositorySourceKind::File => {
+                url::Url::parse(&request.repository.source.uri)
+                    .ok()
+                    .and_then(|uri| uri.to_file_path().ok())
+                    .ok_or_else(|| {
+                        WorkingDirectoryDiagnostic::new(
+                            "working_directory_repository_source_invalid",
+                            "configured file Repository source is invalid",
+                        )
+                    })?
+            }
+            workspace_api::RepositorySourceKind::Ssh
+            | workspace_api::RepositorySourceKind::Http
+            | workspace_api::RepositorySourceKind::Https => {
+                return Err(WorkingDirectoryDiagnostic::new(
+                    "working_directory_remote_repository_access_required",
+                    "remote Repository materialization requires an explicit authenticated access and trust handle",
+                ));
+            }
+            workspace_api::RepositorySourceKind::Invalid => {
+                return Err(WorkingDirectoryDiagnostic::new(
+                    "working_directory_repository_source_invalid",
+                    "configured Repository source is invalid and cannot be materialized",
+                ));
+            }
+        };
         let source_root = git_stdout(&source_path, ["rev-parse", "--show-toplevel"])
             .map(|value| PathBuf::from(value.trim()))
             .map_err(|_| {
@@ -661,10 +679,6 @@ fn path_str(path: &Path) -> Result<String, WorkingDirectoryDiagnostic> {
     })
 }
 
-fn is_remote_uri(uri: &str) -> bool {
-    uri.contains("://") || uri.starts_with("git@") || uri.starts_with("ssh:")
-}
-
 fn sanitize_path_component(value: &str) -> String {
     let sanitized = value
         .chars()
@@ -793,8 +807,12 @@ mod tests {
             repository: WorkingDirectoryRepository {
                 id: "repo-main".to_string(),
                 provider: "git".to_string(),
-                uri: ".".to_string(),
-                local_path: Some(repo.to_path_buf()),
+                source: workspace_api::RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::LocalPath,
+                    uri: repo.display().to_string(),
+                },
+                source_revision: 1,
+                source_fingerprint: "sha256:test".to_string(),
                 selector: Some(RepositorySelector::from("HEAD")),
             },
             materializer: MaterializerKind::LocalGitWorktree,
@@ -908,19 +926,21 @@ mod tests {
         let runtime_root = tempfile::tempdir().unwrap();
         let materializer = LocalGitWorktreeMaterializer::new(runtime_root.path());
         let mut remote = request(Path::new("."));
-        remote.repository.local_path = None;
-        remote.repository.uri = "https://example.invalid/repo.git".to_string();
+        remote.repository.source = workspace_api::RepositorySource {
+            kind: workspace_api::RepositorySourceKind::Https,
+            uri: "https://example.invalid/repo.git".to_string(),
+        };
         let error = materializer
             .materialize(&worker_ref(1), &remote)
             .unwrap_err();
         assert_eq!(
             error.code,
-            "working_directory_remote_repository_unsupported"
+            "working_directory_remote_repository_access_required"
         );
 
         let mut non_git = remote;
         non_git.repository.provider = "archive".to_string();
-        non_git.repository.uri = ".".to_string();
+        non_git.repository.source.uri = ".".to_string();
         let error = materializer
             .materialize(&worker_ref(2), &non_git)
             .unwrap_err();

crates/worker/src/feature/builtin/manage_workdir.rs

@@ -398,24 +398,35 @@ impl WorkspaceHttpWorkdirBackend {
         workdir_output(format!("Listed {count} Workdir(s)"), &response)
     }
 
-    fn create(&self, input: WorkdirCreateInput) -> Result<ToolOutput, ToolError> {
-        let runtime_id = validate_identity(&input.runtime_id, CREATE_TOOL, "runtime_id")?;
+    fn create(
+        &self,
+        input: WorkdirCreateInput,
+        operation_id: String,
+    ) -> Result<ToolOutput, ToolError> {
+        let runtime_id = input
+            .runtime_id
+            .as_deref()
+            .map(|value| validate_identity(value, CREATE_TOOL, "runtime_id"))
+            .transpose()?;
         let repository_id = validate_identity(&input.repository_id, CREATE_TOOL, "repository_id")?;
         let selector = validate_optional_selector(input.selector)?;
         let workspace_id = encode_path_segment(self.workspace_id()?);
-        let runtime_path = encode_path_segment(runtime_id);
         let request = WorkdirCreateRequest {
-            runtime_id: runtime_id.to_string(),
+            runtime_id: runtime_id.map(str::to_string),
             repository_id: repository_id.to_string(),
             selector,
+            operation_id,
         };
         let response = self.execute_json::<WorkdirDetailResponse>(WorkspaceRequest::json(
             WorkspaceRequestMethod::Post,
-            format!("/api/w/{workspace_id}/runtimes/{runtime_path}/working-directories"),
+            format!("/api/w/{workspace_id}/working-directories"),
             serde_json::to_string(&request).map_err(decode_error)?,
         ))?;
         workdir_output(
-            format!("Created Workdir {}", response.item.working_directory_id),
+            format!(
+                "Created Workdir {} on Runtime {}",
+                response.item.working_directory_id, response.runtime_id
+            ),
             &response,
         )
     }
@@ -518,16 +529,17 @@ impl Tool for WorkspaceHttpWorkdirTool {
     async fn execute(
         &self,
         input_json: &str,
-        _ctx: ToolExecutionContext,
+        ctx: ToolExecutionContext,
     ) -> Result<ToolOutput, ToolError> {
         match self.operation {
             WorkdirOperation::List => {
                 let _input = parse_input::<WorkdirListInput>(input_json)?;
                 self.backend.list()
             }
-            WorkdirOperation::Create => self
-                .backend
-                .create(parse_input::<WorkdirCreateInput>(input_json)?),
+            WorkdirOperation::Create => self.backend.create(
+                parse_input::<WorkdirCreateInput>(input_json)?,
+                ctx.call_id.to_string(),
+            ),
             WorkdirOperation::Attach => self
                 .backend
                 .attach(parse_input::<WorkdirAttachInput>(input_json)?),
@@ -635,9 +647,9 @@ fn create_schema() -> serde_json::Value {
     json!({
         "type": "object",
         "additionalProperties": false,
-        "required": ["runtime_id", "repository_id"],
+        "required": ["repository_id"],
         "properties": {
-            "runtime_id": {"type": "string", "minLength": 1},
+            "runtime_id": {"type": ["string", "null"], "minLength": 1},
             "repository_id": {"type": "string", "minLength": 1},
             "selector": {"type": ["string", "null"], "minLength": 1}
         }
@@ -677,7 +689,8 @@ struct WorkdirListInput {}
 #[derive(Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct WorkdirCreateInput {
-    runtime_id: String,
+    #[serde(default)]
+    runtime_id: Option<String>,
     repository_id: String,
     #[serde(default)]
     selector: Option<String>,
@@ -685,10 +698,12 @@ struct WorkdirCreateInput {
 
 #[derive(Debug, Serialize)]
 struct WorkdirCreateRequest {
-    runtime_id: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    runtime_id: Option<String>,
     repository_id: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     selector: Option<String>,
+    operation_id: String,
 }
 
 #[derive(Debug, Deserialize)]
@@ -916,7 +931,11 @@ mod tests {
     #[test]
     fn schemas_expose_identities_without_paths_or_session_handles() {
         let create = create_schema();
-        assert_eq!(create["required"], json!(["runtime_id", "repository_id"]));
+        assert_eq!(create["required"], json!(["repository_id"]));
+        assert_eq!(
+            create["properties"]["runtime_id"]["type"],
+            json!(["string", "null"])
+        );
         assert!(create["properties"].get("path").is_none());
         assert!(create["properties"].get("session_id").is_none());
         assert_eq!(attach_schema()["required"], json!(["workdir_id"]));
@@ -946,6 +965,7 @@ mod tests {
             })),
             response(json!({
                 "workspace_id": "workspace/test",
+                "runtime_id": "runtime/one",
                 "item": workdir_json("wd-created"),
                 "diagnostics": []
             })),
@@ -961,6 +981,7 @@ mod tests {
             })),
             response(json!({
                 "workspace_id": "workspace/test",
+                "runtime_id": "runtime/one",
                 "item": {
                     "working_directory_id": "wd-created",
                     "repository_id": "main",
@@ -985,13 +1006,19 @@ mod tests {
                 .is_none()
         );
         let created = backend
-            .create(WorkdirCreateInput {
-                runtime_id: "runtime/one".to_string(),
-                repository_id: "main".to_string(),
-                selector: Some("refs/heads/topic".to_string()),
-            })
+            .create(
+                WorkdirCreateInput {
+                    runtime_id: Some("runtime/one".to_string()),
+                    repository_id: "main".to_string(),
+                    selector: Some("refs/heads/topic".to_string()),
+                },
+                "call-create-1".to_string(),
+            )
             .unwrap();
-        assert_eq!(created.summary, "Created Workdir wd-created");
+        assert_eq!(
+            created.summary,
+            "Created Workdir wd-created on Runtime runtime/one"
+        );
         let created: serde_json::Value =
             serde_json::from_str(created.content.as_deref().unwrap()).unwrap();
         assert_eq!(created["item"]["working_directory_id"], "wd-created");
@@ -1017,12 +1044,14 @@ mod tests {
         assert_eq!(requests[0].method, WorkspaceRequestMethod::Get);
         assert_eq!(
             requests[1].path,
-            "/api/w/workspace%2Ftest/runtimes/runtime%2Fone/working-directories"
+            "/api/w/workspace%2Ftest/working-directories"
         );
         assert_eq!(requests[1].method, WorkspaceRequestMethod::Post);
         let body: serde_json::Value =
             serde_json::from_str(requests[1].body.as_deref().unwrap()).unwrap();
         assert_eq!(body["repository_id"], "main");
+        assert_eq!(body["runtime_id"], "runtime/one");
+        assert_eq!(body["operation_id"], "call-create-1");
         assert_eq!(body["selector"], "refs/heads/topic");
         assert_eq!(
             requests[2].path,
@@ -1216,16 +1245,50 @@ mod tests {
         assert_eq!(body["operation"]["request"]["path"], "file");
     }
 
+    #[test]
+    fn create_omits_runtime_for_backend_default_resolution() {
+        let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
+            "workspace_id": "workspace/test",
+            "runtime_id": "arcadia",
+            "item": workdir_json("wd-default"),
+            "diagnostics": []
+        }))]));
+        let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
+
+        let created = backend
+            .create(
+                WorkdirCreateInput {
+                    runtime_id: None,
+                    repository_id: "main".to_string(),
+                    selector: None,
+                },
+                "call-default".to_string(),
+            )
+            .unwrap();
+        assert_eq!(
+            created.summary,
+            "Created Workdir wd-default on Runtime arcadia"
+        );
+        let requests = client.requests();
+        let body: serde_json::Value =
+            serde_json::from_str(requests[0].body.as_deref().unwrap()).unwrap();
+        assert!(body.get("runtime_id").is_none());
+        assert_eq!(body["operation_id"], "call-default");
+    }
+
     #[test]
     fn invalid_or_extra_inputs_are_rejected_before_workspace_request() {
         let client = Arc::new(RecordingWorkspaceClient::new(Vec::new()));
         let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
         let error = backend
-            .create(WorkdirCreateInput {
-                runtime_id: " ".to_string(),
-                repository_id: "main".to_string(),
-                selector: None,
-            })
+            .create(
+                WorkdirCreateInput {
+                    runtime_id: Some(" ".to_string()),
+                    repository_id: "main".to_string(),
+                    selector: None,
+                },
+                "call-invalid".to_string(),
+            )
             .unwrap_err();
         assert!(matches!(error, ToolError::InvalidArgument(_)));
         assert!(client.requests().is_empty());

crates/workspace-api/src/lib.rs

@@ -7,6 +7,88 @@
 use serde::{Deserialize, Serialize};
 use workdir::workspace::WorkingDirectorySummary;
 
+/// Provider-neutral classification of an authoritative Repository source.
+///
+/// Local paths remain distinct from network Git transports so callers cannot
+/// accidentally treat an unmaterialized remote as a server-local filesystem path.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RepositorySourceKind {
+    LocalPath,
+    File,
+    Ssh,
+    Http,
+    Https,
+    /// A legacy value that could not be classified during migration. It remains
+    /// inspectable but every provider operation must fail closed.
+    Invalid,
+}
+
+impl RepositorySourceKind {
+    pub const fn is_remote(self) -> bool {
+        matches!(self, Self::Ssh | Self::Http | Self::Https)
+    }
+
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::LocalPath => "local_path",
+            Self::File => "file",
+            Self::Ssh => "ssh",
+            Self::Http => "http",
+            Self::Https => "https",
+            Self::Invalid => "invalid",
+        }
+    }
+
+    pub fn parse(value: &str) -> Option<Self> {
+        Some(match value {
+            "local_path" => Self::LocalPath,
+            "file" => Self::File,
+            "ssh" => Self::Ssh,
+            "http" => Self::Http,
+            "https" => Self::Https,
+            "invalid" => Self::Invalid,
+            _ => return None,
+        })
+    }
+}
+
+/// Stable Repository source identity stored by Workspace authority.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RepositorySource {
+    pub kind: RepositorySourceKind,
+    /// Canonical source representation. This is an absolute local path for
+    /// `local_path`, and a normalized URI/remote specification otherwise.
+    pub uri: String,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RepositoryObservedStatus {
+    Unverified,
+    Ready,
+    Invalid,
+}
+
+impl RepositoryObservedStatus {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Unverified => "unverified",
+            Self::Ready => "ready",
+            Self::Invalid => "invalid",
+        }
+    }
+
+    pub fn parse(value: &str) -> Option<Self> {
+        Some(match value {
+            "unverified" => Self::Unverified,
+            "ready" => Self::Ready,
+            "invalid" => Self::Invalid,
+            _ => return None,
+        })
+    }
+}
+
 pub const TICKET_RELATIONS_QUERY_PATH: &str = "/tickets/relations/search";
 pub const TICKET_ORCHESTRATION_PLANS_QUERY_PATH: &str = "/tickets/orchestration-plans/search";
 

crates/workspace-server/src/auth.rs

@@ -176,8 +176,28 @@ fn actor_for_user<S: ControlPlaneStore + ?Sized>(
     }))
 }
 
-pub fn session_set_cookie(cookie_name: &str, token: &str, max_age_seconds: i64) -> String {
-    format!("{cookie_name}={token}; Max-Age={max_age_seconds}; Path=/; HttpOnly; SameSite=Lax")
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct SessionCookiePolicy<'a> {
+    pub cookie_name: &'a str,
+    pub path: &'a str,
+    pub domain: Option<&'a str>,
+    pub secure: bool,
+}
+
+pub fn session_set_cookie(
+    policy: SessionCookiePolicy<'_>,
+    token: &str,
+    max_age_seconds: i64,
+) -> String {
+    let domain = policy
+        .domain
+        .map(|domain| format!("; Domain={domain}"))
+        .unwrap_or_default();
+    let secure = if policy.secure { "; Secure" } else { "" };
+    format!(
+        "{}={token}; Max-Age={max_age_seconds}; Path={}; HttpOnly; SameSite=Lax{domain}{secure}",
+        policy.cookie_name, policy.path
+    )
 }
 
 pub fn auth_error(code: &str, message: &str) -> Error {

crates/workspace-server/src/config.rs

@@ -471,14 +471,18 @@ fn resolve_repository(
     let provider =
         normalize_required_string("repository provider", &config.provider)?.to_ascii_lowercase();
     let uri = normalize_required_string("repository uri", &config.uri)?;
-    let path = resolve_repository_uri(workspace_root, &id, &uri)?;
+    let (source, path) = resolve_repository_source(workspace_root, &id, &uri)?;
     let display_name = normalize_optional_string(config.display_name.as_deref());
     let default_selector = normalize_optional_string(config.default_selector.as_deref());
 
     Ok(ConfiguredRepository {
         id,
         provider,
-        uri,
+        source_fingerprint: crate::repository_source::repository_source_fingerprint(&source),
+        source,
+        source_revision: 1,
+        observed_status: workspace_api::RepositoryObservedStatus::Unverified,
+        observed_at: None,
         path,
         display_name,
         default_selector,
@@ -517,13 +521,41 @@ fn validate_repository_id(id: &str) -> Result<()> {
     }
 }
 
-fn resolve_repository_uri(workspace_root: &Path, id: &str, uri: &str) -> Result<PathBuf> {
-    if uri.contains("://") {
-        return Err(Error::Config(format!(
-            "repository `{id}` uses a remote URI, but remote repository materialization is not implemented"
-        )));
+fn resolve_repository_source(
+    workspace_root: &Path,
+    id: &str,
+    uri: &str,
+) -> Result<(workspace_api::RepositorySource, Option<PathBuf>)> {
+    match crate::repository_source::parse_repository_source(uri) {
+        Ok(source) => {
+            let path = match source.kind {
+                workspace_api::RepositorySourceKind::LocalPath => Some(PathBuf::from(&source.uri)),
+                workspace_api::RepositorySourceKind::File => url::Url::parse(&source.uri)
+                    .ok()
+                    .and_then(|uri| uri.to_file_path().ok()),
+                workspace_api::RepositorySourceKind::Ssh
+                | workspace_api::RepositorySourceKind::Http
+                | workspace_api::RepositorySourceKind::Https => None,
+                workspace_api::RepositorySourceKind::Invalid => {
+                    return Err(Error::Config(format!(
+                        "repository `{id}` has an invalid source"
+                    )));
+                }
+            };
+            Ok((source, path))
+        }
+        Err(_) if !Path::new(uri).is_absolute() && !uri.contains("://") => {
+            let path = resolve_workspace_path(workspace_root, Path::new(uri));
+            let source = workspace_api::RepositorySource {
+                kind: workspace_api::RepositorySourceKind::LocalPath,
+                uri: path.to_string_lossy().into_owned(),
+            };
+            Ok((source, Some(path)))
+        }
+        Err(error) => Err(Error::Config(format!(
+            "repository `{id}` has an invalid source: {error}"
+        ))),
     }
-    Ok(resolve_workspace_path(workspace_root, Path::new(uri)))
 }
 
 pub(crate) fn resolve_remote_runtime(
@@ -741,13 +773,13 @@ default_selector = "HEAD"
 
         assert_eq!(repository.id, "main");
         assert_eq!(repository.provider, "git");
-        assert_eq!(repository.path, dir.path());
+        assert_eq!(repository.path.as_deref(), Some(dir.path()));
         assert_eq!(repository.display_name.as_deref(), Some("Main"));
         assert_eq!(repository.default_selector.as_deref(), Some("HEAD"));
     }
 
     #[test]
-    fn remote_repository_uri_fails_closed() {
+    fn remote_repository_source_is_preserved_without_a_local_path() {
         let dir = tempfile::tempdir().unwrap();
         let config = WorkspaceBackendConfigFile::parse_str(
             r#"
@@ -759,17 +791,15 @@ uri = "https://example.com/org/repo.git"
             "test",
         )
         .unwrap();
-        let error = match config.resolve(dir.path(), identity()) {
-            Ok(_) => panic!("remote repository URI should fail closed"),
-            Err(error) => error,
-        };
+        let resolved = config.resolve(dir.path(), identity()).unwrap();
+        let repository = &resolved.server.repositories[0];
 
-        assert!(
-            error
-                .to_string()
-                .contains("remote repository materialization is not implemented"),
-            "unexpected error: {error}"
+        assert_eq!(
+            repository.source.kind,
+            workspace_api::RepositorySourceKind::Https
         );
+        assert_eq!(repository.source.uri, "https://example.com/org/repo.git");
+        assert!(repository.path.is_none());
     }
 
     #[test]

crates/workspace-server/src/lib.rs

@@ -20,12 +20,15 @@ pub mod records;
 #[cfg(feature = "typescript")]
 pub use records::ticket_api_typescript;
 pub mod repositories;
+pub mod repository_source;
 pub mod resource_broker;
 pub mod retention;
+pub mod runtime_settings;
 pub mod runtime_subscription;
 pub mod server;
 pub mod skills;
 pub mod store;
+pub mod workdir_create_operations;
 pub mod worker_source;
 pub mod workspace_catalog;
 mod workspace_subscription;

crates/workspace-server/src/main.rs

@@ -711,14 +711,15 @@ fn infer_workspace_root_from_repositories(
         )));
     };
 
-    let repository_path = PathBuf::from(&repository.uri);
-    if !repository_path.is_absolute() {
+    if repository.source.kind == workspace_api::RepositorySourceKind::Invalid {
         return Err(CliError(format!(
-            "repository `{}` has relative URI `{}`; repository records used by serve must be absolute paths",
-            repository.repository_id, repository.uri
+            "repository `{}` has an invalid migrated source and cannot be used by serve",
+            repository.repository_id
         )));
     }
-    Ok(repository_path)
+    Ok(ServerConfig::default_workspace_backend_data_root(
+        &workspace.workspace_id,
+    ))
 }
 
 fn parse_config_command(args: &[String]) -> Result<Command, CliError> {

crates/workspace-server/src/repositories.rs

@@ -5,6 +5,7 @@ use std::{
 };
 
 use serde::{Deserialize, Serialize};
+use workspace_api::{RepositoryObservedStatus, RepositorySource};
 
 pub type RepositoryId = String;
 pub type RepositorySelector = String;
@@ -13,8 +14,12 @@ pub type RepositorySelector = String;
 pub struct ConfiguredRepository {
     pub id: RepositoryId,
     pub provider: String,
-    pub uri: String,
-    pub path: PathBuf,
+    pub source: RepositorySource,
+    pub source_revision: u64,
+    pub source_fingerprint: String,
+    pub observed_status: RepositoryObservedStatus,
+    pub observed_at: Option<String>,
+    pub path: Option<PathBuf>,
     pub display_name: Option<String>,
     pub default_selector: Option<RepositorySelector>,
 }
@@ -25,6 +30,12 @@ pub struct RepositorySummary {
     pub display_name: String,
     pub kind: String,
     pub provider: String,
+    pub source: RepositorySource,
+    pub source_revision: u64,
+    pub source_fingerprint: String,
+    pub observed_status: RepositoryObservedStatus,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub observed_at: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub default_selector: Option<RepositorySelector>,
     pub record_authority: String,
@@ -262,9 +273,17 @@ impl RepositoryRegistryReader {
         descendant: &str,
     ) -> Result<(), RepositoryLookupError> {
         let repository = self.merge_repository(id)?;
+        let repository_path =
+            repository
+                .path
+                .as_ref()
+                .ok_or_else(|| RepositoryLookupError::ProviderFailure {
+                    id: id.into(),
+                    operation: "repository source is not materialized for local Git access".into(),
+                })?;
         let status = Command::new("git")
             .arg("-C")
-            .arg(&repository.path)
+            .arg(repository_path)
             .args(["merge-base", "--is-ancestor", ancestor, descendant])
             .status()
             .map_err(|_| RepositoryLookupError::ProviderFailure {
@@ -306,7 +325,24 @@ impl RepositoryRegistryReader {
             .clone()
             .unwrap_or_else(|| repository.id.clone());
         let mut diagnostics = Vec::new();
+        if repository.source.kind == workspace_api::RepositorySourceKind::Http {
+            diagnostics.push(RepositoryDiagnostic {
+                severity: "warning".to_string(),
+                code: "repository_source_insecure_http".to_string(),
+                message:
+                    "HTTP Repository source is unencrypted; prefer HTTPS or SSH when available."
+                        .to_string(),
+            });
+        }
         let git = match repository.provider.as_str() {
+            "git" if repository.path.is_none() => {
+                diagnostics.push(RepositoryDiagnostic {
+                    severity: "info".to_string(),
+                    code: "repository_source_unverified".to_string(),
+                    message: "Remote Repository source is registered but is not materialized for server-local inspection.".to_string(),
+                });
+                None
+            }
             "git" => match self.inspect_git(repository) {
                 Ok(git) => Some(git),
                 Err(message) => {
@@ -335,6 +371,11 @@ impl RepositoryRegistryReader {
             display_name,
             kind: repository.provider.clone(),
             provider: repository.provider.clone(),
+            source: repository.source.clone(),
+            source_revision: repository.source_revision,
+            source_fingerprint: repository.source_fingerprint.clone(),
+            observed_status: repository.observed_status,
+            observed_at: repository.observed_at.clone(),
             default_selector: repository.default_selector.clone(),
             record_authority: "workspace-control-plane".to_string(),
             git,
@@ -346,12 +387,15 @@ impl RepositoryRegistryReader {
         &self,
         repository: &ConfiguredRepository,
     ) -> Result<GitRepositorySummary, String> {
-        let head = git_stdout(&repository.path, ["rev-parse", "HEAD"])?;
-        let branch = git_stdout(&repository.path, ["branch", "--show-current"])
+        let path = repository.path.as_ref().ok_or_else(|| {
+            "Repository source is not materialized for local Git inspection.".to_string()
+        })?;
+        let head = git_stdout(path, ["rev-parse", "HEAD"])?;
+        let branch = git_stdout(path, ["branch", "--show-current"])
             .ok()
             .and_then(|value| non_empty_string(value.trim()));
-        let status = git_stdout(&repository.path, ["status", "--porcelain"])?;
-        let remotes = git_stdout(&repository.path, ["remote", "-v"])
+        let status = git_stdout(path, ["status", "--porcelain"])?;
+        let remotes = git_stdout(path, ["remote", "-v"])
             .map(|raw| parse_remotes(&raw))
             .unwrap_or_default();
         Ok(GitRepositorySummary {
@@ -369,8 +413,11 @@ impl RepositoryRegistryReader {
         limit: usize,
     ) -> Result<Vec<GitCommitSummary>, String> {
         let limit_arg = format!("-{limit}");
+        let path = repository.path.as_ref().ok_or_else(|| {
+            "Repository source is not materialized for local Git log access.".to_string()
+        })?;
         let output = git_stdout(
-            &repository.path,
+            path,
             [
                 "log",
                 "--date=iso-strict",
@@ -419,11 +466,16 @@ fn merge_git_stdout(
     operation: &str,
     args: &[&str],
 ) -> Result<String, RepositoryLookupError> {
-    git_stdout(&repository.path, args.iter().copied()).map_err(|_| {
-        RepositoryLookupError::ProviderFailure {
+    let path = repository
+        .path
+        .as_ref()
+        .ok_or_else(|| RepositoryLookupError::ProviderFailure {
             id: repository.id.clone(),
-            operation: operation.into(),
-        }
+            operation: "repository source is not materialized for local Git access".into(),
+        })?;
+    git_stdout(path, args.iter().copied()).map_err(|_| RepositoryLookupError::ProviderFailure {
+        id: repository.id.clone(),
+        operation: operation.into(),
     })
 }
 
@@ -593,6 +645,47 @@ mod tests {
         assert_eq!(projection.diagnostics[0].code, "repository_config_empty");
     }
 
+    #[test]
+    fn remote_source_is_visible_but_local_provider_operations_fail_closed() {
+        let source = RepositorySource {
+            kind: workspace_api::RepositorySourceKind::Ssh,
+            uri: "git@example.test:org/repository.git".to_string(),
+        };
+        let reader = RepositoryRegistryReader::new(vec![ConfiguredRepository {
+            id: "remote".into(),
+            display_name: Some("Remote".into()),
+            provider: "git".into(),
+            source_fingerprint: crate::repository_source::repository_source_fingerprint(&source),
+            source,
+            source_revision: 1,
+            observed_status: RepositoryObservedStatus::Unverified,
+            observed_at: None,
+            path: None,
+            default_selector: Some("main".into()),
+        }]);
+
+        let projection = reader.list();
+        let summary = &projection.items[0];
+        assert_eq!(
+            summary.source.kind,
+            workspace_api::RepositorySourceKind::Ssh
+        );
+        assert_eq!(
+            summary.observed_status,
+            RepositoryObservedStatus::Unverified
+        );
+        assert!(summary.git.is_none());
+        assert_eq!(summary.diagnostics[0].code, "repository_source_unverified");
+
+        let repository = reader.merge_repository("remote").unwrap();
+        let error = merge_git_stdout(&repository, "inspect", &["rev-parse", "HEAD"]).unwrap_err();
+        assert!(matches!(
+            error,
+            RepositoryLookupError::ProviderFailure { operation, .. }
+                if operation.contains("not materialized")
+        ));
+    }
+
     #[test]
     fn merge_evidence_is_resolved_by_repository_identity() {
         let temp = tempfile::tempdir().unwrap();
@@ -675,12 +768,22 @@ mod tests {
                 .success()
         );
 
+        let source_descriptor = RepositorySource {
+            kind: workspace_api::RepositorySourceKind::LocalPath,
+            uri: path.display().to_string(),
+        };
         let reader = RepositoryRegistryReader::new(vec![ConfiguredRepository {
             id: "main".into(),
             display_name: Some("Main".into()),
             provider: "git".into(),
-            path: path.to_path_buf(),
-            uri: path.display().to_string(),
+            source_fingerprint: crate::repository_source::repository_source_fingerprint(
+                &source_descriptor,
+            ),
+            source: source_descriptor,
+            source_revision: 1,
+            observed_status: RepositoryObservedStatus::Unverified,
+            observed_at: None,
+            path: Some(path.to_path_buf()),
             default_selector: Some("main".into()),
         }]);
         let target = reader.observe_merge_target("main", Some("main")).unwrap();

crates/workspace-server/src/repository_source.rs

@@ -0,0 +1,230 @@
+use std::path::Path;
+
+use sha2::{Digest, Sha256};
+use url::Url;
+use workspace_api::{RepositorySource, RepositorySourceKind};
+
+use crate::{Error, Result};
+
+const MAX_REPOSITORY_SOURCE_BYTES: usize = 4096;
+
+/// Parse and canonicalize a user-authored Git source without accessing the
+/// filesystem or network.
+pub fn parse_repository_source(value: &str) -> Result<RepositorySource> {
+    let value = value.trim();
+    if value.is_empty() || value.len() > MAX_REPOSITORY_SOURCE_BYTES {
+        return Err(Error::InvalidInput(format!(
+            "initial repository source must be between 1 and {MAX_REPOSITORY_SOURCE_BYTES} bytes"
+        )));
+    }
+    if value.chars().any(char::is_control) {
+        return Err(Error::InvalidInput(
+            "initial repository source must not contain control characters".to_string(),
+        ));
+    }
+
+    if Path::new(value).is_absolute() {
+        return Ok(RepositorySource {
+            kind: RepositorySourceKind::LocalPath,
+            uri: value.to_string(),
+        });
+    }
+
+    if is_scp_like_ssh(value) {
+        validate_scp_like_ssh(value)?;
+        return Ok(RepositorySource {
+            kind: RepositorySourceKind::Ssh,
+            uri: value.to_string(),
+        });
+    }
+
+    let parsed = Url::parse(value).map_err(|_| {
+        Error::InvalidInput(
+            "initial repository source must be an absolute local path or a supported Git URI"
+                .to_string(),
+        )
+    })?;
+    if parsed.query().is_some() || parsed.fragment().is_some() {
+        return Err(Error::InvalidInput(
+            "initial repository source must not contain query parameters or fragments".to_string(),
+        ));
+    }
+    if parsed.password().is_some() {
+        return Err(Error::InvalidInput(
+            "initial repository source must not embed a password or token".to_string(),
+        ));
+    }
+
+    let kind = match parsed.scheme() {
+        "file" => {
+            if !parsed.username().is_empty() {
+                return Err(Error::InvalidInput(
+                    "file repository URI must not contain user information".to_string(),
+                ));
+            }
+            if parsed.host_str().is_some_and(|host| host != "localhost") {
+                return Err(Error::InvalidInput(
+                    "file repository URI host must be empty or localhost".to_string(),
+                ));
+            }
+            parsed.to_file_path().map_err(|_| {
+                Error::InvalidInput("file repository URI must contain an absolute path".to_string())
+            })?;
+            RepositorySourceKind::File
+        }
+        "ssh" => {
+            require_remote_host_and_path(&parsed)?;
+            RepositorySourceKind::Ssh
+        }
+        "http" | "https" => {
+            if !parsed.username().is_empty() {
+                return Err(Error::InvalidInput(
+                    "HTTP repository URI must not contain user information".to_string(),
+                ));
+            }
+            require_remote_host_and_path(&parsed)?;
+            if parsed.scheme() == "http" {
+                RepositorySourceKind::Http
+            } else {
+                RepositorySourceKind::Https
+            }
+        }
+        scheme => {
+            return Err(Error::InvalidInput(format!(
+                "unsupported initial repository source scheme `{scheme}`"
+            )));
+        }
+    };
+
+    Ok(RepositorySource {
+        kind,
+        uri: parsed.to_string(),
+    })
+}
+
+/// Classify persisted pre-source-contract rows without guessing a usable remote
+/// when the legacy value is malformed. No filesystem or network access occurs.
+pub fn classify_legacy_repository_source(value: &str) -> RepositorySource {
+    parse_repository_source(value).unwrap_or_else(|_| RepositorySource {
+        kind: RepositorySourceKind::Invalid,
+        uri: value.trim().to_string(),
+    })
+}
+
+pub fn repository_source_fingerprint(source: &RepositorySource) -> String {
+    let payload = serde_json::to_vec(source).expect("Repository source serializes");
+    let mut hasher = Sha256::new();
+    hasher.update(b"yoi.repository-source.v1\0");
+    hasher.update(payload);
+    let digest = hasher.finalize();
+    let mut encoded = String::with_capacity(digest.len() * 2);
+    for byte in digest {
+        use std::fmt::Write as _;
+        write!(&mut encoded, "{byte:02x}").expect("writing to String cannot fail");
+    }
+    format!("sha256:{encoded}")
+}
+
+fn require_remote_host_and_path(parsed: &Url) -> Result<()> {
+    if parsed.host_str().is_none() || parsed.path().is_empty() || parsed.path() == "/" {
+        return Err(Error::InvalidInput(
+            "remote repository URI must contain a host and repository path".to_string(),
+        ));
+    }
+    Ok(())
+}
+
+fn is_scp_like_ssh(value: &str) -> bool {
+    !value.contains("://")
+        && value
+            .split_once(':')
+            .is_some_and(|(identity, _)| identity.contains('@'))
+}
+
+fn validate_scp_like_ssh(value: &str) -> Result<()> {
+    let (identity, path) = value.split_once(':').ok_or_else(|| {
+        Error::InvalidInput("scp-like SSH source must contain `host:path`".to_string())
+    })?;
+    let (username, host) = identity.split_once('@').ok_or_else(|| {
+        Error::InvalidInput("scp-like SSH source must contain `user@host:path`".to_string())
+    })?;
+    if username.is_empty()
+        || host.is_empty()
+        || path.is_empty()
+        || username.contains('@')
+        || username.contains(':')
+        || host.contains('@')
+        || path.starts_with('-')
+        || value.contains('?')
+        || value.contains('#')
+        || value.chars().any(char::is_whitespace)
+    {
+        return Err(Error::InvalidInput(
+            "scp-like SSH source must use `user@host:path` without credentials or parameters"
+                .to_string(),
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_local_file_ssh_http_and_https_sources_without_io() {
+        let cases = [
+            ("/runtime/repos/project", RepositorySourceKind::LocalPath),
+            ("file:///runtime/repos/project", RepositorySourceKind::File),
+            (
+                "ssh://git@example.test/org/project.git",
+                RepositorySourceKind::Ssh,
+            ),
+            (
+                "git@example.test:org/project.git",
+                RepositorySourceKind::Ssh,
+            ),
+            (
+                "http://git.test/org/project.git",
+                RepositorySourceKind::Http,
+            ),
+            (
+                "https://git.test/org/project.git",
+                RepositorySourceKind::Https,
+            ),
+        ];
+        for (source, expected_kind) in cases {
+            assert_eq!(parse_repository_source(source).unwrap().kind, expected_kind);
+        }
+    }
+
+    #[test]
+    fn rejects_relative_unsupported_and_credential_bearing_sources() {
+        for source in [
+            "relative/project",
+            "ftp://git.test/project.git",
+            "https://user@git.test/project.git",
+            "https://git.test/project.git?token=secret",
+            "ssh://git:secret@git.test/project.git",
+            "git@example.test:",
+            "git:secret@example.test:org/project.git",
+            "https://git.test/project.git\nother",
+        ] {
+            assert!(
+                parse_repository_source(source).is_err(),
+                "accepted {source:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn fingerprint_uses_canonical_source_identity() {
+        let first = parse_repository_source(" https://EXAMPLE.test/a/../project.git ").unwrap();
+        let second = parse_repository_source("https://example.test/project.git").unwrap();
+        assert_eq!(first, second);
+        assert_eq!(
+            repository_source_fingerprint(&first),
+            repository_source_fingerprint(&second)
+        );
+    }
+}

crates/workspace-server/src/runtime_settings.rs

@@ -0,0 +1,229 @@
+use config_source::ConfigSchemaContribution;
+use serde::Deserialize;
+
+use crate::config_source::{
+    WorkspaceConfigSchemaProvider, WorkspaceConfigState, evaluate_workspace_config_state,
+};
+use crate::{Error, Result};
+
+const RUNTIME_SCHEMA_SOURCE: &str = r#"{
+    runtime = {
+        default_runtime_id = String default "";
+    };
+}"#;
+
+#[derive(Debug, Default)]
+pub struct RuntimeConfigSchemaProvider;
+
+impl WorkspaceConfigSchemaProvider for RuntimeConfigSchemaProvider {
+    fn contribution(&self) -> Result<ConfigSchemaContribution> {
+        ConfigSchemaContribution::new("builtin:runtime", "runtime", "1", RUNTIME_SCHEMA_SOURCE)
+            .map_err(|error| Error::Config(error.to_string()))
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RuntimeConfigProjection {
+    pub config_revision: u64,
+    pub projection_digest: String,
+    pub default_runtime_id: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+struct VirtualRuntimeConfig {
+    runtime: VirtualRuntimeSection,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct VirtualRuntimeSection {
+    default_runtime_id: String,
+}
+
+pub fn project_runtime_from_workspace_config(
+    workspace_id: &str,
+    state: &WorkspaceConfigState,
+) -> Result<RuntimeConfigProjection> {
+    let has_runtime_schema = state
+        .contract
+        .schema_bundle
+        .contributions
+        .iter()
+        .any(|entry| entry.provider_id == "builtin:runtime");
+    if !has_runtime_schema {
+        return Ok(RuntimeConfigProjection {
+            config_revision: state.snapshot.revision,
+            projection_digest: state.projection_digest.clone(),
+            default_runtime_id: None,
+        });
+    }
+
+    let evaluation = evaluate_workspace_config_state(state, state.contract.schema_bundle.clone())?;
+    if evaluation.projection_digest != state.projection_digest {
+        return Err(Error::RegistryInconsistency(format!(
+            "Runtime projection digest mismatch for Workspace {workspace_id}"
+        )));
+    }
+    let projected = evaluation.projections.first().ok_or_else(|| {
+        Error::RegistryInconsistency("Workspace config has no active projection".to_string())
+    })?;
+    let config: VirtualRuntimeConfig = serde_json::from_value(projected.data_json.clone())
+        .map_err(|error| Error::RegistryInconsistency(error.to_string()))?;
+    let default_runtime_id = normalize_runtime_id(&config.runtime.default_runtime_id)?;
+    Ok(RuntimeConfigProjection {
+        config_revision: state.snapshot.revision,
+        projection_digest: evaluation.projection_digest,
+        default_runtime_id,
+    })
+}
+
+fn normalize_runtime_id(value: &str) -> Result<Option<String>> {
+    let value = value.trim();
+    if value.is_empty() {
+        return Ok(None);
+    }
+    if value.chars().any(char::is_control) {
+        return Err(Error::InvalidRuntimeIdentifier {
+            kind: "runtime_id".to_string(),
+            value: "[redacted invalid value]".to_string(),
+        });
+    }
+    Ok(Some(value.to_string()))
+}
+
+#[cfg(test)]
+mod tests {
+    use config_source::{ConfigContentType, ConfigEntry, ConfigTreeSnapshot, VirtualPath};
+
+    use super::*;
+
+    fn state(source: &str) -> WorkspaceConfigState {
+        let bundle =
+            config_source::WorkspaceConfigSchemaBundle::compose([RuntimeConfigSchemaProvider
+                .contribution()
+                .unwrap()])
+            .unwrap();
+        let snapshot = ConfigTreeSnapshot::from_entries(
+            7,
+            [ConfigEntry::new(
+                VirtualPath::parse("main.dcdl").unwrap(),
+                ConfigContentType::Decodal,
+                source,
+            )
+            .unwrap()],
+        )
+        .unwrap();
+        let contract = config_source::ToolchainContract::with_schema_bundle(
+            config_source::DEFAULT_SCHEMA_VERSION,
+            vec![VirtualPath::parse("main.dcdl").unwrap()],
+            config_source::DEFAULT_IMPORT_POLICY_VERSION,
+            bundle,
+        );
+        let projection_digest = config_source::SnapshotEnvironment::new(snapshot.clone())
+            .evaluate_contract(&contract)
+            .unwrap()
+            .projection_digest;
+        WorkspaceConfigState {
+            snapshot,
+            contract,
+            projection_digest,
+        }
+    }
+
+    #[test]
+    fn runtime_projection_reads_default_and_preserves_revision_evidence() {
+        let projection = project_runtime_from_workspace_config(
+            "workspace",
+            &state(
+                r#"{ runtime = { default_runtime_id = "arcadia"; }; } as WorkspaceConfigSchema"#,
+            ),
+        )
+        .unwrap();
+        assert_eq!(projection.default_runtime_id.as_deref(), Some("arcadia"));
+        assert_eq!(projection.config_revision, 7);
+        assert!(!projection.projection_digest.is_empty());
+    }
+
+    #[test]
+    fn runtime_projection_treats_missing_default_as_unconfigured() {
+        let projection = project_runtime_from_workspace_config(
+            "workspace",
+            &state("{} as WorkspaceConfigSchema"),
+        )
+        .unwrap();
+        assert_eq!(projection.default_runtime_id, None);
+    }
+
+    #[test]
+    fn runtime_projection_treats_pre_runtime_schema_bundle_as_unconfigured() {
+        let bundle =
+            config_source::WorkspaceConfigSchemaBundle::compose([ConfigSchemaContribution::new(
+                "builtin:legacy",
+                "legacy",
+                "1",
+                "{ legacy = { enabled = Bool default false; }; }",
+            )
+            .unwrap()])
+            .unwrap();
+        let snapshot = ConfigTreeSnapshot::from_entries(
+            6,
+            [ConfigEntry::new(
+                VirtualPath::parse("main.dcdl").unwrap(),
+                ConfigContentType::Decodal,
+                "{ legacy = { enabled = true; }; } as WorkspaceConfigSchema",
+            )
+            .unwrap()],
+        )
+        .unwrap();
+        let contract = config_source::ToolchainContract::with_schema_bundle(
+            config_source::DEFAULT_SCHEMA_VERSION,
+            vec![VirtualPath::parse("main.dcdl").unwrap()],
+            config_source::DEFAULT_IMPORT_POLICY_VERSION,
+            bundle,
+        );
+        let projection_digest = config_source::SnapshotEnvironment::new(snapshot.clone())
+            .evaluate_contract(&contract)
+            .unwrap()
+            .projection_digest;
+        let state = WorkspaceConfigState {
+            snapshot,
+            contract,
+            projection_digest: projection_digest.clone(),
+        };
+
+        let projection = project_runtime_from_workspace_config("workspace", &state).unwrap();
+        assert_eq!(projection.default_runtime_id, None);
+        assert_eq!(projection.config_revision, 6);
+        assert_eq!(projection.projection_digest, projection_digest);
+    }
+
+    #[test]
+    fn runtime_schema_rejects_non_string_default() {
+        let bundle =
+            config_source::WorkspaceConfigSchemaBundle::compose([RuntimeConfigSchemaProvider
+                .contribution()
+                .unwrap()])
+            .unwrap();
+        let snapshot = ConfigTreeSnapshot::from_entries(
+            1,
+            [ConfigEntry::new(
+                VirtualPath::parse("main.dcdl").unwrap(),
+                ConfigContentType::Decodal,
+                "{ runtime = { default_runtime_id = 42; }; } as WorkspaceConfigSchema",
+            )
+            .unwrap()],
+        )
+        .unwrap();
+        let contract = config_source::ToolchainContract::with_schema_bundle(
+            config_source::DEFAULT_SCHEMA_VERSION,
+            vec![VirtualPath::parse("main.dcdl").unwrap()],
+            config_source::DEFAULT_IMPORT_POLICY_VERSION,
+            bundle,
+        );
+        assert!(
+            config_source::SnapshotEnvironment::new(snapshot)
+                .evaluate_contract(&contract)
+                .is_err()
+        );
+    }
+}

crates/workspace-server/src/store.rs

@@ -14,6 +14,7 @@ use uuid::Uuid;
 use worker_runtime::identity::{
     LegacyWorkerIdentityMapping, RuntimeWorkerRef, WorkerId, legacy_worker_identity_mapping_digest,
 };
+use workspace_api::{RepositoryObservedStatus, RepositorySource};
 
 use crate::{Error, Result};
 
@@ -241,6 +242,16 @@ const MIGRATIONS: &[Migration] = &[
         name: "generalize Ticket assignments to role principals",
         apply: generalize_ticket_role_assignments,
     },
+    Migration {
+        version: 44,
+        name: "create Repository source authority",
+        apply: create_repository_source_authority,
+    },
+    Migration {
+        version: 45,
+        name: "create Workdir create operations",
+        apply: create_workdir_create_operations,
+    },
 ];
 
 struct Migration {
@@ -295,10 +306,12 @@ pub struct RepositoryRecord {
     pub name: String,
     pub kind: String,
     pub provider: Option<String>,
-    pub uri: String,
+    pub source: RepositorySource,
     pub default_ref: Option<String>,
-    pub auth_ref_kind: Option<String>,
-    pub auth_ref_key: Option<String>,
+    pub source_revision: u64,
+    pub source_fingerprint: String,
+    pub observed_status: RepositoryObservedStatus,
+    pub observed_at: Option<String>,
     pub created_at: String,
     pub updated_at: String,
 }
@@ -561,6 +574,24 @@ pub struct TicketWorkerAssignmentUpdate {
     pub previous: Option<TicketCoderAssignmentRecord>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct WorkdirCreateOperationRecord {
+    pub workspace_id: String,
+    pub operation_id: String,
+    pub request_fingerprint: String,
+    pub repository_id: String,
+    pub selector: Option<String>,
+    pub requested_runtime_id: Option<String>,
+    pub resolved_runtime_id: String,
+    pub config_revision: u64,
+    pub config_projection_digest: String,
+    pub working_directory_id: String,
+    pub state: String,
+    pub failure: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct WorkdirRegistryRecord {
     pub workspace_id: String,
@@ -730,6 +761,7 @@ pub trait ControlPlaneStore: Send + Sync {
         record: &WorkspaceBootstrapRecord,
     ) -> Result<WorkspaceBootstrapResult>;
     async fn get_trusted_runtime(&self, runtime_id: &str) -> Result<Option<TrustedRuntimeRecord>>;
+    async fn upsert_trusted_runtime_record(&self, record: &TrustedRuntimeRecord) -> Result<()>;
     async fn consume_worker_mutation_source_jti(
         &self,
         runtime_id: &str,
@@ -1770,8 +1802,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                     read_workspace_record,
                 )?;
                 let repository = tx.query_row(
-                    r#"SELECT workspace_id, repository_id, name, kind, provider, uri, default_ref,
-                              auth_ref_kind, auth_ref_key, created_at, updated_at
+                    r#"SELECT workspace_id, repository_id, name, kind, provider,
+                              source_kind, source_uri, default_ref, source_revision,
+                              source_fingerprint, observed_status, observed_at, created_at, updated_at
                        FROM repositories WHERE workspace_id = ?1 AND repository_id = ?2"#,
                     params![workspace.workspace_id, record.repository.repository_id],
                     read_repository_record,
@@ -1822,8 +1855,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                 }
                 let existing_repository = tx
                     .query_row(
-                        r#"SELECT workspace_id, repository_id, name, kind, provider, uri, default_ref,
-                                  auth_ref_kind, auth_ref_key, created_at, updated_at
+                        r#"SELECT workspace_id, repository_id, name, kind, provider,
+                                  source_kind, source_uri, default_ref, source_revision,
+                                  source_fingerprint, observed_status, observed_at, created_at, updated_at
                            FROM repositories WHERE workspace_id = ?1 AND repository_id = ?2"#,
                         params![record.repository.workspace_id, record.repository.repository_id],
                         read_repository_record,
@@ -1861,19 +1895,24 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                 )?;
                 tx.execute(
                     r#"INSERT INTO repositories (
-                        workspace_id, repository_id, name, kind, provider, uri, default_ref,
-                        auth_ref_kind, auth_ref_key, created_at, updated_at
-                    ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)"#,
+                        workspace_id, repository_id, name, kind, provider, uri,
+                        source_kind, source_uri, default_ref, source_revision,
+                        source_fingerprint, observed_status, observed_at, created_at, updated_at
+                    ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)"#,
                     params![
                         record.repository.workspace_id,
                         record.repository.repository_id,
                         record.repository.name,
                         record.repository.kind,
                         record.repository.provider,
-                        record.repository.uri,
+                        record.repository.source.uri,
+                        record.repository.source.kind.as_str(),
+                        record.repository.source.uri,
                         record.repository.default_ref,
-                        record.repository.auth_ref_kind,
-                        record.repository.auth_ref_key,
+                        record.repository.source_revision,
+                        record.repository.source_fingerprint,
+                        record.repository.observed_status.as_str(),
+                        record.repository.observed_at,
                         record.repository.created_at,
                         record.repository.updated_at,
                     ],
@@ -1937,6 +1976,10 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         })
     }
 
+    async fn upsert_trusted_runtime_record(&self, record: &TrustedRuntimeRecord) -> Result<()> {
+        SqliteWorkspaceStore::upsert_trusted_runtime(self, record)
+    }
+
     async fn consume_worker_mutation_source_jti(
         &self,
         runtime_id: &str,
@@ -2053,17 +2096,17 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         self.with_conn(|conn| {
             conn.execute(
                 r#"INSERT INTO repositories (
-                    workspace_id, repository_id, name, kind, provider, uri, default_ref,
-                    auth_ref_kind, auth_ref_key, created_at, updated_at
-                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
+                    workspace_id, repository_id, name, kind, provider, uri,
+                    source_kind, source_uri, default_ref, source_revision,
+                    source_fingerprint, observed_status, observed_at, created_at, updated_at
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)
                 ON CONFLICT(workspace_id, repository_id) DO UPDATE SET
                     name = excluded.name,
                     kind = excluded.kind,
                     provider = excluded.provider,
-                    uri = excluded.uri,
                     default_ref = excluded.default_ref,
-                    auth_ref_kind = excluded.auth_ref_kind,
-                    auth_ref_key = excluded.auth_ref_key,
+                    observed_status = excluded.observed_status,
+                    observed_at = excluded.observed_at,
                     updated_at = excluded.updated_at"#,
                 params![
                     record.workspace_id,
@@ -2071,10 +2114,14 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                     record.name,
                     record.kind,
                     record.provider,
-                    record.uri,
+                    record.source.uri,
+                    record.source.kind.as_str(),
+                    record.source.uri,
                     record.default_ref,
-                    record.auth_ref_kind,
-                    record.auth_ref_key,
+                    record.source_revision,
+                    record.source_fingerprint,
+                    record.observed_status.as_str(),
+                    record.observed_at,
                     record.created_at,
                     record.updated_at,
                 ],
@@ -2090,8 +2137,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
     ) -> Result<Option<RepositoryRecord>> {
         self.with_conn(|conn| {
             conn.query_row(
-                r#"SELECT workspace_id, repository_id, name, kind, provider, uri, default_ref,
-                          auth_ref_kind, auth_ref_key, created_at, updated_at
+                r#"SELECT workspace_id, repository_id, name, kind, provider,
+                          source_kind, source_uri, default_ref, source_revision,
+                          source_fingerprint, observed_status, observed_at, created_at, updated_at
                    FROM repositories
                    WHERE workspace_id = ?1 AND repository_id = ?2"#,
                 params![workspace_id, repository_id],
@@ -2105,8 +2153,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
     fn list_repositories(&self, workspace_id: &str) -> Result<Vec<RepositoryRecord>> {
         self.with_conn(|conn| {
             let mut stmt = conn.prepare(
-                r#"SELECT workspace_id, repository_id, name, kind, provider, uri, default_ref,
-                          auth_ref_kind, auth_ref_key, created_at, updated_at
+                r#"SELECT workspace_id, repository_id, name, kind, provider,
+                          source_kind, source_uri, default_ref, source_revision,
+                          source_fingerprint, observed_status, observed_at, created_at, updated_at
                    FROM repositories
                    WHERE workspace_id = ?1
                    ORDER BY repository_id ASC"#,
@@ -5069,18 +5118,38 @@ fn read_workspace_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<WorkspaceR
 }
 
 fn read_repository_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<RepositoryRecord> {
+    let source_kind_value = row.get::<_, String>(5)?;
+    let source_kind = workspace_api::RepositorySourceKind::parse(&source_kind_value)
+        .unwrap_or(workspace_api::RepositorySourceKind::Invalid);
+    let source_revision = row.get::<_, u64>(8)?;
+    let source_fingerprint = row.get::<_, String>(9)?;
+    let mut source = RepositorySource {
+        kind: source_kind,
+        uri: row.get(6)?,
+    };
+    let observed_status_value = row.get::<_, String>(10)?;
+    let mut observed_status = RepositoryObservedStatus::parse(&observed_status_value)
+        .unwrap_or(RepositoryObservedStatus::Invalid);
+    if source_revision == 0
+        || crate::repository_source::repository_source_fingerprint(&source) != source_fingerprint
+    {
+        source.kind = workspace_api::RepositorySourceKind::Invalid;
+        observed_status = RepositoryObservedStatus::Invalid;
+    }
     Ok(RepositoryRecord {
         workspace_id: row.get(0)?,
         repository_id: row.get(1)?,
         name: row.get(2)?,
         kind: row.get(3)?,
         provider: row.get(4)?,
-        uri: row.get(5)?,
-        default_ref: row.get(6)?,
-        auth_ref_kind: row.get(7)?,
-        auth_ref_key: row.get(8)?,
-        created_at: row.get(9)?,
-        updated_at: row.get(10)?,
+        source,
+        default_ref: row.get(7)?,
+        source_revision,
+        source_fingerprint,
+        observed_status,
+        observed_at: row.get(11)?,
+        created_at: row.get(12)?,
+        updated_at: row.get(13)?,
     })
 }
 
@@ -6399,6 +6468,87 @@ CREATE UNIQUE INDEX ux_worker_workdir_attachment_reservation_id
     Ok(())
 }
 
+fn create_repository_source_authority(conn: &Connection) -> Result<()> {
+    conn.execute_batch(
+        r#"
+        ALTER TABLE repositories ADD COLUMN source_kind TEXT NOT NULL DEFAULT 'invalid';
+        ALTER TABLE repositories ADD COLUMN source_uri TEXT NOT NULL DEFAULT '';
+        ALTER TABLE repositories ADD COLUMN source_revision INTEGER NOT NULL DEFAULT 1;
+        ALTER TABLE repositories ADD COLUMN source_fingerprint TEXT NOT NULL DEFAULT '';
+        ALTER TABLE repositories ADD COLUMN observed_status TEXT NOT NULL DEFAULT 'unverified';
+        ALTER TABLE repositories ADD COLUMN observed_at TEXT;
+        "#,
+    )?;
+
+    let legacy = {
+        let mut stmt = conn.prepare(
+            "SELECT workspace_id, repository_id, uri FROM repositories ORDER BY workspace_id, repository_id",
+        )?;
+        stmt.query_map([], |row| {
+            Ok((
+                row.get::<_, String>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, String>(2)?,
+            ))
+        })?
+        .collect::<std::result::Result<Vec<_>, _>>()?
+    };
+    for (workspace_id, repository_id, uri) in legacy {
+        let source = crate::repository_source::classify_legacy_repository_source(&uri);
+        let fingerprint = crate::repository_source::repository_source_fingerprint(&source);
+        let observed_status = if source.kind == workspace_api::RepositorySourceKind::Invalid {
+            RepositoryObservedStatus::Invalid
+        } else {
+            RepositoryObservedStatus::Unverified
+        };
+        conn.execute(
+            r#"UPDATE repositories
+               SET source_kind = ?3,
+                   source_uri = ?4,
+                   source_revision = 1,
+                   source_fingerprint = ?5,
+                   observed_status = ?6,
+                   observed_at = NULL
+               WHERE workspace_id = ?1 AND repository_id = ?2"#,
+            params![
+                workspace_id,
+                repository_id,
+                source.kind.as_str(),
+                source.uri,
+                fingerprint,
+                observed_status.as_str(),
+            ],
+        )?;
+    }
+    Ok(())
+}
+
+fn create_workdir_create_operations(conn: &Connection) -> Result<()> {
+    conn.execute_batch(
+        r#"
+        CREATE TABLE workdir_create_operations (
+            workspace_id TEXT NOT NULL,
+            operation_id TEXT NOT NULL,
+            request_fingerprint TEXT NOT NULL,
+            repository_id TEXT NOT NULL,
+            selector TEXT,
+            requested_runtime_id TEXT,
+            resolved_runtime_id TEXT NOT NULL,
+            config_revision INTEGER NOT NULL,
+            config_projection_digest TEXT NOT NULL,
+            working_directory_id TEXT NOT NULL,
+            state TEXT NOT NULL CHECK (state IN ('pending', 'succeeded', 'failed')),
+            failure TEXT,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            PRIMARY KEY (workspace_id, operation_id),
+            UNIQUE (workspace_id, working_directory_id)
+        );
+        "#,
+    )?;
+    Ok(())
+}
+
 fn create_workspace_catalog_operations(conn: &Connection) -> Result<()> {
     conn.execute_batch(
         r#"
@@ -9234,6 +9384,72 @@ mod tests {
     use super::*;
     use std::collections::BTreeSet;
 
+    #[test]
+    fn schema_v44_migrates_repository_sources_without_promoting_legacy_auth_refs() {
+        let conn = Connection::open_in_memory().unwrap();
+        configure_sqlite(&conn).unwrap();
+        apply_migrations_through(&conn, 43).unwrap();
+        conn.execute(
+            "INSERT INTO workspaces(workspace_id, display_name, state, created_at, updated_at) \
+             VALUES ('workspace-a', 'Workspace A', 'active', '1', '1')",
+            [],
+        )
+        .unwrap();
+        for (repository_id, uri, auth_kind, auth_key) in [
+            (
+                "remote",
+                "https://example.test/org/repository.git",
+                Some("secret_store"),
+                Some("legacy/key"),
+            ),
+            ("invalid", "relative/repository", Some("file"), Some("/key")),
+        ] {
+            conn.execute(
+                r#"INSERT INTO repositories(
+                        workspace_id, repository_id, name, kind, provider, uri, default_ref,
+                        auth_ref_kind, auth_ref_key, created_at, updated_at
+                    ) VALUES ('workspace-a', ?1, ?1, 'git', 'git', ?2, 'main', ?3, ?4, '1', '1')"#,
+                params![repository_id, uri, auth_kind, auth_key],
+            )
+            .unwrap();
+        }
+
+        apply_migrations(&conn).unwrap();
+
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
+        let remote = conn
+            .query_row(
+                "SELECT source_kind, source_uri, source_revision, source_fingerprint, observed_status \
+                 FROM repositories WHERE workspace_id = 'workspace-a' AND repository_id = 'remote'",
+                [],
+                |row| {
+                    Ok((
+                        row.get::<_, String>(0)?,
+                        row.get::<_, String>(1)?,
+                        row.get::<_, i64>(2)?,
+                        row.get::<_, String>(3)?,
+                        row.get::<_, String>(4)?,
+                    ))
+                },
+            )
+            .unwrap();
+        assert_eq!(remote.0, "https");
+        assert_eq!(remote.1, "https://example.test/org/repository.git");
+        assert_eq!(remote.2, 1);
+        assert!(remote.3.starts_with("sha256:"));
+        assert_eq!(remote.4, "unverified");
+
+        let invalid = conn
+            .query_row(
+                "SELECT source_kind, observed_status FROM repositories \
+                 WHERE workspace_id = 'workspace-a' AND repository_id = 'invalid'",
+                [],
+                |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
+            )
+            .unwrap();
+        assert_eq!(invalid, ("invalid".to_string(), "invalid".to_string()));
+    }
+
     #[test]
     fn startup_composes_ticket_migrations_when_control_plane_is_current() {
         let conn = Connection::open_in_memory().unwrap();
@@ -9278,7 +9494,7 @@ mod tests {
         let before = std::fs::read(&path).unwrap();
         let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
         assert_eq!(plan.current_schema_version, 36);
-        assert_eq!(plan.target_schema_version, 43);
+        assert_eq!(plan.target_schema_version, 45);
         assert!(plan.migration_required);
         assert_eq!(plan.worker_count, 1);
         assert_eq!(plan.mappings[0].legacy_worker_id, 7);
@@ -9292,7 +9508,7 @@ mod tests {
         store
             .with_conn(|conn| {
                 assert!(table_exists(conn, "worker_diagnostics_archives")?);
-                assert_eq!(current_schema_version(conn)?, 43);
+                assert_eq!(current_schema_version(conn)?, 45);
                 Ok(())
             })
             .unwrap();
@@ -9371,7 +9587,7 @@ mod tests {
                 ),
             ]
         );
-        assert_eq!(current_schema_version(&conn).unwrap(), 43);
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
         let foreign_key_error: Option<String> = conn
             .query_row("PRAGMA foreign_key_check", [], |row| row.get(0))
             .optional()
@@ -9500,7 +9716,7 @@ INSERT INTO worker_orphan_diagnostics (
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 43);
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
         assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap());
         let controller_worker_id: String = conn
             .query_row(
@@ -9618,7 +9834,7 @@ INSERT INTO worker_orphan_diagnostics (
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 43);
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
         assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
     }
 
@@ -9636,7 +9852,7 @@ INSERT INTO worker_orphan_diagnostics (
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 43);
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
         let settings = conn
             .query_row(
                 "SELECT settings_revision, language FROM workspace_memory_settings \
@@ -9677,7 +9893,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 43);
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
         assert!(table_exists(&conn, "flow_sources").unwrap());
         assert!(table_exists(&conn, "flow_source_revisions").unwrap());
         assert!(!table_exists(&conn, "flow_instances").unwrap());
@@ -9744,7 +9960,7 @@ INSERT INTO worker_workdir_attachment_reservations (
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 43);
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
         let repositories_sql: String = conn
             .query_row(
                 "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'",
@@ -9889,10 +10105,15 @@ INSERT INTO workdir_registry (
                 name: "Main".to_string(),
                 kind: "git".to_string(),
                 provider: Some("git".to_string()),
-                uri: "/repo-a".to_string(),
+                source: RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::LocalPath,
+                    uri: "/repo-a".to_string(),
+                },
                 default_ref: Some("HEAD".to_string()),
-                auth_ref_kind: None,
-                auth_ref_key: None,
+                source_revision: 1,
+                source_fingerprint: "sha256:test".to_string(),
+                observed_status: RepositoryObservedStatus::Unverified,
+                observed_at: None,
                 created_at: "1".to_string(),
                 updated_at: "1".to_string(),
             })
@@ -9922,7 +10143,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(), 43);
+        assert_eq!(store.schema_version().await.unwrap(), 45);
         assert!(
             !store
                 .with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -9939,7 +10160,7 @@ INSERT INTO workdir_registry (
         store.upsert_workspace(&record).await.unwrap();
 
         let reopened = SqliteWorkspaceStore::open(&db).unwrap();
-        assert_eq!(reopened.schema_version().await.unwrap(), 43);
+        assert_eq!(reopened.schema_version().await.unwrap(), 45);
         assert_eq!(
             reopened.get_workspace("local-dev").await.unwrap(),
             Some(record)
@@ -10693,7 +10914,7 @@ INSERT INTO worker_registry (
         let migrated = SqliteWorkspaceStore::open(&db_path).unwrap();
         migrated
             .with_conn(|conn| {
-                assert_eq!(current_schema_version(conn)?, 43);
+                assert_eq!(current_schema_version(conn)?, 45);
                 assert_eq!(
                     conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?,
                     1,
@@ -10783,10 +11004,15 @@ INSERT INTO worker_registry (
                 name: "Main".to_string(),
                 kind: "git".to_string(),
                 provider: Some("git".to_string()),
-                uri: "file:///tmp/main".to_string(),
+                source: RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::File,
+                    uri: "file:///tmp/main".to_string(),
+                },
                 default_ref: Some("develop".to_string()),
-                auth_ref_kind: None,
-                auth_ref_key: None,
+                source_revision: 1,
+                source_fingerprint: "sha256:test".to_string(),
+                observed_status: RepositoryObservedStatus::Unverified,
+                observed_at: None,
                 created_at: "2026-09-01T00:00:00Z".to_string(),
                 updated_at: "2026-09-01T00:00:00Z".to_string(),
             })
@@ -10991,19 +11217,51 @@ INSERT INTO worker_registry (
         );
     }
 
+    #[test]
+    fn schema_v45_adds_workdir_create_operations_to_v44_database() {
+        let conn = Connection::open_in_memory().unwrap();
+        configure_sqlite(&conn).unwrap();
+        apply_migrations(&conn).unwrap();
+        conn.execute_batch(
+            "DROP TABLE workdir_create_operations;
+             DELETE FROM __yoi_schema_migrations WHERE version = 45;",
+        )
+        .unwrap();
+        assert_eq!(current_schema_version(&conn).unwrap(), 44);
+
+        apply_migrations(&conn).unwrap();
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
+        assert!(table_exists(&conn, "workdir_create_operations").unwrap());
+        let columns = table_columns(&conn, "workdir_create_operations").unwrap();
+        for required in [
+            "operation_id",
+            "request_fingerprint",
+            "resolved_runtime_id",
+            "config_revision",
+            "config_projection_digest",
+            "working_directory_id",
+            "state",
+        ] {
+            assert!(
+                columns.iter().any(|column| column == required),
+                "missing column {required}"
+            );
+        }
+    }
+
     #[test]
     fn server_refuses_a_database_from_a_newer_schema_generation() {
         let conn = Connection::open_in_memory().unwrap();
         configure_sqlite(&conn).unwrap();
         apply_migrations(&conn).unwrap();
         conn.execute(
-            "INSERT INTO __yoi_schema_migrations (version, name) VALUES (44, 'future')",
+            "INSERT INTO __yoi_schema_migrations (version, name) VALUES (46, 'future')",
             [],
         )
         .unwrap();
 
         let error = apply_migrations(&conn).unwrap_err().to_string();
-        assert!(error.contains("schema version 44 is newer"), "{error}");
+        assert!(error.contains("schema version 46 is newer"), "{error}");
         assert!(error.contains("refusing to serve"), "{error}");
     }
 
@@ -11224,7 +11482,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026-
 
         apply_migrations(&mut conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 43);
+        assert_eq!(current_schema_version(&conn).unwrap(), 45);
         let workspace_id: Option<String> = conn
             .query_row(
                 "SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'",
@@ -11727,6 +11985,12 @@ WHERE workspace_id = 'workspace-a'
                 "auth_ref_key",
                 "created_at",
                 "updated_at",
+                "source_kind",
+                "source_uri",
+                "source_revision",
+                "source_fingerprint",
+                "observed_status",
+                "observed_at",
             ],
         );
         assert_columns(
@@ -11841,7 +12105,7 @@ WHERE workspace_id = 'workspace-a'
         .unwrap();
 
         let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 43);
+        assert_eq!(store.schema_version().await.unwrap(), 45);
 
         store
             .with_conn(|conn| {
@@ -12030,7 +12294,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(), 43);
+        assert_eq!(store.schema_version().await.unwrap(), 45);
         let workspace = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
             owner_account_id: None,
@@ -12047,10 +12311,20 @@ CREATE TABLE ticket_assignment_operations (
             name: "Yoi".to_string(),
             kind: "git".to_string(),
             provider: Some("git".to_string()),
-            uri: ".".to_string(),
+            source: RepositorySource {
+                kind: workspace_api::RepositorySourceKind::LocalPath,
+                uri: "/repo".to_string(),
+            },
             default_ref: Some("HEAD".to_string()),
-            auth_ref_kind: None,
-            auth_ref_key: None,
+            source_revision: 1,
+            source_fingerprint: crate::repository_source::repository_source_fingerprint(
+                &RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::LocalPath,
+                    uri: "/repo".to_string(),
+                },
+            ),
+            observed_status: RepositoryObservedStatus::Unverified,
+            observed_at: None,
             created_at: "2".to_string(),
             updated_at: "2".to_string(),
         };
@@ -12080,7 +12354,9 @@ CREATE TABLE ticket_assignment_operations (
         let mut other_repository = repository.clone();
         other_repository.workspace_id = other_workspace.workspace_id.clone();
         other_repository.name = "Other Yoi".to_string();
-        other_repository.uri = "/other/yoi".to_string();
+        other_repository.source.uri = "/other/yoi".to_string();
+        other_repository.source_fingerprint =
+            crate::repository_source::repository_source_fingerprint(&other_repository.source);
         store.upsert_repository(&other_repository).unwrap();
 
         assert_eq!(
@@ -12096,7 +12372,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(), 43);
+        assert_eq!(store.schema_version().await.unwrap(), 45);
         let workspace = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
             owner_account_id: None,
@@ -12189,10 +12465,15 @@ CREATE TABLE ticket_assignment_operations (
                 name: "Repository".to_string(),
                 kind: "git".to_string(),
                 provider: Some("git".to_string()),
-                uri: ".".to_string(),
+                source: RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::LocalPath,
+                    uri: "/repo".to_string(),
+                },
                 default_ref: Some("HEAD".to_string()),
-                auth_ref_kind: None,
-                auth_ref_key: None,
+                source_revision: 1,
+                source_fingerprint: "sha256:test".to_string(),
+                observed_status: RepositoryObservedStatus::Unverified,
+                observed_at: None,
                 created_at: "1".to_string(),
                 updated_at: "1".to_string(),
             })
@@ -12498,7 +12779,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(), 43);
+        assert_eq!(store.schema_version().await.unwrap(), 45);
         let now = "2026-07-22T00:00:00Z".to_string();
         let account = AccountRecord {
             account_id: "acct-user-alice".to_string(),

crates/workspace-server/src/workdir_create_operations.rs

@@ -0,0 +1,246 @@
+use rusqlite::{OptionalExtension, params};
+use sha2::{Digest, Sha256};
+
+use crate::store::WorkdirCreateOperationRecord;
+use crate::{Error, Result, SqliteWorkspaceStore};
+
+pub fn request_fingerprint(
+    repository_id: &str,
+    selector: Option<&str>,
+    requested_runtime_id: Option<&str>,
+) -> String {
+    let mut hasher = Sha256::new();
+    for value in [Some(repository_id), selector, requested_runtime_id] {
+        match value {
+            Some(value) => {
+                hasher.update([1]);
+                hasher.update((value.len() as u64).to_be_bytes());
+                hasher.update(value.as_bytes());
+            }
+            None => hasher.update([0]),
+        }
+    }
+    let digest = hasher.finalize();
+    let mut encoded = String::with_capacity(digest.len() * 2);
+    for byte in digest {
+        use std::fmt::Write as _;
+        write!(&mut encoded, "{byte:02x}").expect("writing to String cannot fail");
+    }
+    format!("sha256:{encoded}")
+}
+
+impl SqliteWorkspaceStore {
+    pub fn reserve_workdir_create_operation(
+        &self,
+        record: &WorkdirCreateOperationRecord,
+    ) -> Result<WorkdirCreateOperationRecord> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction()?;
+            tx.execute(
+                r#"INSERT OR IGNORE INTO workdir_create_operations (
+                    workspace_id, operation_id, request_fingerprint, repository_id, selector,
+                    requested_runtime_id, resolved_runtime_id, config_revision,
+                    config_projection_digest, working_directory_id, state, failure,
+                    created_at, updated_at
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)"#,
+                params![
+                    record.workspace_id,
+                    record.operation_id,
+                    record.request_fingerprint,
+                    record.repository_id,
+                    record.selector,
+                    record.requested_runtime_id,
+                    record.resolved_runtime_id,
+                    record.config_revision as i64,
+                    record.config_projection_digest,
+                    record.working_directory_id,
+                    record.state,
+                    record.failure,
+                    record.created_at,
+                    record.updated_at,
+                ],
+            )?;
+            let persisted =
+                read_workdir_create_operation(&tx, &record.workspace_id, &record.operation_id)?
+                    .ok_or_else(|| {
+                        Error::RegistryInconsistency(format!(
+                            "Workdir create operation `{}` was not persisted",
+                            record.operation_id
+                        ))
+                    })?;
+            if persisted.request_fingerprint != record.request_fingerprint {
+                return Err(Error::InvalidInput(format!(
+                    "Workdir create operation `{}` was reused with different input",
+                    record.operation_id
+                )));
+            }
+            tx.commit()?;
+            Ok(persisted)
+        })
+    }
+
+    pub fn finish_workdir_create_operation(
+        &self,
+        workspace_id: &str,
+        operation_id: &str,
+        request_fingerprint: &str,
+        succeeded: bool,
+        failure: Option<&str>,
+        updated_at: &str,
+    ) -> Result<WorkdirCreateOperationRecord> {
+        self.with_conn_mut(|conn| {
+            let changed = conn.execute(
+                r#"UPDATE workdir_create_operations
+                   SET state = ?1, failure = ?2, updated_at = ?3
+                   WHERE workspace_id = ?4 AND operation_id = ?5
+                     AND request_fingerprint = ?6"#,
+                params![
+                    if succeeded { "succeeded" } else { "failed" },
+                    failure,
+                    updated_at,
+                    workspace_id,
+                    operation_id,
+                    request_fingerprint,
+                ],
+            )?;
+            if changed != 1 {
+                return Err(Error::RegistryInconsistency(format!(
+                    "Workdir create operation `{operation_id}` could not be finalized"
+                )));
+            }
+            read_workdir_create_operation(conn, workspace_id, operation_id)?.ok_or_else(|| {
+                Error::RegistryInconsistency(format!(
+                    "Workdir create operation `{operation_id}` disappeared"
+                ))
+            })
+        })
+    }
+
+    pub fn load_workdir_create_operation(
+        &self,
+        workspace_id: &str,
+        operation_id: &str,
+    ) -> Result<Option<WorkdirCreateOperationRecord>> {
+        self.with_conn(|conn| read_workdir_create_operation(conn, workspace_id, operation_id))
+    }
+}
+
+fn read_workdir_create_operation(
+    conn: &rusqlite::Connection,
+    workspace_id: &str,
+    operation_id: &str,
+) -> Result<Option<WorkdirCreateOperationRecord>> {
+    conn.query_row(
+        r#"SELECT workspace_id, operation_id, request_fingerprint, repository_id, selector,
+                  requested_runtime_id, resolved_runtime_id, config_revision,
+                  config_projection_digest, working_directory_id, state, failure,
+                  created_at, updated_at
+           FROM workdir_create_operations
+           WHERE workspace_id = ?1 AND operation_id = ?2"#,
+        params![workspace_id, operation_id],
+        |row| {
+            Ok(WorkdirCreateOperationRecord {
+                workspace_id: row.get(0)?,
+                operation_id: row.get(1)?,
+                request_fingerprint: row.get(2)?,
+                repository_id: row.get(3)?,
+                selector: row.get(4)?,
+                requested_runtime_id: row.get(5)?,
+                resolved_runtime_id: row.get(6)?,
+                config_revision: row.get::<_, i64>(7)? as u64,
+                config_projection_digest: row.get(8)?,
+                working_directory_id: row.get(9)?,
+                state: row.get(10)?,
+                failure: row.get(11)?,
+                created_at: row.get(12)?,
+                updated_at: row.get(13)?,
+            })
+        },
+    )
+    .optional()
+    .map_err(Error::from)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::store::{ControlPlaneStore, RepositoryRecord, WorkspaceRecord};
+
+    #[test]
+    fn retry_keeps_resolved_config_evidence_and_rejects_changed_input() {
+        let store = SqliteWorkspaceStore::in_memory().unwrap();
+        futures::executor::block_on(store.upsert_workspace(&WorkspaceRecord {
+            workspace_id: "workspace".to_string(),
+            owner_account_id: None,
+            display_name: "Workspace".to_string(),
+            state: "active".to_string(),
+            created_at: "2026-08-24T00:00:00Z".to_string(),
+            updated_at: "2026-08-24T00:00:00Z".to_string(),
+        }))
+        .unwrap();
+        store
+            .upsert_repository(&RepositoryRecord {
+                workspace_id: "workspace".to_string(),
+                repository_id: "main".to_string(),
+                name: "main".to_string(),
+                kind: "git".to_string(),
+                provider: Some("git".to_string()),
+                source: workspace_api::RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::LocalPath,
+                    uri: "/tmp/main".to_string(),
+                },
+                default_ref: Some("develop".to_string()),
+                source_revision: 1,
+                source_fingerprint: "sha256:test".to_string(),
+                observed_status: workspace_api::RepositoryObservedStatus::Unverified,
+                observed_at: None,
+                created_at: "2026-08-24T00:00:00Z".to_string(),
+                updated_at: "2026-08-24T00:00:00Z".to_string(),
+            })
+            .unwrap();
+        let record = WorkdirCreateOperationRecord {
+            workspace_id: "workspace".to_string(),
+            operation_id: "call-1".to_string(),
+            request_fingerprint: request_fingerprint("main", Some("develop"), None),
+            repository_id: "main".to_string(),
+            selector: Some("develop".to_string()),
+            requested_runtime_id: None,
+            resolved_runtime_id: "arcadia".to_string(),
+            config_revision: 7,
+            config_projection_digest: "sha256:projection".to_string(),
+            working_directory_id: "wd-1".to_string(),
+            state: "pending".to_string(),
+            failure: None,
+            created_at: "2026-08-24T00:00:00Z".to_string(),
+            updated_at: "2026-08-24T00:00:00Z".to_string(),
+        };
+        assert_eq!(
+            store.reserve_workdir_create_operation(&record).unwrap(),
+            record
+        );
+        let mut changed_resolution = record.clone();
+        changed_resolution.resolved_runtime_id = "other".to_string();
+        changed_resolution.config_revision = 8;
+        assert_eq!(
+            store
+                .reserve_workdir_create_operation(&changed_resolution)
+                .unwrap(),
+            record
+        );
+        assert_eq!(
+            store
+                .load_workdir_create_operation("workspace", "call-1")
+                .unwrap(),
+            Some(record.clone())
+        );
+        let mut changed_input = record.clone();
+        changed_input.request_fingerprint = request_fingerprint("main", Some("main"), None);
+        assert!(
+            store
+                .reserve_workdir_create_operation(&changed_input)
+                .unwrap_err()
+                .to_string()
+                .contains("reused with different input")
+        );
+    }
+}

crates/workspace-server/src/worker_source.rs

@@ -3,14 +3,118 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use axum::http::HeaderMap;
 use worker_runtime::auth::{
-    WorkerMutationActorKind, WorkerMutationOperation, WorkerMutationSourceClaims,
-    WorkerMutationSourceExpectation, decode_worker_mutation_source_claims,
-    verify_worker_mutation_source_proof,
+    RuntimeRequestSourceExpectation, WorkerMutationActorKind, WorkerMutationOperation,
+    WorkerMutationSourceClaims, WorkerMutationSourceExpectation,
+    decode_runtime_request_source_claims, decode_worker_mutation_source_claims,
+    verify_runtime_request_source, verify_worker_mutation_source_proof,
 };
 use worker_runtime::worker_source::InProcessWorkerMutationProof;
 
 use crate::hosts::RemoteRuntimeConfig;
-use crate::server::WorkspaceApi;
+use crate::server::{ServerConfig, WorkspaceApi};
+use crate::store::ControlPlaneStore;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct VerifiedRuntimeRequestSource {
+    pub runtime_id: String,
+    pub worker_id: Option<String>,
+}
+
+pub async fn verify_runtime_request_source_proof(
+    api: &WorkspaceApi,
+    proof: &str,
+    workspace_id: &str,
+    permission: &str,
+    method: &str,
+    path: &str,
+    body_digest: &str,
+) -> Result<VerifiedRuntimeRequestSource, WorkerMutationSourceProofError> {
+    verify_runtime_request_source_proof_with_store(
+        api.store.as_ref(),
+        &api.config,
+        proof,
+        workspace_id,
+        permission,
+        method,
+        path,
+        body_digest,
+    )
+    .await
+}
+
+#[allow(clippy::too_many_arguments)]
+pub async fn verify_runtime_request_source_proof_with_store(
+    store: &dyn ControlPlaneStore,
+    config: &ServerConfig,
+    proof: &str,
+    workspace_id: &str,
+    permission: &str,
+    method: &str,
+    path: &str,
+    body_digest: &str,
+) -> Result<VerifiedRuntimeRequestSource, WorkerMutationSourceProofError> {
+    let unverified = decode_runtime_request_source_claims(proof)
+        .map_err(|_| WorkerMutationSourceProofError::Invalid)?;
+    let audience = remote_audience(config, &unverified.iss, workspace_id)?;
+    let trusted = store
+        .get_trusted_runtime(&unverified.iss)
+        .await
+        .map_err(|error| WorkerMutationSourceProofError::Authority(error.to_string()))?
+        .filter(|record| record.revoked_at.is_none())
+        .ok_or(WorkerMutationSourceProofError::RevokedRuntimeTrust)?;
+    let trusted_for_workspace = trusted.workspace_id.as_deref() == Some(workspace_id)
+        || (unverified.iss == crate::hosts::EMBEDDED_RUNTIME_ID && trusted.workspace_id.is_none());
+    if !trusted_for_workspace {
+        return Err(WorkerMutationSourceProofError::WrongWorkspace);
+    }
+    let expected = RuntimeRequestSourceExpectation {
+        identity_id: &unverified.iss,
+        audience: audience.as_ref(),
+        workspace_id,
+        worker_id: unverified.worker_id.as_deref(),
+        permission,
+        method,
+        path,
+        body_digest,
+        now_unix: i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX),
+    };
+    let claims = verify_runtime_request_source(proof, &trusted.public_key, &expected)
+        .map_err(map_auth_error)?;
+    let now_seconds = u64::try_from(expected.now_unix).unwrap_or(u64::MAX);
+    let expires_at = u64::try_from(claims.exp).unwrap_or(0);
+    let consumed_at = chrono::DateTime::from_timestamp(expected.now_unix, 0)
+        .ok_or(WorkerMutationSourceProofError::Expired)?
+        .to_rfc3339();
+    if !store
+        .consume_worker_mutation_source_jti(
+            &claims.iss,
+            &claims.jti,
+            expires_at,
+            now_seconds,
+            &consumed_at,
+        )
+        .await
+        .map_err(|error| WorkerMutationSourceProofError::Authority(error.to_string()))?
+    {
+        return Err(WorkerMutationSourceProofError::Replay);
+    }
+    if let Some(worker_id) = claims.worker_id.as_deref() {
+        let worker = worker_runtime::identity::RuntimeWorkerRef {
+            runtime_id: claims.iss.clone(),
+            worker_id: worker_id.to_owned(),
+        };
+        let member = store
+            .get_worker_registry(workspace_id, &worker)
+            .map_err(|error| WorkerMutationSourceProofError::Authority(error.to_string()))?;
+        if member.is_none() {
+            return Err(WorkerMutationSourceProofError::WorkerCatalogMembership);
+        }
+    }
+    Ok(VerifiedRuntimeRequestSource {
+        runtime_id: claims.iss,
+        worker_id: claims.worker_id,
+    })
+}
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum PresentedWorkerMutationSourceProof<'a> {
@@ -97,16 +201,19 @@ async fn verify_worker_remove_source_with(
         PresentedWorkerMutationSourceProof::Remote(token) => {
             let unverified = decode_worker_mutation_source_claims(token)
                 .map_err(|_| WorkerMutationSourceProofError::Invalid)?;
-            let audience = remote_audience(config, &unverified.iss)?;
+            let audience = remote_audience(config, &unverified.iss, &config.workspace_id)?;
             let trusted = store
                 .get_trusted_runtime(&unverified.iss)
                 .await
                 .map_err(|error| WorkerMutationSourceProofError::Authority(error.to_string()))?
                 .filter(|record| record.revoked_at.is_none())
                 .ok_or(WorkerMutationSourceProofError::RevokedRuntimeTrust)?;
+            if trusted.workspace_id.as_deref() != Some(config.workspace_id.as_str()) {
+                return Err(WorkerMutationSourceProofError::WrongWorkspace);
+            }
             let expected = WorkerMutationSourceExpectation {
                 runtime_id: &unverified.iss,
-                audience,
+                audience: audience.as_ref(),
                 workspace_id: &config.workspace_id,
                 worker_id: None,
                 actor_kind: WorkerMutationActorKind::Worker,
@@ -247,13 +354,17 @@ impl worker_runtime::worker_source::EmbeddedWorkerMutationDispatcher
 fn remote_audience<'a>(
     config: &'a crate::server::ServerConfig,
     runtime_id: &str,
-) -> Result<&'a str, WorkerMutationSourceProofError> {
+    workspace_id: &str,
+) -> Result<std::borrow::Cow<'a, str>, WorkerMutationSourceProofError> {
+    if runtime_id == crate::hosts::EMBEDDED_RUNTIME_ID {
+        return Ok(std::borrow::Cow::Owned(format!("embedded:{workspace_id}")));
+    }
     config
         .remote_runtime_sources
         .iter()
         .find(|runtime| runtime.runtime_id == runtime_id)
         .and_then(|runtime: &RemoteRuntimeConfig| runtime.auth.as_ref())
-        .map(|auth| auth.server_id.as_str())
+        .map(|auth| std::borrow::Cow::Borrowed(auth.server_id.as_str()))
         .ok_or(WorkerMutationSourceProofError::RevokedRuntimeTrust)
 }
 

crates/workspace-server/src/workspace_catalog.rs

@@ -1,4 +1,3 @@
-use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use chrono::{SecondsFormat, Utc};
@@ -6,6 +5,9 @@ use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use uuid::Uuid;
 
+use workspace_api::{RepositoryObservedStatus, RepositorySource};
+
+use crate::repository_source::{parse_repository_source, repository_source_fingerprint};
 use crate::store::{
     ControlPlaneStore, RepositoryRecord, WorkspaceBootstrapRecord, WorkspaceRecord,
 };
@@ -109,8 +111,8 @@ impl WorkspaceCatalogService {
         )?;
         let display_name =
             normalize_required("display_name", request.display_name, MAX_DISPLAY_NAME_BYTES)?;
-        let repository_path = validate_repository_uri(&request.repository.uri)?;
-        let repository_uri = repository_path.to_string_lossy().into_owned();
+        let repository_source = validate_repository_source(&request.repository.uri)?;
+        let repository_uri = repository_source.uri.clone();
         let repository_name = request
             .repository
             .display_name
@@ -166,10 +168,12 @@ impl WorkspaceCatalogService {
                     name: repository_name,
                     kind: "git".to_string(),
                     provider: Some("git".to_string()),
-                    uri: repository_uri,
+                    source: repository_source.clone(),
                     default_ref: Some(default_ref),
-                    auth_ref_kind: None,
-                    auth_ref_key: None,
+                    source_revision: 1,
+                    source_fingerprint: repository_source_fingerprint(&repository_source),
+                    observed_status: RepositoryObservedStatus::Unverified,
+                    observed_at: None,
                     created_at: now.clone(),
                     updated_at: now,
                 },
@@ -194,35 +198,8 @@ fn normalize_required(field: &str, value: String, max_bytes: usize) -> Result<St
     Ok(value.to_string())
 }
 
-fn validate_repository_uri(uri: &str) -> Result<PathBuf> {
-    let uri = uri.trim();
-    if uri.is_empty() || uri.contains("://") {
-        return Err(Error::InvalidInput(
-            "initial repository uri must be an absolute server-local path".to_string(),
-        ));
-    }
-    let path = Path::new(uri);
-    if !path.is_absolute() {
-        return Err(Error::InvalidInput(
-            "initial repository uri must be an absolute server-local path".to_string(),
-        ));
-    }
-    let path = path.canonicalize().map_err(|error| {
-        Error::InvalidInput(format!("initial repository path is unavailable: {error}"))
-    })?;
-    if !path.is_dir() {
-        return Err(Error::InvalidInput(
-            "initial repository path must be a directory".to_string(),
-        ));
-    }
-    let normal_git = path.join(".git").exists();
-    let bare_git = path.join("HEAD").is_file() && path.join("objects").is_dir();
-    if !normal_git && !bare_git {
-        return Err(Error::InvalidInput(
-            "initial repository path is not a Git repository".to_string(),
-        ));
-    }
-    Ok(path)
+fn validate_repository_source(uri: &str) -> Result<RepositorySource> {
+    parse_repository_source(uri)
 }
 
 fn workspace_create_fingerprint(
@@ -259,6 +236,7 @@ fn workspace_create_fingerprint(
 mod tests {
     use super::*;
     use crate::store::SqliteWorkspaceStore;
+    use workspace_api::RepositorySourceKind;
 
     fn git_repository() -> tempfile::TempDir {
         let dir = tempfile::tempdir().unwrap();
@@ -383,12 +361,53 @@ mod tests {
     }
 
     #[test]
-    fn repository_intent_rejects_remote_and_non_git_paths() {
-        let remote = validate_repository_uri("https://example.test/repo.git").unwrap_err();
-        assert!(remote.to_string().contains("server-local path"));
+    fn repository_intent_accepts_unavailable_local_sources_without_server_io() {
+        let remote = validate_repository_source("https://example.test/repo.git").unwrap();
+        assert_eq!(remote.kind, RepositorySourceKind::Https);
 
-        let dir = tempfile::tempdir().unwrap();
-        let non_git = validate_repository_uri(&dir.path().display().to_string()).unwrap_err();
-        assert!(non_git.to_string().contains("not a Git repository"));
+        let local = validate_repository_source("/runtime-only/missing/repository").unwrap();
+        assert_eq!(local.kind, RepositorySourceKind::LocalPath);
+        assert_eq!(local.uri, "/runtime-only/missing/repository");
+
+        let file = validate_repository_source("file:///runtime-only/missing/repository").unwrap();
+        assert_eq!(file.kind, RepositorySourceKind::File);
+
+        assert!(validate_repository_source("relative/repository").is_err());
+    }
+
+    #[test]
+    fn remote_repository_creation_persists_typed_source_without_auth_metadata() {
+        let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap());
+        let service = WorkspaceCatalogService::new(store.clone());
+        let result = service
+            .create_first_ownerless(WorkspaceCreateRequest {
+                operation_key: "remote-create".to_string(),
+                display_name: "Remote Workspace".to_string(),
+                repository: InitialRepositoryIntent {
+                    uri: "ssh://git@example.test/org/repository.git".to_string(),
+                    display_name: Some("Remote Repository".to_string()),
+                    default_ref: Some("main".to_string()),
+                },
+            })
+            .unwrap();
+
+        let persisted = store
+            .get_repository(
+                &result.workspace.workspace_id,
+                &result.repository.repository_id,
+            )
+            .unwrap()
+            .unwrap();
+        assert_eq!(persisted.source.kind, RepositorySourceKind::Ssh);
+        assert_eq!(persisted.source_revision, 1);
+        assert!(persisted.source_fingerprint.starts_with("sha256:"));
+        assert_eq!(
+            persisted.observed_status,
+            RepositoryObservedStatus::Unverified
+        );
+        let json = serde_json::to_value(&persisted).unwrap();
+        assert!(json.get("source").is_some());
+        assert!(json.get("auth_ref_kind").is_none());
+        assert!(json.get("auth_ref_key").is_none());
     }
 }

web/workspace/src/lib/workspace/api/workspace-catalog.ts

@@ -7,13 +7,29 @@ export type WorkspaceCatalogRecord = {
   updated_at: string;
 };
 
+export type RepositorySourceKind =
+  | "local_path"
+  | "file"
+  | "ssh"
+  | "http"
+  | "https"
+  | "invalid";
+
 export type WorkspaceRepositoryRecord = {
   workspace_id: string;
   repository_id: string;
   name: string;
   kind: string;
-  uri: string;
+  provider: string | null;
+  source: {
+    kind: RepositorySourceKind;
+    uri: string;
+  };
   default_ref: string | null;
+  source_revision: number;
+  source_fingerprint: string;
+  observed_status: "unverified" | "ready" | "invalid";
+  observed_at: string | null;
 };
 
 export type WorkspaceCatalogItem = WorkspaceCatalogRecord & {

web/workspace/src/lib/workspace/sidebar/types.ts

@@ -269,6 +269,14 @@ export type RepositorySummary = {
   display_name: string;
   kind: string;
   provider: string;
+  source: {
+    kind: "local_path" | "file" | "ssh" | "http" | "https" | "invalid";
+    uri: string;
+  };
+  source_revision: number;
+  source_fingerprint: string;
+  observed_status: "unverified" | "ready" | "invalid";
+  observed_at?: string | null;
   default_selector?: string | null;
   record_authority: string;
   git?: GitRepositorySummary | null;

web/workspace/src/routes/+page.svelte

@@ -155,8 +155,7 @@
       <p class="workspace-catalog-eyebrow">New team space</p>
       <h2 id="workspace-create-title">Create Workspace</h2>
       <p>
-        Repository paths and URIs are interpreted by the Backend. Browser-local paths are
-        not authority.
+        Repository sources are interpreted by Backend authority. Supported Git sources are absolute local paths, file://, ssh://, http(s)://, and user@host:path; Browser-local paths and embedded credentials are not authority. Plain HTTP is unencrypted, so prefer HTTPS or SSH.
       </p>
     </div>
     <form onsubmit={submitCreation}>

web/workspace/src/routes/w/[workspaceId]/repositories/[repositoryId]/+page.svelte

@@ -16,7 +16,7 @@
       <div>
         <h3>{data.repository.item.display_name}</h3>
       </div>
-      <span class="status-pill" class:warn={data.repository.item.git?.status !== 'clean'}>{data.repository.item.git?.status ?? 'not observed'}</span>
+      <span class="status-pill" class:warn={data.repository.item.observed_status !== 'ready'}>{data.repository.item.observed_status}</span>
     </div>
     <dl>
       <div>
@@ -27,6 +27,18 @@
         <dt>Provider</dt>
         <dd>{data.repository.item.provider}</dd>
       </div>
+      <div>
+        <dt>Source</dt>
+        <dd>{data.repository.item.source.kind} · {data.repository.item.source.uri}</dd>
+      </div>
+      <div>
+        <dt>Source revision</dt>
+        <dd>{data.repository.item.source_revision} · {data.repository.item.source_fingerprint}</dd>
+      </div>
+      <div>
+        <dt>Observed</dt>
+        <dd>{data.repository.item.observed_at ?? 'not observed'}</dd>
+      </div>
       <div>
         <dt>Record authority</dt>
         <dd>{data.repository.item.record_authority}</dd>