use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use crate::workspace_request::{RuntimeWorkspaceRequest, RuntimeWorkspaceRequestClient};
use worker::{
    WorkspaceClient, WorkspaceClientError, WorkspacePromptCatalogResolution,
    WorkspacePromptProjection, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse,
};

use crate::auth::{
    RuntimeAuthError, RuntimeIdentityMaterial, RuntimeWorkerMutationSourceSigner,
    WORKER_REMOVE_PERMISSION, WORKSPACE_REQUEST_PERMISSION, WORKSPACE_WORKER_DISCOVERY_PERMISSION,
    WorkerMutationActorKind, WorkerMutationOperation, WorkerMutationSourceClaims, new_token_id,
};
use crate::runtime::RuntimeWorkspaceScope;
use crate::worker_backend::WorkspacePromptProjectionCache;

pub const DEFAULT_WORKER_MUTATION_SOURCE_TTL_SECONDS: u64 = 60;

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RuntimeOwnedWorkerMutationProof {
    Remote(String),
    InProcess(InProcessWorkerMutationProof),
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InProcessWorkerMutationProof {
    claims: WorkerMutationSourceClaims,
}

impl InProcessWorkerMutationProof {
    pub fn claims(&self) -> &WorkerMutationSourceClaims {
        &self.claims
    }

    pub fn into_claims(self) -> WorkerMutationSourceClaims {
        self.claims
    }
}

#[derive(Clone)]
pub struct RuntimeWorkerMutationSourceAuthority {
    mode: RuntimeWorkerMutationSourceMode,
}

#[derive(Clone)]
enum RuntimeWorkerMutationSourceMode {
    Remote {
        signer: RuntimeWorkerMutationSourceSigner,
    },
    Embedded {
        runtime_id: String,
        audience: String,
    },
}

impl RuntimeWorkerMutationSourceAuthority {
    pub fn remote(identity: &RuntimeIdentityMaterial) -> Self {
        Self {
            mode: RuntimeWorkerMutationSourceMode::Remote {
                signer: RuntimeWorkerMutationSourceSigner::from_identity(identity),
            },
        }
    }

    pub fn embedded(runtime_id: impl Into<String>, workspace_id: impl AsRef<str>) -> Self {
        Self {
            mode: RuntimeWorkerMutationSourceMode::Embedded {
                runtime_id: runtime_id.into(),
                audience: format!("embedded:{}", workspace_id.as_ref()),
            },
        }
    }

    pub fn issue_worker_remove(
        &self,
        scope: &RuntimeWorkspaceScope,
        source_worker_id: &str,
        target_runtime_id: &str,
        target_worker_id: &str,
    ) -> Result<RuntimeOwnedWorkerMutationProof, RuntimeAuthError> {
        match &self.mode {
            RuntimeWorkerMutationSourceMode::Remote { signer } => {
                let token = signer.issue_worker_remove(
                    &scope.server_id,
                    &scope.workspace_id,
                    source_worker_id,
                    target_runtime_id,
                    target_worker_id,
                    DEFAULT_WORKER_MUTATION_SOURCE_TTL_SECONDS,
                )?;
                Ok(RuntimeOwnedWorkerMutationProof::Remote(token))
            }
            RuntimeWorkerMutationSourceMode::Embedded {
                runtime_id,
                audience,
            } => {
                let issued_at = unix_now_seconds();
                Ok(RuntimeOwnedWorkerMutationProof::InProcess(
                    InProcessWorkerMutationProof {
                        claims: WorkerMutationSourceClaims {
                            iss: runtime_id.clone(),
                            aud: audience.clone(),
                            workspace_id: scope.workspace_id.clone(),
                            worker_id: source_worker_id.to_string(),
                            actor_kind: WorkerMutationActorKind::Worker,
                            operation: WorkerMutationOperation::WorkerRemove,
                            target_runtime_id: target_runtime_id.to_string(),
                            target_worker_id: target_worker_id.to_string(),
                            permission: WORKER_REMOVE_PERMISSION.to_string(),
                            iat: issued_at,
                            exp: issued_at
                                .saturating_add(DEFAULT_WORKER_MUTATION_SOURCE_TTL_SECONDS),
                            jti: new_token_id()?,
                        },
                    },
                ))
            }
        }
    }
}

pub trait EmbeddedWorkerMutationDispatcher: Send + Sync {
    fn execute_worker_remove(
        &self,
        proof: InProcessWorkerMutationProof,
        target_runtime_id: &str,
        target_worker_id: &str,
        reason: &str,
    ) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError>;
}

#[derive(Clone)]
enum RuntimeWorkerMutationTransport {
    Remote {
        request_client: RuntimeWorkspaceRequestClient,
    },
    Embedded {
        dispatcher: Arc<dyn EmbeddedWorkerMutationDispatcher>,
    },
}

#[derive(Clone)]
pub struct RuntimeWorkerMutationForwarder {
    authority: RuntimeWorkerMutationSourceAuthority,
    scope: RuntimeWorkspaceScope,
    source_worker_id: String,
    transport: RuntimeWorkerMutationTransport,
}

impl RuntimeWorkerMutationForwarder {
    pub fn remote(
        identity: &RuntimeIdentityMaterial,
        scope: RuntimeWorkspaceScope,
        source_worker_id: impl Into<String>,
        request_client: RuntimeWorkspaceRequestClient,
    ) -> Self {
        Self {
            authority: RuntimeWorkerMutationSourceAuthority::remote(identity),
            scope: scope.clone(),
            source_worker_id: source_worker_id.into(),
            transport: RuntimeWorkerMutationTransport::Remote { request_client },
        }
    }

    pub fn embedded(
        runtime_id: impl Into<String>,
        scope: RuntimeWorkspaceScope,
        source_worker_id: impl Into<String>,
        dispatcher: Arc<dyn EmbeddedWorkerMutationDispatcher>,
    ) -> Self {
        let runtime_id = runtime_id.into();
        Self {
            authority: RuntimeWorkerMutationSourceAuthority::embedded(
                &runtime_id,
                &scope.workspace_id,
            ),
            scope,
            source_worker_id: source_worker_id.into(),
            transport: RuntimeWorkerMutationTransport::Embedded { dispatcher },
        }
    }

    pub fn execute_worker_remove(
        &self,
        target_runtime_id: &str,
        target_worker_id: &str,
        reason: &str,
    ) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
        let proof = self.authority.issue_worker_remove(
            &self.scope,
            &self.source_worker_id,
            target_runtime_id,
            target_worker_id,
        )?;
        match (&self.transport, proof) {
            (
                RuntimeWorkerMutationTransport::Remote { request_client },
                RuntimeOwnedWorkerMutationProof::Remote(token),
            ) => execute_remote_worker_remove_http(RemoteWorkerRemoveHttpRequest {
                request_client: request_client.clone(),
                source_worker_id: self.source_worker_id.clone(),
                token,
                target_runtime_id: target_runtime_id.to_string(),
                target_worker_id: target_worker_id.to_string(),
                reason: reason.to_string(),
            }),
            (
                RuntimeWorkerMutationTransport::Embedded { dispatcher },
                RuntimeOwnedWorkerMutationProof::InProcess(claims),
            ) => dispatcher.execute_worker_remove(
                claims,
                target_runtime_id,
                target_worker_id,
                reason,
            ),
            _ => Err(RuntimeWorkerMutationForwardError::AuthorityTransportMismatch),
        }
    }
}

struct RemoteWorkerRemoveHttpRequest {
    request_client: RuntimeWorkspaceRequestClient,
    source_worker_id: String,
    token: String,
    target_runtime_id: String,
    target_worker_id: String,
    reason: String,
}

fn execute_remote_worker_remove_http(
    request: RemoteWorkerRemoveHttpRequest,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
    if tokio::runtime::Handle::try_current().is_ok() {
        return std::thread::Builder::new()
            .name("yoi-worker-mutation-http".to_string())
            .spawn(move || execute_remote_worker_remove_http_blocking(request))
            .map_err(|error| {
                RuntimeWorkerMutationForwardError::Transport(format!(
                    "failed to start Worker mutation HTTP thread: {error}"
                ))
            })?
            .join()
            .map_err(|_| {
                RuntimeWorkerMutationForwardError::Transport(
                    "Worker mutation HTTP thread panicked".to_string(),
                )
            })?;
    }

    execute_remote_worker_remove_http_blocking(request)
}

fn execute_remote_worker_remove_http_blocking(
    request: RemoteWorkerRemoveHttpRequest,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
    let path = format!(
        "/api/w/{}/workers/remove",
        request.request_client.workspace_id()
    );
    let body = serde_json::to_vec(&serde_json::json!({
        "target_runtime_id": request.target_runtime_id,
        "target_worker_id": request.target_worker_id,
        "reason": request.reason,
    }))
    .map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
    let mut headers = reqwest::header::HeaderMap::new();
    headers.insert(
        crate::auth::WORKER_MUTATION_SOURCE_PROOF_HEADER,
        reqwest::header::HeaderValue::from_str(&request.token)
            .map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?,
    );
    headers.insert(
        reqwest::header::CONTENT_TYPE,
        reqwest::header::HeaderValue::from_static("application/json"),
    );
    let response = request
        .request_client
        .execute_blocking(RuntimeWorkspaceRequest {
            method: reqwest::Method::POST,
            path_and_query: path,
            body,
            headers,
            permission: WORKSPACE_REQUEST_PERMISSION.to_string(),
            worker_id: Some(request.source_worker_id),
            timeout: Some(Duration::from_secs(5)),
            max_response_bytes: 8 * 1024 * 1024,
        })
        .map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
    let body = String::from_utf8(response.body)
        .map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
    Ok(WorkspaceResponse {
        status: response.status.as_u16(),
        body,
    })
}

#[derive(Clone)]
pub struct RuntimeOwnedWorkspaceClient {
    workspace_id: String,
    request_client: RuntimeWorkspaceRequestClient,
    worker_id: String,
    request_timeout: Option<Duration>,
    worker_remove: Option<RuntimeWorkerMutationForwarder>,
    prompt_projection_cache: Option<Arc<WorkspacePromptProjectionCache>>,
}

impl RuntimeOwnedWorkspaceClient {
    pub fn new(
        workspace_id: impl Into<String>,
        base_url: impl Into<String>,
        runtime_id: impl Into<String>,
        worker_id: impl Into<String>,
    ) -> Self {
        let workspace_id = workspace_id.into();
        Self {
            request_client: RuntimeWorkspaceRequestClient::new(
                workspace_id.clone(),
                base_url,
                runtime_id,
            ),
            workspace_id,
            worker_id: worker_id.into(),
            request_timeout: None,
            worker_remove: None,
            prompt_projection_cache: None,
        }
    }

    pub(crate) fn from_request_client(
        request_client: RuntimeWorkspaceRequestClient,
        worker_id: impl Into<String>,
    ) -> Self {
        let workspace_id = request_client.workspace_id().to_string();
        Self {
            workspace_id,
            request_client,
            worker_id: worker_id.into(),
            request_timeout: None,
            worker_remove: None,
            prompt_projection_cache: None,
        }
    }

    pub fn with_worker_remove(mut self, worker_remove: RuntimeWorkerMutationForwarder) -> Self {
        self.worker_remove = Some(worker_remove);
        self
    }

    pub fn with_runtime_request_source(
        mut self,
        identity: &RuntimeIdentityMaterial,
        audience: impl Into<String>,
    ) -> Self {
        self.request_client = self
            .request_client
            .with_runtime_request_source(identity, audience);
        self
    }

    pub(crate) fn with_prompt_projection_cache(
        mut self,
        cache: Arc<WorkspacePromptProjectionCache>,
    ) -> Self {
        self.prompt_projection_cache = Some(cache);
        self
    }

    #[cfg(test)]
    fn with_request_timeout(mut self, request_timeout: Option<Duration>) -> Self {
        self.request_timeout = request_timeout;
        self
    }

    fn execute_with_permission(
        &self,
        request: WorkspaceRequest,
        permission: &'static str,
    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
        let method = match request.method {
            WorkspaceRequestMethod::Get => reqwest::Method::GET,
            WorkspaceRequestMethod::Post => reqwest::Method::POST,
            WorkspaceRequestMethod::Put => reqwest::Method::PUT,
            WorkspaceRequestMethod::Patch => reqwest::Method::PATCH,
            WorkspaceRequestMethod::Delete => reqwest::Method::DELETE,
        };
        let body = request.body.unwrap_or_default().into_bytes();
        let mut headers = reqwest::header::HeaderMap::new();
        if !body.is_empty() {
            headers.insert(
                reqwest::header::CONTENT_TYPE,
                reqwest::header::HeaderValue::from_static("application/json"),
            );
        }
        let request_label = format!("{method} {}", request.path);
        let response = self
            .request_client
            .execute_blocking(RuntimeWorkspaceRequest {
                method,
                path_and_query: request.path,
                body,
                headers,
                permission: permission.to_string(),
                worker_id: Some(self.worker_id.clone()),
                timeout: self.request_timeout,
                max_response_bytes: 8 * 1024 * 1024,
            })
            .map_err(|error| {
                WorkspaceClientError::Request(format!("{request_label} failed: {error}"))
            })?;
        let body = String::from_utf8(response.body)
            .map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
        Ok(WorkspaceResponse {
            status: response.status.as_u16(),
            body,
        })
    }
}

impl std::fmt::Debug for RuntimeOwnedWorkspaceClient {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("RuntimeOwnedWorkspaceClient")
            .field("workspace_id", &self.request_client.workspace_id())
            .field("base_url", &self.request_client.base_url())
            .field("source", &"Runtime-owned")
            .field(
                "worker_remove",
                &self.worker_remove.as_ref().map(|_| "enabled"),
            )
            .finish()
    }
}

impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
    fn workspace_id(&self) -> Option<&str> {
        Some(self.request_client.workspace_id())
    }

    fn kind(&self) -> &str {
        "runtime-owned-workspace-client"
    }

    fn is_available(&self) -> bool {
        true
    }

    fn execute(
        &self,
        request: WorkspaceRequest,
    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
        self.execute_with_permission(request, WORKSPACE_REQUEST_PERMISSION)
    }

    fn list_workspace_workers(
        &self,
        request: worker::WorkspaceWorkerDiscoveryRequest,
    ) -> Result<workspace_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
        let mut path = format!(
            "/api/w/{}/worker-discovery/workers?limit={}",
            self.workspace_id, request.limit
        );
        if let Some(cursor) = request.cursor.as_deref() {
            path.push_str("&cursor=");
            path.push_str(&percent_encode_query(cursor));
        }
        if let Some(query) = request.query.as_deref() {
            path.push_str("&query=");
            path.push_str(&percent_encode_query(query));
        }
        let response = self.execute_with_permission(
            WorkspaceRequest::get(path),
            WORKSPACE_WORKER_DISCOVERY_PERMISSION,
        )?;
        if !(200..300).contains(&response.status) {
            return Err(WorkspaceClientError::Request(format!(
                "Workspace Worker discovery failed with HTTP {}: {}",
                response.status, response.body
            )));
        }
        serde_json::from_str(&response.body).map_err(|error| {
            WorkspaceClientError::Request(format!(
                "invalid Workspace Worker discovery response: {error}"
            ))
        })
    }

    fn current_prompt_projection(
        &self,
        minimum_revision: Option<u64>,
    ) -> Result<Option<WorkspacePromptCatalogResolution>, WorkspaceClientError> {
        let Some(cache) = self.prompt_projection_cache.as_ref() else {
            return Ok(None);
        };
        if let Some(resolution) = cache
            .active(&self.workspace_id)
            .map_err(WorkspaceClientError::Request)?
            .filter(|resolution| {
                minimum_revision
                    .map(|minimum| resolution.projection.config_revision >= minimum)
                    .unwrap_or(true)
            })
        {
            return Ok(Some((*resolution).clone()));
        }
        let fetch_gate = cache
            .fetch_gate(&self.workspace_id)
            .map_err(WorkspaceClientError::Request)?;
        let _fetch_guard = fetch_gate.lock().map_err(|_| {
            WorkspaceClientError::Request(
                "Workspace Prompt projection fetch gate was poisoned".to_string(),
            )
        })?;
        if let Some(resolution) = cache
            .active(&self.workspace_id)
            .map_err(WorkspaceClientError::Request)?
            .filter(|resolution| {
                minimum_revision
                    .map(|minimum| resolution.projection.config_revision >= minimum)
                    .unwrap_or(true)
            })
        {
            return Ok(Some((*resolution).clone()));
        }
        let response = self.execute(WorkspaceRequest::get(format!(
            "/api/w/{}/config/projections/prompts",
            self.workspace_id
        )))?;
        if !(200..300).contains(&response.status) {
            return Err(WorkspaceClientError::Request(format!(
                "active Workspace Prompt projection request failed with HTTP {}: {}",
                response.status, response.body
            )));
        }
        let projection: WorkspacePromptProjection =
            serde_json::from_str(&response.body).map_err(|error| {
                WorkspaceClientError::Request(format!(
                    "invalid active Workspace Prompt projection response: {error}"
                ))
            })?;
        if projection.workspace_id != self.workspace_id {
            return Err(WorkspaceClientError::Request(format!(
                "active Workspace Prompt projection scope mismatch: expected {}, got {}",
                self.workspace_id, projection.workspace_id
            )));
        }
        let resolution = cache
            .observe(projection)
            .map_err(WorkspaceClientError::Request)?;
        if let Some(minimum_revision) = minimum_revision
            && resolution.projection.config_revision < minimum_revision
        {
            return Err(WorkspaceClientError::Request(format!(
                "active Workspace Prompt projection is stale: required revision {minimum_revision}, got {}",
                resolution.projection.config_revision
            )));
        }
        Ok(Some((*resolution).clone()))
    }

    fn execute_worker_remove(
        &self,
        target_runtime_id: &str,
        target_worker_id: &str,
        reason: &str,
    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
        self.worker_remove
            .as_ref()
            .ok_or_else(|| {
                WorkspaceClientError::Unavailable(
                    "Runtime-owned WorkerRemove forwarding is unavailable".to_string(),
                )
            })?
            .execute_worker_remove(target_runtime_id, target_worker_id, reason)
            .map_err(|error| WorkspaceClientError::Request(error.to_string()))
    }
}

fn percent_encode_query(value: &str) -> String {
    let mut encoded = String::with_capacity(value.len());
    for byte in value.bytes() {
        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
            encoded.push(char::from(byte));
        } else {
            use std::fmt::Write as _;
            let _ = write!(encoded, "%{byte:02X}");
        }
    }
    encoded
}

#[derive(Debug, thiserror::Error)]
pub enum RuntimeWorkerMutationForwardError {
    #[error(transparent)]
    Auth(#[from] RuntimeAuthError),
    #[error("Worker mutation forwarding transport failed: {0}")]
    Transport(String),
    #[error("Worker mutation source authority does not match its forwarding transport")]
    AuthorityTransportMismatch,
    #[error("embedded Worker mutation dispatcher failed: {0}")]
    Embedded(String),
}

fn unix_now_seconds() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::auth::{
        RUNTIME_REQUEST_SOURCE_PROOF_HEADER, WorkerMutationSourceExpectation,
        decode_runtime_request_source_claims, decode_worker_mutation_source_claims,
        request_body_digest, verify_worker_mutation_source_proof,
    };

    #[test]
    fn current_prompt_projection_uses_the_shared_runtime_cache_without_http() {
        let cache = Arc::new(WorkspacePromptProjectionCache::default());
        let catalog = worker::EffectivePromptCatalog::new(
            std::collections::BTreeMap::from([(
                "default".to_string(),
                "workspace prompt".to_string(),
            )]),
            3,
            "schema",
            "toolchain",
        )
        .unwrap();
        let projection = WorkspacePromptProjection::new(
            "workspace-a",
            "source-3",
            catalog.catalog_digest.clone(),
            catalog,
        )
        .unwrap();
        cache.observe(projection).unwrap();
        let client = RuntimeOwnedWorkspaceClient::new(
            "workspace-a",
            "http://127.0.0.1:1",
            "runtime-a",
            "worker-a",
        )
        .with_prompt_projection_cache(cache);

        let projection = client.current_prompt_projection(None).unwrap().unwrap();
        let second = client.current_prompt_projection(None).unwrap().unwrap();

        assert_eq!(projection.projection.config_revision, 3);
        assert_eq!(projection.projection.source_digest, "source-3");
        assert!(Arc::ptr_eq(&projection.catalog, &second.catalog));
    }

    #[test]
    fn prompt_projection_minimum_revision_rejects_stale_server_response() {
        use std::io::{Read, Write};
        use std::net::TcpListener;

        let catalog = worker::EffectivePromptCatalog::new(
            std::collections::BTreeMap::from([("default".to_string(), "stale prompt".to_string())]),
            3,
            "schema",
            "toolchain",
        )
        .unwrap();
        let projection = WorkspacePromptProjection::new(
            "workspace-a",
            "source-3",
            catalog.catalog_digest.clone(),
            catalog,
        )
        .unwrap();
        let body = serde_json::to_string(&projection).unwrap();
        let cache = Arc::new(WorkspacePromptProjectionCache::default());
        cache.observe(projection).unwrap();
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let base_url = format!("http://{}", listener.local_addr().unwrap());
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut request = [0_u8; 4096];
            let _ = stream.read(&mut request).unwrap();
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            )
            .unwrap();
        });
        let client =
            RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a")
                .with_prompt_projection_cache(cache);

        let error = client.current_prompt_projection(Some(4)).unwrap_err();
        server.join().unwrap();

        assert!(error.to_string().contains("required revision 4, got 3"));
    }

    #[test]
    fn concurrent_prompt_projection_miss_fetches_once_and_shares_catalog() {
        use std::io::{Read, Write};
        use std::net::TcpListener;
        use std::sync::Barrier;

        let catalog = worker::EffectivePromptCatalog::new(
            std::collections::BTreeMap::from([(
                "default".to_string(),
                "shared prompt".to_string(),
            )]),
            5,
            "schema",
            "toolchain",
        )
        .unwrap();
        let projection = WorkspacePromptProjection::new(
            "workspace-a",
            "source-5",
            catalog.catalog_digest.clone(),
            catalog,
        )
        .unwrap();
        let body = serde_json::to_string(&projection).unwrap();
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let base_url = format!("http://{}", listener.local_addr().unwrap());
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut request = [0_u8; 4096];
            let read = stream.read(&mut request).unwrap();
            let request = std::str::from_utf8(&request[..read]).unwrap();
            assert!(
                request.contains("GET /api/w/workspace-a/config/projections/prompts "),
                "unexpected Prompt projection request: {request}"
            );
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            )
            .unwrap();
        });
        let cache = Arc::new(WorkspacePromptProjectionCache::default());
        let client =
            RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a")
                .with_prompt_projection_cache(cache);
        let barrier = Arc::new(Barrier::new(8));
        let threads = (0..8)
            .map(|_| {
                let client = client.clone();
                let barrier = barrier.clone();
                std::thread::spawn(move || {
                    barrier.wait();
                    client.current_prompt_projection(None).unwrap().unwrap()
                })
            })
            .collect::<Vec<_>>();
        let resolutions = threads
            .into_iter()
            .map(|thread| thread.join().unwrap())
            .collect::<Vec<_>>();
        server.join().unwrap();

        let first = &resolutions[0].catalog;
        assert!(
            resolutions
                .iter()
                .all(|resolution| Arc::ptr_eq(first, &resolution.catalog))
        );
    }

    #[test]
    fn current_prompt_projection_rejects_cross_workspace_response() {
        use std::io::{Read, Write};
        use std::net::TcpListener;

        let catalog = worker::EffectivePromptCatalog::new(
            std::collections::BTreeMap::from([(
                "default".to_string(),
                "foreign prompt".to_string(),
            )]),
            4,
            "schema",
            "toolchain",
        )
        .unwrap();
        let projection = WorkspacePromptProjection::new(
            "workspace-b",
            "source-4",
            catalog.catalog_digest.clone(),
            catalog,
        )
        .unwrap();
        let body = serde_json::to_string(&projection).unwrap();
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let base_url = format!("http://{}", listener.local_addr().unwrap());
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut request = [0_u8; 4096];
            let _ = stream.read(&mut request).unwrap();
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            )
            .unwrap();
        });
        let client =
            RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a")
                .with_prompt_projection_cache(Arc::new(WorkspacePromptProjectionCache::default()));

        let error = client.current_prompt_projection(None).unwrap_err();
        server.join().unwrap();

        assert!(error.to_string().contains("scope mismatch"));
    }

    #[test]
    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;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        let received = Arc::new(Mutex::new(String::new()));
        let received_for_server = received.clone();
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut bytes = [0_u8; 4096];
            let count = stream.read(&mut bytes).unwrap();
            *received_for_server.lock().unwrap() =
                String::from_utf8_lossy(&bytes[..count]).into_owned();
            stream
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}")
                .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?state=planning&limit=20",
            ))
            .unwrap();
        assert_eq!(response.status, 200);
        server.join().unwrap();
        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]
    fn workspace_worker_discovery_signs_dedicated_permission_and_encoded_query() {
        use std::io::{Read, Write};
        use std::net::TcpListener;
        use std::sync::Mutex;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        let received = Arc::new(Mutex::new(String::new()));
        let received_for_server = received.clone();
        let body = serde_json::json!({
            "workers": [{
                "subject": {
                    "kind": "runtime_worker",
                    "runtime_id": "runtime-b",
                    "worker_id": "worker-b"
                },
                "resource_key": "W-2",
                "display_name": "coder two",
                "profile": "builtin:coder",
                "status": "idle"
            }],
            "next_cursor": "v1:1"
        })
        .to_string();
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut bytes = [0_u8; 4096];
            let count = stream.read(&mut bytes).unwrap();
            *received_for_server.lock().unwrap() =
                String::from_utf8_lossy(&bytes[..count]).into_owned();
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}",
                body.len(),
                body
            )
            .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 page = client
            .list_workspace_workers(worker::WorkspaceWorkerDiscoveryRequest {
                cursor: Some("v1:0".to_string()),
                limit: 1,
                query: Some("coder two".to_string()),
            })
            .unwrap();
        assert_eq!(page.workers[0].resource_key, "W-2");
        server.join().unwrap();

        let request = received.lock().unwrap().clone();
        assert!(request.contains(
            "GET /api/w/workspace-a/worker-discovery/workers?limit=1&cursor=v1%3A0&query=coder%20two "
        ));
        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())
                })
            })
            .unwrap();
        let claims = decode_runtime_request_source_claims(token).unwrap();
        assert_eq!(claims.permission, WORKSPACE_WORKER_DISCOVERY_PERMISSION);
        assert_eq!(
            claims.path,
            "/api/w/workspace-a/worker-discovery/workers?limit=1&cursor=v1%3A0&query=coder%20two"
        );
    }

    #[test]
    fn remote_authority_stamps_and_signs_worker_remove_without_caller_claim_choices() {
        let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap();
        let authority = RuntimeWorkerMutationSourceAuthority::remote(&identity);
        let scope = RuntimeWorkspaceScope {
            workspace_id: "workspace-a".to_string(),
            server_id: "server-a".to_string(),
        };

        let RuntimeOwnedWorkerMutationProof::Remote(token) = authority
            .issue_worker_remove(&scope, "worker-source", "runtime-b", "worker-target")
            .unwrap()
        else {
            panic!("remote authority must produce a signed proof");
        };
        let claims = decode_worker_mutation_source_claims(&token).unwrap();
        let expected = WorkerMutationSourceExpectation {
            runtime_id: "runtime-a",
            audience: "server-a",
            workspace_id: "workspace-a",
            worker_id: Some("worker-source"),
            actor_kind: WorkerMutationActorKind::Worker,
            operation: WorkerMutationOperation::WorkerRemove,
            target_runtime_id: "runtime-b",
            target_worker_id: "worker-target",
            permission: WORKER_REMOVE_PERMISSION,
        };
        assert_eq!(
            verify_worker_mutation_source_proof(
                &identity.public_key,
                &token,
                &expected,
                claims.iat
            )
            .unwrap(),
            claims
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn remote_forwarder_is_safe_in_async_runtime_and_stamps_signed_proof() {
        use std::io::{Read, Write};
        use std::net::TcpListener;
        use std::sync::Mutex;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        let received = Arc::new(Mutex::new(String::new()));
        let received_for_server = received.clone();
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut bytes = [0_u8; 8192];
            let count = stream.read(&mut bytes).unwrap();
            *received_for_server.lock().unwrap() =
                String::from_utf8_lossy(&bytes[..count]).into_owned();
            stream
                .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n")
                .unwrap();
        });

        let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap();
        let scope = RuntimeWorkspaceScope::new("workspace-a", "server-a");
        let forwarder = RuntimeWorkerMutationForwarder::remote(
            &identity,
            scope,
            "worker-source",
            RuntimeWorkspaceRequestClient::new(
                "workspace-a",
                format!("http://{address}"),
                "runtime-a",
            )
            .with_runtime_request_source(&identity, "server-a"),
        );
        let response = forwarder
            .execute_worker_remove("runtime-target", "worker-target", "retire obsolete Worker")
            .unwrap();
        assert_eq!(response.status, 204);
        server.join().unwrap();

        let request = received.lock().unwrap().clone();
        assert!(request.starts_with("POST /api/w/workspace-a/workers/remove HTTP/1.1"));
        assert!(request.contains("\"target_runtime_id\":\"runtime-target\""));
        assert!(request.contains("\"target_worker_id\":\"worker-target\""));
        assert!(!request.contains("expected_worker_revision"));
        assert!(request.contains("\"reason\":\"retire obsolete Worker\""));
        let request_source_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 request source proof header");
        let request_source_claims =
            decode_runtime_request_source_claims(request_source_token).unwrap();
        assert_eq!(request_source_claims.iss, "runtime-a");
        assert_eq!(request_source_claims.aud, "server-a");
        assert_eq!(request_source_claims.workspace_id, "workspace-a");
        assert_eq!(
            request_source_claims.worker_id.as_deref(),
            Some("worker-source")
        );
        assert_eq!(
            request_source_claims.permission,
            WORKSPACE_REQUEST_PERMISSION
        );
        assert_eq!(request_source_claims.method, "POST");
        assert_eq!(
            request_source_claims.path,
            "/api/w/workspace-a/workers/remove"
        );
        let request_body = request
            .split_once("\r\n\r\n")
            .map(|(_, body)| body)
            .expect("WorkerRemove request body");
        assert_eq!(
            request_source_claims.body_digest,
            request_body_digest(request_body.as_bytes())
        );
        let token = request
            .lines()
            .find_map(|line| {
                line.split_once(':').and_then(|(name, value)| {
                    name.eq_ignore_ascii_case(crate::auth::WORKER_MUTATION_SOURCE_PROOF_HEADER)
                        .then(|| value.trim())
                })
            })
            .expect("proof header");
        let claims = decode_worker_mutation_source_claims(token).unwrap();
        let expected = WorkerMutationSourceExpectation {
            runtime_id: "runtime-a",
            audience: "server-a",
            workspace_id: "workspace-a",
            worker_id: Some("worker-source"),
            actor_kind: WorkerMutationActorKind::Worker,
            operation: WorkerMutationOperation::WorkerRemove,
            target_runtime_id: "runtime-target",
            target_worker_id: "worker-target",
            permission: WORKER_REMOVE_PERMISSION,
        };
        verify_worker_mutation_source_proof(&identity.public_key, token, &expected, claims.iat)
            .unwrap();
    }

    #[test]
    fn embedded_forwarder_delivers_in_process_proof_with_the_request() {
        use std::sync::Mutex;

        #[derive(Default)]
        struct RecordingDispatcher {
            seen: Mutex<Option<(WorkerMutationSourceClaims, String, String, String)>>,
        }
        impl EmbeddedWorkerMutationDispatcher for RecordingDispatcher {
            fn execute_worker_remove(
                &self,
                proof: InProcessWorkerMutationProof,
                target_runtime_id: &str,
                target_worker_id: &str,
                reason: &str,
            ) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
                *self.seen.lock().unwrap() = Some((
                    proof.into_claims(),
                    target_runtime_id.to_string(),
                    target_worker_id.to_string(),
                    reason.to_string(),
                ));
                Ok(WorkspaceResponse {
                    status: 202,
                    body: "accepted".to_string(),
                })
            }
        }

        let dispatcher = Arc::new(RecordingDispatcher::default());
        let scope = RuntimeWorkspaceScope::new("workspace-a", "server-a");
        let forwarder = RuntimeWorkerMutationForwarder::embedded(
            "runtime-embedded",
            scope,
            "worker-source",
            dispatcher.clone(),
        );
        let response = forwarder
            .execute_worker_remove("runtime-target", "worker-target", "retire obsolete Worker")
            .unwrap();
        assert_eq!(response.status, 202);
        let (claims, target_runtime_id, target_worker_id, reason) =
            dispatcher.seen.lock().unwrap().take().unwrap();
        assert_eq!(claims.iss, "runtime-embedded");
        assert_eq!(claims.worker_id, "worker-source");
        assert_eq!(claims.target_runtime_id, "runtime-target");
        assert_eq!(claims.target_worker_id, "worker-target");
        assert_eq!(target_runtime_id, "runtime-target");
        assert_eq!(target_worker_id, "worker-target");
        assert_eq!(reason, "retire obsolete Worker");
    }

    #[test]
    fn runtime_owned_workspace_client_has_no_fixed_request_timeout() {
        let (base_url, server) = delayed_workspace_response(Duration::from_millis(75));
        let client =
            RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a");
        assert_eq!(client.request_timeout, None);

        let response = client
            .execute(WorkspaceRequest::json(
                WorkspaceRequestMethod::Post,
                "/api/test",
                "{}",
            ))
            .unwrap();
        assert_eq!(response.status, 200);
        assert_eq!(response.body, r#"{"ok":true}"#);
        server.join().unwrap();
    }

    #[test]
    fn runtime_owned_workspace_client_reports_request_timeouts() {
        let (base_url, server) = delayed_workspace_response(Duration::from_millis(75));
        let client =
            RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a")
                .with_request_timeout(Some(Duration::from_millis(20)));

        let error = client
            .execute(WorkspaceRequest::json(
                WorkspaceRequestMethod::Post,
                "/api/test",
                "{}",
            ))
            .unwrap_err();
        assert!(matches!(error, WorkspaceClientError::Request(_)));
        let message = error.to_string();
        assert!(message.contains("POST /api/test"), "{message}");
        assert!(message.contains("timed out"), "{message}");
        server.join().unwrap();
    }

    fn delayed_workspace_response(delay: Duration) -> (String, std::thread::JoinHandle<()>) {
        use std::io::{Read, Write};
        use std::net::TcpListener;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut bytes = [0_u8; 8192];
            let _ = stream.read(&mut bytes);
            std::thread::sleep(delay);
            let _ = stream.write_all(
                b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}",
            );
        });
        (format!("http://{address}"), server)
    }

    #[test]
    fn embedded_authority_uses_the_same_claim_contract_without_a_credential() {
        let authority =
            RuntimeWorkerMutationSourceAuthority::embedded("runtime-embedded", "workspace-a");
        let scope = RuntimeWorkspaceScope {
            workspace_id: "workspace-a".to_string(),
            server_id: "server-unused-for-embedded".to_string(),
        };

        let RuntimeOwnedWorkerMutationProof::InProcess(proof) = authority
            .issue_worker_remove(&scope, "worker-source", "runtime-b", "worker-target")
            .unwrap()
        else {
            panic!("embedded authority must produce an in-process proof");
        };
        let claims = proof.claims();
        assert_eq!(claims.iss, "runtime-embedded");
        assert_eq!(claims.aud, "embedded:workspace-a");
        assert_eq!(claims.workspace_id, "workspace-a");
        assert_eq!(claims.worker_id, "worker-source");
        assert_eq!(claims.operation, WorkerMutationOperation::WorkerRemove);
        assert_eq!(claims.target_runtime_id, "runtime-b");
        assert_eq!(claims.target_worker_id, "worker-target");
        assert_eq!(claims.permission, WORKER_REMOVE_PERMISSION);
        assert!(!claims.jti.is_empty());
    }
}